Access Plugins: Support dynamic credential reloading

access/email/app.go

@@ -24,21 +24,16 @@ import (
 	"github.com/gravitational/teleport/api/client/proto"
 	"github.com/gravitational/teleport/api/types"
 	"github.com/gravitational/teleport/integrations/lib"
-	"github.com/gravitational/teleport/integrations/lib/credentials"
 	"github.com/gravitational/teleport/integrations/lib/logger"
 	"github.com/gravitational/teleport/integrations/lib/watcherjob"
 	"github.com/gravitational/trace"
-	"google.golang.org/grpc"
-	grpcbackoff "google.golang.org/grpc/backoff"
 )
 
 const (
 	// minServerVersion is the minimal teleport version the plugin supports.
 	minServerVersion = "6.1.0-beta.1"
 	// pluginName is used to tag PluginData and as a Delegator in Audit log.
 	pluginName = "email"
-	// backoffMaxDelay is a maximum time GRPC client waits before reconnection attempt.
-	grpcBackoffMaxDelay = time.Second * 2
 	// initTimeout is used to bound execution time of health check and teleport version check.
 	initTimeout = time.Second * 10
 	// handlerTimeout is used to bound the execution time of watcher event handler.
@@ -127,44 +122,18 @@ func (a *App) run(ctx context.Context) error {
 func (a *App) init(ctx context.Context) error {
 	ctx, cancel := context.WithTimeout(ctx, initTimeout)
 	defer cancel()
-	log := logger.Get(ctx)
 
-	if validCred, err := credentials.CheckIfExpired(a.conf.Teleport.Credentials()); err != nil {
-		log.Warn(err)
-		if !validCred {
-			return trace.BadParameter(
-				"No valid credentials found, this likely means credentials are expired. In this case, please sign new credentials and increase their TTL if needed.",
-			)
-		}
-		log.Info("At least one non-expired credential has been found, continuing startup")
-	}
-
-	var (
-		err          error
-		webProxyAddr string
-		pong         proto.PingResponse
-	)
-
-	bk := grpcbackoff.DefaultConfig
-	bk.MaxDelay = grpcBackoffMaxDelay
-	if a.apiClient, err = client.New(ctx, client.Config{
-		Addrs:       a.conf.Teleport.GetAddrs(),
-		Credentials: a.conf.Teleport.Credentials(),
-		DialOpts: []grpc.DialOption{
-			grpc.WithConnectParams(grpc.ConnectParams{Backoff: bk, MinConnectTimeout: initTimeout}),
-			grpc.WithDefaultCallOptions(
-				grpc.WaitForReady(true),
-			),
-			grpc.WithReturnConnectionError(),
-		},
-	}); err != nil {
+	var err error
+	if a.apiClient, err = a.conf.Teleport.NewClient(ctx); err != nil {
 		return trace.Wrap(err)
 	}
 
-	if pong, err = a.checkTeleportVersion(ctx); err != nil {
+	pong, err := a.checkTeleportVersion(ctx)
+	if err != nil {
 		return trace.Wrap(err)
 	}
 
+	var webProxyAddr string
 	if pong.ServerFeatures.AdvancedAccessWorkflows {
 		webProxyAddr = pong.ProxyPublicAddr
 	}

access/msteams/app.go

@@ -22,25 +22,20 @@ import (
 	"github.com/gravitational/teleport/api/client/proto"
 	"github.com/gravitational/teleport/api/types"
 	"github.com/gravitational/teleport/integrations/lib"
-	"github.com/gravitational/teleport/integrations/lib/credentials"
 	"github.com/gravitational/teleport/integrations/lib/logger"
 	pd "github.com/gravitational/teleport/integrations/lib/plugindata"
 	"github.com/gravitational/teleport/integrations/lib/stringset"
 	"github.com/gravitational/teleport/integrations/lib/watcherjob"
 	"github.com/gravitational/trace"
-	"google.golang.org/grpc"
-	grpcbackoff "google.golang.org/grpc/backoff"
 )
 
 const (
 	// pluginName used as Teleport plugin identifier
 	pluginName = "msteams"
 	// minServerVersion is the minimal teleport version the plugin supports.
 	minServerVersion = "8.0.0"
-	// grpcBackoffMaxDelay is a maximum time GRPC client waits before reconnection attempt.
-	grpcBackoffMaxDelay = time.Second * 2
-	// minConnectTimeout GRPC connect timeout.
-	minConnectTimeout = time.Second * 10
+	// initTimeout is used to bound execution time of health check and teleport version check.
+	initTimeout = time.Second * 10
 	// handlerTimeout is used to bound the execution time of watcher event handler.
 	handlerTimeout = time.Second * 5
 )
@@ -106,36 +101,10 @@ func (a *App) WaitReady(ctx context.Context) (bool, error) {
 
 // init initializes the application
 func (a *App) init(ctx context.Context) error {
-	log := logger.Get(ctx)
-	if validCred, err := credentials.CheckIfExpired(a.conf.Teleport.Credentials()); err != nil {
-		log.Warn(err)
-		if !validCred {
-			return trace.BadParameter(
-				"No valid credentials found, this likely means credentials are expired. In this case, please sign new credentials and increase their TTL if needed.",
-			)
-		}
-		log.Info("At least one non-expired credential has been found, continuing startup")
-	}
+	ctx, cancel := context.WithTimeout(ctx, initTimeout)
+	defer cancel()
 
-	bk := grpcbackoff.DefaultConfig
-	bk.MaxDelay = grpcBackoffMaxDelay
-
-	apiClient, err := client.New(ctx, client.Config{
-		Addrs:       a.conf.Teleport.GetAddrs(),
-		Credentials: a.conf.Teleport.Credentials(),
-		DialOpts: []grpc.DialOption{
-			grpc.WithReturnConnectionError(),
-			grpc.WithDefaultCallOptions(
-				grpc.WaitForReady(true),
-			),
-			grpc.WithConnectParams(
-				grpc.ConnectParams{
-					Backoff:           bk,
-					MinConnectTimeout: minConnectTimeout,
-				},
-			),
-		},
-	})
+	apiClient, err := a.conf.Teleport.NewClient(ctx)
 	if err != nil {
 		return trace.Wrap(err)
 	}

event-handler/README.md

@@ -213,7 +213,7 @@ storage = "./storage" # Plugin will save it's state here
 timeout = "10s"
 batch = 20
 
-[fluentd]
+[forward.fluentd]
 cert = "client.crt"
 key = "client.key"
 ca = "ca.crt"
@@ -278,26 +278,28 @@ Note that start time can be set only once, on the first run of the tool. If you
 
 You may specify configuration options via command line arguments, environment variables or TOML file.
 
-| CLI arg name        | Description                                         | Env var name              |
-| --------------------|-----------------------------------------------------|---------------------------|
-| teleport-addr       | Teleport host and port                              | FDFWD_TELEPORT_ADDR       |
-| teleport-ca         | Teleport TLS CA file                                | FDFWD_TELEPORT_CA         |
-| teleport-cert       | Teleport TLS certificate file                       | FDWRD_TELEPORT_CERT       |
-| teleport-key        | Teleport TLS key file                               | FDFWD_TELEPORT_KEY        |
-| teleport-identity   | Teleport identity file                              | FDFWD_TELEPORT_IDENTITY   |
-| fluentd-url         | Fluentd URL                                         | FDFWD_FLUENTD_URL         |
-| fluentd-session-url | Fluentd session URL                                 | FDFWD_FLUENTD_SESSION_URL |
-| fluentd-ca          | fluentd TLS CA file                                 | FDFWD_FLUENTD_CA          |
-| fluentd-cert        | Fluentd TLS certificate file                        | FDFWD_FLUENTD_CERT        |
-| fluentd-key         | Fluentd TLS key file                                | FDFWD_FLUENTD_KEY         |
-| storage             | Storage directory                                   | FDFWD_STORAGE             |
-| batch               | Fetch batch size                                    | FDFWD_BATCH               |
-| types               | Comma-separated list of event types to forward      | FDFWD_TYPES               |
-| skip-session-types  | Comma-separated list of session event types to skip | FDFWD_SKIP_SESSION_TYPES  |
-| start-time          | Minimum event time (RFC3339 format)                 | FDFWD_START_TIME          |
-| timeout             | Polling timeout                                     | FDFWD_TIMEOUT             |
-| cursor              | Start cursor value                                  | FDFWD_CURSOR              |
-| debug               | Debug logging                                       | FDFWD_DEBUG               |
+| CLI arg name              | Description                                                                                           | Env var name                    |
+|---------------------------|-------------------------------------------------------------------------------------------------------|---------------------------------|
+| teleport-addr             | Teleport host and port                                                                                | FDFWD_TELEPORT_ADDR             |
+| teleport-ca               | Teleport TLS CA file                                                                                  | FDFWD_TELEPORT_CA               |
+| teleport-cert             | Teleport TLS certificate file                                                                         | FDWRD_TELEPORT_CERT             |
+| teleport-key              | Teleport TLS key file                                                                                 | FDFWD_TELEPORT_KEY              |
+| teleport-identity         | Teleport identity file                                                                                | FDFWD_TELEPORT_IDENTITY         |
+| teleport-refresh-enabled  | Controls if the identity file should be reloaded from disk after the initial start on interval.       | FDFWD_TELEPORT_REFRESH_ENABLED  |
+| teleport-refresh-interval | How often to load the identity file from disk when teleport-refresh-enabled is specified. Default: 1m | FDFWD_TELEPORT_REFRESH_INTERVAL |
+| fluentd-url               | Fluentd URL                                                                                           | FDFWD_FLUENTD_URL               |
+| fluentd-session-url       | Fluentd session URL                                                                                   | FDFWD_FLUENTD_SESSION_URL       |
+| fluentd-ca                | fluentd TLS CA file                                                                                   | FDFWD_FLUENTD_CA                |
+| fluentd-cert              | Fluentd TLS certificate file                                                                          | FDFWD_FLUENTD_CERT              |
+| fluentd-key               | Fluentd TLS key file                                                                                  | FDFWD_FLUENTD_KEY               |
+| storage                   | Storage directory                                                                                     | FDFWD_STORAGE                   |
+| batch                     | Fetch batch size                                                                                      | FDFWD_BATCH                     |
+| types                     | Comma-separated list of event types to forward                                                        | FDFWD_TYPES                     |
+| skip-session-types        | Comma-separated list of session event types to skip                                                   | FDFWD_SKIP_SESSION_TYPES        |
+| start-time                | Minimum event time (RFC3339 format)                                                                   | FDFWD_START_TIME                |
+| timeout                   | Polling timeout                                                                                       | FDFWD_TIMEOUT                   |
+| cursor                    | Start cursor value                                                                                    | FDFWD_CURSOR                    |
+| debug                     | Debug logging                                                                                         | FDFWD_DEBUG                     |
 
 TOML configuration keys are the same as CLI args. Teleport and Fluentd variables can be grouped into sections. See [example TOML](example/config.toml). You can specify TOML file location using `--config` CLI flag.
 

event-handler/cli.go

@@ -55,6 +55,14 @@ type TeleportConfig struct {
 	// TeleportIdentityFile is a path to Teleport identity file
 	TeleportIdentityFile string `help:"Teleport identity file" type:"existingfile" name:"teleport-identity" env:"FDFWD_TELEPORT_IDENTITY"`
 
+	// TeleportRefreshEnabled will reload the identity file from disk on the
+	// configured interval.
+	TeleportRefreshEnabled bool `help:"Configures the identity file to be reloaded from disk at a configured interval." env:"FDFWD_TELEPORT_REFRESH_ENABLED"`
+
+	// TeleportRefreshInterval is how often the identity file should
+	// be reloaded from disk.
+	TeleportRefreshInterval time.Duration `help:"Configures how often the identity file should be reloaded from disk." env:"FDFWD_TELEPORT_REFRESH_INTERVAL" default:"1m"`
+
 	// TeleportCA is a path to Teleport CA file
 	TeleportCA string `help:"Teleport TLS CA file" type:"existingfile" env:"FDFWD_TELEPORT_CA"`
 
@@ -241,6 +249,9 @@ func (c *StartCmdConfig) Dump(ctx context.Context) {
 	if c.TeleportIdentityFile != "" {
 		log.WithField("file", c.TeleportIdentityFile).Info("Using Teleport identity file")
 	}
+	if c.TeleportRefreshEnabled {
+		log.WithField("interval", c.TeleportRefreshInterval).Info("Using Teleport identity file refresh")
+	}
 
 	if c.TeleportKey != "" {
 		log.WithField("addr", c.TeleportAddr).Info("Using Teleport addr")

event-handler/cli_test.go

@@ -0,0 +1,76 @@
+package main
+
+import (
+	"github.com/alecthomas/kong"
+	"github.com/stretchr/testify/require"
+	"os"
+	"path"
+	"testing"
+	"time"
+)
+
+// StartCmdConfig is mostly to test that the TOML file parsing works as
+// expected.
+func TestStartCmdConfig(t *testing.T) {
+	wd, err := os.Getwd()
+	require.NoError(t, err)
+
+	testCases := []struct {
+		name string
+		args []string
+
+		want StartCmdConfig
+	}{
+		{
+			name: "standard",
+			args: []string{"start", "--config", "testdata/config.toml"},
+			want: StartCmdConfig{
+				FluentdConfig: FluentdConfig{
+					FluentdURL:        "https://localhost:8888/test.log",
+					FluentdSessionURL: "https://localhost:8888/session",
+					FluentdCert:       path.Join(wd, "testdata", "fake-file"),
+					FluentdKey:        path.Join(wd, "testdata", "fake-file"),
+					FluentdCA:         path.Join(wd, "testdata", "fake-file"),
+				},
+				TeleportConfig: TeleportConfig{
+					TeleportAddr:            "localhost:3025",
+					TeleportIdentityFile:    path.Join(wd, "testdata", "fake-file"),
+					TeleportRefreshEnabled:  true,
+					TeleportRefreshInterval: 2 * time.Minute,
+				},
+				IngestConfig: IngestConfig{
+					StorageDir:          "./storage",
+					BatchSize:           20,
+					SkipSessionTypesRaw: []string{"print"},
+					SkipSessionTypes: map[string]struct{}{
+						"print": {},
+					},
+					Timeout:     10 * time.Second,
+					Concurrency: 5,
+				},
+				LockConfig: LockConfig{
+					LockFailedAttemptsCount: 3,
+					LockPeriod:              time.Minute,
+				},
+			},
+		},
+	}
+
+	for _, tc := range testCases {
+		t.Run(tc.name, func(t *testing.T) {
+			cli := CLI{}
+			parser, err := kong.New(
+				&cli,
+				kong.UsageOnError(),
+				kong.Configuration(KongTOMLResolver),
+				kong.Name(pluginName),
+				kong.Description(pluginDescription),
+			)
+			require.NoError(t, err)
+			_, err = parser.Parse(tc.args)
+			require.NoError(t, err)
+
+			require.Equal(t, tc.want, cli.Start)
+		})
+	}
+}

event-handler/event_handler_test.go

@@ -241,7 +241,7 @@ func (s *EventHandlerSuite) TestEvents() {
 	require.Contains(t, evt, `"user":"`+s.userNames.plugin+`"`)
 	require.Contains(t, evt, `"roles":["access-event-handler"]`)
 
-	err = s.ruler().CreateUser(s.Context(), &types.UserV2{
+	_, err = s.ruler().CreateUser(s.Context(), &types.UserV2{
 		Metadata: types.Metadata{
 			Name: "fake-ruler",
 		},

event-handler/kong_toml_resolver.go

@@ -49,10 +49,10 @@ func KongTOMLResolver(r io.Reader) (kong.Resolver, error) {
 		}
 
 		value := config.Get(name)
-		valueWithinSeciton := config.Get(strings.ReplaceAll(name, "-", "."))
+		valueWithinSection := config.Get(strings.ReplaceAll(name, "-", "."))
 
-		if valueWithinSeciton != nil {
-			return valueWithinSeciton, nil
+		if valueWithinSection != nil {
+			return valueWithinSection, nil
 		}
 
 		return value, nil

event-handler/teleport_events_watcher.go

@@ -25,6 +25,7 @@ import (
 	auditlogpb "github.com/gravitational/teleport/api/gen/proto/go/teleport/auditlog/v1"
 	"github.com/gravitational/teleport/api/types"
 	"github.com/gravitational/teleport/api/types/events"
+	"github.com/gravitational/teleport/integrations/lib"
 	"github.com/gravitational/teleport/integrations/lib/credentials"
 	"github.com/gravitational/teleport/integrations/lib/logger"
 	"github.com/gravitational/trace"
@@ -80,17 +81,23 @@ func NewTeleportEventsWatcher(
 	cursor string,
 	id string,
 ) (*TeleportEventsWatcher, error) {
-	var err error
-
-	config := client.Config{
-		Addrs: []string{c.TeleportAddr},
-		Credentials: []client.Credentials{
-			client.LoadIdentityFile(c.TeleportIdentityFile),
-			client.LoadKeyPair(c.TeleportCert, c.TeleportKey, c.TeleportCA),
-		},
+	var creds []client.Credentials
+	switch {
+	case c.TeleportIdentityFile != "" && !c.TeleportRefreshEnabled:
+		creds = []client.Credentials{client.LoadIdentityFile(c.TeleportIdentityFile)}
+	case c.TeleportIdentityFile != "" && c.TeleportRefreshEnabled:
+		cred, err := lib.NewIdentityFileWatcher(ctx, c.TeleportIdentityFile, c.TeleportRefreshInterval)
+		if err != nil {
+			return nil, trace.Wrap(err)
+		}
+		creds = []client.Credentials{cred}
+	case c.TeleportCert != "" && c.TeleportKey != "" && c.TeleportCA != "":
+		creds = []client.Credentials{client.LoadKeyPair(c.TeleportCert, c.TeleportKey, c.TeleportCA)}
+	default:
+		return nil, trace.BadParameter("no credentials configured")
 	}
 
-	if validCred, err := credentials.CheckIfExpired(config.Credentials); err != nil {
+	if validCred, err := credentials.CheckIfExpired(creds); err != nil {
 		log.Warn(err)
 		if !validCred {
 			return nil, trace.BadParameter(
@@ -100,13 +107,18 @@ func NewTeleportEventsWatcher(
 		log.Info("At least one non-expired credential has been found, continuing startup")
 	}
 
-	client, err := client.New(ctx, config)
+	config := client.Config{
+		Addrs:       []string{c.TeleportAddr},
+		Credentials: creds,
+	}
+
+	teleportClient, err := client.New(ctx, config)
 	if err != nil {
 		return nil, trace.Wrap(err)
 	}
 
 	tc := TeleportEventsWatcher{
-		client:    client,
+		client:    teleportClient,
 		pos:       -1,
 		cursor:    cursor,
 		config:    c,