[sec_scan][12] add cache and events support for AccessGraphSettings (…

…#44016)

* [sec_scan][12] add cache and events support for `AccessGraphSettings`

This PR adds the cache and events support for the new resource `AccessGraphSettings`.

This PR is part of gravitational/access-graph#637.

Signed-off-by: Tiago Silva <[email protected]>

* add tests

---------

Signed-off-by: Tiago Silva <[email protected]>

api/client/events.go

@@ -19,6 +19,7 @@ import (
 
 	"github.com/gravitational/teleport/api/client/proto"
 	accessmonitoringrulesv1 "github.com/gravitational/teleport/api/gen/proto/go/teleport/accessmonitoringrules/v1"
+	clusterconfigpb "github.com/gravitational/teleport/api/gen/proto/go/teleport/clusterconfig/v1"
 	crownjewelv1 "github.com/gravitational/teleport/api/gen/proto/go/teleport/crownjewel/v1"
 	dbobjectv1 "github.com/gravitational/teleport/api/gen/proto/go/teleport/dbobject/v1"
 	kubewaitingcontainerpb "github.com/gravitational/teleport/api/gen/proto/go/teleport/kubewaitingcontainer/v1"
@@ -80,6 +81,10 @@ func EventToGRPC(in types.Event) (*proto.Event, error) {
 			out.Resource = &proto.Event_DatabaseObject{
 				DatabaseObject: r,
 			}
+		case *clusterconfigpb.AccessGraphSettings:
+			out.Resource = &proto.Event_AccessGraphSettings{
+				AccessGraphSettings: r,
+			}
 		default:
 			return nil, trace.BadParameter("resource type %T is not supported", r)
 		}
@@ -511,6 +516,9 @@ func EventFromGRPC(in *proto.Event) (*types.Event, error) {
 	} else if r := in.GetDatabaseObject(); r != nil {
 		out.Resource = types.Resource153ToLegacy(r)
 		return &out, nil
+	} else if r := in.GetAccessGraphSettings(); r != nil {
+		out.Resource = types.Resource153ToLegacy(r)
+		return &out, nil
 	} else {
 		return nil, trace.BadParameter("received unsupported resource %T", in.Resource)
 	}

api/proto/teleport/legacy/client/proto/event.proto

@@ -18,6 +18,7 @@ package proto;
 
 import "teleport/accesslist/v1/accesslist.proto";
 import "teleport/accessmonitoringrules/v1/access_monitoring_rules.proto";
+import "teleport/clusterconfig/v1/access_graph_settings.proto";
 import "teleport/crownjewel/v1/crownjewel.proto";
 import "teleport/dbobject/v1/dbobject.proto";
 import "teleport/discoveryconfig/v1/discoveryconfig.proto";
@@ -165,5 +166,7 @@ message Event {
     teleport.crownjewel.v1.CrownJewel CrownJewel = 58;
     // DatabaseObject is a database object resource.
     teleport.dbobject.v1.DatabaseObject DatabaseObject = 59;
+    // AccessGraphSettings is a resource for access graph settings.
+    teleport.clusterconfig.v1.AccessGraphSettings AccessGraphSettings = 61;
   }
 }

lib/auth/authclient/api.go

@@ -28,6 +28,7 @@ import (
 
 	"github.com/gravitational/teleport/api/client/proto"
 	accessmonitoringrules "github.com/gravitational/teleport/api/gen/proto/go/teleport/accessmonitoringrules/v1"
+	clusterconfigpb "github.com/gravitational/teleport/api/gen/proto/go/teleport/clusterconfig/v1"
 	crownjewelv1 "github.com/gravitational/teleport/api/gen/proto/go/teleport/crownjewel/v1"
 	integrationpb "github.com/gravitational/teleport/api/gen/proto/go/teleport/integration/v1"
 	kubewaitingcontainerpb "github.com/gravitational/teleport/api/gen/proto/go/teleport/kubewaitingcontainer/v1"
@@ -1172,6 +1173,9 @@ type Cache interface {
 
 	// DatabaseObjectsGetter defines methods for fetching database objects.
 	services.DatabaseObjectsGetter
+
+	// GetAccessGraphSettings returns the access graph settings.
+	GetAccessGraphSettings(context.Context) (*clusterconfigpb.AccessGraphSettings, error)
 }
 
 type NodeWrapper struct {

lib/auth/clusterconfig/clusterconfigv1/service.go

@@ -38,13 +38,15 @@ type Cache interface {
 	GetAuthPreference(context.Context) (types.AuthPreference, error)
 	GetClusterNetworkingConfig(ctx context.Context) (types.ClusterNetworkingConfig, error)
 	GetSessionRecordingConfig(ctx context.Context) (types.SessionRecordingConfig, error)
+	GetAccessGraphSettings(context.Context) (*clusterconfigpb.AccessGraphSettings, error)
 }
 
 // ReadOnlyCache abstracts over the required methods of [readonly.Cache].
 type ReadOnlyCache interface {
 	GetReadOnlyAuthPreference(context.Context) (readonly.AuthPreference, error)
 	GetReadOnlyClusterNetworkingConfig(ctx context.Context) (readonly.ClusterNetworkingConfig, error)
 	GetReadOnlySessionRecordingConfig(ctx context.Context) (readonly.SessionRecordingConfig, error)
+	GetReadOnlyAccessGraphSettings(context.Context) (readonly.AccessGraphSettings, error)
 }
 
 // Backend is used by the [Service] to mutate cluster config resources.

lib/authz/permissions.go

@@ -36,6 +36,7 @@ import (
 	"github.com/gravitational/teleport"
 	"github.com/gravitational/teleport/api/client/proto"
 	"github.com/gravitational/teleport/api/defaults"
+	clusterconfigpb "github.com/gravitational/teleport/api/gen/proto/go/teleport/clusterconfig/v1"
 	mfav1 "github.com/gravitational/teleport/api/gen/proto/go/teleport/mfa/v1"
 	"github.com/gravitational/teleport/api/mfa"
 	"github.com/gravitational/teleport/api/types"
@@ -113,7 +114,7 @@ func NewAuthorizer(opts AuthorizerOpts) (Authorizer, error) {
 		// cannot handle caching, and will fail if caching is enabled.
 		var err error
 		opts.ReadOnlyAccessPoint, err = readonly.NewCache(readonly.CacheConfig{
-			Upstream: opts.AccessPoint,
+			Upstream: accessPointWrapper{opts.AccessPoint},
 			Disabled: !opts.PermitCaching,
 		})
 		if err != nil {
@@ -133,6 +134,15 @@ func NewAuthorizer(opts AuthorizerOpts) (Authorizer, error) {
 	}, nil
 }
 
+type accessPointWrapper struct {
+	AuthorizerAccessPoint
+}
+
+// GetAccessGraphSettings returns the access graph settings.
+func (accessPointWrapper) GetAccessGraphSettings(ctx context.Context) (*clusterconfigpb.AccessGraphSettings, error) {
+	return nil, trace.NotImplemented("GetAccessGraphSettings is not implemented")
+}
+
 // Authorizer authorizes identity and returns auth context
 type Authorizer interface {
 	// Authorize authorizes user based on identity supplied via context

lib/cache/cache.go

@@ -32,11 +32,13 @@ import (
 	"go.opentelemetry.io/otel/attribute"
 	oteltrace "go.opentelemetry.io/otel/trace"
 	"golang.org/x/sync/errgroup"
+	protobuf "google.golang.org/protobuf/proto"
 
 	"github.com/gravitational/teleport"
 	"github.com/gravitational/teleport/api/client/proto"
 	apidefaults "github.com/gravitational/teleport/api/defaults"
 	accessmonitoringrulesv1 "github.com/gravitational/teleport/api/gen/proto/go/teleport/accessmonitoringrules/v1"
+	clusterconfigpb "github.com/gravitational/teleport/api/gen/proto/go/teleport/clusterconfig/v1"
 	crownjewelv1 "github.com/gravitational/teleport/api/gen/proto/go/teleport/crownjewel/v1"
 	dbobjectv1 "github.com/gravitational/teleport/api/gen/proto/go/teleport/dbobject/v1"
 	kubewaitingcontainerpb "github.com/gravitational/teleport/api/gen/proto/go/teleport/kubewaitingcontainer/v1"
@@ -179,6 +181,7 @@ func ForAuth(cfg Config) Config {
 		{Kind: types.KindGlobalNotification},
 		{Kind: types.KindAccessMonitoringRule},
 		{Kind: types.KindDatabaseObject},
+		{Kind: types.KindAccessGraphSettings},
 	}
 	cfg.QueueSize = defaults.AuthQueueSize
 	// We don't want to enable partial health for auth cache because auth uses an event stream
@@ -3261,3 +3264,28 @@ func (c *Cache) ListResources(ctx context.Context, req proto.ListResourcesReques
 
 	return rg.reader.ListResources(ctx, req)
 }
+
+// GetAccessGraphSettings gets AccessGraphSettings from the backend.
+func (c *Cache) GetAccessGraphSettings(ctx context.Context) (*clusterconfigpb.AccessGraphSettings, error) {
+	ctx, span := c.Tracer.Start(ctx, "cache/GetAccessGraphSettings")
+	defer span.End()
+
+	rg, err := readCollectionCache(c, c.collections.accessGraphSettings)
+	if err != nil {
+		return nil, trace.Wrap(err)
+	}
+	defer rg.Release()
+	if !rg.IsCacheRead() {
+		cachedCfg, err := utils.FnCacheGet(ctx, c.fnCache, clusterConfigCacheKey{"access_graph_settings"}, func(ctx context.Context) (*clusterconfigpb.AccessGraphSettings, error) {
+			cfg, err := rg.reader.GetAccessGraphSettings(ctx)
+			return cfg, err
+		})
+		if err != nil {
+			return nil, trace.Wrap(err)
+		}
+
+		clone := protobuf.Clone(cachedCfg).(*clusterconfigpb.AccessGraphSettings)
+		return clone, nil
+	}
+	return rg.reader.GetAccessGraphSettings(ctx)
+}

lib/cache/cache_test.go

@@ -43,13 +43,15 @@ import (
 	"github.com/gravitational/teleport/api/client/proto"
 	apidefaults "github.com/gravitational/teleport/api/defaults"
 	accessmonitoringrulesv1 "github.com/gravitational/teleport/api/gen/proto/go/teleport/accessmonitoringrules/v1"
+	clusterconfigpb "github.com/gravitational/teleport/api/gen/proto/go/teleport/clusterconfig/v1"
 	crownjewelv1 "github.com/gravitational/teleport/api/gen/proto/go/teleport/crownjewel/v1"
 	dbobjectv1 "github.com/gravitational/teleport/api/gen/proto/go/teleport/dbobject/v1"
 	headerv1 "github.com/gravitational/teleport/api/gen/proto/go/teleport/header/v1"
 	kubewaitingcontainerpb "github.com/gravitational/teleport/api/gen/proto/go/teleport/kubewaitingcontainer/v1"
 	notificationsv1 "github.com/gravitational/teleport/api/gen/proto/go/teleport/notifications/v1"
 	"github.com/gravitational/teleport/api/types"
 	"github.com/gravitational/teleport/api/types/accesslist"
+	"github.com/gravitational/teleport/api/types/clusterconfig"
 	"github.com/gravitational/teleport/api/types/discoveryconfig"
 	"github.com/gravitational/teleport/api/types/header"
 	"github.com/gravitational/teleport/api/types/kubewaitingcontainer"
@@ -3217,6 +3219,7 @@ func TestCacheWatchKindExistsInEvents(t *testing.T) {
 		types.KindAccessMonitoringRule:    types.Resource153ToLegacy(newAccessMonitoringRule(t)),
 		types.KindCrownJewel:              types.Resource153ToLegacy(newCrownJewel(t, "test")),
 		types.KindDatabaseObject:          types.Resource153ToLegacy(newDatabaseObject(t, "test")),
+		types.KindAccessGraphSettings:     types.Resource153ToLegacy(newAccessGraphSettings(t)),
 	}
 
 	for name, cfg := range cases {
@@ -3689,6 +3692,16 @@ func newDatabaseObject(t *testing.T, name string) *dbobjectv1.DatabaseObject {
 	return r
 }
 
+func newAccessGraphSettings(t *testing.T) *clusterconfigpb.AccessGraphSettings {
+	t.Helper()
+
+	r, err := clusterconfig.NewAccessGraphSettings(&clusterconfigpb.AccessGraphSettingsSpec{
+		SecretsScanConfig: clusterconfigpb.AccessGraphSecretsScanConfig_ACCESS_GRAPH_SECRETS_SCAN_CONFIG_ENABLED,
+	})
+	require.NoError(t, err)
+	return r
+}
+
 func newUserNotification(t *testing.T, name string) *notificationsv1.Notification {
 	t.Helper()
 

lib/cache/collections.go

@@ -29,6 +29,7 @@ import (
 	"github.com/gravitational/teleport/api/client/proto"
 	apidefaults "github.com/gravitational/teleport/api/defaults"
 	accessmonitoringrulesv1 "github.com/gravitational/teleport/api/gen/proto/go/teleport/accessmonitoringrules/v1"
+	clusterconfigpb "github.com/gravitational/teleport/api/gen/proto/go/teleport/clusterconfig/v1"
 	crownjewelv1 "github.com/gravitational/teleport/api/gen/proto/go/teleport/crownjewel/v1"
 	dbobjectv1 "github.com/gravitational/teleport/api/gen/proto/go/teleport/dbobject/v1"
 	kubewaitingcontainerpb "github.com/gravitational/teleport/api/gen/proto/go/teleport/kubewaitingcontainer/v1"
@@ -246,6 +247,7 @@ type cacheCollections struct {
 	windowsDesktops          collectionReader[windowsDesktopsGetter]
 	windowsDesktopServices   collectionReader[windowsDesktopServiceGetter]
 	userNotifications        collectionReader[notificationGetter]
+	accessGraphSettings      collectionReader[accessGraphSettingsGetter]
 	globalNotifications      collectionReader[notificationGetter]
 	accessMonitoringRules    collectionReader[accessMonitoringRuleGetter]
 }
@@ -731,6 +733,15 @@ func setupCollections(c *Cache, watches []types.WatchKind) (*cacheCollections, e
 			}
 			collections.accessMonitoringRules = &genericCollection[*accessmonitoringrulesv1.AccessMonitoringRule, accessMonitoringRuleGetter, accessMonitoringRulesExecutor]{cache: c, watch: watch}
 			collections.byKind[resourceKind] = collections.accessMonitoringRules
+		case types.KindAccessGraphSettings:
+			if c.ClusterConfig == nil {
+				return nil, trace.BadParameter("missing parameter ClusterConfig")
+			}
+			collections.accessGraphSettings = &genericCollection[*clusterconfigpb.AccessGraphSettings, accessGraphSettingsGetter, accessGraphSettingsExecutor]{
+				cache: c,
+				watch: watch,
+			}
+			collections.byKind[resourceKind] = collections.accessGraphSettings
 		default:
 			return nil, trace.BadParameter("resource %q is not supported", watch.Kind)
 		}
@@ -3228,3 +3239,42 @@ type accessMonitoringRuleGetter interface {
 	ListAccessMonitoringRules(ctx context.Context, limit int, startKey string) ([]*accessmonitoringrulesv1.AccessMonitoringRule, string, error)
 	ListAccessMonitoringRulesWithFilter(ctx context.Context, pageSize int, nextToken string, subjects []string, notificationName string) ([]*accessmonitoringrulesv1.AccessMonitoringRule, string, error)
 }
+
+type accessGraphSettingsExecutor struct{}
+
+func (accessGraphSettingsExecutor) getAll(ctx context.Context, cache *Cache, _ bool) ([]*clusterconfigpb.AccessGraphSettings, error) {
+	set, err := cache.ClusterConfig.GetAccessGraphSettings(ctx)
+	if err != nil {
+		return nil, trace.Wrap(err)
+	}
+
+	return []*clusterconfigpb.AccessGraphSettings{set}, nil
+}
+
+func (accessGraphSettingsExecutor) upsert(ctx context.Context, cache *Cache, resource *clusterconfigpb.AccessGraphSettings) error {
+	_, err := cache.clusterConfigCache.UpsertAccessGraphSettings(ctx, resource)
+	return trace.Wrap(err)
+}
+
+func (accessGraphSettingsExecutor) deleteAll(ctx context.Context, cache *Cache) error {
+	return trace.Wrap(cache.clusterConfigCache.DeleteAccessGraphSettings(ctx))
+}
+
+func (accessGraphSettingsExecutor) delete(ctx context.Context, cache *Cache, resource types.Resource) error {
+	return trace.Wrap(cache.clusterConfigCache.DeleteAccessGraphSettings(ctx))
+}
+
+func (accessGraphSettingsExecutor) isSingleton() bool { return false }
+
+func (accessGraphSettingsExecutor) getReader(cache *Cache, cacheOK bool) accessGraphSettingsGetter {
+	if cacheOK {
+		return cache.clusterConfigCache
+	}
+	return cache.Config.ClusterConfig
+}
+
+type accessGraphSettingsGetter interface {
+	GetAccessGraphSettings(context.Context) (*clusterconfigpb.AccessGraphSettings, error)
+}
+
+var _ executor[*clusterconfigpb.AccessGraphSettings, accessGraphSettingsGetter] = accessGraphSettingsExecutor{}

lib/services/local/events.go

@@ -219,6 +219,8 @@ func (e *EventsService) NewWatcher(ctx context.Context, watch types.Watch) (type
 			parser = newAccessGraphSecretPrivateKeyParser()
 		case types.KindAccessGraphSecretAuthorizedKey:
 			parser = newAccessGraphSecretAuthorizedKeyParser()
+		case types.KindAccessGraphSettings:
+			parser = newAccessGraphSettingsParser()
 		default:
 			if watch.AllowPartialSuccess {
 				continue
@@ -2377,3 +2379,37 @@ func baseTwoKeys(key []byte) (string, string, error) {
 	}
 	return string(parts[len(parts)-2]), string(parts[len(parts)-1]), nil
 }
+
+func newAccessGraphSettingsParser() *accessGraphSettingsParser {
+	return &accessGraphSettingsParser{
+		baseParser: newBaseParser(backend.Key(clusterConfigPrefix, accessGraphSettingsPrefix)),
+	}
+}
+
+type accessGraphSettingsParser struct {
+	baseParser
+}
+
+func (p *accessGraphSettingsParser) parse(event backend.Event) (types.Resource, error) {
+	switch event.Type {
+	case types.OpDelete:
+		h, err := resourceHeader(event, types.KindAccessGraphSettings, types.V1, 0)
+		if err != nil {
+			return nil, trace.Wrap(err)
+		}
+		h.SetName(types.MetaNameAccessGraphSettings)
+		return h, nil
+	case types.OpPut:
+		settings, err := services.UnmarshalAccessGraphSettings(
+			event.Item.Value,
+			services.WithExpires(event.Item.Expires),
+			services.WithRevision(event.Item.Revision),
+		)
+		if err != nil {
+			return nil, trace.Wrap(err)
+		}
+		return types.Resource153ToLegacy(settings), nil
+	default:
+		return nil, trace.BadParameter("event %v is not supported", event.Type)
+	}
+}

lib/services/readonly/cache.go

@@ -24,6 +24,7 @@ import (
 
 	"github.com/gravitational/trace"
 
+	clusterconfigpb "github.com/gravitational/teleport/api/gen/proto/go/teleport/clusterconfig/v1"
 	"github.com/gravitational/teleport/api/types"
 	"github.com/gravitational/teleport/lib/utils"
 )
@@ -33,6 +34,7 @@ type Upstream interface {
 	GetAuthPreference(ctx context.Context) (types.AuthPreference, error)
 	GetClusterNetworkingConfig(ctx context.Context) (types.ClusterNetworkingConfig, error)
 	GetSessionRecordingConfig(ctx context.Context) (types.SessionRecordingConfig, error)
+	GetAccessGraphSettings(ctx context.Context) (*clusterconfigpb.AccessGraphSettings, error)
 }
 
 // Cache provides simple ttl-based in-memory caching for select resources that are frequently accessed
@@ -123,3 +125,16 @@ func (c *Cache) GetReadOnlySessionRecordingConfig(ctx context.Context) (SessionR
 	})
 	return cfg, trace.Wrap(err)
 }
+
+// GetReadOnlyAccessGraphSettings returns a read-only shared reference to the dynamic access graph settings resource.
+func (c *Cache) GetReadOnlyAccessGraphSettings(ctx context.Context) (AccessGraphSettings, error) {
+	if c.cfg.Disabled {
+		cfg, err := c.cfg.Upstream.GetAccessGraphSettings(ctx)
+		return sealAccessGraphSettings(cfg), trace.Wrap(err)
+	}
+	cfg, err := utils.FnCacheGet(ctx, c.ttlCache, ttlCacheKey{kind: types.KindAccessGraphSettings}, func(ctx context.Context) (AccessGraphSettings, error) {
+		cfg, err := c.cfg.Upstream.GetAccessGraphSettings(ctx)
+		return sealAccessGraphSettings(cfg), trace.Wrap(err)
+	})
+	return cfg, trace.Wrap(err)
+}