feat(authz): cache entitlement policy within authorization service

opentdf-dev.yaml

@@ -48,8 +48,12 @@ services:
   # policy is enabled by default in mode 'all'
   # policy:
   #   enabled: true
-  # list_request_limit_default: 1000
-  # list_request_limit_max: 2500
+  #   list_request_limit_default: 1000
+  #   list_request_limit_max: 2500
+  # authorization:
+  #   entitlement_policy_cache:
+  #     enabled: false
+  #     refresh_interval: 30s
 server:
   public_hostname: localhost
   tls:

opentdf-example.yaml

@@ -35,8 +35,12 @@ services:
   # policy is enabled by default in mode 'all'
   # policy:
   #   enabled: true
-  # list_request_limit_default: 1000
-  # list_request_limit_max: 2500
+  #   list_request_limit_default: 1000
+  #   list_request_limit_max: 2500
+  # authorization:
+  #   entitlement_policy_cache:
+  #     enabled: false
+  #     refresh_interval: 30s
 server:
   auth:
     enabled: true

service/authorization/v2/authorization.go

@@ -3,14 +3,19 @@ package authorization
 import (
 	"context"
 	"errors"
+	"fmt"
 	"log/slog"
+	"time"
 
 	"connectrpc.com/connect"
+	"github.com/creasty/defaults"
+	"github.com/go-viper/mapstructure/v2"
 	authzV2 "github.com/opentdf/platform/protocol/go/authorization/v2"
 	authzV2Connect "github.com/opentdf/platform/protocol/go/authorization/v2/authorizationv2connect"
 	otdf "github.com/opentdf/platform/sdk"
 	"github.com/opentdf/platform/service/internal/access/v2"
 	"github.com/opentdf/platform/service/logger"
+	"github.com/opentdf/platform/service/pkg/cache"
 	"github.com/opentdf/platform/service/pkg/serviceregistry"
 	"go.opentelemetry.io/otel"
 	"go.opentelemetry.io/otel/propagation"
@@ -23,41 +28,93 @@ type Service struct {
 	config *Config
 	logger *logger.Logger
 	trace.Tracer
+	cache *EntitlementPolicyCache
 }
 
-type Config struct{}
-
 func NewRegistration() *serviceregistry.Service[authzV2Connect.AuthorizationServiceHandler] {
 	as := new(Service)
 
 	return &serviceregistry.Service[authzV2Connect.AuthorizationServiceHandler]{
+		Close: as.Close,
 		ServiceOptions: serviceregistry.ServiceOptions[authzV2Connect.AuthorizationServiceHandler]{
 			Namespace:      "authorization",
 			Version:        "v2",
 			ServiceDesc:    &authzV2.AuthorizationService_ServiceDesc,
 			ConnectRPCFunc: authzV2Connect.NewAuthorizationServiceHandler,
 			RegisterFunc: func(srp serviceregistry.RegistrationParams) (authzV2Connect.AuthorizationServiceHandler, serviceregistry.HandlerServer) {
 				authZCfg := new(Config)
-
-				logger := srp.Logger
+				l := srp.Logger
 
 				as.sdk = srp.SDK
-				as.logger = logger
+				as.logger = l
+				as.config = authZCfg
+				as.Tracer = srp.Tracer
+
+				err := defaults.Set(authZCfg)
+				if err != nil {
+					l.Error("failed to set defaults for authorization service config", slog.Any("error", err))
+					panic(fmt.Errorf("failed to set defaults for authorization service config: %w", err))
+				}
+
+				// Only decode config if it exists
+				if srp.Config != nil {
+					if err := mapstructure.Decode(srp.Config, &authZCfg); err != nil {
+						l.Error("failed to decode authorization service config", slog.Any("error", err))
+						panic(fmt.Errorf("invalid authorization svc config [%v] %w", srp.Config, err))
+					}
+				}
+
+				if err := authZCfg.Validate(); err != nil {
+					l.Error("invalid authorization service config",
+						slog.Any("config", authZCfg.LogValue()),
+						slog.Any("error", err),
+					)
+					panic(fmt.Errorf("invalid authorization svc config %w", err))
+				}
+				l.Debug("authorization service config", slog.Any("config", authZCfg.LogValue()))
+
+				if !authZCfg.Cache.Enabled {
+					l.Debug("entitlement policy cache is disabled")
+					return as, nil
+				}
+
+				cacheClient, err := srp.NewCacheClient(cache.Options{})
+				if err != nil || cacheClient == nil {
+					l.Error("failed to create platform cache client", slog.Any("error", err))
+					panic(fmt.Errorf("failed to create platform cache client: %w", err))
+				}
+
+				refreshInterval, err := time.ParseDuration(authZCfg.Cache.RefreshInterval)
+				if err != nil {
+					l.Error("failed to parse entitlement policy cache refresh interval", slog.Any("error", err))
+					panic(fmt.Errorf("failed to parse entitlement policy cache refresh interval [%s]: %w", authZCfg.Cache.RefreshInterval, err))
+				}
+
+				retriever := access.NewEntitlementPolicyRetriever(as.sdk)
+				as.cache, err = NewEntitlementPolicyCache(context.Background(), l, retriever, cacheClient, refreshInterval)
+				if err != nil {
+					l.Error("failed to create entitlement policy cache", slog.Any("error", err))
+					panic(fmt.Errorf("failed to create entitlement policy cache: %w", err))
+				}
 
 				// if err := srp.RegisterReadinessCheck("authorization", as.IsReady); err != nil {
 				// 	logger.Error("failed to register authorization readiness check", slog.String("error", err.Error()))
 				// }
 
-				as.config = authZCfg
-				as.Tracer = srp.Tracer
-				logger.Debug("authorization v2 service register func")
-
 				return as, nil
 			},
 		},
 	}
 }
 
+// Close gracefully shuts down the authorization service, closing the entitlement policy cache.
+func (as *Service) Close() {
+	as.logger.Info("gracefully shutting down authorization service")
+	if as.cache != nil {
+		as.cache.Stop()
+	}
+}
+
 // TODO: uncomment after v1 is deprecated, as cannot have more than one readiness check under a namespace
 // func (as Service) IsReady(ctx context.Context) error {
 // 	as.logger.TraceContext(ctx, "checking readiness of authorization service")
@@ -79,7 +136,7 @@ func (as *Service) GetEntitlements(ctx context.Context, req *connect.Request[aut
 	withComprehensiveHierarchy := req.Msg.GetWithComprehensiveHierarchy()
 
 	// When authorization service can consume cached policy, switch to the other PDP (process based on policy passed in)
-	pdp, err := access.NewJustInTimePDP(ctx, as.logger, as.sdk)
+	pdp, err := access.NewJustInTimePDP(ctx, as.logger, as.sdk, as.cache)
 	if err != nil {
 		as.logger.ErrorContext(ctx, "failed to create JIT PDP", slog.Any("error", err))
 		return nil, connect.NewError(connect.CodeInternal, err)
@@ -91,7 +148,6 @@ func (as *Service) GetEntitlements(ctx context.Context, req *connect.Request[aut
 		as.logger.ErrorContext(ctx, "failed to get entitlements", slog.Any("error", err))
 		return nil, connect.NewError(connect.CodeInternal, err)
 	}
-
 	rsp := &authzV2.GetEntitlementsResponse{
 		Entitlements: entitlements,
 	}
@@ -110,7 +166,7 @@ func (as *Service) GetDecision(ctx context.Context, req *connect.Request[authzV2
 	propagator := otel.GetTextMapPropagator()
 	ctx = propagator.Extract(ctx, propagation.HeaderCarrier(req.Header()))
 
-	pdp, err := access.NewJustInTimePDP(ctx, as.logger, as.sdk)
+	pdp, err := access.NewJustInTimePDP(ctx, as.logger, as.sdk, as.cache)
 	if err != nil {
 		as.logger.ErrorContext(ctx, "failed to create JIT PDP", slog.Any("error", err))
 		return nil, connect.NewError(connect.CodeInternal, err)
@@ -148,7 +204,7 @@ func (as *Service) GetDecisionMultiResource(ctx context.Context, req *connect.Re
 	propagator := otel.GetTextMapPropagator()
 	ctx = propagator.Extract(ctx, propagation.HeaderCarrier(req.Header()))
 
-	pdp, err := access.NewJustInTimePDP(ctx, as.logger, as.sdk)
+	pdp, err := access.NewJustInTimePDP(ctx, as.logger, as.sdk, as.cache)
 	if err != nil {
 		return nil, statusifyError(ctx, as.logger, errors.Join(errors.New("failed to create JIT PDP"), err))
 	}
@@ -159,12 +215,12 @@ func (as *Service) GetDecisionMultiResource(ctx context.Context, req *connect.Re
 
 	decisions, allPermitted, err := pdp.GetDecision(ctx, entityIdentifier, action, resources)
 	if err != nil {
-		return nil, statusifyError(ctx, as.logger, errors.Join(errors.New("failed to get decision"), err), slog.Any("request", request))
+		return nil, statusifyError(ctx, as.logger, errors.Join(errors.New("failed to get decision"), err))
 	}
 
 	resourceDecisions, err := rollupMultiResourceDecisions(decisions)
 	if err != nil {
-		return nil, statusifyError(ctx, as.logger, errors.Join(errors.New("failed to rollup multi-resource decision"), err), slog.Any("request", request))
+		return nil, statusifyError(ctx, as.logger, errors.Join(errors.New("failed to rollup multi-resource decision"), err))
 	}
 
 	resp := &authzV2.GetDecisionMultiResourceResponse{
@@ -188,7 +244,7 @@ func (as *Service) GetDecisionBulk(ctx context.Context, req *connect.Request[aut
 	propagator := otel.GetTextMapPropagator()
 	ctx = propagator.Extract(ctx, propagation.HeaderCarrier(req.Header()))
 
-	pdp, err := access.NewJustInTimePDP(ctx, as.logger, as.sdk)
+	pdp, err := access.NewJustInTimePDP(ctx, as.logger, as.sdk, as.cache)
 	if err != nil {
 		return nil, statusifyError(ctx, as.logger, errors.Join(errors.New("failed to create JIT PDP"), err))
 	}
@@ -205,12 +261,12 @@ func (as *Service) GetDecisionBulk(ctx context.Context, req *connect.Request[aut
 
 		decisions, allPermitted, err := pdp.GetDecision(ctx, entityIdentifier, action, resources)
 		if err != nil {
-			return nil, statusifyError(ctx, as.logger, errors.Join(errors.New("failed to get bulk decision"), err), slog.Any("request", request))
+			return nil, statusifyError(ctx, as.logger, errors.Join(errors.New("failed to get bulk decision"), err))
 		}
 
 		resourceDecisions, err := rollupMultiResourceDecisions(decisions)
 		if err != nil {
-			return nil, statusifyError(ctx, as.logger, errors.Join(errors.New("failed to rollup bulk multi-resource decision"), err), slog.Any("request", request), slog.Int("index", idx))
+			return nil, statusifyError(ctx, as.logger, errors.Join(errors.New("failed to rollup bulk multi-resource decision"), err), slog.Int("index", idx))
 		}
 
 		decisionResponse := &authzV2.GetDecisionMultiResourceResponse{