diff --git a/module/CALLABLE_SOURCES.md b/module/CALLABLE_SOURCES.md index ee45981b6..63f208f4e 100644 --- a/module/CALLABLE_SOURCES.md +++ b/module/CALLABLE_SOURCES.md @@ -17,6 +17,18 @@ Released CLI v0.1.185 uses it to derive packages only on Connect. Discovery projects individual declaration schemas/effects/digests over this one binding. The caller never chooses a destination, HTTP header, credential or method. +`ListSourceOperations` reads the current declarations through the same Connect +endpoint. Interactive readers retain the membership and source-boundary gates. +A Work Context reader presents an idempotent `saas-datasource` capability with +an explicit `datasource.sources:read` scope naming the requested source. The +host verifies the capability and current authority on every request, then checks +its tenant, exact source scope, any borrowed-source delegation, membership, +personal ownership and source-boundary read permission before reading declarations. +An invoke-only scope or an unconstrained source kind cannot list declarations. +The result includes each declaration's schemas, effect, bound and canonical digest; +it grants no invocation authority and is not itself a model-facing tool projection. +Consumers must freeze the admitted declaration and separately authorize execution. + Invoke takes `org_id`, `source_id`, `operation`, `input_json`, and `effect_id`. `input_json` is a string containing exactly one JSON object. The complete protojson request (proto field names, unpopulated fields included) is limited to diff --git a/module/services/accounts/code/pkg/adapters/source_operation_authority.go b/module/services/accounts/code/pkg/adapters/source_operation_authority.go index 50542c479..e6b4c7b56 100644 --- a/module/services/accounts/code/pkg/adapters/source_operation_authority.go +++ b/module/services/accounts/code/pkg/adapters/source_operation_authority.go @@ -33,6 +33,7 @@ type sourceOperationDelegationKey struct{} func sourceOperationProcedure(method string) bool { switch method { case business.SourceOperationMethod, + "/saas.accounts.v1.DatasourceService/ListSourceOperations", "/saas.accounts.v1.DatasourceService/LookupInvokeSourceOperation", business.SourceReceiptRetentionMethod, "/saas.accounts.v1.DatasourceService/LookupPruneSourceOperationReceipts", diff --git a/module/services/accounts/code/pkg/adapters/source_operation_discovery_test.go b/module/services/accounts/code/pkg/adapters/source_operation_discovery_test.go new file mode 100644 index 000000000..3ca99c4a7 --- /dev/null +++ b/module/services/accounts/code/pkg/adapters/source_operation_discovery_test.go @@ -0,0 +1,281 @@ +package adapters + +import ( + "context" + "net/http" + "net/http/httptest" + "sync" + "sync/atomic" + "testing" + "time" + + "accounts/pkg/auth" + "accounts/pkg/business" + "accounts/pkg/datasource/connector/connectortest" + "accounts/pkg/datasource/operations" + gen "accounts/pkg/gen/saas/accounts/v1" + "accounts/pkg/gen/saas/accounts/v1/accountsv1connect" + + "connectrpc.com/connect" + basev0 "github.com/codefly-dev/core/generated/go/codefly/base/v0" + "github.com/codefly-dev/core/workcontext" + "github.com/stretchr/testify/require" + "google.golang.org/protobuf/proto" +) + +const ( + discoveryOrg = "11111111-1111-4111-8111-111111111111" + discoveryOwner = "22222222-2222-4222-8222-222222222222" + discoverySource = "33333333-3333-4333-8333-333333333333" + discoveryOther = "44444444-4444-4444-8444-444444444444" +) + +// The transport and signed capability verification are real. These fixtures +// replace storage, current authority facts and seals; no database or provider +// is exercised and this is not a composed consuming application. +type sourceDiscoveryStore struct { + sourceWireStore + mu sync.Mutex + readable bool + member bool + declarationReads atomic.Int64 +} + +func (s *sourceDiscoveryStore) GetOrgMembership(context.Context, string, string) (*gen.OrgMembership, error) { + s.mu.Lock() + defer s.mu.Unlock() + if !s.member { + return nil, nil + } + return &gen.OrgMembership{Role: gen.OrgRole_ORG_ROLE_MEMBER}, nil +} + +func (s *sourceDiscoveryStore) CanReadScopeNode(ctx context.Context, org, actor string, _ gen.SubjectKind, _, _, _ string) (bool, error) { + if err := auth.RequireVerifiedDatabaseScope(ctx, org, actor); err != nil { + return false, err + } + s.mu.Lock() + defer s.mu.Unlock() + return s.readable, nil +} + +func (s *sourceDiscoveryStore) ListSourceOperations(ctx context.Context, org, source string) ([]operations.Declaration, error) { + s.declarationReads.Add(1) + return s.sourceWireStore.ListSourceOperations(ctx, org, source) +} + +func sourceDiscoveryFixture(t *testing.T) (*sourceDiscoveryStore, *sealAnswersFake, accountsv1connect.DatasourceServiceClient, func(testCapability) string) { + t.Helper() + declaration, err := operations.Admit(connectortest.OperationDeclaration()) + require.NoError(t, err) + store := &sourceDiscoveryStore{member: true, readable: true, sourceWireStore: sourceWireStore{ + declaration: declaration, + source: &business.DatasourceSource{ID: discoverySource, OrgID: discoveryOrg, BoundaryNodeID: "boundary"}, + }} + svc, err := business.NewService(store) + require.NoError(t, err) + previousService, previousAuthority, previousCache := service, workContextSingleton, orgMembershipCache + t.Cleanup(func() { + service, workContextSingleton, orgMembershipCache = previousService, previousAuthority, previousCache + }) + service, orgMembershipCache = svc, nil + seals := newSealAnswersFake(1) + workContextSingleton = newCoreTestServer(t, sourceReadIdentityAuthority{&workContextAuthorityFake{ + facts: &business.WorkContextAuthorityFacts{OrganizationRevision: 1}, + }}, seals) + mux := http.NewServeMux() + path, handler := accountsv1connect.NewDatasourceServiceHandler(&datasourceConnectHandler{svc: svc}, connect.WithInterceptors(connectAuthInterceptor(nil))) + mux.Handle(path, handler) + server := httptest.NewServer(mux) + t.Cleanup(server.Close) + client := accountsv1connect.NewDatasourceServiceClient(server.Client(), server.URL) + mint := func(c testCapability) string { + if c.Audience == "" { + c.Audience = business.SourceOperationAudience + } + if c.Tenant == "" { + c.Tenant = discoveryOrg + } + if c.Owner == "" { + c.Owner = discoveryOwner + } + if c.Scopes == nil { + c.Scopes = []*basev0.WorkScopeV1{{ResourceKind: "datasource.sources", Actions: []string{"read"}, ResourceIds: []string{discoverySource}}} + } + token, _ := mintTestCapability(t, workContextSingleton, c) + return token + } + return store, seals, client, mint +} + +func discoveryRequest(token string) *connect.Request[gen.ListSourceOperationsRequest] { + req := connect.NewRequest(&gen.ListSourceOperationsRequest{OrgId: discoveryOrg, SourceId: discoverySource}) + if token != "" { + req.Header().Set(workContextHeader, token) + } + return req +} + +func TestSourceOperationDiscoveryConnectAuthority(t *testing.T) { + for _, tc := range []struct { + name string + capability testCapability + change func(*sourceDiscoveryStore, *connect.Request[gen.ListSourceOperationsRequest]) + code connect.Code + }{ + {name: "exact read"}, + {name: "another source", change: func(_ *sourceDiscoveryStore, r *connect.Request[gen.ListSourceOperationsRequest]) { + r.Msg.SourceId = discoveryOther + }, code: connect.CodePermissionDenied}, + {name: "another tenant", change: func(_ *sourceDiscoveryStore, r *connect.Request[gen.ListSourceOperationsRequest]) { + r.Msg.OrgId = discoveryOther + }, code: connect.CodePermissionDenied}, + {name: "invoke is not read", capability: testCapability{Scopes: []*basev0.WorkScopeV1{{ResourceKind: "datasource.sources", Actions: []string{"invoke"}, ResourceIds: []string{discoverySource}}}}, code: connect.CodePermissionDenied}, + {name: "empty scopes", capability: testCapability{Scopes: []*basev0.WorkScopeV1{}}, code: connect.CodePermissionDenied}, + {name: "unrelated scopes", capability: testCapability{Scopes: []*basev0.WorkScopeV1{{ResourceKind: "example.items", Actions: []string{"read"}, ResourceIds: []string{discoverySource}}}}, code: connect.CodePermissionDenied}, + {name: "unconstrained source", capability: testCapability{Scopes: []*basev0.WorkScopeV1{{ResourceKind: "datasource.sources", Actions: []string{"read"}}}}, code: connect.CodePermissionDenied}, + {name: "wrong audience", capability: testCapability{Audience: "example"}, code: connect.CodeUnauthenticated}, + {name: "module identity is not source authority", capability: testCapability{Audience: ModuleWorkContextAudience, Owner: "module:example", OwnerKind: business.PrincipalKindService, Scopes: []*basev0.WorkScopeV1{}}, code: connect.CodeUnauthenticated}, + {name: "expired credential", change: func(_ *sourceDiscoveryStore, _ *connect.Request[gen.ListSourceOperationsRequest]) { + workContextSingleton.now = func() time.Time { return time.Now().Add(time.Hour) } + }, code: connect.CodeUnauthenticated}, + {name: "single use", capability: testCapability{Replay: workcontext.ReplaySingleUse}, code: connect.CodePermissionDenied}, + {name: "no membership", change: func(s *sourceDiscoveryStore, _ *connect.Request[gen.ListSourceOperationsRequest]) { s.member = false }, code: connect.CodePermissionDenied}, + {name: "no source access", change: func(s *sourceDiscoveryStore, _ *connect.Request[gen.ListSourceOperationsRequest]) { s.readable = false }, code: connect.CodePermissionDenied}, + {name: "other person's source", change: func(s *sourceDiscoveryStore, _ *connect.Request[gen.ListSourceOperationsRequest]) { + s.source.PersonalOwnerUserID = discoveryOther + }, code: connect.CodePermissionDenied}, + {name: "invalid credential", change: func(_ *sourceDiscoveryStore, r *connect.Request[gen.ListSourceOperationsRequest]) { + r.Header().Set(workContextHeader, "invalid") + }, code: connect.CodeUnauthenticated}, + {name: "duplicate credential", change: func(_ *sourceDiscoveryStore, r *connect.Request[gen.ListSourceOperationsRequest]) { + r.Header().Add(workContextHeader, r.Header().Get(workContextHeader)) + }, code: connect.CodeUnauthenticated}, + {name: "missing credential with forged identity", change: func(_ *sourceDiscoveryStore, r *connect.Request[gen.ListSourceOperationsRequest]) { + r.Header().Del(workContextHeader) + r.Header().Set("X-User-Id", discoveryOwner) + r.Header().Set("X-Org-Id", discoveryOrg) + }, code: connect.CodeUnauthenticated}, + } { + t.Run(tc.name, func(t *testing.T) { + store, _, client, mint := sourceDiscoveryFixture(t) + req := discoveryRequest(mint(tc.capability)) + if tc.change != nil { + tc.change(store, req) + } + out, err := client.ListSourceOperations(t.Context(), req) + if tc.code != 0 { + require.Equal(t, tc.code, connect.CodeOf(err), "%v", err) + require.Nil(t, out) + require.Zero(t, store.declarationReads.Load(), "refused request read declarations") + return + } + require.NoError(t, err) + require.Len(t, out.Msg.Operations, 1) + expected, err := declarationToProto(store.declaration) + require.NoError(t, err) + require.True(t, proto.Equal(expected, out.Msg.Operations[0]), "discovery must preserve the admitted declaration") + require.EqualValues(t, 1, store.declarationReads.Load()) + }) + } +} + +func TestSourceOperationDiscoveryRechecksAfterSuccessfulRead(t *testing.T) { + for _, revoke := range []string{"source permission", "membership", "capability revision"} { + t.Run(revoke, func(t *testing.T) { + store, seals, client, mint := sourceDiscoveryFixture(t) + token := mint(testCapability{}) + _, err := client.ListSourceOperations(t.Context(), discoveryRequest(token)) + require.NoError(t, err) + switch revoke { + case "source permission": + store.mu.Lock() + store.readable = false + store.mu.Unlock() + case "membership": + store.mu.Lock() + store.member = false + store.mu.Unlock() + default: + seals.setRevision(2) + } + _, err = client.ListSourceOperations(t.Context(), discoveryRequest(token)) + require.Error(t, err, "a prior listing must convey no continuing authority") + require.EqualValues(t, 1, store.declarationReads.Load()) + }) + } +} + +func TestSourceOperationDiscoveryHandlerUsesFinalActorScope(t *testing.T) { + store, _, _, _ := sourceDiscoveryFixture(t) + ownerScope := &basev0.WorkScopeV1{ResourceKind: "datasource.sources", Actions: []string{"read"}, ResourceIds: []string{discoverySource, discoveryOther}} + actorScope := &basev0.WorkScopeV1{ResourceKind: "datasource.sources", Actions: []string{"read"}, ResourceIds: []string{discoverySource}} + _, claims := mintTestCapability(t, workContextSingleton, testCapability{Audience: business.SourceOperationAudience, Tenant: discoveryOrg, Owner: discoveryOwner, + Scopes: []*basev0.WorkScopeV1{ownerScope}, Actors: []delegationHop{agentHop("example-agent", "example-delegation", actorScope)}}) + // The handler receives verified identity/claims at its trust seam. Actor + // admission is outside this test; it must still honor the final attenuation. + ctx := stampVerifiedIdentity(t.Context(), discoveryOwner, discoveryOrg, auth.Assurance{}) + ctx = context.WithValue(ctx, sourceOperationContextKey{}, claims) + req := discoveryRequest("") + req.Msg.SourceId = discoveryOther + _, err := (&datasourceConnectHandler{svc: service}).ListSourceOperations(ctx, req) + require.Equal(t, connect.CodePermissionDenied, connect.CodeOf(err)) + require.Zero(t, store.declarationReads.Load()) + req.Msg.SourceId = discoverySource + _, err = (&datasourceConnectHandler{svc: service}).ListSourceOperations(ctx, req) + require.NoError(t, err) +} + +func TestSourceOperationDiscoveryInteractiveReadRemainsAvailable(t *testing.T) { + store, _, _, _ := sourceDiscoveryFixture(t) + ctx := stampVerifiedIdentity(t.Context(), discoveryOwner, discoveryOrg, auth.Assurance{}) + out, err := (&datasourceConnectHandler{svc: service}).ListSourceOperations(ctx, discoveryRequest("")) + require.NoError(t, err) + require.Len(t, out.Msg.Operations, 1) + require.EqualValues(t, 1, store.declarationReads.Load()) +} + +func TestSourceOperationDiscoveryHandlerAttenuatesVerifiedContext(t *testing.T) { + store, _, _, mint := sourceDiscoveryFixture(t) + headers := http.Header{} + headers.Set(workContextHeader, mint(testCapability{})) + ctx, err := authenticateSourceOperationContext(t.Context(), headers) + require.NoError(t, err) + req := discoveryRequest("") + req.Msg.SourceId = discoveryOther + _, err = (&datasourceConnectHandler{svc: service}).ListSourceOperations(ctx, req) + require.Equal(t, connect.CodePermissionDenied, connect.CodeOf(err)) + require.Zero(t, store.declarationReads.Load()) +} + +func TestSourceOperationDiscoveryHandlerBindsBorrowedSource(t *testing.T) { + // The binding legitimately names two sources, but the signed capability + // borrows authority from only one. The production verifier must retain that + // distinction for the read handler rather than relying on scope IDs alone. + principals := `{"docstore":{"workload":{"service_account":"module","namespace":"acme-prod","container":"app"},` + + `"tenant":"` + moduleWorkContextTenant + `","cross_tenant":true,"operation_audiences":{` + + `"source-sync":{"audience":"` + business.SourceOperationAudience + `",` + + `"invoke_scopes":[{"resource_kind":"datasource.sources","actions":["invoke","read"],"resource_ids":["` + discoverySource + `","` + discoveryOther + `"]}],` + + `"source_delegation_scopes":[{"resource_kind":"datasource.sources","actions":["invoke","read"],"resource_ids":["` + discoverySource + `","` + discoveryOther + `"]}]}}}}` + installDelegationServiceWith(t, principals, discoverySource) + minted, err := mintSourceContext("docstore", moduleToken(t, "docstore"), business.SourceDelegationRef{SourceID: discoverySource}) + require.NoError(t, err) + claims := verifySourceContext2(t, minted.GetToken()) + require.ElementsMatch(t, []string{discoverySource, discoveryOther}, claims.GetAuthorityScopes()[0].GetResourceIds()) + headers := http.Header{} + headers.Set(workContextHeader, minted.GetToken()) + ctx, err := authenticateSourceOperationContext(t.Context(), headers) + require.NoError(t, err) + // Only storage is replaced after production authentication established the + // borrowed-source context. No hand-stamped source delegation key is used. + store, _, _, _ := sourceDiscoveryFixture(t) + store.source.OrgID = delegationOrg + req := discoveryRequest("") + req.Msg.OrgId = delegationOrg + _, err = (&datasourceConnectHandler{svc: service}).ListSourceOperations(ctx, req) + require.NoError(t, err) + req.Msg.SourceId = discoveryOther + _, err = (&datasourceConnectHandler{svc: service}).ListSourceOperations(ctx, req) + require.Equal(t, connect.CodePermissionDenied, connect.CodeOf(err)) + require.EqualValues(t, 1, store.declarationReads.Load()) +} diff --git a/module/services/accounts/code/pkg/adapters/source_operations.go b/module/services/accounts/code/pkg/adapters/source_operations.go index 82ce62c74..3d50216b5 100644 --- a/module/services/accounts/code/pkg/adapters/source_operations.go +++ b/module/services/accounts/code/pkg/adapters/source_operations.go @@ -95,6 +95,9 @@ func (h *datasourceConnectHandler) ListSourceOperations(ctx context.Context, req if err != nil { return nil, err } + if err := requireSourceOperationScope(ctx, req.Msg.OrgId, req.Msg.SourceId, "read"); err != nil { + return nil, err + } if err := requireOrgMember(ctx, actor, req.Msg.OrgId); err != nil { return nil, translateGRPCError(err) }