GODRIVER-OPENMIRAI-KEK: native-v2 facade and readiness support

openmirai/mongo/change_stream.go

@@ -503,7 +503,7 @@ func (stream *ChangeStream) validateEventOwnership(operationType string, fullDoc
 	}
 	if len(fullDocument) > 0 {
 		value, ok := lookupChangePath(fullDocument, stream.policy.TenantPath)
-		if !ok || value.Type != bson.TypeString || value.StringValue() != stream.scope.TenantID {
+		if !ok || !tenantScopeMatches(value, stream.policy, stream.scope.TenantID) {
 			return newError(ErrorOperationScopeInvalid, ErrorClassAuthorization, nil)
 		}
 		return nil
@@ -512,7 +512,7 @@ func (stream *ChangeStream) validateEventOwnership(operationType string, fullDoc
 		return nil
 	}
 	value, ok := lookupChangePath(documentKey, stream.policy.TenantPath)
-	if !ok || value.Type != bson.TypeString || value.StringValue() != stream.scope.TenantID {
+	if !ok || !tenantScopeMatches(value, stream.policy, stream.scope.TenantID) {
 		return newError(ErrorOperationScopeInvalid, ErrorClassAuthorization, nil)
 	}
 	return nil

openmirai/mongo/collection.go

@@ -885,7 +885,7 @@ func prependAggregateScopeMatch(value bson.A, policy PolicyRow, scope OperationS
 		if strings.TrimSpace(policy.TenantPath) == "" || scope.TenantID == "" {
 			return nil, newError(ErrorOperationScopeInvalid, ErrorClassAuthorization, nil)
 		}
-		owned = append(owned, bson.E{Key: policy.TenantPath, Value: scope.TenantID})
+		owned = append(owned, bson.E{Key: policy.TenantPath, Value: tenantScopeValue(policy, scope.TenantID)})
 	}
 	owned = append(owned, bson.E{Key: "_pii.state", Value: "active"})
 	result := make(bson.A, 0, len(value)+1)

openmirai/mongo/enroll.go

@@ -10,6 +10,8 @@ package mongo
 
 import (
 	"context"
+	"crypto/sha256"
+	"encoding/hex"
 	"errors"
 	"time"
 
@@ -157,21 +159,22 @@ func EnrollActiveRecords(ctx context.Context, config RuntimeConfig, scope Operat
 				continue
 			}
 			index := field.ExactBlindIndex.Value
+			scope := scopeBytes(scope)
 			bikQuery := keys.ActiveQuery{
 				RecordType: keys.RecordTypeBIK, Environment: string(snapshot.Environment), Service: snapshot.ServiceID,
 				LogicalKeyDatabase: snapshot.KeyDatabase, Region: snapshot.LogicalRegion,
-				LogicalResidency: string(snapshot.Residency), TenantScope: []byte(scope.TenantID),
+				LogicalResidency: string(snapshot.Residency), TenantScope: scope,
 				Database: snapshot.ApplicationDatabase, Collection: collectionName, Purpose: index.Purpose,
 			}
-			recordID := "poc-bik-" + index.IndexID
+			recordID := enrollmentBIKRecordID(collectionName, index.IndexID, scope)
 			bikID, bikSkipped, bikErr := enrollActive(ctx, manager, bikQuery, func() (keys.KeyRecord, error) {
 				input := enrollmentRecordBase(snapshot, recordID, keys.RecordTypeBIK, keys.ChildBIK, index.Purpose)
 				input.ParentRef, input.KEKRef = kekID, kekID
 				input.ParentType, input.KEKGeneration = keys.ChildKEK, 1
 				input.ParentGeneration, input.ChildGeneration, input.BIKGeneration = 1, 1, 1
 				input.Database = snapshot.ApplicationDatabase
 				input.Collection = collectionName
-				input.TenantScope = []byte(scope.TenantID)
+				input.TenantScope = scope
 				input.DataClass = string(field.DataClass)
 				input.QueryPurpose = index.Purpose
 				input.NormalizationVersion = index.Version
@@ -210,6 +213,11 @@ func EnrollActiveRecords(ctx context.Context, config RuntimeConfig, scope Operat
 	return report, nil
 }
 
+func enrollmentBIKRecordID(collectionName, indexID string, scope []byte) string {
+	scopeDigest := sha256.Sum256(scope)
+	return "poc-bik-" + sanitizeIndexName(collectionName) + "-" + sanitizeIndexName(indexID) + "-" + hex.EncodeToString(scopeDigest[:])
+}
+
 func sanitizeIndexName(path string) string {
 	result := make([]byte, 0, len(path))
 	for _, char := range path {

openmirai/mongo/enroll_test.go

@@ -0,0 +1,30 @@
+// Copyright 2026 OpenMirai
+//
+// Licensed under the Apache License, Version 2.0 (the "License"); you may
+// not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+
+package mongo
+
+import (
+	"strings"
+	"testing"
+)
+
+func TestEnrollmentBIKRecordIDIsScopedAndCollectionSpecific(t *testing.T) {
+	first := enrollmentBIKRecordID("users", "EMAIL", []byte("tenant-a"))
+	secondCollection := enrollmentBIKRecordID("learners", "EMAIL", []byte("tenant-a"))
+	secondScope := enrollmentBIKRecordID("users", "EMAIL", []byte("tenant-b"))
+
+	if first == secondCollection || first == secondScope {
+		t.Fatalf("BIK record IDs must distinguish collection and scope: %q", first)
+	}
+	if strings.Contains(first, "tenant-a") {
+		t.Fatalf("BIK record ID must not expose tenant scope: %q", first)
+	}
+	if !strings.HasPrefix(first, "poc-bik-users-EMAIL-") {
+		t.Fatalf("unexpected BIK record ID format: %q", first)
+	}
+}

openmirai/mongo/g020_lane_b_test.go

@@ -15,6 +15,8 @@ import (
 	"go.mongodb.org/mongo-driver/v2/bson"
 	drivermongo "go.mongodb.org/mongo-driver/v2/mongo"
 	"go.mongodb.org/mongo-driver/v2/mongo/options"
+	"go.mongodb.org/mongo-driver/v2/openmirai/mongo/internal/compiler"
+	"go.mongodb.org/mongo-driver/v2/openmirai/mongo/internal/compiler/query"
 	"go.mongodb.org/mongo-driver/v2/openmirai/mongo/internal/compiler/update"
 	"go.mongodb.org/mongo-driver/v2/openmirai/mongo/internal/limits"
 )
@@ -56,6 +58,45 @@ func TestG020LaneBAggregatePrependsScopeBeforeEveryCallerPipelineShape(t *testin
 	}
 }
 
+func TestG020LaneBAggregateUnwindCompilerBoundary(t *testing.T) {
+	driver := newG020Driver()
+	client := newG020Client(&g020CompilerPipelineRuntime{}, driver)
+	client.policies["users"] = g020AggregatePolicy()
+	collection, err := client.ApplicationDatabase().Collection("users")
+	if err != nil {
+		t.Fatal(err)
+	}
+
+	pipeline := bson.A{bson.D{{Key: "$unwind", Value: bson.D{
+		{Key: "path", Value: "$tags"},
+		{Key: "includeArrayIndex", Value: "tagIndex"},
+	}}}}
+	cursor, err := collection.Aggregate(context.Background(), g020Scope(), pipeline, ResultPolicy{ID: "G020_UNWIND"})
+	if err != nil || cursor == nil {
+		t.Fatalf("safe unwind Aggregate = %#v, %v", cursor, err)
+	}
+	if len(driver.aggregatePipeline) != 4 {
+		t.Fatalf("aggregate pipeline = %#v, want scope, input bound, unwind, output bound", driver.aggregatePipeline)
+	}
+	assertG020OwnedAggregateStage(t, driver.aggregatePipeline[0])
+	if stage, ok := driver.aggregatePipeline[1].(bson.D); !ok || len(stage) != 1 || stage[0].Key != "$limit" {
+		t.Fatalf("aggregate input bound = %#v", driver.aggregatePipeline[1])
+	}
+	if stage, ok := driver.aggregatePipeline[2].(bson.D); !ok || len(stage) != 1 || stage[0].Key != "$unwind" {
+		t.Fatalf("aggregate unwind stage = %#v", driver.aggregatePipeline[2])
+	}
+
+	before := len(driver.aggregatePipeline)
+	_, err = collection.Aggregate(context.Background(), g020Scope(), bson.A{bson.D{{Key: "$unwind", Value: "$email"}}}, ResultPolicy{ID: "G020_REJECT"})
+	if !hasG020ErrorCode(err, ErrorUnsupportedOperation) || len(driver.aggregatePipeline) != before {
+		t.Fatalf("protected unwind err=%v pipeline=%#v, want pre-effect unsupported rejection", err, driver.aggregatePipeline)
+	}
+	_, err = collection.Aggregate(context.Background(), g020Scope(), bson.A{bson.D{{Key: "$group", Value: bson.D{{Key: "_id", Value: "$status"}, {Key: "count", Value: bson.D{{Key: "$sum", Value: int32(1)}}}}}}}, ResultPolicy{ID: "G020_GROUP_REJECT"})
+	if !hasG020ErrorCode(err, ErrorUnsupportedOperation) || len(driver.aggregatePipeline) != before {
+		t.Fatalf("group result-shape err=%v pipeline=%#v, want pre-effect unsupported rejection", err, driver.aggregatePipeline)
+	}
+}
+
 func TestG020LaneBWatchUsesEventScopeAndRejectsUnsafeDeleteRequests(t *testing.T) {
 	driver := newG020Driver()
 	client := newG020Client(&g020PipelineRuntime{}, driver)
@@ -337,6 +378,18 @@ func TestG020LaneBPublicAdmissionRejectsEmptyOrWrongPolicyBeforeDriver(t *testin
 	}
 }
 
+type g020CompilerPipelineRuntime struct{ g020PipelineRuntime }
+
+func (*g020CompilerPipelineRuntime) CompilePipeline(_ context.Context, row PolicyRow, scope OperationScope, value any) (bson.A, error) {
+	collection, err := normalizedCollection(row)
+	if err != nil {
+		return nil, err
+	}
+	return compiler.CompileAggregate(value, collection, query.Context{
+		Environment: "TEST", Service: "openmirai", Database: "app", Collection: "users", TenantScope: []byte(scope.TenantID), Limits: limits.DefaultEffective(),
+	})
+}
+
 type g020PipelineRuntime struct{}
 
 type g020BulkPipelineRuntime struct{ g020PipelineRuntime }
@@ -458,6 +511,15 @@ func g020Policy() PolicyRow {
 	return PolicyRow{Version: PolicyVersion1, PolicyID: "USERS", Namespace: "app.users", Scope: PolicyScopeTenant, TenantPath: "tenant.id", DocumentIDPath: "_id", Generation: 1}
 }
 
+func g020AggregatePolicy() PolicyRow {
+	row := g020Policy()
+	row.ProtectedFields = []ProtectedFieldPolicy{{
+		Path: "email", BSONType: ProtectedBSONString, DataClass: DataClassPIIRestricted, EncryptionRequired: true,
+		Algorithm: FieldAlgorithmAES256GCM, Normalization: NormalizationNone, AADProfile: AADProfileOMAAD1FieldV1, MaximumPlaintextBytes: 256,
+	}}
+	return row
+}
+
 func g020Scope() OperationScope {
 	return OperationScope{PolicyID: "USERS", TenantID: "tenant-a"}
 }