feat(event-ledger): authorize NVCA writes via SIS PSAT introspection

src/control-plane-services/event-ledger/cmd/api/service/v3.go

@@ -297,7 +297,7 @@ func (s *Server) processOTLPEvents(traceCtx context.Context, req *collectorlogsv
 	for _, rl := range req.ResourceLogs {
 		for _, sl := range rl.ScopeLogs {
 			for _, lr := range sl.LogRecords {
-				event, err := extractK8sEvent(lr)
+				event, err := extractK8sEvent(traceCtx, lr)
 				if err != nil {
 					logger.WarnContext(traceCtx, "Skipping event", zap.Error(err))
 					result.FailureCount++
@@ -442,6 +442,25 @@ func eventContextToCanonical(eventContext ContextV3) (string, error) {
 	return strings.Join(parts, ","), nil
 }
 
+// bindNVCAClusterID makes an SIS-verified NVCA cluster identity authoritative
+// over whatever cluster_id a request payload claims: a missing payload value
+// is populated from it, and a mismatching one is rejected outright, so a PSAT
+// valid for one cluster cannot write events attributed to another. Requests
+// with no NVCA identity in context (SIS/Spot JWT callers) are unaffected.
+func bindNVCAClusterID(ctx context.Context, payloadClusterID string) (string, error) {
+	identity, ok := middleware.NVCAIdentityFromContext(ctx)
+	if !ok {
+		return payloadClusterID, nil
+	}
+	if payloadClusterID == "" {
+		return identity.ClusterID, nil
+	}
+	if payloadClusterID != identity.ClusterID {
+		return "", fmt.Errorf("cluster_id %q does not match the authorized cluster", payloadClusterID)
+	}
+	return payloadClusterID, nil
+}
+
 // extractK8sEvent converts an OTLP log record to EventV3
 // Expected OTLP attributes:
 //   - event_name (string): Event type
@@ -450,7 +469,7 @@ func eventContextToCanonical(eventContext ContextV3) (string, error) {
 //   - Context fields (optional): instance_id, deployment_id, gpu_specification_id, cluster_id
 //   - resource_id (optional): generic unique identifier for events that have no
 //     other distinguishing context field (e.g. an ICMSRequest keyed by its request id).
-func extractK8sEvent(lr *logsv1.LogRecord) (*EventV3, error) {
+func extractK8sEvent(ctx context.Context, lr *logsv1.LogRecord) (*EventV3, error) {
 	// Step 1: Convert OTLP protobuf attributes to map
 	attrs := make(map[string]any)
 	for _, attr := range lr.Attributes {
@@ -464,11 +483,15 @@ func extractK8sEvent(lr *logsv1.LogRecord) (*EventV3, error) {
 	}
 
 	// Step 3: Convert wire format to internal context representation
+	clusterID, err := bindNVCAClusterID(ctx, wireFormat.ClusterID)
+	if err != nil {
+		return nil, err
+	}
 	contextV3 := ContextV3{
 		InstanceID:         wireFormat.InstanceID,
 		DeploymentID:       wireFormat.DeploymentID,
 		GPUSpecificationID: wireFormat.GPUSpecificationID,
-		ClusterID:          wireFormat.ClusterID,
+		ClusterID:          clusterID,
 		ResourceID:         wireFormat.ResourceID,
 	}
 
@@ -522,7 +545,7 @@ func extractK8sEvent(lr *logsv1.LogRecord) (*EventV3, error) {
 //   - namespace (required)
 //   - Context fields (optional, camelCase): instanceId, deploymentId, gpuSpecificationId, clusterId
 //     Note: CloudEvents spec forbids underscores in extension names, so we use camelCase
-func extractCloudEvent(ce *cloudevents.Event) (*EventV3, error) {
+func extractCloudEvent(ctx context.Context, ce *cloudevents.Event) (*EventV3, error) {
 	// Validate required CloudEvents fields per spec (using CloudEvents field names in errors)
 	if strings.TrimSpace(ce.ID()) == "" {
 		return nil, errors.New("missing required field: id")
@@ -541,11 +564,15 @@ func extractCloudEvent(ce *cloudevents.Event) (*EventV3, error) {
 	}
 
 	// Convert wire format to internal context representation
+	clusterID, err := bindNVCAClusterID(ctx, wireFormat.ClusterID)
+	if err != nil {
+		return nil, err
+	}
 	contextV3 := ContextV3{
 		InstanceID:         wireFormat.InstanceID,
 		DeploymentID:       wireFormat.DeploymentID,
 		GPUSpecificationID: wireFormat.GPUSpecificationID,
-		ClusterID:          wireFormat.ClusterID,
+		ClusterID:          clusterID,
 		ResourceID:         wireFormat.ResourceID,
 	}
 
@@ -595,7 +622,7 @@ func (s *Server) processCloudEvents(traceCtx context.Context, cloudEvents []*clo
 			continue
 		}
 
-		event, err := extractCloudEvent(cloudEvent)
+		event, err := extractCloudEvent(traceCtx, cloudEvent)
 		if err != nil {
 			logger.WarnContext(traceCtx, "Skipping event", zap.Error(err))
 			result.FailureCount++

src/control-plane-services/event-ledger/cmd/api/service/v3_test.go

@@ -40,6 +40,8 @@ import (
 	commonv1 "go.opentelemetry.io/proto/otlp/common/v1"
 	logsv1 "go.opentelemetry.io/proto/otlp/logs/v1"
 
+	"github.com/NVIDIA/nvcf/src/control-plane-services/event-ledger/internal/middleware"
+
 	"github.com/NVIDIA/nvcf/src/control-plane-services/event-ledger/internal/observability/logging"
 
 	"github.com/NVIDIA/nvcf/src/control-plane-services/event-ledger/common/core/types"
@@ -611,7 +613,7 @@ func TestExtractK8sEvent(t *testing.T) {
 		"extra_field": "extra_value",
 	})
 
-	event, err := extractK8sEvent(lr)
+	event, err := extractK8sEvent(context.Background(), lr)
 	require.NoError(t, err)
 
 	// Check struct fields
@@ -680,13 +682,57 @@ func TestExtractK8sEvent_ResourceID(t *testing.T) {
 		"resource_id": "icms-abc",
 	})
 
-	event, err := extractK8sEvent(lr)
+	event, err := extractK8sEvent(context.Background(), lr)
 	require.NoError(t, err)
 
 	// resource_id participates in the context (sorted last), keeping the row unique.
 	assert.Equal(t, "cluster_id=clus-1,resource_id=icms-abc", event.Context)
 }
 
+// TestExtractK8sEvent_NVCAClusterBinding verifies that an SIS-verified NVCA
+// cluster identity is authoritative over the payload: a matching cluster_id
+// is accepted, a missing one is populated, and a mismatched one is rejected
+// so a PSAT valid for one cluster cannot write events for another.
+func TestExtractK8sEvent_NVCAClusterBinding(t *testing.T) {
+	nvcaCtx := middleware.WithNVCAIdentity(context.Background(), middleware.NVCAIdentity{
+		Subject:   "system:serviceaccount:customer-ns:nvca",
+		ClusterID: "cluster-a",
+	})
+
+	t.Run("matching payload cluster_id is accepted", func(t *testing.T) {
+		lr := createOTLPLogRecord("pod.ready", "tenant-123", "nvca", "pod-1", map[string]string{
+			"cluster_id": "cluster-a",
+		})
+		event, err := extractK8sEvent(nvcaCtx, lr)
+		require.NoError(t, err)
+		assert.Contains(t, event.Context, "cluster_id=cluster-a")
+	})
+
+	t.Run("missing payload cluster_id is populated from the verified identity", func(t *testing.T) {
+		lr := createOTLPLogRecord("pod.ready", "tenant-123", "nvca", "pod-1", nil)
+		event, err := extractK8sEvent(nvcaCtx, lr)
+		require.NoError(t, err)
+		assert.Contains(t, event.Context, "cluster_id=cluster-a")
+	})
+
+	t.Run("mismatched payload cluster_id is rejected", func(t *testing.T) {
+		lr := createOTLPLogRecord("pod.ready", "tenant-123", "nvca", "pod-1", map[string]string{
+			"cluster_id": "cluster-b",
+		})
+		_, err := extractK8sEvent(nvcaCtx, lr)
+		assert.Error(t, err)
+	})
+
+	t.Run("no NVCA identity leaves the payload cluster_id untouched", func(t *testing.T) {
+		lr := createOTLPLogRecord("pod.ready", "tenant-123", "sis", "pod-1", map[string]string{
+			"cluster_id": "cluster-a",
+		})
+		event, err := extractK8sEvent(context.Background(), lr)
+		require.NoError(t, err)
+		assert.Contains(t, event.Context, "cluster_id=cluster-a")
+	})
+}
+
 // TestExtractK8sEvent_DistinctResourceIDsDoNotCollide verifies two resources with
 // the same non-resource context but different resource_id produce distinct contexts.
 func TestExtractK8sEvent_DistinctResourceIDsDoNotCollide(t *testing.T) {
@@ -695,7 +741,7 @@ func TestExtractK8sEvent_DistinctResourceIDsDoNotCollide(t *testing.T) {
 			"cluster_id":  "clus-1",
 			"resource_id": resourceID,
 		})
-		event, err := extractK8sEvent(lr)
+		event, err := extractK8sEvent(context.Background(), lr)
 		require.NoError(t, err)
 		return event.Context
 	}
@@ -712,7 +758,7 @@ func TestExtractK8sEvent_PodKeepsUnmappedAttrsInDetails(t *testing.T) {
 		"icms_request_id": "icms-xyz",
 	})
 
-	event, err := extractK8sEvent(lr)
+	event, err := extractK8sEvent(context.Background(), lr)
 	require.NoError(t, err)
 
 	// Pod context stays the original shape and excludes the unmapped attribute.
@@ -733,7 +779,7 @@ func TestExtractCloudEvent_SourceRequired(t *testing.T) {
 	ce.SetSource("") // Empty source
 	ce.SetExtension("namespace", "test-namespace")
 
-	_, err := extractCloudEvent(&ce)
+	_, err := extractCloudEvent(context.Background(), &ce)
 	require.Error(t, err)
 	assert.Contains(t, err.Error(), "missing required field: source")
 }
@@ -746,7 +792,7 @@ func TestExtractCloudEvent_TypeRequired(t *testing.T) {
 	ce.SetSource("/test")
 	ce.SetExtension("namespace", "test-namespace")
 
-	_, err := extractCloudEvent(&ce)
+	_, err := extractCloudEvent(context.Background(), &ce)
 	require.Error(t, err)
 	assert.Contains(t, err.Error(), "missing required field: type")
 }
@@ -759,7 +805,7 @@ func TestExtractCloudEvent_IdRequired(t *testing.T) {
 	ce.SetSource("/test")
 	ce.SetExtension("namespace", "test-namespace")
 
-	_, err := extractCloudEvent(&ce)
+	_, err := extractCloudEvent(context.Background(), &ce)
 	require.Error(t, err)
 	assert.Contains(t, err.Error(), "missing required field: id")
 }
@@ -775,11 +821,61 @@ func TestExtractCloudEvent_ResourceID(t *testing.T) {
 	ce.SetExtension("clusterId", "clus-1")
 	ce.SetExtension("resourceId", "icms-1")
 
-	event, err := extractCloudEvent(&ce)
+	event, err := extractCloudEvent(context.Background(), &ce)
 	require.NoError(t, err)
 	assert.Equal(t, "cluster_id=clus-1,resource_id=icms-1", event.Context)
 }
 
+// TestExtractCloudEvent_NVCAClusterBinding mirrors
+// TestExtractK8sEvent_NVCAClusterBinding: bindNVCAClusterID is wired into
+// both extractK8sEvent and extractCloudEvent, so both need the same
+// match/populate/reject/no-identity coverage.
+func TestExtractCloudEvent_NVCAClusterBinding(t *testing.T) {
+	nvcaCtx := middleware.WithNVCAIdentity(context.Background(), middleware.NVCAIdentity{
+		Subject:   "system:serviceaccount:customer-ns:nvca",
+		ClusterID: "cluster-a",
+	})
+
+	newEvent := func(clusterID string) cloudevents.Event {
+		ce := cloudevents.NewEvent()
+		ce.SetID("test-id")
+		ce.SetType("test.event")
+		ce.SetSource("/test")
+		ce.SetExtension("namespace", "tenant-123")
+		if clusterID != "" {
+			ce.SetExtension("clusterId", clusterID)
+		}
+		return ce
+	}
+
+	t.Run("matching payload cluster_id is accepted", func(t *testing.T) {
+		ce := newEvent("cluster-a")
+		event, err := extractCloudEvent(nvcaCtx, &ce)
+		require.NoError(t, err)
+		assert.Contains(t, event.Context, "cluster_id=cluster-a")
+	})
+
+	t.Run("missing payload cluster_id is populated from the verified identity", func(t *testing.T) {
+		ce := newEvent("")
+		event, err := extractCloudEvent(nvcaCtx, &ce)
+		require.NoError(t, err)
+		assert.Contains(t, event.Context, "cluster_id=cluster-a")
+	})
+
+	t.Run("mismatched payload cluster_id is rejected", func(t *testing.T) {
+		ce := newEvent("cluster-b")
+		_, err := extractCloudEvent(nvcaCtx, &ce)
+		assert.Error(t, err)
+	})
+
+	t.Run("no NVCA identity leaves the payload cluster_id untouched", func(t *testing.T) {
+		ce := newEvent("cluster-a")
+		event, err := extractCloudEvent(context.Background(), &ce)
+		require.NoError(t, err)
+		assert.Contains(t, event.Context, "cluster_id=cluster-a")
+	})
+}
+
 // ======================
 // CloudEvents Endpoint Validation Tests
 // ======================

src/control-plane-services/event-ledger/cmd/api/startup/BUILD.bazel

@@ -16,6 +16,7 @@ go_library(
         "//src/control-plane-services/event-ledger/internal/data_access",
         "//src/control-plane-services/event-ledger/internal/interfaces",
         "//src/control-plane-services/event-ledger/internal/middleware",
+        "//src/control-plane-services/event-ledger/internal/nvca",
         "//src/control-plane-services/event-ledger/internal/observability/logging",
         "//src/control-plane-services/event-ledger/internal/observability/tracing",
         "//src/control-plane-services/event-ledger/internal/policy",

src/control-plane-services/event-ledger/cmd/api/startup/run_service.go

@@ -46,6 +46,7 @@ import (
 	"github.com/NVIDIA/nvcf/src/control-plane-services/event-ledger/internal/data_access"
 	"github.com/NVIDIA/nvcf/src/control-plane-services/event-ledger/internal/interfaces"
 	"github.com/NVIDIA/nvcf/src/control-plane-services/event-ledger/internal/middleware"
+	"github.com/NVIDIA/nvcf/src/control-plane-services/event-ledger/internal/nvca"
 	"github.com/NVIDIA/nvcf/src/control-plane-services/event-ledger/internal/observability/logging"
 	"github.com/NVIDIA/nvcf/src/control-plane-services/event-ledger/internal/observability/tracing"
 	"github.com/NVIDIA/nvcf/src/control-plane-services/event-ledger/internal/policy"
@@ -297,6 +298,23 @@ func runService(cfg config.Config) error {
 				jwtOpts = &opts
 			}
 
+			var introspector nvca.Introspector
+			if cfg.Auth.Introspection.Enabled {
+				introspectionCfg := cfg.Auth.Introspection.WithDefaults()
+				logger.Warn("nvca psat introspection enabled", zap.String("url", introspectionCfg.URL))
+
+				introspectionClient, err := nvca.NewClient(
+					introspectionCfg.URL,
+					time.Duration(introspectionCfg.TimeoutSeconds)*time.Second,
+					time.Duration(introspectionCfg.CacheTTLSeconds)*time.Second,
+				)
+				if err != nil {
+					logger.Error("failed to create nvca introspection client", zap.Error(err))
+					return fmt.Errorf("failed to create nvca introspection client: %w", err)
+				}
+				introspector = introspectionClient
+			}
+
 			requireLocalScopeCheck = cfg.SelfManaged
 
 			authRouter.Use(middleware.NewAuthMiddleware(
@@ -305,6 +323,7 @@ func runService(cfg config.Config) error {
 				jwtOpts,
 				jwkCache,
 				cfg.SelfManaged,
+				introspector,
 				logger,
 			))
 		default: