Draft: bounded calls during installation setup

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.

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

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)
+}

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)
+	}
+}

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()

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)