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

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

@@ -282,10 +282,11 @@ func (s *Server) PostK8sEventV3(w http.ResponseWriter, r *http.Request) {
 
 // EventProcessingResult holds the results of processing a batch of events
 type EventProcessingResult struct {
-	SuccessCount    int
-	FailureCount    int
-	ProcessedEvents []ProcessedEventSummary
-	LastError       error
+	SuccessCount     int
+	FailureCount     int
+	ProcessedEvents  []ProcessedEventSummary
+	LastError        error
+	AuthorizationErr error
 }
 
 // processOTLPEvents extracts and stores K8s events from OTLP log records
@@ -297,8 +298,13 @@ 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 {
+					if errors.Is(err, errClusterAuthorization) {
+						logger.WarnContext(traceCtx, "Aborting batch", zap.Error(err))
+						result.AuthorizationErr = err
+						return result
+					}
 					logger.WarnContext(traceCtx, "Skipping event", zap.Error(err))
 					result.FailureCount++
 					result.LastError = err
@@ -442,6 +448,30 @@ func eventContextToCanonical(eventContext ContextV3) (string, error) {
 	return strings.Join(parts, ","), nil
 }
 
+// errClusterAuthorization is checked with errors.Is by the batch processors
+// to abort on a bindNVCAClusterID rejection, instead of skipping it like an
+// ordinary per-event validation error.
+var errClusterAuthorization = errors.New("cluster_id does not match the authorized cluster")
+
+// 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("%w: %q", errClusterAuthorization, payloadClusterID)
+	}
+	return payloadClusterID, nil
+}
+
 // extractK8sEvent converts an OTLP log record to EventV3
 // Expected OTLP attributes:
 //   - event_name (string): Event type
@@ -450,7 +480,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 +494,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 +556,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 +575,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,8 +633,13 @@ func (s *Server) processCloudEvents(traceCtx context.Context, cloudEvents []*clo
 			continue
 		}
 
-		event, err := extractCloudEvent(cloudEvent)
+		event, err := extractCloudEvent(traceCtx, cloudEvent)
 		if err != nil {
+			if errors.Is(err, errClusterAuthorization) {
+				logger.WarnContext(traceCtx, "Aborting batch", zap.Error(err))
+				result.AuthorizationErr = err
+				return result
+			}
 			logger.WarnContext(traceCtx, "Skipping event", zap.Error(err))
 			result.FailureCount++
 			result.LastError = err
@@ -722,6 +765,12 @@ func (s *Server) storeK8sEvent(traceCtx context.Context, event *EventV3) error {
 func (s *Server) sendEventResponse(w http.ResponseWriter, traceCtx context.Context, result EventProcessingResult) {
 	logger := logging.GetLogger(traceCtx)
 
+	if result.AuthorizationErr != nil {
+		logger.WarnContext(traceCtx, "Rejecting batch", zap.Error(result.AuthorizationErr))
+		sendProblemDetail(w, http.StatusForbidden, "Forbidden", result.AuthorizationErr.Error())
+		return
+	}
+
 	// Prepare response based on results
 	if result.SuccessCount == 0 && result.FailureCount == 0 {
 		logger.ErrorContext(traceCtx, "No events found in OTLP request")