diff --git a/README.md b/README.md index ae2f27a..fcd2e7b 100644 --- a/README.md +++ b/README.md @@ -207,11 +207,8 @@ a `scope` among them, is refused 400. verified task id), which a host retry of the same attempt keeps. The rest of the consent receipt (its intent, the delivered generation and the offer digest) is the authority's to check, through the step's signed binding and - its current state. Nothing else is passed on. -- **The step holds nothing.** It is given no Gateway (`ViewerFromContext` - errors inside it), and the setup credential is never handed to it or - forwarded. A pending installation is active nowhere, so there is nothing to - act under until it is. + its current state. The callback also carries a private attempt handle for declared setup calls. +- **Setup calls are explicit.** `OnInstall(step, SetupModule{As: ..., Authority: ..., Methods: ...})` declares exact unary RPCs. `installation.Call(ctx, alias, method, request, response)` exchanges the verified callback credential through Authority for that one setup method. Composition must declare the same setup binding/method. No Gateway or ordinary pending authority is granted; the callback credential is private and never sent to the destination. Calls expire with the callback and cannot be used from a retained handle. - **One attempt runs once per process.** Concurrent deliveries of one (installation, revision, consent operation) wait for the first, and an acknowledgement is answered from memory for the most recent 1024 attempts. diff --git a/installation.go b/installation.go index 5958079..0a78416 100644 --- a/installation.go +++ b/installation.go @@ -31,7 +31,8 @@ import ( // // The step is given no Gateway. The installation is pending, so nothing it // holds is active yet, and the setup credential grants delivery of the step -// and nothing else: it is never handed to the step and never forwarded. +// and explicitly declared setup methods, issued separately by Authority. The +// callback credential is never exposed to solution code or sent to a module. // Installation is one pending installation of this solution, as the authority // verified the step for it. @@ -44,6 +45,7 @@ type Installation struct { // whose intent, delivered generation and digest the authority checks. A // host retry of the same attempt carries the same operation. OperationID string + setup *installationSetup } // ErrInstallRefused marks an intentional install refusal. Use RefuseInstall @@ -75,8 +77,9 @@ func (*installRefusal) Unwrap() error { return ErrInstallRefused } // A solution that declares no step is not installable: the step is refused // rather than acknowledged, because an acknowledgement is what activates the // installation, and nothing here set it up. -func (s *Server) OnInstall(step func(context.Context, Installation) error) *Server { +func (s *Server) OnInstall(step func(context.Context, Installation) error, modules ...SetupModule) *Server { s.installStep = step + s.setupModules = append([]SetupModule(nil), modules...) return s } @@ -119,7 +122,10 @@ func (s *Server) handleInstallation(w http.ResponseWriter, r *http.Request) { return } installation.OperationID = verified.GetTaskId() - s.installOnce(r.Context(), installation).write(w) + ctx, cancel := context.WithCancel(r.Context()) + defer cancel() + installation.setup = &installationSetup{ctx: ctx, server: s, token: tokens[0]} + s.installOnce(ctx, installation).write(w) } // decodeInstallation bounds the entire body, including whitespace after the diff --git a/installation_setup.go b/installation_setup.go new file mode 100644 index 0000000..54087cc --- /dev/null +++ b/installation_setup.go @@ -0,0 +1,95 @@ +package solution + +import ( + "context" + "fmt" + "strings" + + "google.golang.org/protobuf/proto" +) + +// SetupModule names exact unary methods used only during OnInstall. Authority +// derives their grants from the delivered setup declaration and callee policy. +// These methods are not exposed by the browser passthrough. +type SetupModule struct { + As string + Authority AuthorityRelay + Methods []string +} + +type installationSetup struct { + ctx context.Context + server *Server + token string +} + +// Call invokes a declared setup method for this verified pending attempt. The +// authority issues a fresh, method-confined capability for every call. A retained +// Installation cannot call after its callback has returned. +func (i Installation) Call(ctx context.Context, as, method string, request, response proto.Message) error { + if i.setup == nil { + return fmt.Errorf("no verified installation attempt") + } + a := i.setup + if err := a.ctx.Err(); err != nil { + return err + } + declared := false + for _, module := range a.server.setupModules { + if module.As != as { + continue + } + for _, allowed := range module.Methods { + if allowed == method { + declared = true + } + } + } + if !declared { + return fmt.Errorf("setup method %q is not declared for %q", method, as) + } + relay := a.server.setupRelays[as] + if relay == nil { + return fmt.Errorf("setup module %q is not connected", as) + } + md, err := methodDescriptor(method) + if err != nil { + return err + } + if request == nil || response == nil || !request.ProtoReflect().IsValid() || !response.ProtoReflect().IsValid() || request.ProtoReflect().Descriptor().FullName() != md.Input().FullName() || response.ProtoReflect().Descriptor().FullName() != md.Output().FullName() { + return fmt.Errorf("setup method %q requires its declared request and response types", method) + } + callCtx, cancel := context.WithCancel(ctx) + stop := context.AfterFunc(a.ctx, cancel) + defer stop() + defer cancel() + issued, err := a.server.host.outbound.IssueForInstallationSetup(callCtx, a.token, relay.binding, method) + if err != nil { + return err + } + return relay.conn.Invoke(issued.Carry(callCtx), method, request, response) +} + +func (s *Server) openSetupRelays(lookup func(string, string) (string, error)) (map[string]*moduleRelay, error) { + modules := make([]ConsumedModule, 0, len(s.setupModules)) + seen := map[string]bool{} + for _, m := range s.setupModules { + a := m.Authority + if m.As == "" || seen[m.As] || a.Group == "" || a.EndpointKey == "" || a.BindingKey == "" || len(m.Methods) == 0 { + return nil, fmt.Errorf("setup module %q requires a unique alias, authority slots and exact methods", m.As) + } + seen[m.As] = true + for _, method := range m.Methods { + md, err := methodDescriptor(method) + if err != nil { + return nil, err + } + canonical := "/" + string(md.Parent().FullName()) + "/" + string(md.Name()) + if method != canonical || strings.ContainsAny(method, "* ") || md.IsStreamingClient() || md.IsStreamingServer() { + return nil, fmt.Errorf("setup method %q must name an exact unary RPC", method) + } + } + modules = append(modules, ConsumedModule{As: m.As, Authority: a}) + } + return s.openModuleRelays(modules, lookup) +} diff --git a/installation_setup_test.go b/installation_setup_test.go new file mode 100644 index 0000000..16fbea0 --- /dev/null +++ b/installation_setup_test.go @@ -0,0 +1,198 @@ +package solution + +import ( + "context" + "fmt" + "net/http" + "net/http/httptest" + "strings" + "sync" + "testing" + "time" + + authorityv1 "github.com/codefly-dev/authority/libraries/authority-client/go/gen/host/authority/v1" + "github.com/codefly-dev/sdk-go/workcontext" + "google.golang.org/protobuf/proto" + "google.golang.org/protobuf/reflect/protoreflect" +) + +func setupReceiver(t *testing.T, module *relayedModule, step func(context.Context, Installation) error) (*Server, *interfaceAuthority) { + t.Helper() + s, stand := authorityServer(t) + stand.step = verifiedStep(nil) + s.cfg.transport = meshTransport{local: true} + s.OnInstall(step, SetupModule{As: "things", Authority: AuthorityRelay{Group: "page", EndpointKey: "THINGS_ENDPOINT", BindingKey: "THINGS_BINDING"}, Methods: []string{searchName}}) + address := module.serve(t) + var err error + s.setupRelays, err = s.openSetupRelays(func(_, key string) (string, error) { + if key == "THINGS_ENDPOINT" { + return "grpc://" + address, nil + } + return "things-binding", nil + }) + if err != nil { + t.Fatal(err) + } + t.Cleanup(func() { + for _, r := range s.setupRelays { + _ = r.conn.Close() + } + }) + return s, stand +} + +func TestSetupCallbackUsesOnlyIssuedMethodAuthority(t *testing.T) { + module := &relayedModule{} + var saved Installation + s, stand := setupReceiver(t, module, func(ctx context.Context, i Installation) error { + saved = i + if _, err := ViewerFromContext(ctx); err == nil { + return fmt.Errorf("setup received a viewer") + } + in, out := thingMessage("GetThingRequest"), thingMessage("Thing") + setField(in, "entry_id", protoreflect.ValueOfString("setup-item")) + if err := i.Call(ctx, "things", searchName, in, out); err != nil { + return err + } + if out.Get(out.Descriptor().Fields().ByName("entry_id")).String() != "setup-item" { + return fmt.Errorf("response lost") + } + if err := i.Call(ctx, "things", getName, in, out); err == nil { + return fmt.Errorf("undeclared method accepted") + } + if err := i.Call(ctx, "things", searchName, out, in); err == nil { + return fmt.Errorf("wrong types accepted") + } + return nil + }) + if answer := install(s, stepRequest(stepCredential)); answer.Code != http.StatusNoContent { + t.Fatalf("setup: %d %s", answer.Code, answer.Body) + } + asks := stand.asked() + if len(asks) != 1 || asks[0].GetSetup().GetStepToken() != stepCredential || asks[0].GetSetup().GetMethod() != searchName || asks[0].GetOperationBindingId() != "things-binding" || asks[0].GetActor().GetCredential() != "workload-credential" || len(asks[0].GetScopes()) != 0 { + t.Fatalf("issue: %v", asks) + } + calls := module.observed() + if len(calls) != 1 || calls[0].method != searchName || calls[0].workContext != "issued-things-binding" || strings.Contains(calls[0].workContext, stepCredential) { + t.Fatalf("destination: %+v", calls) + } + if err := saved.Call(context.Background(), "things", searchName, thingMessage("GetThingRequest"), thingMessage("Thing")); err == nil { + t.Fatal("retained handle accepted") + } + if len(stand.asked()) != 1 { + t.Fatal("retained handle issued authority") + } + mux := http.NewServeMux() + if err := s.mountPassthrough(mux); err != nil { + t.Fatal(err) + } + r := httptest.NewRequest(http.MethodPost, "/v1/page"+searchPath, strings.NewReader(`{}`)) + r.Header.Set(WorkContextHeader, edgeContext) + w := httptest.NewRecorder() + mux.ServeHTTP(w, r) + if w.Code != http.StatusNotFound { + t.Fatalf("setup method exposed to viewer: %d", w.Code) + } +} + +func TestSetupCallbackCancellation(t *testing.T) { + module := &relayedModule{} + s, stand := setupReceiver(t, module, func(ctx context.Context, i Installation) error { + canceled, cancel := context.WithCancel(ctx) + cancel() + return i.Call(canceled, "things", searchName, thingMessage("GetThingRequest"), thingMessage("Thing")) + }) + if answer := install(s, stepRequest(stepCredential)); answer.Code != http.StatusServiceUnavailable { + t.Fatalf("cancellation: %d", answer.Code) + } + if len(stand.asked()) != 0 || len(module.observed()) != 0 { + t.Fatal("canceled call reached authority or destination") + } +} + +func TestSetupConcurrentAttemptsKeepTheirCredentials(t *testing.T) { + module := &relayedModule{} + entered := make(chan struct{}, 2) + release := make(chan struct{}) + s, stand := setupReceiver(t, module, func(ctx context.Context, i Installation) error { + entered <- struct{}{} + <-release + return i.Call(ctx, "things", searchName, thingMessage("GetThingRequest"), thingMessage("Thing")) + }) + stand.step = func(r *authorityv1.CheckRequest) (*authorityv1.CheckResponse, error) { + copy := proto.Clone(r).(*authorityv1.CheckRequest) + copy.Token = stepCredential + result, err := verifiedStep(nil)(copy) + if err != nil { + return nil, err + } + if r.GetToken() == "step-2" { + result.Context.Seal.InstallationId = "inst-2" + result.Context.AuthorityScopes[0].ResourceIds = []string{"inst-2"} + result.Context.TaskId = "operation-2" + } + return result, nil + } + second := httptest.NewRequest(http.MethodPost, installationPath, strings.NewReader(`{"installation_id":"inst-2","organization_id":"org-1","revision":3}`)) + second.Header.Set(workcontext.HeaderName, "step-2") + second.Header.Set(workcontext.InstallationIDHeaderName, "inst-2") + second.Header.Set(workcontext.InstallationRevisionHeaderName, "3") + var wg sync.WaitGroup + answers := make(chan int, 2) + for _, r := range []*http.Request{stepRequest(stepCredential), second} { + wg.Add(1) + go func(r *http.Request) { defer wg.Done(); answers <- install(s, r).Code }(r) + } + <-entered + <-entered + close(release) + wg.Wait() + close(answers) + for code := range answers { + if code != http.StatusNoContent { + t.Fatalf("attempt: %d", code) + } + } + tokens := map[string]bool{} + for _, ask := range stand.asked() { + tokens[ask.GetSetup().GetStepToken()] = true + } + if len(stand.asked()) != 2 || !tokens[stepCredential] || !tokens["step-2"] { + t.Fatalf("attempt credentials: %v", tokens) + } +} + +func TestSetupCancellationStopsAnInFlightCall(t *testing.T) { + module := &relayedModule{unaryStarted: make(chan struct{}), unaryCanceled: make(chan struct{})} + s, _ := setupReceiver(t, module, func(ctx context.Context, i Installation) error { + callCtx, cancel := context.WithCancel(ctx) + done := make(chan error, 1) + go func() { + done <- i.Call(callCtx, "things", searchName, thingMessage("GetThingRequest"), thingMessage("Thing")) + }() + select { + case <-module.unaryStarted: + case <-time.After(5 * time.Second): + cancel() + return fmt.Errorf("module did not receive call") + } + cancel() + select { + case err := <-done: + if err == nil { + return fmt.Errorf("canceled call succeeded") + } + case <-time.After(5 * time.Second): + return fmt.Errorf("call did not cancel") + } + select { + case <-module.unaryCanceled: + return nil + case <-time.After(5 * time.Second): + return fmt.Errorf("destination did not observe cancellation") + } + }) + if answer := install(s, stepRequest(stepCredential)); answer.Code != http.StatusNoContent { + t.Fatalf("cancellation: %d %s", answer.Code, answer.Body) + } +} diff --git a/installation_test.go b/installation_test.go index 2bda7e3..b1208ba 100644 --- a/installation_test.go +++ b/installation_test.go @@ -88,7 +88,7 @@ func TestAnInstallStepRunsOnlyWhatTheAuthorityVerified(t *testing.T) { t.Fatalf("a verified step answered %d: %s", answer.Code, answer.Body) } want := Installation{ID: "inst-1", OrganizationID: "org-1", Revision: 3, OperationID: "operation-1"} - if len(*ran) != 1 || (*ran)[0] != want { + if len(*ran) != 1 || (*ran)[0].ID != want.ID || (*ran)[0].OrganizationID != want.OrganizationID || (*ran)[0].Revision != want.Revision || (*ran)[0].OperationID != want.OperationID || (*ran)[0].setup == nil { t.Fatalf("the step ran with %+v, want once with %+v", *ran, want) } checks := stand.checks() diff --git a/passthrough_authority.go b/passthrough_authority.go index ba0366a..8b5d583 100644 --- a/passthrough_authority.go +++ b/passthrough_authority.go @@ -62,8 +62,12 @@ func (s *Server) authorityRoute() string { return "/v1/" + s.manifest.ID } // refuses the boot naming what is missing: a backend that cannot name a // binding cannot be issued a capability, and that belongs at start-up. func (s *Server) openRelays(lookup func(group, key string) (string, error)) (map[string]*moduleRelay, error) { + return s.openModuleRelays(s.consumed, lookup) +} + +func (s *Server) openModuleRelays(modules []ConsumedModule, lookup func(group, key string) (string, error)) (map[string]*moduleRelay, error) { relays := map[string]*moduleRelay{} - for _, module := range s.consumed { + for _, module := range modules { a := module.Authority slot := func(key, what string) (string, error) { value, err := lookup(a.Group, key) diff --git a/passthrough_authority_test.go b/passthrough_authority_test.go index 4236c06..e97c35f 100644 --- a/passthrough_authority_test.go +++ b/passthrough_authority_test.go @@ -33,9 +33,11 @@ type relayedModule struct { streamed int hold bool // secret is the Thing's undeclared field, which never reaches the page. - secret string - mu sync.Mutex - calls []relayedCall + secret string + unaryStarted chan struct{} + unaryCanceled chan struct{} + mu sync.Mutex + calls []relayedCall } type relayedCall struct { @@ -71,6 +73,12 @@ func (m *relayedModule) serve(t *testing.T) string { return out } if method != watchName { + if m.unaryStarted != nil { + close(m.unaryStarted) + <-stream.Context().Done() + close(m.unaryCanceled) + return stream.Context().Err() + } if refuse != nil { return refuse } diff --git a/solution.go b/solution.go index 9f4a303..551385a 100644 --- a/solution.go +++ b/solution.go @@ -148,7 +148,9 @@ type Server struct { passthrough map[string]passthroughRoute // relays is each relayed consumed module's binding and connection // (passthrough_authority.go). - relays map[string]*moduleRelay + relays map[string]*moduleRelay + setupModules []SetupModule + setupRelays map[string]*moduleRelay // mcp is the MCP server the solution declared with ServeMCP, nil when it // declared none; mcpRedeclared records a second ServeMCP, which the boot // refuses rather than letting the last call quietly be the only surface. @@ -563,6 +565,9 @@ func (s *Server) start(ctx context.Context) (net.Listener, error) { if s.relays, err = s.openRelays(codefly.For(ctx).WorkspaceConfiguration); err != nil { return nil, fmt.Errorf("solution %q: %w", s.manifest.ID, err) } + if s.setupRelays, err = s.openSetupRelays(codefly.For(ctx).WorkspaceConfiguration); err != nil { + return nil, fmt.Errorf("solution %q: %w", s.manifest.ID, err) + } // The contract this solution publishes is resolved from the same // declaration, under the profile this process runs in, and every value it // requires is named in the refusal when it is absent.