feat(core): add global audit config and add configured JWT claims to audit logs

opentdf-core-mode.yaml

@@ -11,6 +11,14 @@ logger:
   level: debug
   type: text
   output: stderr
+audit:
+  # Optional JWT claim mappings for audit enrichment.
+  # Paths use dot notation over the emitted audit log shape.
+  # jwt_claim_mappings:
+  #   - claim: sub
+  #     path: eventMetaData.requester.sub
+  #   - claim: realm_access.roles
+  #     path: eventMetaData.requester.roles
 # DB and Server configurations are defaulted for local development
 # db:
 #   host: localhost
@@ -57,4 +65,4 @@ server:
     maxage: 3600
   grpc:
     reflectionEnabled: true # Default is false
-  port: 8383
+  port: 8383

opentdf-dev.yaml

@@ -2,6 +2,14 @@ logger:
   level: debug
   type: text
   output: stderr
+audit:
+  # Optional JWT claim mappings for audit enrichment.
+  # Paths use dot notation over the emitted audit log shape.
+  # jwt_claim_mappings:
+  #   - claim: sub
+  #     path: eventMetaData.requester.sub
+  #   - claim: realm_access.roles
+  #     path: eventMetaData.requester.roles
 # DB and Server configurations are defaulted for local development
 # db:
 #   host: localhost

opentdf-ers-mode.yaml

@@ -5,6 +5,14 @@ logger:
   level: debug
   type: text
   output: stderr
+audit:
+  # Optional JWT claim mappings for audit enrichment.
+  # Paths use dot notation over the emitted audit log shape.
+  # jwt_claim_mappings:
+  #   - claim: sub
+  #     path: eventMetaData.requester.sub
+  #   - claim: realm_access.roles
+  #     path: eventMetaData.requester.roles
 services:
   entityresolution:
     log_level: info

opentdf-ers-test.yaml

@@ -12,6 +12,14 @@ mode: standalone
 logger:
   level: info
   type: json
+audit:
+  # Optional JWT claim mappings for audit enrichment.
+  # Paths use dot notation over the emitted audit log shape.
+  # jwt_claim_mappings:
+  #   - claim: sub
+  #     path: eventMetaData.requester.sub
+  #   - claim: realm_access.roles
+  #     path: eventMetaData.requester.roles
 
 crypto:
   type: standard
@@ -256,4 +264,4 @@ development:
   # Allow insecure connections for testing
   allow_insecure: true
   # Disable some validations for easier testing
-  strict_validation: false
+  strict_validation: false

opentdf-example.yaml

@@ -2,6 +2,14 @@ logger:
   level: debug
   type: text
   output: stdout
+audit:
+  # Optional JWT claim mappings for audit enrichment.
+  # Paths use dot notation over the emitted audit log shape.
+  # jwt_claim_mappings:
+  #   - claim: sub
+  #     path: eventMetaData.requester.sub
+  #   - claim: realm_access.roles
+  #     path: eventMetaData.requester.roles
 # DB and Server configurations are defaulted for local development
 db:
   host: opentdfdb

opentdf-kas-mode.yaml

@@ -11,6 +11,14 @@ logger:
   level: debug
   type: text
   output: stderr
+audit:
+  # Optional JWT claim mappings for audit enrichment.
+  # Paths use dot notation over the emitted audit log shape.
+  # jwt_claim_mappings:
+  #   - claim: sub
+  #     path: eventMetaData.requester.sub
+  #   - claim: realm_access.roles
+  #     path: eventMetaData.requester.roles
 security:
   unsafe:
     # Increase only when diagnosing clock drift issues; default is 1m

service/internal/auth/authn.go

@@ -438,10 +438,52 @@ func IPCMetadataClientInterceptor(log *logger.Logger) connect.UnaryInterceptorFu
 	})
 }
 
+// rehydrateIPCAuthContext reconstructs pkg/auth context from propagated IPC metadata.
+// It only transports an already-authenticated token across an internal hop; it does not
+// validate the token again. Network-facing requests must continue to rely on the
+// ConnectUnaryServerInterceptor auth middleware for validation.
+func rehydrateIPCAuthContext(ctx context.Context, l *logger.Logger) (context.Context, error) {
+	if ctxAuth.GetAccessTokenFromContext(ctx, l) != nil && ctxAuth.GetRawAccessTokenFromContext(ctx, l) != "" {
+		return ctx, nil
+	}
+
+	md, ok := metadata.FromIncomingContext(ctx)
+	if !ok {
+		return ctx, nil
+	}
+
+	rawToken := rawAccessTokenFromIncomingMetadata(md)
+	if rawToken == "" {
+		return ctx, nil
+	}
+
+	parsed, err := jwt.Parse([]byte(rawToken), jwt.WithVerify(false), jwt.WithValidate(false))
+	if err != nil {
+		if l != nil {
+			l.ErrorContext(ctx, "failed to rehydrate IPC access token from metadata", slog.Any("error", err))
+		}
+		return ctx, fmt.Errorf("rehydrate IPC access token from metadata: %w", err)
+	}
+
+	return ctxAuth.ContextWithAuthNInfo(ctx, ctxAuth.GetJWKFromContext(ctx, l), parsed, rawToken), nil
+}
+
+func rawAccessTokenFromIncomingMetadata(md metadata.MD) string {
+	if accessTokens := md.Get(ctxAuth.AccessTokenKey); len(accessTokens) > 0 && accessTokens[0] != "" {
+		return accessTokens[0]
+	}
+
+	authHeaders := md.Get("authorization")
+	if len(authHeaders) == 0 || authHeaders[0] == "" {
+		return ""
+	}
+	return strings.TrimPrefix(strings.TrimPrefix(authHeaders[0], "Bearer "), "DPoP ")
+}
+
 // IPCUnaryServerInterceptor is a grpc interceptor that:
