feat(authz): wire up obligations enforcement in auth service (#2756)

### Proposed Changes

* Wires up obligation PDP to JustInTimePDP within Auth Service
* EntitlementPolicyCache fetches all obligations from policy alongside
attributes/subjectmappings/registeredresources
* Interceptors updated so that IPC requests transmit auth state to
downstream called RPCs
* Update tests
* Improved Auth Service errors (sanitizing internal information from
error responses)
* Improved Auth Service logging
* Improved [context metadata
implementation](https://github.com/grpc/grpc-go/blob/master/Documentation/grpc-metadata.md)

### Checklist

- [X] I have added or updated unit tests
- [X] I have added or updated integration tests (if appropriate)
- [ ] I have added or updated documentation

### Testing Instructions

Author: jakedoublev

service/authorization/v2/authorization.go

@@ -12,9 +12,11 @@ import (
 	"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"
+	"github.com/opentdf/platform/protocol/go/policy"
 	otdf "github.com/opentdf/platform/sdk"
 	"github.com/opentdf/platform/service/internal/access/v2"
 	"github.com/opentdf/platform/service/logger"
+	ctxAuth "github.com/opentdf/platform/service/pkg/auth"
 	"github.com/opentdf/platform/service/pkg/cache"
 	"github.com/opentdf/platform/service/pkg/serviceregistry"
 	"go.opentelemetry.io/otel"
@@ -23,6 +25,13 @@ import (
 	"google.golang.org/protobuf/types/known/wrapperspb"
 )
 
+var (
+	ErrFailedToBuildRequestContext = errors.New("failed to contextualize decision request")
+	ErrFailedToInitPDP             = errors.New("failed to create JIT PDP")
+	ErrFailedToGetDecision         = errors.New("failed to get decision")
+	ErrFailedToGetEntitlements     = errors.New("failed to get entitlements")
+)
+
 type Service struct {
 	sdk    *otdf.SDK
 	config *Config
@@ -138,15 +147,12 @@ func (as *Service) GetEntitlements(ctx context.Context, req *connect.Request[aut
 	// 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, 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)
+		return nil, statusifyError(ctx, as.logger, errors.Join(ErrFailedToGetEntitlements, ErrFailedToInitPDP, err))
 	}
 
 	entitlements, err := pdp.GetEntitlements(ctx, entityIdentifier, withComprehensiveHierarchy)
 	if err != nil {
-		// TODO: any bad request errors that aren't 500s?
-		as.logger.ErrorContext(ctx, "failed to get entitlements", slog.Any("error", err))
-		return nil, connect.NewError(connect.CodeInternal, err)
+		return nil, statusifyError(ctx, as.logger, errors.Join(ErrFailedToGetEntitlements, err))
 	}
 	rsp := &authzV2.GetEntitlementsResponse{
 		Entitlements: entitlements,
@@ -168,27 +174,34 @@ func (as *Service) GetDecision(ctx context.Context, req *connect.Request[authzV2
 
 	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)
+		return nil, statusifyError(ctx, as.logger, errors.Join(ErrFailedToInitPDP, err))
 	}
 
 	request := req.Msg
 	entityIdentifier := request.GetEntityIdentifier()
 	action := request.GetAction()
 	resource := request.GetResource()
+	fulfillableObligations := request.GetFulfillableObligationFqns()
 
-	decisions, permitted, err := pdp.GetDecision(ctx, entityIdentifier, action, []*authzV2.Resource{resource})
+	reqContext, err := as.getDecisionRequestContext(ctx)
 	if err != nil {
-		as.logger.ErrorContext(ctx, "failed to get decision", slog.Any("error", err))
-		if errors.Is(err, access.ErrFQNNotFound) || errors.Is(err, access.ErrDefinitionNotFound) {
-			return nil, connect.NewError(connect.CodeNotFound, err)
-		}
-		return nil, connect.NewError(connect.CodeInternal, err)
+		return nil, statusifyError(ctx, as.logger, err)
+	}
+
+	decisions, permitted, err := pdp.GetDecision(
+		ctx,
+		entityIdentifier,
+		action,
+		[]*authzV2.Resource{resource},
+		reqContext,
+		fulfillableObligations,
+	)
+	if err != nil {
+		return nil, statusifyError(ctx, as.logger, err)
 	}
 	resp, err := rollupSingleResourceDecision(permitted, decisions)
 	if err != nil {
-		as.logger.ErrorContext(ctx, "failed to rollup single-resource decision", slog.Any("error", err))
-		return nil, connect.NewError(connect.CodeInternal, err)
+		return nil, statusifyError(ctx, as.logger, err)
 	}
 	return connect.NewResponse(resp), nil
 }
@@ -206,21 +219,34 @@ func (as *Service) GetDecisionMultiResource(ctx context.Context, req *connect.Re
 
 	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))
+		return nil, statusifyError(ctx, as.logger, errors.Join(ErrFailedToInitPDP, err))
 	}
 	request := req.Msg
 	entityIdentifier := request.GetEntityIdentifier()
 	action := request.GetAction()
 	resources := request.GetResources()
+	fulfillableObligations := request.GetFulfillableObligationFqns()
+
+	reqContext, err := as.getDecisionRequestContext(ctx)
+	if err != nil {
+		return nil, statusifyError(ctx, as.logger, err)
+	}
 
-	decisions, allPermitted, err := pdp.GetDecision(ctx, entityIdentifier, action, resources)
+	decisions, allPermitted, err := pdp.GetDecision(
+		ctx,
+		entityIdentifier,
+		action,
+		resources,
+		reqContext,
+		fulfillableObligations,
+	)
 	if err != nil {
-		return nil, statusifyError(ctx, as.logger, errors.Join(errors.New("failed to get decision"), err))
+		return nil, statusifyError(ctx, as.logger, errors.Join(ErrFailedToGetDecision, 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))
+		return nil, statusifyError(ctx, as.logger, err)
 	}
 
 	resp := &authzV2.GetDecisionMultiResourceResponse{
@@ -246,27 +272,33 @@ func (as *Service) GetDecisionBulk(ctx context.Context, req *connect.Request[aut
 
 	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))
+		return nil, statusifyError(ctx, as.logger, errors.Join(ErrFailedToInitPDP, err))
 	}
 
 	multiRequests := req.Msg.GetDecisionRequests()
 	decisionResponses := make([]*authzV2.GetDecisionMultiResourceResponse, len(multiRequests))
 
+	reqContext, err := as.getDecisionRequestContext(ctx)
+	if err != nil {
+		return nil, statusifyError(ctx, as.logger, err)
+	}
+
 	// TODO: revisit performance of this loop after introduction of caching and registered resource values within decisioning,
 	// as the same entity in multiple requests should only be resolved JIT once, not once per request if the same in each.
 	for idx, request := range multiRequests {
 		entityIdentifier := request.GetEntityIdentifier()
 		action := request.GetAction()
 		resources := request.GetResources()
+		fulfillableObligations := request.GetFulfillableObligationFqns()
 
-		decisions, allPermitted, err := pdp.GetDecision(ctx, entityIdentifier, action, resources)
+		decisions, allPermitted, err := pdp.GetDecision(ctx, entityIdentifier, action, resources, reqContext, fulfillableObligations)
 		if err != nil {
-			return nil, statusifyError(ctx, as.logger, errors.Join(errors.New("failed to get bulk decision"), err))
+			return nil, statusifyError(ctx, as.logger, errors.Join(ErrFailedToGetDecision, 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.Int("index", idx))
+			return nil, statusifyError(ctx, as.logger, err, slog.Int("index", idx))
 		}
 
 		decisionResponse := &authzV2.GetDecisionMultiResourceResponse{
@@ -283,3 +315,17 @@ func (as *Service) GetDecisionBulk(ctx context.Context, req *connect.Request[aut
 	}
 	return connect.NewResponse(rsp), nil
 }
+
+// Builds a decision request context out of contextual metadata for the downstream obligation trigger/fulfillment decisioning
+func (as *Service) getDecisionRequestContext(ctx context.Context) (*policy.RequestContext, error) {
+	incoming := true
+	clientID, err := ctxAuth.GetClientIDFromContext(ctx, incoming)
+	if err != nil {
+		return nil, errors.Join(ErrFailedToBuildRequestContext, err)
+	}
+	return &policy.RequestContext{
+		Pep: &policy.PolicyEnforcementPoint{
+			ClientId: clientID,
+		},
+	}, nil
+}