[v14] Fix the access list reminder pagination loop. (#36407)

* Fix the access list reminder pagination loop.

On error in the access list reminder pagination loop, the loop now breaks
instead of continuing, which is the intended behavior. A test has been added
to exercise this. This test will time out with the previous behavior. A light
refactoring has been done to make it easier to mock the ListAccessLists
response.

Users running their own Slack plugins (outside of cloud) were running into an
issue here due to the plugin user not having permissions to read access lists.
On cloud tenants, the plugins should have this access, so the impact there is
minimal.

Additionally, the docs have been updated to include the access list
permissions.

* Test fixes.

* Tests are more consistent.

* Update docs/pages/includes/plugins/rbac-with-friendly-name.mdx

Co-authored-by: Paul Gottschling <paul.gottschling@goteleport.com>

* Add in initial duration/trigger.

* Output migration instructions, add jitter back in.

* Don't return error when no recipients.

* Tune the error return.

---------

Co-authored-by: Paul Gottschling <paul.gottschling@goteleport.com>

docs/pages/includes/plugins/rbac-with-friendly-name.mdx

@@ -19,6 +19,11 @@ spec:
       - resources: ['access_plugin_data']
         verbs: ['update']
 
+      # In order to provide access list review reminders, permissions to list and read access lists
+      # are necessary. This is currently only supported for a subset of plugins.
+      - resources: ['access_list']
+        verbs: ['list', 'read']
+
     # Optional: To display user-friendly names in resource-based Access
     # Requests instead of resource IDs, the plugin also needs permission
     # to list the resources being requested. Include this along with the

integrations/access/accesslist/app.go

@@ -31,15 +31,15 @@ import (
 	"github.com/gravitational/teleport/integrations/lib"
 	"github.com/gravitational/teleport/integrations/lib/logger"
 	pd "github.com/gravitational/teleport/integrations/lib/plugindata"
-	"github.com/gravitational/teleport/lib/utils"
-	"github.com/gravitational/teleport/lib/utils/interval"
 )
 
 const (
 	// oneDay is the number of hours in a day.
 	oneDay = 24 * time.Hour
 	// oneWeek is the number of days in a week.
 	oneWeek = oneDay * 7
+	// reminderInterval is the interval for sending access list reminders.
+	reminderInterval = 3 * time.Hour
 )
 
 // App is the access list application for plugins. This will notify access list owners
@@ -114,61 +114,86 @@ func (a *App) run(ctx context.Context) error {
 		return nil
 	})
 
-	remindInterval := interval.New(interval.Config{
-		Duration:      time.Hour * 3,
-		FirstDuration: utils.FullJitter(time.Second * 30),
-		Jitter:        retryutils.NewSeventhJitter(),
-		Clock:         a.clock,
-	})
-	defer remindInterval.Stop()
 	log := logger.Get(ctx)
 
 	log.Info("Access list monitor is running")
 
 	a.job.SetReady(true)
+
+	jitter := retryutils.NewSeventhJitter()
+	timer := a.clock.NewTimer(jitter(30 * time.Second))
+	defer timer.Stop()
+
 	for {
 		select {
-		case <-remindInterval.Next():
-			log.Info("Looking for Access List Review reminders")
-
-			var nextToken string
-			var err error
-			for {
-				var accessLists []*accesslist.AccessList
-				accessLists, nextToken, err = a.apiClient.AccessListClient().ListAccessLists(ctx, 0 /* default page size */, nextToken)
-				if err != nil {
-					if trace.IsNotImplemented(err) {
-						log.Errorf("access list endpoint is not implemented on this auth server, so the access list app is ceasing to run.")
-						return nil
-					}
-					log.Errorf("error listing access lists: %v", err)
-					continue
-				}
-
-				for _, accessList := range accessLists {
-					if err := a.notifyForAccessListReviews(ctx, accessList); err != nil {
-						log.WithError(err).Warn("Error notifying for access list reviews")
-					}
-				}
-
-				if nextToken == "" {
-					break
-				}
+		case <-timer.Chan():
+			if err := a.remindIfNecessary(ctx); err != nil {
+				return trace.Wrap(err)
 			}
+
+			timer.Reset(jitter(reminderInterval))
 		case <-ctx.Done():
 			log.Info("Access list monitor is finished")
 			return nil
 		}
 	}
 }
 