-// 1. verifies the token in the metadata
-// 2. reauthorizes the token if the route is in the list
-// 3. translates known IPC Connect request headers back to context metadata for downstream consumers
+// 1. translates known IPC Connect request headers back to incoming metadata
+// 2. reauthorizes routes that are configured for IPC reauth
+// 3. rehydrates auth context from propagated incoming metadata without revalidating it
 func (a Authentication) IPCUnaryServerInterceptor() connect.UnaryInterceptorFunc {
 	interceptor := func(next connect.UnaryFunc) connect.UnaryFunc {
 		return connect.UnaryFunc(func(
@@ -467,6 +509,10 @@ func (a Authentication) IPCUnaryServerInterceptor() connect.UnaryInterceptorFunc
 			if err != nil {
 				return nil, err
 			}
+			nextCtx, err = rehydrateIPCAuthContext(nextCtx, a.logger)
+			if err != nil {
+				return nil, connect.NewError(connect.CodeUnauthenticated, errors.New("invalid IPC authentication context"))
+			}
 			return next(nextCtx, req)
 		})
 	}

service/internal/auth/authn_ipc_metadata_interceptor_test.go

@@ -2,6 +2,7 @@ package auth
 
 import (
 	"context"
+	"errors"
 	"testing"
 
 	"connectrpc.com/connect"
@@ -184,6 +185,10 @@ func (m *mockAnyRequest) Any() any {
 
 func TestIPCUnaryServerInterceptor(t *testing.T) {
 	testLogger := logger.CreateTestLogger()
+	validJWT, err := jwt.NewBuilder().Subject("ipc-user").Build()
+	require.NoError(t, err)
+	validRawToken, err := jwt.Sign(validJWT, jwt.WithInsecureNoSignature())
+	require.NoError(t, err)
 
 	// Create a minimal authentication instance
 	auth := &Authentication{
@@ -201,7 +206,7 @@ func TestIPCUnaryServerInterceptor(t *testing.T) {
 			setupRequest: func() connect.AnyRequest {
 				req := connect.NewRequest(&kas.PublicKeyRequest{})
 				req.Header().Set(canonicalIPCHeaderClientID, "test-client-from-header")
-				req.Header().Set(canonicalIPCHeaderAccessToken, "test-token-from-header")
+				req.Header().Set(canonicalIPCHeaderAccessToken, string(validRawToken))
 				return &mockAnyRequest{
 					Request:  req,
 					isClient: false,
@@ -225,7 +230,7 @@ func TestIPCUnaryServerInterceptor(t *testing.T) {
 			setupRequest: func() connect.AnyRequest {
 				req := connect.NewRequest(&kas.PublicKeyRequest{})
 				req.Header().Set(canonicalIPCHeaderClientID, "merged-client-id")
-				req.Header().Set(canonicalIPCHeaderAccessToken, "merged-token")
+				req.Header().Set(canonicalIPCHeaderAccessToken, string(validRawToken))
 				return &mockAnyRequest{
 					Request:  req,
 					isClient: false,
@@ -251,8 +256,13 @@ func TestIPCUnaryServerInterceptor(t *testing.T) {
 					for _, key := range tt.expectedIncomingMDKeys {
 						assert.NotEmpty(t, md.Get(key), "metadata key %s should exist", key)
 					}
+					retrievedJWT := ctxAuth.GetAccessTokenFromContext(postInterceptorCtx, testLogger)
+					require.NotNil(t, retrievedJWT)
+					assert.Equal(t, "ipc-user", retrievedJWT.Subject())
+					assert.Equal(t, string(validRawToken), ctxAuth.GetRawAccessTokenFromContext(postInterceptorCtx, testLogger))
 				} else {
 					assert.Zero(t, md.Len())
+					assert.Nil(t, ctxAuth.GetAccessTokenFromContext(postInterceptorCtx, testLogger))
 				}
 				return connect.NewResponse(&kas.PublicKeyResponse{}), nil
 			}
@@ -267,19 +277,22 @@ func TestIPCUnaryServerInterceptor(t *testing.T) {
 
 func TestIPCUnaryServerInterceptor_Integration(t *testing.T) {
 	testLogger := logger.CreateTestLogger()
+	mockJWT, err := jwt.NewBuilder().Subject("integration-user").Build()
+	require.NoError(t, err)
+	rawToken, err := jwt.Sign(mockJWT, jwt.WithInsecureNoSignature())
+	require.NoError(t, err)
 
 	auth := &Authentication{
 		logger:          testLogger,
 		ipcReauthRoutes: []string{},
 	}
 
-	t.Run("clientID and access token from headers available in context metadata", func(t *testing.T) {
+	t.Run("clientID and access token from headers available in context metadata and auth context", func(t *testing.T) {
 		clientID := "integration-client-id"
-		accessToken := "integration-access-token"
 
 		req := connect.NewRequest(&kas.PublicKeyRequest{})
 		req.Header().Set(canonicalIPCHeaderClientID, clientID)
-		req.Header().Set(canonicalIPCHeaderAccessToken, accessToken)
+		req.Header().Set(canonicalIPCHeaderAccessToken, string(rawToken))
 
 		wrappedReq := &mockAnyRequest{
 			Request:  req,
@@ -290,7 +303,7 @@ func TestIPCUnaryServerInterceptor_Integration(t *testing.T) {
 
 		ctx := t.Context()
 
-		var receivedClientID, receivedAccessToken string
+		var receivedClientID, receivedAccessToken, receivedSubject string
 		mockNext := func(ctx context.Context, _ connect.AnyRequest) (connect.AnyResponse, error) {
 			md, ok := metadata.FromIncomingContext(ctx)
 			require.True(t, ok)
@@ -302,6 +315,9 @@ func TestIPCUnaryServerInterceptor_Integration(t *testing.T) {
 			if len(accessTokens) > 0 {
 				receivedAccessToken = accessTokens[0]
 			}
+			retrievedJWT := ctxAuth.GetAccessTokenFromContext(ctx, testLogger)
+			require.NotNil(t, retrievedJWT)
+			receivedSubject = retrievedJWT.Subject()
 			return connect.NewResponse(&kas.PublicKeyResponse{}), nil
 		}
 
@@ -310,6 +326,52 @@ func TestIPCUnaryServerInterceptor_Integration(t *testing.T) {
 
 		require.NoError(t, err)
 		assert.Equal(t, clientID, receivedClientID)
-		assert.Equal(t, accessToken, receivedAccessToken)
+		assert.Equal(t, string(rawToken), receivedAccessToken)
+		assert.Equal(t, "integration-user", receivedSubject)
+	})
+
+	t.Run("invalid access token header fails rehydration", func(t *testing.T) {
+		req := connect.NewRequest(&kas.PublicKeyRequest{})
+		req.Header().Set(canonicalIPCHeaderAccessToken, "not-a-jwt")
+
+		wrappedReq := &mockAnyRequest{
+			Request:  req,
+			isClient: false,
+		}
+
+		interceptor := auth.IPCUnaryServerInterceptor()
+		interceptorFunc := interceptor(func(context.Context, connect.AnyRequest) (connect.AnyResponse, error) {
+			t.Fatal("next handler should not be called on invalid token")
+			return nil, errors.New("unreachable")
+		})
+
+		_, err := interceptorFunc(t.Context(), wrappedReq)
+		require.Error(t, err)
+
+		var connectErr *connect.Error
+		require.ErrorAs(t, err, &connectErr)
+		assert.Equal(t, connect.CodeUnauthenticated, connectErr.Code())
 	})
 }
+
+func TestRehydrateIPCAuthContextPreservesExistingJWK(t *testing.T) {
+	testLogger := logger.CreateTestLogger()
+	mockJWT, err := jwt.NewBuilder().Subject("rehydrated-user").Build()
+	require.NoError(t, err)
+	rawToken, err := jwt.Sign(mockJWT, jwt.WithInsecureNoSignature())
+	require.NoError(t, err)
+	mockJWK, err := jwk.FromRaw([]byte("existing-jwk"))
+	require.NoError(t, err)
+
+	ctx := metadata.NewIncomingContext(t.Context(), metadata.Pairs(ctxAuth.AccessTokenKey, string(rawToken)))
+	ctx = ctxAuth.ContextWithAuthNInfo(ctx, mockJWK, nil, "")
+
+	rehydratedCtx, err := rehydrateIPCAuthContext(ctx, testLogger)
+	require.NoError(t, err)
+	require.Same(t, mockJWK, ctxAuth.GetJWKFromContext(rehydratedCtx, testLogger))
+
+	retrievedJWT := ctxAuth.GetAccessTokenFromContext(rehydratedCtx, testLogger)
+	require.NotNil(t, retrievedJWT)
+	assert.Equal(t, "rehydrated-user", retrievedJWT.Subject())
+	assert.Equal(t, string(rawToken), ctxAuth.GetRawAccessTokenFromContext(rehydratedCtx, testLogger))
+}