//go:build crm && runtime_l1_integration package crm import ( "bytes" "context" "crypto/ed25519" "database/sql" "encoding/json" "errors" "fmt" "net" "net/http" "os" "os/exec" "os/signal" "path/filepath" "reflect" "strings" "sync" "sync/atomic" "syscall" "testing" "time" "github.com/go-sql-driver/mysql" issuetrackerconformance "github.com/nexuscrm/app-packs-conformance/issue-tracker" featureauth "github.com/nexuscrm/auth" "github.com/nexuscrm/backend/internal/runtimeapi" "github.com/nexuscrm/backend/internal/runtimeassembly" "github.com/nexuscrm/backend/internal/runtimeidentity" kernelauth "github.com/nexuscrm/kernel/pkg/auth" "github.com/nexuscrm/kernel/pkg/ports" domainrls "github.com/nexuscrm/kernel/pkg/rls" "github.com/nexuscrm/metadata/catalogsnapshot" "github.com/nexuscrm/permissions" "github.com/nexuscrm/persistence" "github.com/nexuscrm/shared/pkg/constants" "github.com/nexuscrm/shared/pkg/contract" "github.com/nexuscrm/shared/pkg/models" "github.com/pingcap-inc/nexuscrm-api-host/processhttp" "github.com/pingcap-inc/nexuscrm-api-host/runtimegrpc" "github.com/pingcap-inc/nexuscrm-contract/pkg/runtime/delegationv1" querysemantic "github.com/pingcap-inc/nexuscrm-contract/pkg/runtime/queryv1" accesssemantic "github.com/pingcap-inc/nexuscrm-contract/pkg/runtime/recordaccessv1" semantic "github.com/pingcap-inc/nexuscrm-contract/pkg/runtime/recordsv1" "github.com/pingcap-inc/nexuscrm-contract/wire-go/envelope" runtimev1 "github.com/pingcap-inc/nexuscrm-contract/wire-go/gen/runtime/v1" "github.com/pingcap-inc/nexuscrm-controlplane/catalogapp" "github.com/pingcap-inc/nexuscrm-controlplane/catalogread" "google.golang.org/grpc" "google.golang.org/grpc/credentials/insecure" "google.golang.org/grpc/metadata" ) const ( p4ChildModeEnv = "NEXUSCRM_P4_RUNTIME_CHILD" p4ReadyFileEnv = "NEXUSCRM_P4_RUNTIME_READY_FILE" p4ResultFileEnv = "NEXUSCRM_P4_RUNTIME_RESULT_FILE" p4GrantReadEnv = "NEXUSCRM_P4_RUNTIME_GRANT_READ" p4SecurityScenarioEnv = "NEXUSCRM_P4_RUNTIME_SECURITY_SCENARIO" p4SessionTokenEnv = "NEXUSCRM_P4_RUNTIME_SESSION_TOKEN" p4RecordID = "ISS-P4-1" p4RLSRecordID = "ISS-P4-RLS" p4WorkloadID = "spiffe://example.test/ns/runtime/sa/p4-client" p4OrgID = "org-p4" p4InstallID = "installation-p4" p4Audience = "runtime-p4" p4SessionID = "session-p4" p4SigningKeyID = "p4-test-key" d2aQueryAppCapability = "io.nexuscrm.fixture.issue-tracker.query.v1" ) var p4SelectedFields = []string{ "UpdatedAt", "SecretNote", "ZeroEstimate", "StoryPoints", "Subject", "RetryCount", "ProjectId", "OptionalValue", "EmptySummary", "DueDate", "Disabled", "Archived", "ZeroCount", } func TestCRMAssignmentPolicyUserProfileReadSerializesConcurrentRename(t *testing.T) { if os.Getenv("TIDB_HOST") == "" { t.Skip("runtime_l1_integration requires the declared TiDB environment") } db, cleanup := openP4ScratchDB(t) defer cleanup() db.SetMaxOpenConns(4) if _, err := db.Exec(`CREATE TABLE Profile (Id VARCHAR(255) PRIMARY KEY, Name VARCHAR(255) NOT NULL)`); err != nil { t.Fatal(err) } if _, err := db.Exec(`INSERT INTO Profile (Id, Name) VALUES ('profile-sales', 'Sales')`); err != nil { t.Fatal(err) } readTx, err := db.BeginTx(t.Context(), &sql.TxOptions{Isolation: sql.LevelReadCommitted}) if err != nil { t.Fatal(err) } defer func() { _ = readTx.Rollback() }() hook, err := permissions.NewAssignmentPolicyEligibilityLifecycleHook( lifecycleTxManagerStub{tx: readTx, active: true, pessimistic: true}, &assignmentPolicyEligibilityMutationHandlerStub{}, ) if err != nil { t.Fatal(err) } policy := crmUserAssignmentPolicyEligibilityTestPolicy() // This test exercises the independent Runtime interpreter, whose lifecycle // events carry the Catalog-compiled physical projection rather than the live // authored metadata consumed by cmd/crm-server. policy.AssignmentPolicyEligibility.Selector.Reference.ResolvedTargetTable = "Profile" policy.AssignmentPolicyEligibility.Selector.Reference.ResolvedTargetIDColumn = "Id" policy.AssignmentPolicyEligibility.Selector.Reference.ResolvedTargetFieldColumn = "Name" if err := hook(t.Context(), []contract.RecordLifecycleBatchEvent{{ Phase: contract.RecordLifecycleAfterCreateInTx, Policy: policy, Event: contract.RecordLifecycleEvent{ ObjectName: "User", RecordID: "user-sales", Record: contract.Record{ "IsActive": true, "IsDeleted": false, "ProfileId": "profile-sales", }, }, }}); err != nil { t.Fatal(err) } renameStarted := make(chan struct{}) renameFinished := make(chan error, 1) renameCtx, cancelRename := context.WithTimeout(context.Background(), 5*time.Second) defer cancelRename() go func() { close(renameStarted) _, renameErr := db.ExecContext( renameCtx, `UPDATE Profile SET Name = 'Renamed Sales' WHERE Id = 'profile-sales'`, ) renameFinished <- renameErr }() <-renameStarted select { case renameErr := <-renameFinished: t.Fatalf("concurrent Profile rename completed before eligibility read transaction released its lock: %v", renameErr) case <-time.After(250 * time.Millisecond): } if err := readTx.Commit(); err != nil { t.Fatal(err) } select { case renameErr := <-renameFinished: if renameErr != nil { t.Fatal(renameErr) } case <-renameCtx.Done(): t.Fatalf("concurrent Profile rename did not resume after eligibility read committed: %v", renameCtx.Err()) } var renamed string if err := db.QueryRow(`SELECT Name FROM Profile WHERE Id = 'profile-sales'`).Scan(&renamed); err != nil { t.Fatal(err) } if renamed != "Renamed Sales" { t.Fatalf("committed Profile name = %q, want Renamed Sales", renamed) } } func TestNeutralRecordsConsumersCrossSeparateRuntimeProcess(t *testing.T) { if os.Getenv("TIDB_HOST") == "" { t.Skip("runtime_l1_integration requires the declared TiDB environment") } granted := startP4RuntimeChild(t, true) client, grantedRPCCalls, closeClient := newP4Client(t, granted.grpcAddress) defer closeClient() fixture := newP4FixtureConsumer(t, client) fixtureContext, fixtureCancel := context.WithTimeout(context.Background(), time.Second) assertP4FixtureResult(t, fixture, p4RecordID, fixtureContext, issuetrackerconformance.OutcomeOK, p4ExpectedFixtureFields()) fixtureCancel() missingContext, missingCancel := context.WithTimeout(context.Background(), time.Second) assertP4FixtureResult(t, fixture, "ISS-MISSING", missingContext, issuetrackerconformance.OutcomeMissing, nil) missingCancel() rlsContext, rlsCancel := context.WithTimeout(context.Background(), time.Second) assertP4FixtureResult(t, fixture, p4RLSRecordID, rlsContext, issuetrackerconformance.OutcomeMissing, nil) rlsCancel() canceled, cancelDeadline := context.WithTimeout(context.Background(), time.Second) cancelDeadline() assertP4FixtureResult(t, fixture, p4RecordID, canceled, issuetrackerconformance.OutcomeCanceled, nil) accessTransport, accessRPCCalls, closeAccessClient := newP4RecordAccessClient(t, granted.grpcAddress) defer closeAccessClient() objectID := "issue" accessContext, accessCancel := context.WithTimeout(context.Background(), time.Second) accessContext = metadata.AppendToOutgoingContext(accessContext, envelope.RuntimeDelegationMetadataKey, p4DelegationToken(t)) access, err := accessTransport.EvaluateRecordAccess(accessContext, accesssemantic.Request{ ObjectID: &objectID, RecordIDs: []string{p4RLSRecordID, p4RecordID, "ISS-MISSING", p4RecordID}, }) accessCancel() if err != nil { t.Fatal(err) } if accessRPCCalls.Load() != 1 || access.Receipts.CapabilityVersion != accesssemantic.CapabilityName || len(access.Results) != 1 || access.Results[0].RecordID != p4RecordID || !access.Results[0].Access.Read || !access.Results[0].Access.Update || !access.Results[0].Access.Delete || access.Results[0].Access.Locked { t.Fatalf("record-access calls=%d response=%#v", accessRPCCalls.Load(), access) } grantedResult := granted.stop(t) if grantedRPCCalls.Load() != 3 { t.Fatalf("granted Runtime data RPCs = %d, want 3", grantedRPCCalls.Load()) } if grantedResult.DatabaseQueries != 4 { t.Fatalf("granted Runtime database queries = %d, want 4", grantedResult.DatabaseQueries) } denied := startP4RuntimeChild(t, false) deniedClient, deniedRPCCalls, closeDeniedClient := newP4Client(t, denied.grpcAddress) defer closeDeniedClient() deniedFixture := newP4FixtureConsumer(t, deniedClient) deniedContext, deniedCancel := context.WithTimeout(context.Background(), time.Second) assertP4FixtureResult(t, deniedFixture, p4RecordID, deniedContext, issuetrackerconformance.OutcomeForbidden, nil) deniedCancel() deniedResult := denied.stop(t) if deniedRPCCalls.Load() != 1 { t.Fatalf("forbidden Runtime data RPCs = %d, want 1", deniedRPCCalls.Load()) } if deniedResult.DatabaseQueries != 0 { t.Fatalf("forbidden Runtime reached database: %d queries", deniedResult.DatabaseQueries) } for _, securityCase := range []struct { name string scenario string want semantic.FailureCategory wantReady bool }{ {name: "cross-org", scenario: "cross-org", want: semantic.FailureForbidden, wantReady: true}, {name: "cross-AppInstallation", scenario: "cross-app", want: semantic.FailureForbidden, wantReady: true}, {name: "workload mismatch", scenario: "workload-mismatch", want: semantic.FailureForbidden, wantReady: true}, {name: "wrong audience", scenario: "wrong-audience", want: semantic.FailureUnauthenticated, wantReady: true}, {name: "inactive AppInstallation", scenario: "inactive-installation", want: semantic.FailureForbidden, wantReady: true}, {name: "removed capability", scenario: "removed-capability", want: semantic.FailureForbidden, wantReady: true}, {name: "delegation revocation", scenario: "delegation-revoked", want: semantic.FailureUnauthenticated, wantReady: true}, {name: "retired signing key", scenario: "retired-key", want: semantic.FailureUnauthenticated, wantReady: true}, {name: "session revocation", scenario: "session-revoked", want: semantic.FailureUnauthenticated, wantReady: true}, {name: "inactive user", scenario: "user-inactive", want: semantic.FailureUnauthenticated, wantReady: true}, {name: "Connected App disabled", scenario: "connected-app-disabled", want: semantic.FailureUnauthenticated, wantReady: true}, {name: "old AuthEpoch", scenario: "old-auth-epoch", want: semantic.FailureUnauthenticated, wantReady: true}, {name: "old SecurityEpoch", scenario: "old-security-epoch", want: semantic.FailureStaleState, wantReady: true}, {name: "authority partition", scenario: "authority-partition", want: semantic.FailureUnavailable, wantReady: true}, {name: "expired SecurityLease", scenario: "expired-lease", want: semantic.FailureStaleState}, } { t.Run(securityCase.name, func(t *testing.T) { child := startP4RuntimeChildScenario(t, true, securityCase.scenario, securityCase.wantReady) client, calls, closeClient := newP4Client(t, child.grpcAddress) defer closeClient() ctx, cancel := context.WithTimeout(context.Background(), time.Second) _, err := client.ReadSelectedRoot(ctx, semantic.Request{ObjectID: "Issue", RecordID: p4RecordID, SelectedRootFields: []string{"Subject"}}) cancel() if got := p4FailureCategory(err); got != securityCase.want { t.Fatalf("security failure = %s (%v), want %s", got, err, securityCase.want) } result := child.stop(t) if calls.Load() != 1 || result.DatabaseQueries != 0 { t.Fatalf("security denial calls=%d database queries=%d, want 1/0", calls.Load(), result.DatabaseQueries) } }) } } // TestD2AIssueQueryConsumerCrossesSeparateRuntimeProcess is the first // capability-specific non-CRM proof for RecordQueryService. The fixture owns // only an Issue-shaped neutral request; Catalog artifacts, Runtime admission, // metadata/RLS/FLS execution, transport, and TiDB remain in their real owners. func TestD2AIssueQueryConsumerCrossesSeparateRuntimeProcess(t *testing.T) { if os.Getenv("TIDB_HOST") == "" { t.Skip("runtime_l1_integration requires the declared TiDB environment") } child := startP4RuntimeChildCommand(t, "TestD2AQueryRuntimeChildProcess", true, "", true) client, calls, closeClient := newP4QueryClient(t, child.grpcAddress) defer closeClient() consumer, err := issuetrackerconformance.NewIssueQueryConsumer(func(ctx context.Context, request querysemantic.Request) (querysemantic.Response, error) { return client.QueryRecords(ctx, request) }) if err != nil { t.Fatal(err) } ctx, cancel := context.WithTimeout(context.Background(), time.Second) response, err := consumer.ListByRetryFloor(ctx, 0) cancel() if err != nil { t.Fatalf("%v\nRuntime child logs:\n%s", err, child.logs) } if calls.Load() != 1 || response.TotalSize != 1 || !response.Done || len(response.Records) != 1 || response.Receipts.CapabilityVersion != querysemantic.CapabilityName || response.Receipts.CatalogRevision != 1 { t.Fatalf("query calls=%d response=%#v", calls.Load(), response) } wantFields := []querysemantic.Field{ {Name: "Id", Value: querysemantic.Value{Kind: querysemantic.ValueString, String: stringPointer(p4RecordID)}}, {Name: "RetryCount", Value: querysemantic.Value{Kind: querysemantic.ValueInteger, Integer: int64Pointer(7)}}, {Name: "Subject", Value: querysemantic.Value{Kind: querysemantic.ValueString, String: stringPointer("Non-CRM point read")}}, } if response.Records[0].ObjectID != issuetrackerconformance.IssueObject || !reflect.DeepEqual(response.Records[0].Fields, wantFields) { t.Fatalf("authorized query record = %#v, want fields %#v", response.Records[0], wantFields) } unsupportedContext, unsupportedCancel := context.WithTimeout(context.Background(), time.Second) unsupportedValue := "Non-CRM point read" _, err = client.QueryRecords(unsupportedContext, querysemantic.Request{ ObjectID: "Issue", Selections: []querysemantic.Selection{ {Kind: querysemantic.SelectionField, Field: "Subject"}, {Kind: querysemantic.SelectionAggregate, Aggregate: querysemantic.AggregateCount, Field: constants.FieldID, Alias: "total"}, }, GroupBy: "Subject", Having: querysemantic.Predicate{ Kind: querysemantic.PredicateComparison, Operator: querysemantic.ComparisonEqual, Left: querysemantic.Operand{Kind: querysemantic.OperandFunction, Function: querysemantic.FunctionString, Args: []querysemantic.Operand{{Kind: querysemantic.OperandField, Field: "Subject"}}}, Right: querysemantic.Operand{Kind: querysemantic.OperandLiteral, Literal: &querysemantic.Scalar{String: &unsupportedValue}}, }, }) unsupportedCancel() var unsupportedFailure *querysemantic.Failure if !errors.As(err, &unsupportedFailure) || unsupportedFailure.Category != querysemantic.FailureInvalidRequest || unsupportedFailure.InvalidComponent != querysemantic.InvalidComponentQueryPlan || calls.Load() != 2 { t.Fatalf("unsupported Runtime function error=%v calls=%d, want typed invalid query plan after one RPC", err, calls.Load()) } unknownContext, unknownCancel := context.WithTimeout(context.Background(), time.Second) _, err = client.QueryRecords(unknownContext, querysemantic.Request{ ObjectID: "MissingObject", Selections: []querysemantic.Selection{{Kind: querysemantic.SelectionField, Field: constants.FieldID}}, Limit: 1, }) unknownCancel() var unknownFailure *querysemantic.Failure if !errors.As(err, &unknownFailure) || unknownFailure.Category != querysemantic.FailureInvalidRequest || unknownFailure.InvalidComponent != querysemantic.InvalidComponentObjectID || calls.Load() != 3 { t.Fatalf("unknown object query error=%v calls=%d, want invalid object identifier after one RPC", err, calls.Load()) } missingRelationshipContext, missingRelationshipCancel := context.WithTimeout(context.Background(), time.Second) _, err = client.QueryRecords(missingRelationshipContext, querysemantic.Request{ ObjectID: "Issue", Selections: []querysemantic.Selection{{Kind: querysemantic.SelectionField, Field: constants.FieldID}}, Limit: 1, Subqueries: []querysemantic.Subquery{{RelationshipName: "MissingChildren", SelectedFields: []string{constants.FieldID}, Limit: 1}}, }) missingRelationshipCancel() var missingRelationshipFailure *querysemantic.Failure if !errors.As(err, &missingRelationshipFailure) || missingRelationshipFailure.Category != querysemantic.FailureInvalidRequest || missingRelationshipFailure.InvalidComponent != querysemantic.InvalidComponentQueryPlan || calls.Load() != 4 { t.Fatalf("missing child relationship error=%v calls=%d, want invalid query plan after one RPC", err, calls.Load()) } canceled, cancel := context.WithCancel(context.Background()) cancel() _, err = client.QueryRecords(canceled, querysemantic.Request{ObjectID: "Issue", Selections: []querysemantic.Selection{{Kind: querysemantic.SelectionField, Field: "Id"}}, Limit: 1}) var canceledFailure *querysemantic.Failure if !errors.As(err, &canceledFailure) || canceledFailure.Category != querysemantic.FailureCanceled || calls.Load() != 4 { t.Fatalf("canceled query error=%v calls=%d, want local cancellation before RPC", err, calls.Load()) } result := child.stop(t) if result.DatabaseQueries != 2 { t.Fatalf("query repository calls = %d, want one successful fixture query plus one fail-closed relationship validation", result.DatabaseQueries) } outageCtx, outageCancel := context.WithTimeout(context.Background(), 250*time.Millisecond) defer outageCancel() _, err = client.QueryRecords(outageCtx, querysemantic.Request{ObjectID: "Issue", Selections: []querysemantic.Selection{{Kind: querysemantic.SelectionField, Field: "Id"}}, Limit: 1}) var outageFailure *querysemantic.Failure if !errors.As(err, &outageFailure) || outageFailure.Category != querysemantic.FailureUnavailable { t.Fatalf("stopped Runtime query error=%v, want fail-closed Unavailable", err) } denied := startP4RuntimeChildCommand(t, "TestD2AQueryRuntimeChildProcess", false, "", true) deniedClient, deniedCalls, closeDenied := newP4QueryClient(t, denied.grpcAddress) defer closeDenied() deniedCtx, deniedCancel := context.WithTimeout(context.Background(), time.Second) _, err = deniedClient.QueryRecords(deniedCtx, querysemantic.Request{ObjectID: "Issue", Selections: []querysemantic.Selection{{Kind: querysemantic.SelectionField, Field: "Id"}}, Limit: 1}) deniedCancel() var deniedFailure *querysemantic.Failure if !errors.As(err, &deniedFailure) || deniedFailure.Category != querysemantic.FailureForbidden || deniedCalls.Load() != 1 { t.Fatalf("object-read denial error=%v calls=%d", err, deniedCalls.Load()) } deniedResult := denied.stop(t) if deniedResult.DatabaseQueries != 0 { t.Fatalf("object-read denial reached TiDB query repository: %d", deniedResult.DatabaseQueries) } } func stringPointer(value string) *string { return &value } func int64Pointer(value int64) *int64 { return &value } func p4FailureCategory(err error) semantic.FailureCategory { var failure *semantic.Failure if errors.As(err, &failure) { return failure.Category } return "" } func TestRecordsRuntimeChildProcess(t *testing.T) { if os.Getenv(p4ChildModeEnv) != "1" { t.Skip("P4 Runtime child is launched by the parent integration test") } grantRead := os.Getenv(p4GrantReadEnv) == "1" securityScenario := os.Getenv(p4SecurityScenarioEnv) db, cleanup := openP4ScratchDB(t) defer cleanup() createP4Dataset(t, db, securityScenario) var databaseQueries atomic.Int64 queryRepo := &p4CountingQueryRepository{QueryRepository: persistence.NewQueryRepository(db), calls: &databaseQueries} recordRepo := &p4CountingRecordRepository{RecordRepository: persistence.NewRecordRepository(db), calls: &databaseQueries} permissionRepo := &p4PermissionRepository{grantRead: grantRead} graph, err := runtimeassembly.NewRecordsReadRuntime(runtimeassembly.RecordsReadDependencies{ DB: db, MetadataRepo: &p4MetadataRepository{schema: p4IssueSchema()}, PermissionRepo: permissionRepo, QueryRepo: queryRepo, RecordRepo: recordRepo, }) if err != nil { t.Fatal(err) } process, err := runtimeapi.Assemble(runtimeapi.ProcessConfig{ Version: "p4-integration", AdminAddress: "127.0.0.1:0", GRPCAddress: "127.0.0.1:0", AdmissionLimit: runtimeapi.DefaultAdmissionLimit, ServingEnabled: true, RecordAccessMaxAdmissionLease: time.Second, WorkloadAuthority: func(context.Context, string) (runtimegrpc.VerifiedCaller, error) { return p4WorkloadID, nil }, SecurityAdmission: p4SecurityAdmission(t, db, securityScenario), Receipts: runtimeapi.ReceiptSourceFunc(runtimeapi.BootstrapReceipts), }, graph.Records) if err != nil { t.Fatal(err) } if err := process.Start(); err != nil { t.Fatal(err) } ready := p4ChildReady{PID: os.Getpid(), GRPCAddress: process.GRPCAddress(), AdminAddress: process.AdminAddress()} writeP4JSON(t, os.Getenv(p4ReadyFileEnv), ready) stop := make(chan os.Signal, 1) signal.Notify(stop, os.Interrupt, syscall.SIGTERM) defer signal.Stop(stop) <-stop shutdownCtx, cancel := context.WithTimeout(context.Background(), 5*time.Second) defer cancel() if err := process.Stop(shutdownCtx); err != nil { t.Fatal(err) } writeP4JSON(t, os.Getenv(p4ResultFileEnv), p4ChildResult{DatabaseQueries: databaseQueries.Load()}) } func TestD2AQueryRuntimeChildProcess(t *testing.T) { if os.Getenv(p4ChildModeEnv) != "1" { t.Skip("D2A Runtime child is launched by the parent integration test") } db, cleanup := openP4ScratchDB(t) defer cleanup() createP4Dataset(t, db, "") createD2AQueryDataset(t, db) var databaseQueries atomic.Int64 queryRepo := &p4CountingQueryRepository{QueryRepository: persistence.NewQueryRepository(db), calls: &databaseQueries} graph, err := runtimeassembly.NewServingRecordsReadRuntime(runtimeassembly.RecordsReadDependencies{ DB: db, PermissionRepo: &p4PermissionRepository{grantRead: os.Getenv(p4GrantReadEnv) == "1"}, QueryRepo: queryRepo, RecordRepo: persistence.NewRecordRepository(db), }, runtimeapi.RuntimeMetadataViewFromContext) if err != nil { t.Fatal(err) } if err := graph.Prepare(t.Context()); err != nil { t.Fatal(err) } servingAdmission := p4QueryServingAdmission(t, db) process, err := runtimeapi.Assemble(runtimeapi.ProcessConfig{ Version: "d2a-query-integration", AdminAddress: "127.0.0.1:0", GRPCAddress: "127.0.0.1:0", AdmissionLimit: runtimeapi.DefaultAdmissionLimit, ServingEnabled: true, QueryMaxAdmissionLease: time.Second, QueryEngine: graph.Query, WorkloadAuthority: func(context.Context, string) (runtimegrpc.VerifiedCaller, error) { return p4WorkloadID, nil }, SecurityAdmission: p4SecurityAdmission(t, db, ""), ServingAdmission: servingAdmission, }, graph.Records) if err != nil { t.Fatal(err) } if err := process.Start(); err != nil { t.Fatal(err) } writeP4JSON(t, os.Getenv(p4ReadyFileEnv), p4ChildReady{PID: os.Getpid(), GRPCAddress: process.GRPCAddress(), AdminAddress: process.AdminAddress()}) stop := make(chan os.Signal, 1) signal.Notify(stop, os.Interrupt, syscall.SIGTERM) defer signal.Stop(stop) <-stop shutdownCtx, cancel := context.WithTimeout(context.Background(), 5*time.Second) defer cancel() if err := process.Stop(shutdownCtx); err != nil { t.Fatal(err) } writeP4JSON(t, os.Getenv(p4ResultFileEnv), p4ChildResult{DatabaseQueries: databaseQueries.Load()}) } func TestP4ChildEnvironmentReplacesOwnedKeys(t *testing.T) { got := p4ChildEnvironment([]string{ "PATH=/bin", p4ChildModeEnv + "=stale", p4GrantReadEnv + "=stale", p4SecurityScenarioEnv + "=stale", p4SessionTokenEnv + "=stale", }, p4ChildModeEnv+"=1", p4GrantReadEnv+"=0", p4SecurityScenarioEnv+"=cross-org", p4SessionTokenEnv+"=current") if !reflect.DeepEqual(got, []string{ "PATH=/bin", p4ChildModeEnv + "=1", p4GrantReadEnv + "=0", p4SecurityScenarioEnv + "=cross-org", p4SessionTokenEnv + "=current", }) { t.Fatalf("child environment = %#v", got) } } func TestD2AQueryServingAdmissionPreparesOnlyTheQueryCapability(t *testing.T) { admission := p4QueryServingAdmission(t, nil) if !admission.Ready() { t.Fatal("D2A query serving admission is not ready after exact artifact reconciliation") } } type p4WirePointReader struct { client *runtimegrpc.SelectedRootClient } func (r *p4WirePointReader) Retrieve(ctx context.Context, objectName, recordID string, _ *models.UserSession) (contract.Record, error) { response, err := r.client.ReadSelectedRoot(ctx, semantic.Request{ ObjectID: objectName, RecordID: recordID, SelectedRootFields: p4SelectedFields, }) if err != nil { return nil, err } record := contract.Record{constants.FieldID: response.RecordID} for _, field := range response.Fields { value, err := p4FixtureValue(field) if err != nil { return nil, err } record[field.Name] = value } return record, nil } func p4FixtureValue(field semantic.Field) (any, error) { if field.Null { return nil, nil } switch field.Kind { case semantic.ValueKindString: return *field.String, nil case semantic.ValueKindBoolean: return *field.Boolean, nil case semantic.ValueKindInteger: return *field.Integer, nil case semantic.ValueKindDecimal: return json.Number(p4DecimalText(*field.Decimal)), nil case semantic.ValueKindDate: return *field.Date, nil case semantic.ValueKindDateTime: return *field.DateTime, nil case semantic.ValueKindReference: return *field.Reference, nil default: return nil, fmt.Errorf("unsupported fixture field kind %q", field.Kind) } } func p4DecimalText(value semantic.Decimal) string { negative := strings.HasPrefix(value.Coefficient, "-") digits := strings.TrimPrefix(value.Coefficient, "-") if value.Scale > 0 { scale := int(value.Scale) if len(digits) <= scale { digits = strings.Repeat("0", scale-len(digits)+1) + digits } point := len(digits) - scale digits = digits[:point] + "." + digits[point:] } if negative { return "-" + digits } return digits } func p4FixtureErrorClassifier(err error) (issuetrackerconformance.Outcome, bool) { var failure *semantic.Failure if !errors.As(err, &failure) { return "", false } switch failure.Category { case semantic.FailureMissing: return issuetrackerconformance.OutcomeMissing, true case semantic.FailureForbidden: return issuetrackerconformance.OutcomeForbidden, true case semantic.FailureCanceled, semantic.FailureDeadlineExceeded: return issuetrackerconformance.OutcomeCanceled, true default: return "", false } } func newP4FixtureConsumer(t *testing.T, client *runtimegrpc.SelectedRootClient) *issuetrackerconformance.Consumer { t.Helper() consumer, err := issuetrackerconformance.NewConsumer(&p4WirePointReader{client: client}, p4FixtureErrorClassifier) if err != nil { t.Fatal(err) } return consumer } func assertP4FixtureResult(t *testing.T, consumer *issuetrackerconformance.Consumer, recordID string, ctx context.Context, wantOutcome issuetrackerconformance.Outcome, wantFields []issuetrackerconformance.RootField) { t.Helper() result, err := consumer.Read(ctx, issuetrackerconformance.Request{RecordID: recordID, Fields: p4FixtureSelection()}, &models.UserSession{ID: "fixture-user"}) if err != nil { t.Fatal(err) } if result.Outcome != wantOutcome { t.Fatalf("record %s outcome = %q, want %q", recordID, result.Outcome, wantOutcome) } if wantFields == nil { wantFields = []issuetrackerconformance.RootField{} } if !reflect.DeepEqual(result.Fields, wantFields) { t.Fatalf("record %s fields = %#v, want %#v", recordID, result.Fields, wantFields) } } func p4FixtureSelection() []issuetrackerconformance.FieldSelection { return []issuetrackerconformance.FieldSelection{ {Name: "UpdatedAt", Kind: issuetrackerconformance.FieldDateTime}, {Name: "SecretNote", Kind: issuetrackerconformance.FieldString}, {Name: "ZeroEstimate", Kind: issuetrackerconformance.FieldDecimal}, {Name: "StoryPoints", Kind: issuetrackerconformance.FieldDecimal}, {Name: "Subject", Kind: issuetrackerconformance.FieldString}, {Name: "RetryCount", Kind: issuetrackerconformance.FieldInteger}, {Name: "ProjectId", Kind: issuetrackerconformance.FieldReference}, {Name: "OptionalValue", Kind: issuetrackerconformance.FieldString}, {Name: "NeverStored", Kind: issuetrackerconformance.FieldString}, {Name: "EmptySummary", Kind: issuetrackerconformance.FieldString}, {Name: "DueDate", Kind: issuetrackerconformance.FieldDate}, {Name: "Disabled", Kind: issuetrackerconformance.FieldBoolean}, {Name: "Archived", Kind: issuetrackerconformance.FieldBoolean}, {Name: "ZeroCount", Kind: issuetrackerconformance.FieldInteger}, } } func p4ExpectedFixtureFields() []issuetrackerconformance.RootField { return []issuetrackerconformance.RootField{ {Name: "Archived", Kind: issuetrackerconformance.FieldBoolean, Value: true}, {Name: "Disabled", Kind: issuetrackerconformance.FieldBoolean, Value: false}, {Name: "DueDate", Kind: issuetrackerconformance.FieldDate, Value: "2026-07-12"}, {Name: "EmptySummary", Kind: issuetrackerconformance.FieldString, Value: ""}, {Name: "OptionalValue", Kind: issuetrackerconformance.FieldString, Null: true}, {Name: "ProjectId", Kind: issuetrackerconformance.FieldReference, Value: "PRJ-001"}, {Name: "RetryCount", Kind: issuetrackerconformance.FieldInteger, Value: int64(7)}, {Name: "StoryPoints", Kind: issuetrackerconformance.FieldDecimal, Value: "12.50"}, {Name: "Subject", Kind: issuetrackerconformance.FieldString, Value: "Non-CRM point read"}, {Name: "UpdatedAt", Kind: issuetrackerconformance.FieldDateTime, Value: "2026-07-12T09:10:11.123456Z"}, {Name: "ZeroCount", Kind: issuetrackerconformance.FieldInteger, Value: int64(0)}, {Name: "ZeroEstimate", Kind: issuetrackerconformance.FieldDecimal, Value: "0.00"}, } } type p4RuntimeChild struct { command *exec.Cmd wait <-chan error logs *bytes.Buffer grpcAddress string resultFile string stopOnce sync.Once result p4ChildResult } type p4ChildReady struct { PID int `json:"pid"` GRPCAddress string `json:"grpc_address"` AdminAddress string `json:"admin_address"` } type p4ChildResult struct { DatabaseQueries int64 `json:"database_queries"` } func startP4RuntimeChild(t *testing.T, grantRead bool) *p4RuntimeChild { return startP4RuntimeChildScenario(t, grantRead, "", true) } func startP4RuntimeChildScenario(t *testing.T, grantRead bool, securityScenario string, wantReady bool) *p4RuntimeChild { return startP4RuntimeChildCommand(t, "TestRecordsRuntimeChildProcess", grantRead, securityScenario, wantReady) } func startP4RuntimeChildCommand(t *testing.T, childTest string, grantRead bool, securityScenario string, wantReady bool) *p4RuntimeChild { t.Helper() if os.Getenv("JWT_SECRET") == "" { t.Fatal("runtime_l1_integration requires one declared JWT_SECRET shared by parent and Runtime child") } dir := t.TempDir() readyFile := filepath.Join(dir, "ready.json") resultFile := filepath.Join(dir, "result.json") command := exec.Command(os.Args[0], "-test.run=^"+childTest+"$", "-test.v") grantValue := "0" if grantRead { grantValue = "1" } command.Env = p4ChildEnvironment(os.Environ(), p4ChildModeEnv+"=1", p4ReadyFileEnv+"="+readyFile, p4ResultFileEnv+"="+resultFile, p4GrantReadEnv+"="+grantValue, p4SecurityScenarioEnv+"="+securityScenario, p4SessionTokenEnv+"="+p4AuthorizingSessionToken(t), ) logs := &bytes.Buffer{} command.Stdout, command.Stderr = logs, logs if err := command.Start(); err != nil { t.Fatal(err) } wait := make(chan error, 1) go func() { wait <- command.Wait() }() child := &p4RuntimeChild{command: command, wait: wait, logs: logs, resultFile: resultFile} t.Cleanup(func() { child.stop(t) }) deadline := time.Now().Add(20 * time.Second) for time.Now().Before(deadline) { var ready p4ChildReady if data, err := os.ReadFile(readyFile); err == nil && json.Unmarshal(data, &ready) == nil && ready.GRPCAddress != "" { if ready.PID == 0 || ready.PID == os.Getpid() { t.Fatalf("Runtime child PID = %d, parent PID = %d", ready.PID, os.Getpid()) } probeClient := http.Client{Timeout: 250 * time.Millisecond} if response, err := probeClient.Get("http://" + ready.AdminAddress + processhttp.ReadyPath); err == nil { _ = response.Body.Close() if (wantReady && response.StatusCode == http.StatusOK) || (!wantReady && response.StatusCode == http.StatusServiceUnavailable) { child.grpcAddress = ready.GRPCAddress return child } } } select { case err := <-wait: t.Fatalf("Runtime child exited before ready: %v\n%s", err, logs) default: } time.Sleep(25 * time.Millisecond) } t.Fatalf("Runtime child did not become ready\n%s", logs) return nil } func (c *p4RuntimeChild) stop(t *testing.T) p4ChildResult { t.Helper() if c == nil { return p4ChildResult{} } c.stopOnce.Do(func() { if err := c.command.Process.Signal(syscall.SIGTERM); err != nil { t.Errorf("signal Runtime child: %v", err) return } select { case err := <-c.wait: if err != nil { t.Errorf("Runtime child exit: %v\n%s", err, c.logs) return } case <-time.After(10 * time.Second): _ = c.command.Process.Kill() t.Errorf("Runtime child did not drain\n%s", c.logs) return } data, err := os.ReadFile(c.resultFile) if err != nil || json.Unmarshal(data, &c.result) != nil { t.Errorf("read Runtime child result: %v data=%q\n%s", err, data, c.logs) } }) return c.result } func newP4Client(t *testing.T, address string) (*runtimegrpc.SelectedRootClient, *atomic.Int64, func()) { t.Helper() connection, err := grpc.NewClient("passthrough:///"+address, grpc.WithTransportCredentials(insecure.NewCredentials())) if err != nil { t.Fatal(err) } generated := runtimev1.NewRecordsServiceClient(connection) calls := &atomic.Int64{} delegation := p4DelegationToken(t) client, err := runtimegrpc.NewSelectedRootClient(func(ctx context.Context, request *runtimev1.ReadSelectedRootRequest, options ...grpc.CallOption) (*runtimev1.ReadSelectedRootResponse, error) { calls.Add(1) ctx = metadata.AppendToOutgoingContext(ctx, envelope.RuntimeDelegationMetadataKey, delegation) return generated.ReadSelectedRoot(ctx, request, options...) }) if err != nil { _ = connection.Close() t.Fatal(err) } return client, calls, func() { _ = connection.Close() } } func newP4RecordAccessClient(t *testing.T, address string) (*runtimegrpc.RecordAccessClient, *atomic.Int64, func()) { t.Helper() connection, err := grpc.NewClient("passthrough:///"+address, grpc.WithTransportCredentials(insecure.NewCredentials())) if err != nil { t.Fatal(err) } generated := runtimev1.NewRecordsServiceClient(connection) calls := &atomic.Int64{} client, err := runtimegrpc.NewRecordAccessClient(func(ctx context.Context, request *runtimev1.EvaluateRecordAccessRequest, options ...grpc.CallOption) (*runtimev1.EvaluateRecordAccessResponse, error) { calls.Add(1) return generated.EvaluateRecordAccess(ctx, request, options...) }) if err != nil { _ = connection.Close() t.Fatal(err) } return client, calls, func() { _ = connection.Close() } } func newP4QueryClient(t *testing.T, address string) (*runtimegrpc.QueryClient, *atomic.Int64, func()) { t.Helper() connection, err := grpc.NewClient("passthrough:///"+address, grpc.WithTransportCredentials(insecure.NewCredentials())) if err != nil { t.Fatal(err) } generated := runtimev1.NewRecordQueryServiceClient(connection) calls := &atomic.Int64{} delegation := p4DelegationToken(t) client, err := runtimegrpc.NewQueryClient( func(ctx context.Context, request *runtimev1.QueryRecordsRequest, options ...grpc.CallOption) (*runtimev1.QueryRecordsResponse, error) { calls.Add(1) ctx = metadata.AppendToOutgoingContext(ctx, envelope.RuntimeDelegationMetadataKey, delegation) return generated.QueryRecords(ctx, request, options...) }, func(ctx context.Context, request *runtimev1.QueryHistoryRequest, options ...grpc.CallOption) (*runtimev1.QueryHistoryResponse, error) { calls.Add(1) ctx = metadata.AppendToOutgoingContext(ctx, envelope.RuntimeDelegationMetadataKey, delegation) return generated.QueryHistory(ctx, request, options...) }, ) if err != nil { _ = connection.Close() t.Fatal(err) } return client, calls, func() { _ = connection.Close() } } func p4QueryServingAdmission(t *testing.T, catalogDB *sql.DB) *runtimeapi.ServingAdmission { t.Helper() lineage := catalogsnapshot.Lineage{CatalogRevision: 1, SchemaGeneration: 1, EngineGeneration: 1} snapshotArtifact, err := catalogsnapshot.CompileCatalogSnapshot(catalogsnapshot.CompileInput{ OrgID: p4OrgID, Lineage: lineage, Objects: []catalogsnapshot.ObjectInput{{CatalogRevision: lineage.CatalogRevision, Metadata: p4IssueSchema()}}, Projections: []catalogsnapshot.RuntimeProjectionInput{{ CatalogRevision: lineage.CatalogRevision, AppInstallationID: p4InstallID, CapabilityID: querysemantic.CapabilityName, Profile: catalogsnapshot.ProjectionRecordQueryV1, ObjectAPINames: []string{issuetrackerconformance.IssueObject}, }}, }) if err != nil { t.Fatal(err) } snapshot, err := catalogsnapshot.OpenCatalogSnapshot(snapshotArtifact) if err != nil { t.Fatal(err) } viewArtifact, err := catalogsnapshot.ProjectRuntimeMetadataView(snapshot, catalogsnapshot.ProjectionKey{ AppInstallationID: p4InstallID, CapabilityID: querysemantic.CapabilityName, }) if err != nil { t.Fatal(err) } appReleaseDigest := catalogsnapshot.ComputeDigest([]byte("d2a-issue-tracker-fixture-release")) serviceReleaseDigest := catalogsnapshot.ComputeDigest([]byte("d2a-runtime-service-release")) receiptArtifact, err := catalogsnapshot.BuildServingCompatibilityReceipt(catalogsnapshot.ServingCompatibilityReceiptInput{ OrgID: p4OrgID, AppInstallationID: p4InstallID, AppReleaseDigest: appReleaseDigest, AppCapabilityID: d2aQueryAppCapability, RuntimeCapabilityID: querysemantic.CapabilityName, WorkloadID: p4WorkloadID, ServiceReleaseDigest: serviceReleaseDigest, OwnershipEpoch: 1, CatalogSnapshotDigest: snapshotArtifact.Digest(), RuntimeMetadataViewDigest: viewArtifact.Digest(), Lineage: lineage, SecurityEpoch: 1, }) if err != nil { t.Fatal(err) } store, err := catalogsnapshot.NewDirectoryStore(t.TempDir()) if err != nil { t.Fatal(err) } for _, artifact := range []catalogsnapshot.Artifact{snapshotArtifact, viewArtifact, receiptArtifact} { if err := store.Put(t.Context(), artifact); err != nil { t.Fatal(err) } } catalogSource := func(_ context.Context, installationID, capabilityID string) (runtimeapi.ServingInstallation, runtimeapi.ServingHandlerBinding, bool, error) { if installationID != p4InstallID || capabilityID != d2aQueryAppCapability { return runtimeapi.ServingInstallation{}, runtimeapi.ServingHandlerBinding{}, false, nil } return runtimeapi.ServingInstallation{ OrgID: p4OrgID, AppInstallationID: p4InstallID, ActiveReleaseDigest: string(appReleaseDigest), CatalogRevision: 1, }, runtimeapi.ServingHandlerBinding{ OrgID: p4OrgID, AppInstallationID: p4InstallID, AppReleaseDigest: string(appReleaseDigest), CapabilityID: d2aQueryAppCapability, WorkloadID: p4WorkloadID, ServiceReleaseDigest: string(serviceReleaseDigest), OwnershipEpoch: 1, CompatibilityReceiptDigest: string(receiptArtifact.Digest()), }, true, nil } if catalogDB != nil { seedP4ActiveCatalogServingCapability(t, catalogDB, appReleaseDigest, serviceReleaseDigest, receiptArtifact.Digest()) reader, err := catalogread.New(catalogDB) if err != nil { t.Fatal(err) } catalogSource, err = runtimeapi.CatalogServingSource(reader) if err != nil { t.Fatal(err) } } state, err := runtimeapi.NewServingState(runtimeapi.ServingStateConfig{ Store: store, Catalog: catalogSource, }) if err != nil { t.Fatal(err) } if err := state.Reconcile(t.Context(), runtimeapi.ServingSelection{ OrgID: p4OrgID, AppInstallationID: p4InstallID, ExpectedWorkloadID: p4WorkloadID, AppCapabilityID: d2aQueryAppCapability, RuntimeCapabilityID: querysemantic.CapabilityName, ExpectedServiceReleaseDigest: serviceReleaseDigest, ExpectedOwnershipEpoch: 1, ExpectedCompatibilityReceiptDigest: receiptArtifact.Digest(), LocalCompatibility: catalogsnapshot.Compatibility{SchemaGeneration: 1, EngineGeneration: 1}, SecurityEpoch: 1, }); err != nil { t.Fatal(err) } admission, err := runtimeapi.NewServingAdmission(state) if err != nil { t.Fatal(err) } return admission } // seedP4ActiveCatalogServingCapability uses Catalog's test schema rather than // copying a reduced table shape into the Runtime walking slice. The Runtime // process reads the resulting active tuple only through catalogread; this test // setup never links the Catalog lifecycle writer into the Runtime closure. func seedP4ActiveCatalogServingCapability(t *testing.T, db *sql.DB, appReleaseDigest, serviceReleaseDigest, receiptDigest catalogsnapshot.Digest) { t.Helper() for _, statement := range catalogapp.SchemaStatementsForTest() { if _, err := db.ExecContext(t.Context(), statement); err != nil { t.Fatal(err) } } now := time.Now().UTC() if _, err := db.ExecContext(t.Context(), `INSERT INTO _System_CatalogAppInstallation (AppInstallationId, OrgId, AppId, State, ActiveReleaseDigest, CatalogRevision, ActivationEpoch, LiveAppKey, CreatedDate, LastModifiedDate) VALUES (?, ?, ?, 'ACTIVE', ?, 1, 1, ?, ?, ?)`, p4InstallID, p4OrgID, "test-issue-tracker", string(appReleaseDigest), strings.Repeat("a", 64), now, now); err != nil { t.Fatal(err) } if _, err := db.ExecContext(t.Context(), `INSERT INTO _System_CatalogHandlerBinding (HandlerBindingId, OrgId, AppInstallationId, AppReleaseDigest, CapabilityId, HandlerRef, WorkloadId, ServiceReleaseDigest, OwnershipEpoch, State, CompatibilityReceiptDigest, NaturalKey, ActiveKey, CreatedDate, LastModifiedDate) VALUES (?, ?, ?, ?, ?, ?, ?, ?, 1, 'ACTIVE', ?, ?, ?, ?, ?)`, "00000000-0000-7000-8000-000000000004", p4OrgID, p4InstallID, string(appReleaseDigest), d2aQueryAppCapability, "test.issue-tracker.query", p4WorkloadID, string(serviceReleaseDigest), string(receiptDigest), strings.Repeat("b", 64), strings.Repeat("c", 64), now, now); err != nil { t.Fatal(err) } } func p4SecurityAdmission(t *testing.T, db *sql.DB, scenario string) *runtimeapi.SecurityAdmission { t.Helper() const replicaID = "replica-p4" privateKey := ed25519.NewKeyFromSeed(bytes.Repeat([]byte{0x24}, ed25519.SeedSize)) verificationKeys := []delegationv1.VerificationKey{{ SigningKeyID: p4SigningKeyID, PublicKey: privateKey.Public().(ed25519.PublicKey), State: delegationv1.KeyStateCurrent, }} if scenario == "retired-key" { currentKey := ed25519.NewKeyFromSeed(bytes.Repeat([]byte{0x25}, ed25519.SeedSize)) verificationKeys = []delegationv1.VerificationKey{ {SigningKeyID: "p4-current-key", PublicKey: currentKey.Public().(ed25519.PublicKey), State: delegationv1.KeyStateCurrent}, {SigningKeyID: p4SigningKeyID, PublicKey: privateKey.Public().(ed25519.PublicKey), State: delegationv1.KeyStateRetired}, } } verifier, err := delegationv1.NewVerifier(delegationv1.VerifierConfig{ Keys: verificationKeys, SupportedCapabilities: p4Capabilities(), }) if err != nil { t.Fatal(err) } securityRepository := persistence.NewRuntimeSecurityRepository(db) now := time.Now().UTC() orgID, installationID, securityEpoch := p4OrgID, p4InstallID, uint64(1) if scenario == "cross-org" { orgID = "org-p4-other" } if scenario == "cross-app" { installationID = "installation-p4-other" } if scenario == "old-security-epoch" { securityEpoch = 2 } runtimeAudience := p4Audience if scenario == "wrong-audience" { runtimeAudience = "runtime-p4-other" } installationActive := scenario != "inactive-installation" capabilities := p4Capabilities() if scenario == "removed-capability" { capabilities = []string{"runtime.records.other.v1"} } if scenario == "cross-org" { if err := securityRepository.ActivateProjection(context.Background(), 0, persistence.RuntimeSecurityProjection{ OrgID: p4OrgID, RuntimeAudience: p4Audience, SecurityEpoch: 1, Bindings: []persistence.RuntimeSecurityBinding{{ OrgID: p4OrgID, AppInstallationID: p4InstallID, CallerWorkloadID: p4WorkloadID, Capabilities: p4Capabilities(), Active: true, }}, }); err != nil { t.Fatal(err) } } if err := securityRepository.ActivateProjection(context.Background(), 0, persistence.RuntimeSecurityProjection{ OrgID: orgID, RuntimeAudience: runtimeAudience, SecurityEpoch: securityEpoch, Bindings: []persistence.RuntimeSecurityBinding{{ OrgID: orgID, AppInstallationID: installationID, CallerWorkloadID: p4WorkloadID, Capabilities: capabilities, Active: installationActive, }}, }); err != nil { t.Fatal(err) } leaseExpiry := now.Add(5 * time.Minute) if scenario == "expired-lease" { leaseExpiry = now.Add(-time.Second) } if err := securityRepository.IssueLease(context.Background(), orgID, replicaID, securityEpoch, leaseExpiry); err != nil { t.Fatal(err) } if scenario == "delegation-revoked" { if err := securityRepository.RevokeDelegation(context.Background(), orgID, "delegation-p4", now.Add(5*time.Minute)); err != nil { t.Fatal(err) } } state, err := runtimeidentity.NewSecurityState(context.Background(), securityRepository, replicaID, time.Now) if err != nil { t.Fatal(err) } if scenario == "authority-partition" { replica, found, readErr := securityRepository.ReadReplicaSecurityContext(context.Background(), replicaID) if readErr != nil || !found { t.Fatalf("read authority-partition replica state: found=%v err=%v", found, readErr) } state, err = runtimeapi.NewRuntimeSecurityState(context.Background(), runtimeapi.RuntimeSecurityStateConfig{ Replica: func(context.Context) (runtimeapi.RuntimeReplicaSecurityState, bool, error) { return runtimeapi.RuntimeReplicaSecurityState{ OrgID: replica.OrgID, RuntimeAudience: replica.RuntimeAudience, SecurityEpoch: replica.SecurityEpoch, LeaseSecurityEpoch: replica.LeaseSecurityEpoch, LeaseExpiresAt: replica.LeaseExpiresAt, }, true, nil }, Admission: func(context.Context, delegationv1.Claims) (runtimeapi.RuntimeAdmissionSecurityState, error) { return runtimeapi.RuntimeAdmissionSecurityState{}, errors.New("runtime security authority unavailable") }, Now: time.Now, }) if err != nil { t.Fatal(err) } } sessionRepository := persistence.NewSessionRepository(db) authService := featureauth.NewAuthService(persistence.NewUserRepository(db), sessionRepository, nil) admission, err := runtimeapi.NewSecurityAdmission(runtimeapi.SecurityAdmissionConfig{ Verifier: verifier, State: state, Bindings: func(_ context.Context, caller runtimegrpc.VerifiedCaller) (runtimeapi.RuntimeCallerBinding, bool, error) { workloadID := p4WorkloadID if scenario == "workload-mismatch" { workloadID = "spiffe://example.test/ns/runtime/sa/other-client" } return runtimeapi.RuntimeCallerBinding{CallerWorkloadID: workloadID}, caller == p4WorkloadID, nil }, Sessions: runtimeidentity.SessionAuthority(authService), }) if err != nil { t.Fatal(err) } return admission } func p4DelegationToken(t *testing.T) string { t.Helper() now := time.Now() privateKey := ed25519.NewKeyFromSeed(bytes.Repeat([]byte{0x24}, ed25519.SeedSize)) token, err := delegationv1.Sign(p4SigningKeyID, privateKey, delegationv1.Claims{ DelegationID: "delegation-p4", OrgID: p4OrgID, PrincipalID: "integration-user", SessionID: p4SessionID, CallerWorkloadID: p4WorkloadID, AppInstallationID: p4InstallID, RuntimeAudience: p4Audience, Capabilities: p4Capabilities(), AuthEpoch: 1, SecurityEpoch: 1, IssuedAt: now.Add(-time.Second).Unix(), ExpiresAt: now.Add(5 * time.Minute).Unix(), }) if err != nil { t.Fatal(err) } return token } func p4AuthorizingSessionToken(t *testing.T) string { t.Helper() roleID := "integration-role" roleName := "Integration Role" token, _, err := kernelauth.GenerateTokenForSession( kernelauth.UserSession{ ID: "integration-user", Name: "Runtime User", Email: "runtime@example.test", ProfileId: "integration-profile", ProfileName: "Integration Profile", RoleId: &roleID, RoleName: &roleName, ConnectedAppId: "connected-app-p4", }, p4SessionID, time.Now(), time.Hour, ) if err != nil { t.Fatalf("mint P4 authorizing session: %v", err) } return token } func p4Capabilities() []string { return []string{accesssemantic.CapabilityName, querysemantic.CapabilityName, semantic.CapabilityName} } type p4CountingQueryRepository struct { ports.QueryRepository calls *atomic.Int64 } func (r *p4CountingQueryRepository) Find(ctx context.Context, schema *models.ObjectMetadata, request models.QueryRequest, result *domainrls.Result, fields []string) (*models.QueryResult, error) { r.calls.Add(1) return r.QueryRepository.Find(ctx, schema, request, result, fields) } func (r *p4CountingQueryRepository) RunAnalytics(ctx context.Context, schema *models.ObjectMetadata, request models.AnalyticsQuery, result *domainrls.Result) (interface{}, error) { r.calls.Add(1) return r.QueryRepository.RunAnalytics(ctx, schema, request, result) } type p4CountingRecordRepository struct { ports.RecordRepository calls *atomic.Int64 } func (r *p4CountingRecordRepository) FindMany(ctx context.Context, object string, ids []string) (map[string]models.SObject, error) { r.calls.Add(1) return r.RecordRepository.FindMany(ctx, object, ids) } func (r *p4CountingRecordRepository) FindObjectTypesForIDs(ctx context.Context, ids []string, objects []string) (map[string]string, error) { r.calls.Add(1) return r.RecordRepository.FindObjectTypesForIDs(ctx, ids, objects) } type p4MetadataRepository struct { ports.MetadataRepository schema *models.ObjectMetadata } func (r *p4MetadataRepository) GetAllSchemas(context.Context) ([]*models.ObjectMetadata, error) { return []*models.ObjectMetadata{r.schema}, nil } func (*p4MetadataRepository) GetAllRelationships(context.Context) ([]*models.Relationship, error) { return []*models.Relationship{}, nil } func (*p4MetadataRepository) GetSharingRules(context.Context, string) ([]*models.SharingRule, error) { return []*models.SharingRule{}, nil } func (r *p4MetadataRepository) GetSchemaByAPIName(_ context.Context, name string) (*models.ObjectMetadata, error) { if strings.EqualFold(name, r.schema.APIName) { return r.schema, nil } return nil, nil } type p4PermissionRepository struct { ports.PermissionRepository grantRead bool } func (*p4PermissionRepository) GetAllRoles(context.Context) ([]*models.UserRole, error) { return []*models.UserRole{}, nil } func (r *p4PermissionRepository) LoadEffectiveObjectPermission(context.Context, *models.UserSession, string) (*models.ObjectPermissions, error) { return &models.ObjectPermissions{PermissionsRead: r.grantRead, PermissionsEdit: r.grantRead, PermissionsDelete: r.grantRead}, nil } func (*p4PermissionRepository) LoadEffectiveFieldPermission(_ context.Context, _ *models.UserSession, _, field string) (*models.FieldPermissions, error) { return &models.FieldPermissions{PermissionsRead: !strings.EqualFold(field, "SecretNote")}, nil } func (*p4PermissionRepository) ListEffectiveFieldPermissionsForUserObject(_ context.Context, _, _, object string) ([]models.FieldPermissions, error) { if !strings.EqualFold(object, "Issue") { return nil, nil } fields := p4IssueSchema().Fields permissions := make([]models.FieldPermissions, 0, len(fields)) for _, field := range fields { permissions = append(permissions, models.FieldPermissions{ SobjectType: "Issue", Field: field.APIName, PermissionsRead: !strings.EqualFold(field.APIName, "SecretNote"), }) } return permissions, nil } func (*p4PermissionRepository) GetEffectiveSystemPermissions(context.Context, string, string) (map[string]bool, error) { return map[string]bool{}, nil } func (*p4PermissionRepository) GetUserGroupIDs(context.Context, string) ([]string, error) { return nil, nil } func (*p4PermissionRepository) GetManualShareRecordIDs(context.Context, string, string) ([]string, error) { return nil, nil } func (*p4PermissionRepository) GetTeamMemberRecordIDs(context.Context, string, string) ([]string, error) { return nil, nil } func (*p4PermissionRepository) GetGroupMemberIDs(context.Context, []string) ([]string, error) { return nil, nil } func (*p4PermissionRepository) IsUserInGroup(context.Context, string, string) (bool, error) { return false, nil } func (*p4PermissionRepository) GetManualShareAccessLevels(context.Context, string, string, string) ([]string, error) { return nil, nil } func (*p4PermissionRepository) GetTeamMemberAccessLevel(context.Context, string, string, string) (*string, error) { return nil, nil } func (*p4PermissionRepository) GetPendingApprovalLockInfo(context.Context, string) (string, string, error) { return "", "", nil } func (*p4PermissionRepository) HasPendingApprovalWorkitemForUser(context.Context, string, string) (bool, error) { return false, nil } func p4IssueSchema() *models.ObjectMetadata { return &models.ObjectMetadata{ APIName: "Issue", SharingModel: constants.SharingModelPrivate, Fields: []models.FieldMetadata{ {APIName: constants.FieldID, Type: constants.FieldTypeString, IsSystem: true}, {APIName: constants.FieldOwnerID, Type: constants.FieldTypeReference, IsSystem: true}, {APIName: "Archived", Type: constants.FieldTypeBoolean}, {APIName: "Disabled", Type: constants.FieldTypeBoolean}, {APIName: "DueDate", Type: constants.FieldTypeDate}, {APIName: "EmptySummary", Type: constants.FieldTypeString}, {APIName: "OptionalValue", Type: constants.FieldTypeString}, {APIName: "Payload", Type: constants.FieldTypeJson}, {APIName: "ProjectId", Type: constants.FieldTypeReference}, {APIName: "RetryCount", Type: constants.SchemaFieldType("integer")}, {APIName: "SecretNote", Type: constants.FieldTypeString}, {APIName: "StoryPoints", Type: constants.FieldTypeDouble}, {APIName: "Subject", Type: constants.FieldTypeString}, {APIName: "UpdatedAt", Type: constants.FieldTypeDatetime}, {APIName: "ZeroCount", Type: constants.SchemaFieldType("integer")}, {APIName: "ZeroEstimate", Type: constants.FieldTypeDouble}, }, } } func createP4Dataset(t *testing.T, db *sql.DB, securityScenario string) { t.Helper() for _, statement := range []string{ `CREATE TABLE User (Id VARCHAR(255) PRIMARY KEY, Username VARCHAR(255) NOT NULL, Email VARCHAR(255) NOT NULL, ProfileId VARCHAR(255) NOT NULL, FirstName VARCHAR(255) NULL, LastName VARCHAR(255) NULL, IsActive BOOL NOT NULL, IsDeleted BOOL NOT NULL DEFAULT 0, FailedLoginAttempts INT NOT NULL DEFAULT 0, LockedUntil DATETIME NULL, UserRoleId VARCHAR(255) NULL)`, `CREATE TABLE _System_Session (Id VARCHAR(255) PRIMARY KEY, UserId VARCHAR(255) NOT NULL, Token VARBINARY(3000) NOT NULL, CurrentBearerToken VARBINARY(3000) NULL, ExpiresAt DATETIME NOT NULL, IpAddress VARCHAR(255), UserAgent VARCHAR(255), IsRevoked BOOL NOT NULL, AuthEpoch DECIMAL(20,0) NOT NULL, ControlProofHash CHAR(64) NULL, LastActivity DATETIME NOT NULL, Scope TEXT NULL, ConnectedAppId VARCHAR(255) NULL, GrantType VARCHAR(50) NULL, CreatedDate DATETIME NOT NULL, LastModifiedDate DATETIME NOT NULL)`, `CREATE TABLE _System_ConnectedApp (Id VARCHAR(255) PRIMARY KEY, IsActive BOOL NOT NULL)`, `CREATE TABLE _System_ConnectedAppDeclaration (ConnectedAppId VARCHAR(255), InstalledPackName VARCHAR(255), IsDeleted BOOL NOT NULL DEFAULT 0)`, `CREATE TABLE _System_OAuthDynamicClientRegistration (ConnectedAppId VARCHAR(255), ScopeOwnerPack VARCHAR(255), IsDeleted BOOL NOT NULL DEFAULT 0)`, `CREATE TABLE _System_PermissionStateVersion (SingletonKey TINYINT PRIMARY KEY, Lineage BINARY(16) NOT NULL, Revision DECIMAL(20,0) NOT NULL, EnforcementMode VARCHAR(16) NOT NULL, CreatedDate DATETIME(6) NOT NULL, LastModifiedDate DATETIME(6) NOT NULL)`, `CREATE TABLE _System_RuntimeSecurityState (OrgId VARCHAR(255) PRIMARY KEY, RuntimeAudience VARCHAR(255) NOT NULL, SecurityEpoch DECIMAL(20,0) NOT NULL, CreatedDate DATETIME(6) NOT NULL, LastModifiedDate DATETIME(6) NOT NULL)`, `CREATE TABLE _System_RuntimeSecurityBinding (Id VARCHAR(255) PRIMARY KEY, OrgId VARCHAR(255) NOT NULL, AppInstallationId VARCHAR(255) NOT NULL, CallerWorkloadId VARCHAR(1024) NOT NULL, CallerWorkloadDigest CHAR(64) NOT NULL, Capabilities JSON NOT NULL, IsActive BOOL NOT NULL, CreatedDate DATETIME(6) NOT NULL, LastModifiedDate DATETIME(6) NOT NULL, INDEX idx_installation (OrgId, AppInstallationId, CallerWorkloadDigest))`, `CREATE TABLE _System_RuntimeSecurityLease (RuntimeReplicaId VARCHAR(255) PRIMARY KEY, OrgId VARCHAR(255) NOT NULL, SecurityEpoch DECIMAL(20,0) NOT NULL, ExpiresAt DATETIME(6) NOT NULL, CreatedDate DATETIME(6) NOT NULL, LastModifiedDate DATETIME(6) NOT NULL)`, `CREATE TABLE _System_RuntimeDelegationRevocation (Id VARCHAR(255) PRIMARY KEY, OrgId VARCHAR(255) NOT NULL, DelegationId VARCHAR(255) NOT NULL, ExpiresAt DATETIME(6) NOT NULL, CreatedDate DATETIME(6) NOT NULL, LastModifiedDate DATETIME(6) NOT NULL, UNIQUE KEY ux_org_delegation (OrgId, DelegationId))`, } { if _, err := db.Exec(statement); err != nil { t.Fatal(err) } } now := time.Now().UTC() if _, err := db.Exec(`INSERT INTO User (Id, Username, Email, ProfileId, FirstName, LastName, IsActive, FailedLoginAttempts, UserRoleId) VALUES ('integration-user', 'runtime-user', 'runtime@example.test', 'integration-profile', 'Runtime', 'User', 1, 0, 'integration-role')`); err != nil { t.Fatal(err) } if _, err := db.Exec(`INSERT INTO _System_ConnectedApp (Id, IsActive) VALUES ('connected-app-p4', 1)`); err != nil { t.Fatal(err) } sessionToken := os.Getenv(p4SessionTokenEnv) if sessionToken == "" { t.Fatal("Runtime child did not receive the parent-issued authorizing session token") } connectedAppID := "connected-app-p4" if err := persistence.NewSessionRepository(db).InsertSession(context.Background(), &models.SystemSession{ ID: p4SessionID, UserID: "integration-user", Token: sessionToken, ExpiresAt: now.Add(time.Hour), LastActivity: now, IPAddress: "127.0.0.1", UserAgent: "integration", AuthEpoch: 1, ConnectedAppID: &connectedAppID, }); err != nil { t.Fatal(err) } if _, err := db.Exec( `INSERT INTO _System_PermissionStateVersion (SingletonKey, Lineage, Revision, EnforcementMode, CreatedDate, LastModifiedDate) VALUES (1, ?, 1, 'dark', ?, ?)`, []byte{1, 2, 3, 4, 5, 6, 7, 8, 9, 10, 11, 12, 13, 14, 15, 16}, now, now, ); err != nil { t.Fatal(err) } switch securityScenario { case "session-revoked": if _, err := db.Exec(`UPDATE _System_Session SET IsRevoked = 1 WHERE Id = ?`, p4SessionID); err != nil { t.Fatal(err) } case "user-inactive": if _, err := db.Exec(`UPDATE User SET IsActive = 0 WHERE Id = 'integration-user'`); err != nil { t.Fatal(err) } case "old-auth-epoch": if _, err := db.Exec(`UPDATE _System_Session SET AuthEpoch = 2 WHERE Id = ?`, p4SessionID); err != nil { t.Fatal(err) } case "connected-app-disabled": if _, err := db.Exec(`UPDATE _System_ConnectedApp SET IsActive = 0 WHERE Id = 'connected-app-p4'`); err != nil { t.Fatal(err) } } if _, err := db.Exec(`CREATE TABLE Issue ( Id VARCHAR(255) PRIMARY KEY, OwnerId VARCHAR(255) NOT NULL, Subject VARCHAR(255), EmptySummary VARCHAR(255), Archived BOOLEAN, Disabled BOOLEAN, RetryCount BIGINT, ZeroCount BIGINT, StoryPoints DECIMAL(30,2), ZeroEstimate DECIMAL(30,2), DueDate DATE, UpdatedAt DATETIME(6), ProjectId VARCHAR(255), OptionalValue VARCHAR(255), SecretNote VARCHAR(255), Payload JSON )`); err != nil { t.Fatal(err) } for _, row := range []struct{ id, owner, subject string }{ {id: p4RecordID, owner: "integration-user", subject: "Non-CRM point read"}, {id: p4RLSRecordID, owner: "different-owner", subject: "masked"}, } { if _, err := db.Exec(`INSERT INTO Issue (Id, OwnerId, Subject, EmptySummary, Archived, Disabled, RetryCount, ZeroCount, StoryPoints, ZeroEstimate, DueDate, UpdatedAt, ProjectId, OptionalValue, SecretNote, Payload) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)`, row.id, row.owner, row.subject, "", true, false, 7, 0, "12.50", "0.00", "2026-07-12", "2026-07-12 09:10:11.123456", "PRJ-001", nil, "hidden", `{"a-b":false,"attributes":{"type":"ordinary-value"},"empty":"","nil":null,"zero":0}`); err != nil { t.Fatal(err) } } } func createD2AQueryDataset(t *testing.T, db *sql.DB) { t.Helper() for index := 0; index < 200; index++ { if _, err := db.Exec(`INSERT INTO Issue (Id, OwnerId, Subject, Archived, Disabled, RetryCount, ZeroCount) VALUES (?, 'integration-user', ?, 0, 0, -1, 0)`, fmt.Sprintf("ISS-PAGE-%03d", index), fmt.Sprintf("pagination row %03d", index)); err != nil { t.Fatal(err) } } for _, statement := range []string{ `CREATE TABLE _System_Object (Id VARCHAR(255) PRIMARY KEY, ApiName VARCHAR(255) NOT NULL, IsDeleted BOOL NOT NULL DEFAULT 0)`, `CREATE TABLE _System_Field (Id VARCHAR(255) PRIMARY KEY, ObjectId VARCHAR(255) NOT NULL, ApiName VARCHAR(255) NOT NULL, IsDeleted BOOL NOT NULL DEFAULT 0)`, `CREATE TABLE _System_FieldHistoryTrackingInterval (Id VARCHAR(255) PRIMARY KEY, FieldId VARCHAR(255) NOT NULL, StartedSequence BIGINT NOT NULL, EndedSequence BIGINT NULL)`, `CREATE TABLE _System_AuditLog (Id VARCHAR(255) PRIMARY KEY, ObjectApiName VARCHAR(255) NOT NULL, RecordId VARCHAR(255) NOT NULL, FieldName VARCHAR(255) NOT NULL, OldValue LONGTEXT NULL, NewValue LONGTEXT NULL, ChangedById VARCHAR(255) NOT NULL, ChangedAt DATETIME(6) NOT NULL, HistorySequence BIGINT NULL, FieldHistoryEligible TINYINT(1) NULL)`, `INSERT INTO _System_Object (Id, ApiName) VALUES ('object-issue', 'Issue')`, `INSERT INTO _System_Field (Id, ObjectId, ApiName) VALUES ('field-subject', 'object-issue', 'Subject'), ('field-secret', 'object-issue', 'SecretNote')`, `INSERT INTO _System_FieldHistoryTrackingInterval (Id, FieldId, StartedSequence) VALUES ('interval-subject', 'field-subject', 10), ('interval-secret', 'field-secret', 20)`, `INSERT INTO _System_AuditLog (Id, ObjectApiName, RecordId, FieldName, OldValue, NewValue, ChangedById, ChangedAt, HistorySequence) VALUES ('history-subject', 'Issue', 'ISS-P4-1', 'Subject', 'before', 'Non-CRM point read', 'integration-user', '2026-07-12 09:10:12.000000', 11), ('history-secret', 'Issue', 'ISS-P4-1', 'SecretNote', 'private-before', 'private-after', 'integration-user', '2026-07-12 09:10:13.000000', 21)`, } { if _, err := db.Exec(statement); err != nil { t.Fatal(err) } } } func openP4ScratchDB(t *testing.T) (*sql.DB, func()) { t.Helper() config := mysql.NewConfig() config.Net = "tcp" config.Addr = net.JoinHostPort(os.Getenv("TIDB_HOST"), p4EnvOr("TIDB_PORT", "4000")) config.User = p4EnvOr("TIDB_USER", "root") config.Passwd = os.Getenv("TIDB_PASSWORD") config.ParseTime = true admin, err := sql.Open("mysql", config.FormatDSN()) if err != nil { t.Fatal(err) } name := fmt.Sprintf("nexuscrm_runtime_p4_%d", time.Now().UnixNano()) if _, err := admin.Exec("CREATE DATABASE `" + name + "`"); err != nil { _ = admin.Close() t.Fatal(err) } config.DBName = name db, err := sql.Open("mysql", config.FormatDSN()) if err != nil { t.Fatal(err) } if err := db.Ping(); err != nil { t.Fatal(err) } return db, func() { _ = db.Close() _, _ = admin.Exec("DROP DATABASE IF EXISTS `" + name + "`") _ = admin.Close() } } func p4EnvOr(key, fallback string) string { if value := os.Getenv(key); value != "" { return value } return fallback } func p4ChildEnvironment(base []string, assignments ...string) []string { owned := map[string]bool{ p4ChildModeEnv: true, p4ReadyFileEnv: true, p4ResultFileEnv: true, p4GrantReadEnv: true, p4SecurityScenarioEnv: true, p4SessionTokenEnv: true, } result := make([]string, 0, len(base)+len(assignments)) for _, entry := range base { key, _, _ := strings.Cut(entry, "=") if !owned[key] { result = append(result, entry) } } return append(result, assignments...) } func writeP4JSON(t *testing.T, path string, value any) { t.Helper() data, err := json.Marshal(value) if err != nil { t.Fatal(err) } if err := os.WriteFile(path, data, 0o600); err != nil { t.Fatal(err) } }