+// remindIfNecessary will create and send reminders if necessary. The only error this returns is
+// notImplemented, which will cease looking for reminders if the auth server does not support
+// access lists.
+func (a *App) remindIfNecessary(ctx context.Context) error {
+	log := logger.Get(ctx)
+
+	log.Info("Looking for Access List Review reminders")
+
+	var nextToken string
+	var err error
+	for {
+		var accessLists []*accesslist.AccessList
+		accessLists, nextToken, err = a.apiClient.ListAccessLists(ctx, 0 /* default page size */, nextToken)
+		if err != nil {
+			if trace.IsNotImplemented(err) {
+				log.Errorf("access list endpoint is not implemented on this auth server, so the access list app is ceasing to run.")
+				return trace.Wrap(err)
+			} else if trace.IsAccessDenied(err) {
+				log.Warnf("Slack bot does not have permissions to list access lists. Please add access_list read and list permissions " +
+					"to the role associated with the Slack bot.")
+			} else {
+				log.Errorf("error listing access lists: %v", err)
+			}
+			break
+		}
+
+		for _, accessList := range accessLists {
+			if err := a.notifyForAccessListReviews(ctx, accessList); err != nil {
+				log.WithError(err).Warn("Error notifying for access list reviews")
+			}
+		}
+
+		if nextToken == "" {
+			break
+		}
+	}
+
+	return nil
+}
+
 // notifyForAccessListReviews will notify if access list review dates are getting close. At the moment, this
 // only supports notifying owners.
 func (a *App) notifyForAccessListReviews(ctx context.Context, accessList *accesslist.AccessList) error {
+	log := logger.Get(ctx)
+
 	// Find the current notification window.
 	now := a.clock.Now()
 	notificationStart := accessList.Spec.Audit.NextAuditDate.Add(-accessList.Spec.Audit.Notifications.Start)
 
+	// If the current time before the notification start time, skip notifications.
+	if now.Before(notificationStart) {
+		log.Debugf("Access list %s is not ready for notifications, notifications start at %s", accessList.GetName(), notificationStart.Format(time.RFC3339))
+		return nil
+	}
+
 	allRecipients := a.fetchRecipients(ctx, accessList, now, notificationStart)
 	if len(allRecipients) == 0 {
 		return trace.NotFound("no recipients could be fetched for access list %s", accessList.GetName())
@@ -198,12 +223,6 @@ func (a *App) fetchRecipients(ctx context.Context, accessList *accesslist.Access
 
 	allRecipients := make(map[string]common.Recipient, len(accessList.Spec.Owners))
 
-	// If the current time before the notification start time, skip notifications.
-	if now.Before(notificationStart) {
-		log.Debugf("Access list %s is not ready for notifications, notifications start at %s", accessList.GetName(), notificationStart.Format(time.RFC3339))
-		return nil
-	}
-
 	// Get the owners from the bot as recipients.
 	for _, owner := range accessList.Spec.Owners {
 		recipient, err := a.bot.FetchRecipient(ctx, owner.Name)

integrations/access/accesslist/app_test.go

@@ -23,7 +23,7 @@ import (
 
 	"github.com/gravitational/trace"
 	"github.com/jonboulle/clockwork"
-	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/mock"
 	"github.com/stretchr/testify/require"
 
 	"github.com/gravitational/teleport/api/types"
@@ -65,12 +65,12 @@ func (m *mockMessagingBot) SupportedApps() []common.App {
 }
 
 type mockPluginConfig struct {
-	as  *auth.Server
-	bot *mockMessagingBot
+	client teleport.Client
+	bot    *mockMessagingBot
 }
 
 func (m *mockPluginConfig) GetTeleportClient(ctx context.Context) (teleport.Client, error) {
-	return m.as, nil
+	return m.client, nil
 }
 
 func (m *mockPluginConfig) GetRecipients() common.RawRecipientsMap {
@@ -93,23 +93,26 @@ func TestAccessListReminders(t *testing.T) {
 	server, err := auth.NewTestServer(auth.TestServerConfig{
 		Auth: auth.TestAuthServerConfig{
 			Dir:   t.TempDir(),
-			Clock: clock,
+			Clock: clockwork.NewFakeClock(),
 		},
 	})
 	require.NoError(t, err)
 	as := server.Auth()
+	t.Cleanup(func() {
+		require.NoError(t, as.Close())
+	})
 
 	bot := &mockMessagingBot{
 		recipients: map[string]*common.Recipient{
 			"owner1": {Name: "owner1"},
 			"owner2": {Name: "owner2"},
 		},
 	}
-	app := common.NewApp(&mockPluginConfig{as: as, bot: bot}, "test-plugin")
+	app := common.NewApp(&mockPluginConfig{client: as, bot: bot}, "test-plugin")
 	app.Clock = clock
 	ctx := context.Background()
 	go func() {
-		require.NoError(t, app.Run(ctx))
+		app.Run(ctx)
 	}()
 
 	ready, err := app.WaitReady(ctx)
@@ -179,6 +182,70 @@ func TestAccessListReminders(t *testing.T) {
 	}
 }
 
+type mockClient struct {
+	mock.Mock
+	teleport.Client
+}
+
+func (m *mockClient) ListAccessLists(ctx context.Context, pageSize int, pageToken string) ([]*accesslist.AccessList, string, error) {
+	args := m.Called(ctx, pageSize, pageToken)
+	return args.Get(0).([]*accesslist.AccessList), args.String(1), args.Error(2)
+}
+
+func TestAccessListReminders_BadClient(t *testing.T) {
+	t.Parallel()
+
+	clock := clockwork.NewFakeClockAt(time.Date(2023, 1, 1, 0, 0, 0, 0, time.UTC))
+
+	server, err := auth.NewTestServer(auth.TestServerConfig{
+		Auth: auth.TestAuthServerConfig{
+			Dir:   t.TempDir(),
+			Clock: clockwork.NewFakeClock(),
+		},
+	})
+	require.NoError(t, err)
+	as := server.Auth()
+	t.Cleanup(func() {
+		require.NoError(t, as.Close())
+	})
+
+	// Use this mock client so that we can force ListAccessLists to return an error.
+	client := &mockClient{
+		Client: as,
+	}
+	client.On("ListAccessLists", mock.Anything, mock.Anything, mock.Anything).Return(([]*accesslist.AccessList)(nil), "", trace.BadParameter("error"))
+
+	bot := &mockMessagingBot{
+		recipients: map[string]*common.Recipient{
+			"owner1": {Name: "owner1"},
+			"owner2": {Name: "owner2"},
+		},
+	}
+	app := common.NewApp(&mockPluginConfig{client: client, bot: bot}, "test-plugin")
+	app.Clock = clock
+	ctx := context.Background()
+	go func() {
+		app.Run(ctx)
+	}()
+
+	ready, err := app.WaitReady(ctx)
+	require.NoError(t, err)
+	require.True(t, ready)
+
+	t.Cleanup(func() {
+		app.Terminate()
+		<-app.Done()
+		require.NoError(t, app.Err())
+	})
+
+	clock.BlockUntil(1)
+	for i := 1; i <= 6; i++ {
+		clock.Advance(3 * time.Hour)
+		clock.BlockUntil(1)
+		client.AssertNumberOfCalls(t, "ListAccessLists", i)
+	}
+}
+
 func advanceAndLookForRecipients(t *testing.T,
 	bot *mockMessagingBot,
 	alSvc services.AccessLists,
@@ -203,8 +270,7 @@ func advanceAndLookForRecipients(t *testing.T,
 		}
 	}
 	clock.Advance(advance)
+	clock.BlockUntil(1)
 
-	require.EventuallyWithT(t, func(t *assert.CollectT) {
-		assert.ElementsMatch(t, expectedRecipients, bot.lastReminderRecipients)
-	}, 5*time.Second, 5*time.Millisecond)
+	require.ElementsMatch(t, expectedRecipients, bot.lastReminderRecipients)
 }

integrations/access/common/config.go

@@ -23,10 +23,10 @@ import (
 
 	"github.com/gravitational/teleport/api/client"
 	"github.com/gravitational/teleport/api/types"
+	"github.com/gravitational/teleport/api/types/accesslist"
 	"github.com/gravitational/teleport/integrations/access/common/teleport"
 	"github.com/gravitational/teleport/integrations/lib"
 	"github.com/gravitational/teleport/integrations/lib/logger"
-	"github.com/gravitational/teleport/lib/services"
 )
 
 type PluginConfiguration interface {
@@ -51,8 +51,8 @@ type wrappedClient struct {
 	*client.Client
 }
 
-func (w *wrappedClient) AccessListClient() services.AccessLists {
-	return w.Client.AccessListClient()
+func (w *wrappedClient) ListAccessLists(ctx context.Context, pageSize int, pageToken string) ([]*accesslist.AccessList, string, error) {
+	return w.Client.AccessListClient().ListAccessLists(ctx, pageSize, pageToken)
 }
 
 // wrapAPIClient will wrap the API client such that it conforms to the Teleport plugin client interface.

integrations/access/common/teleport/client.go

@@ -19,8 +19,8 @@ import (
 
 	"github.com/gravitational/teleport/api/client/proto"
 	"github.com/gravitational/teleport/api/types"
+	"github.com/gravitational/teleport/api/types/accesslist"
 	"github.com/gravitational/teleport/integrations/lib/plugindata"
-	"github.com/gravitational/teleport/lib/services"
 )
 
 // Client aggregates the parts of Teleport API client interface
@@ -34,5 +34,5 @@ type Client interface {
 	SubmitAccessReview(ctx context.Context, params types.AccessReviewSubmission) (types.AccessRequest, error)
 	SetAccessRequestState(ctx context.Context, params types.AccessRequestUpdate) error
 	ListResources(ctx context.Context, req proto.ListResourcesRequest) (*types.ListResourcesResponse, error)
-	AccessListClient() services.AccessLists
+	ListAccessLists(context.Context, int, string) ([]*accesslist.AccessList, string, error)
 }