diff --git a/.github/workflows/go.yml b/.github/workflows/go.yml index d69d509..98484f3 100644 --- a/.github/workflows/go.yml +++ b/.github/workflows/go.yml @@ -21,4 +21,4 @@ jobs: - run: go mod download && go mod verify - run: go test -race ./... -count=1 -timeout=5m - run: go vet ./... - - run: test -z "$(gofmt -l ./*.go)" + - run: test -z "$(gofmt -l .)" diff --git a/AGENTS.md b/AGENTS.md index 07801a0..f16b86c 100644 --- a/AGENTS.md +++ b/AGENTS.md @@ -14,7 +14,10 @@ the gateway; endpoint, port and secret resolution to `codefly-dev/sdk-go`; workspace and manifest types to `codefly-dev/core`; running a composition and provisioning its secrets to `codefly-dev/cli`. Everything here is a library — one Go package at the repository root, consumed as the Go module -`github.com/codefly-dev/solution-runtime-go` at a `vX.Y.Z` tag. +`github.com/codefly-dev/solution-runtime-go` at a `vX.Y.Z` tag, plus +`passthroughtest`, the test seam a consumer runs the root package's +passthrough under against a fake host. It holds no runtime behaviour of its +own: it builds on `Server.PassthroughHandler`, the handler `Serve` mounts. ## Boundaries @@ -117,7 +120,7 @@ gate. It is four commands, and they are the whole local loop: go mod download && go mod verify go test -race ./... -count=1 -timeout=5m go vet ./... -test -z "$(gofmt -l ./*.go)" +test -z "$(gofmt -l .)" ``` Go comes from `go.mod` (1.27). Nothing else is needed: no Docker, no @@ -125,9 +128,8 @@ credentials, no running composition. - The suite takes ~20s, most of it two heartbeat tests that wait on real wall time (10s and 5s). That is not a hang. -- The formatting gate globs `./*.go`, so it covers the root package and only - the root package. A package added in a subdirectory is vetted and tested by - the commands above but not format-checked; widen the glob in the same PR. +- The formatting gate runs over the whole tree (`gofmt -l .`), so a package + added in a subdirectory is format-checked as well as vetted and tested. - Dependency pins are how fixes from `core` and `sdk-go` reach consumers: `go get github.com/codefly-dev/core@vX.Y.Z && go mod tidy`, then the four commands above. `go mod tidy` is a no-op on a clean tree, so a diff it diff --git a/README.md b/README.md index ff361c4..a87e5d6 100644 --- a/README.md +++ b/README.md @@ -515,6 +515,37 @@ In the interface artifact a streamed operation says so in its behaviour and carries `x-streaming: server` (Swagger 2 has no stream; its 200 schema is one message of it). +#### Testing the passthrough: `passthroughtest` + +`Serve` resolves the gateway and api.consumes from the composition and +registers with the host, so a solution cannot run its own passthrough in a +test through it. Package +`github.com/codefly-dev/solution-runtime-go/passthroughtest` serves the real +passthrough — `Server.PassthroughHandler`, the handler `Serve` mounts, behind +the same boot check — over an `httptest` server, against a fake host: + +```go +upstream := httptest.NewServer(fakeWidgets) // the test's stand-in for the module's binding +host := passthroughtest.NewHost(t).Module("widgets", upstream.URL) +page := passthroughtest.Start(t, host, declaration...) // the solution's Consumes +client := widgetsv1connect.NewWidgetsClient(page.Client(), page.BaseURL("widgets")) +``` + +| API | What it is | +| --- | --- | +| `NewHost(t)` | The fake host gateway: it mints the viewer's Work Context (the accounts `StartTask` procedure) and forwards `/v1//*` to each routed module, streams included, flushed as they arrive. | +| `Host.Module(as, upstream)` | Routes a consumed module to the test's own server, path unchanged. Every routed module is also in the solution's api.consumes; one the declaration names but the host does not route is refused as `Serve` refuses it. | +| `Host.RefuseMints(func(Mint) *Refusal)` | Decides each mint: a non-nil `Refusal` refuses it as accounts would (default status 403), so the page sees `permission_denied` with its message. | +| `Host.Mints()`, `Host.Calls()` | Every mint asked for (audience, scopes, org, session, bearer, the token answered), and every call forwarded to a module, each with the `Mint` whose capability it presented — how a test asserts the authority each method was called under. | +| `Start(t, host, consumes...)` / `Handler(host, consumes...)` | The passthrough served for the declaration, or the bare handler and the error `Serve` would refuse the declaration with. | +| `Solution.Client()`, `ClientAs(Viewer)`, `BaseURL(as)` | A client that calls as `DefaultViewer` (or the viewer given), setting the bearer and the `x-org-id` / `x-session-id` headers the gateway stamps; and the base URL a module's generated client takes. | + +What a green test there proves is the solution's side: the declaration, the +authority each method mints, the fields that reach the page, stream bounds and +cancellation. It never proves that a real host admits the solution, that +accounts grants the scopes, or that the real module answers its binding the way +the test's stand-in does. + ### Generated messages in a response A handler's value is encoded with `encoding/json`, which reads a generated diff --git a/passthrough.go b/passthrough.go index 4f35f20..6035b94 100644 --- a/passthrough.go +++ b/passthrough.go @@ -286,6 +286,67 @@ func (s *Server) validatePassthrough() (map[string]passthroughRoute, error) { return routes, checkConsumed(s.consumed, consumed) } +// mountPassthrough mounts the declared routes on mux at PassthroughPathPrefix, +// as serve does and as PassthroughHandler does: the one place the passthrough +// is wired, so a test through PassthroughHandler serves what Serve serves. +// Serve has already resolved and checked the routes; a test that calls serve +// directly has not, and resolves them here. +func (s *Server) mountPassthrough(mux *http.ServeMux) error { + if len(s.consumed) == 0 { + return nil + } + routes := s.passthrough + if routes == nil { + var err error + if routes, err = resolvePassthrough(s.consumed); err != nil { + return err + } + } + mux.Handle(PassthroughPathPrefix, s.passthroughHandler(routes)) + return nil +} + +// PassthroughEnvironment is what Serve resolves from the composition for the +// passthrough, supplied instead by a caller of PassthroughHandler. +type PassthroughEnvironment struct { + // GatewayURL is the host gateway every call goes through: the Work Context + // mint and the consumed modules' /v1/ prefixes. + GatewayURL string + // Consumes is the solution's api.consumes projection, which Serve reads + // from CODEFLY__API_CONSUMES. Every declared module must be in it. + Consumes []manifest.ConsumedAPI +} + +// PassthroughHandler is the consumed-module passthrough exactly as Serve +// serves it — the declaration checked by the same boot check, mounted by the +// same code at PassthroughPathPrefix — against an environment the caller +// supplies instead of the one Serve resolves. Nothing else of the solution is +// served, and nothing registers anywhere. +// +// It exists for tests: package passthroughtest builds on it, with a fake host +// standing in for the gateway. A solution serves with Serve, which overwrites +// the environment set here when it resolves its own. +func (s *Server) PassthroughHandler(env PassthroughEnvironment) (http.Handler, error) { + if len(s.consumed) == 0 { + return nil, fmt.Errorf("solution %q declares no consumed modules (Consumes)", s.manifest.ID) + } + projection, err := json.Marshal(env.Consumes) + if err != nil { + return nil, err + } + s.cfg.gatewayURL, s.cfg.apiConsumes = env.GatewayURL, string(projection) + routes, err := s.validatePassthrough() + if err != nil { + return nil, fmt.Errorf("solution %q: %w", s.manifest.ID, err) + } + s.passthrough = routes + mux := http.NewServeMux() + if err := s.mountPassthrough(mux); err != nil { + return nil, err + } + return mux, nil +} + // passthroughHandler serves every route under PassthroughPathPrefix. Each // route is a Connect unary handler over the method's own descriptor, so every // protocol the module's generated clients speak is accepted and every refusal diff --git a/passthroughtest/passthroughtest.go b/passthroughtest/passthroughtest.go new file mode 100644 index 0000000..48eca10 --- /dev/null +++ b/passthroughtest/passthroughtest.go @@ -0,0 +1,335 @@ +// Package passthroughtest runs a solution's consumed-module passthrough in a +// test, against a fake host, so a consuming solution can exercise the calls its +// page makes — unary and server-streaming — without a composition. +// +// It serves the real passthrough: Server.PassthroughHandler, the handler Serve +// mounts, checked by the same boot check, over an httptest server. Only the +// host is fake. Host stands in for the gateway the passthrough calls through: +// it mints the viewer's Work Context (the accounts StartTask procedure the +// gateway routes) and forwards each /v1//* call to the module address the +// test gave it, typically an httptest server of the test's own that answers +// the method's google.api.http binding. +// +// host := passthroughtest.NewHost(t).Module("widgets", upstream.URL) +// page := passthroughtest.Start(t, host, declaration...) +// client := widgetsv1connect.NewWidgetsClient(page.Client(), page.BaseURL("widgets")) +// +// What a green test here does not prove: that a real host admits the +// solution, that accounts grants the scopes, or that the real module answers +// its binding the way the test's stand-in does. +package passthroughtest + +import ( + "encoding/json" + "fmt" + "net/http" + "net/http/httptest" + "net/http/httputil" + "net/url" + "strings" + "sync" + "testing" + "time" + + "github.com/codefly-dev/core/solution/manifest" + codefly "github.com/codefly-dev/sdk-go" + solution "github.com/codefly-dev/solution-runtime-go" +) + +// startTaskProcedure is the accounts procedure the gateway routes a Work +// Context mint to. It is the host's contract, spoken as Connect JSON. +const startTaskProcedure = "/saas.accounts.v1.WorkContextService/StartTask" + +// Viewer is the identity the page calls as: the bearer the browser sends, and +// the organization and session the gateway stamps from it (x-org-id, +// x-session-id) before the call reaches the solution. +type Viewer struct { + Bearer string + OrgID string + SessionID string +} + +// DefaultViewer is the viewer Solution.Client calls as. +var DefaultViewer = Viewer{ + Bearer: "Bearer passthroughtest-viewer", + OrgID: "passthroughtest-org", + SessionID: "passthroughtest-session", +} + +// Mint is one Work Context the host was asked to mint: the passthrough's +// request, as accounts receives it. +type Mint struct { + // Token is the capability the host answered with ("" for a refused mint). + Token string + Bearer string + OrgID string + SessionID string + // Audience is the module the capability names: its `as`. + Audience string + // Scopes is the authority asked for: the method's own, else its module's. + Scopes []solution.Scope +} + +// Call is one call the host forwarded to a module. +type Call struct { + // Module is the `as` the call was routed under. + Module string + // Method and Path are the HTTP verb and path of the module's binding. + Method string + Path string + // Bearer is the viewer's bearer, as forwarded. + Bearer string + // Mint is the mint whose capability the call presented, nil for a call + // that presented none (a ViewerBearer module) or one this host never + // minted. + Mint *Mint +} + +// Refusal is the host's refusal of a mint, written the way accounts writes +// one: a Connect error body under an HTTP status. +type Refusal struct { + // Status is the HTTP status; zero is 403. + Status int + // Code is the Connect code ("permission_denied"). + Code string + Message string +} + +// Host is a fake host gateway: the Work Context mint and the consumed +// modules' prefixes. Its hooks may be set at any time, from any goroutine. +type Host struct { + server *httptest.Server + + mu sync.Mutex + modules map[string]*httputil.ReverseProxy + refuse func(Mint) *Refusal + mints []Mint + calls []Call +} + +// NewHost starts a fake host, closed when the test ends. +func NewHost(t testing.TB) *Host { + t.Helper() + h := &Host{modules: map[string]*httputil.ReverseProxy{}} + h.server = httptest.NewServer(http.HandlerFunc(h.serveHTTP)) + t.Cleanup(h.server.Close) + return h +} + +// URL is the host gateway's base URL. +func (h *Host) URL() string { return h.server.URL } + +// Module routes the consumed module `as` to upstream, a base URL the host +// forwards every /v1//* call to, path unchanged — as the gateway forwards +// to a registered module. Every module routed here is also in the solution's +// api.consumes projection. It panics on an upstream that is not an absolute +// URL. +func (h *Host) Module(as, upstream string) *Host { + target, err := url.Parse(upstream) + if err != nil || target.Scheme == "" || target.Host == "" { + panic(fmt.Sprintf("passthroughtest: module %q: upstream %q is not an absolute URL", as, upstream)) + } + proxy := &httputil.ReverseProxy{ + Rewrite: func(r *httputil.ProxyRequest) { + r.SetURL(target) + r.Out.Host = target.Host + }, + // A streamed answer is relayed line by line, as it is flushed. + FlushInterval: -1, + } + h.mu.Lock() + defer h.mu.Unlock() + h.modules[as] = proxy + return h +} + +// RefuseMints sets the hook that decides each mint: a non-nil Refusal refuses +// it, the way accounts refuses an authority the viewer does not hold. Nil +// (the default) mints every ask. +func (h *Host) RefuseMints(refuse func(Mint) *Refusal) *Host { + h.mu.Lock() + defer h.mu.Unlock() + h.refuse = refuse + return h +} + +// Mints is every mint the host was asked for, refused ones included, in order. +func (h *Host) Mints() []Mint { + h.mu.Lock() + defer h.mu.Unlock() + return append([]Mint(nil), h.mints...) +} + +// Calls is every call the host forwarded to a module, in the order they +// arrived. +func (h *Host) Calls() []Call { + h.mu.Lock() + defer h.mu.Unlock() + return append([]Call(nil), h.calls...) +} + +// consumes is the api.consumes projection of the routed modules. +func (h *Host) consumes() []manifest.ConsumedAPI { + h.mu.Lock() + defer h.mu.Unlock() + out := make([]manifest.ConsumedAPI, 0, len(h.modules)) + for as := range h.modules { + out = append(out, manifest.ConsumedAPI{ID: as, Module: as, As: as}) + } + return out +} + +func (h *Host) serveHTTP(w http.ResponseWriter, r *http.Request) { + if r.URL.Path == startTaskProcedure { + h.mint(w, r) + return + } + as, _, _ := strings.Cut(strings.TrimPrefix(r.URL.Path, "/v1/"), "/") + h.mu.Lock() + proxy := h.modules[as] + if !strings.HasPrefix(r.URL.Path, "/v1/") || proxy == nil { + h.mu.Unlock() + writeJSON(w, http.StatusNotFound, map[string]any{"code": 5, "message": "no module is routed at " + r.URL.Path}) + return + } + call := Call{Module: as, Method: r.Method, Path: r.URL.Path, Bearer: r.Header.Get("authorization")} + if presented := r.Header.Get(codefly.WorkContextHeaderName); presented != "" { + for i := range h.mints { + if h.mints[i].Token == presented { + mint := h.mints[i] + call.Mint = &mint + break + } + } + } + h.calls = append(h.calls, call) + h.mu.Unlock() + proxy.ServeHTTP(w, r) +} + +// mint answers the StartTask procedure as accounts does. +func (h *Host) mint(w http.ResponseWriter, r *http.Request) { + var ask struct { + OrgID string `json:"orgId"` + SessionID string `json:"sessionId"` + Audience string `json:"audience"` + AuthorityScopes []struct { + ResourceKind string `json:"resourceKind"` + Actions []string `json:"actions"` + ResourceIDs []string `json:"resourceIds"` + } `json:"authorityScopes"` + } + if err := json.NewDecoder(r.Body).Decode(&ask); err != nil { + writeJSON(w, http.StatusBadRequest, map[string]any{"code": "invalid_argument", "message": err.Error()}) + return + } + mint := Mint{Bearer: r.Header.Get("authorization"), OrgID: ask.OrgID, SessionID: ask.SessionID, Audience: ask.Audience} + for _, scope := range ask.AuthorityScopes { + mint.Scopes = append(mint.Scopes, solution.Scope{ResourceKind: scope.ResourceKind, Actions: scope.Actions, ResourceIDs: scope.ResourceIDs}) + } + h.mu.Lock() + refuse := h.refuse + h.mu.Unlock() + if refuse != nil { + if refusal := refuse(mint); refusal != nil { + h.record(&mint, false) + status := refusal.Status + if status == 0 { + status = http.StatusForbidden + } + writeJSON(w, status, map[string]any{"code": refusal.Code, "message": refusal.Message}) + return + } + } + h.record(&mint, true) + writeJSON(w, http.StatusOK, map[string]any{ + "token": mint.Token, + "orgId": mint.OrgID, + "ownerPrincipalId": "passthroughtest-viewer", + "currentActorPrincipalId": "passthroughtest-viewer", + "expiresAt": time.Now().Add(5 * time.Minute).UTC().Format(time.RFC3339Nano), + }) +} + +// record appends mint, naming its capability after its place in the log when +// it is granted, so a forwarded call is traced to the mint it presents. +func (h *Host) record(mint *Mint, granted bool) { + h.mu.Lock() + defer h.mu.Unlock() + if granted { + mint.Token = fmt.Sprintf("passthroughtest-%d.%s", len(h.mints)+1, mint.Audience) + } + h.mints = append(h.mints, *mint) +} + +func writeJSON(w http.ResponseWriter, status int, body any) { + w.Header().Set("content-type", "application/json") + w.WriteHeader(status) + _ = json.NewEncoder(w).Encode(body) +} + +// Solution is a solution's passthrough served for real against a fake host. +type Solution struct { + // URL is the solution backend's base URL, where the passthrough serves + // under solution.PassthroughPathPrefix. + URL string + server *httptest.Server +} + +// Handler builds the passthrough for the declaration against host, with the +// modules host routes as the solution's api.consumes: the real handler, and +// the error Serve would refuse the declaration with at boot. +func Handler(host *Host, consumes ...solution.ConsumedModule) (http.Handler, error) { + server := solution.New(solution.Manifest{ID: "passthroughtest"}).Consumes(consumes...) + return server.PassthroughHandler(solution.PassthroughEnvironment{GatewayURL: host.URL(), Consumes: host.consumes()}) +} + +// Start serves the passthrough for the declaration against host, until the +// test ends. A declaration Serve would refuse fails the test. +func Start(t testing.TB, host *Host, consumes ...solution.ConsumedModule) *Solution { + t.Helper() + handler, err := Handler(host, consumes...) + if err != nil { + t.Fatalf("passthroughtest: %v", err) + } + server := httptest.NewServer(handler) + t.Cleanup(server.Close) + return &Solution{URL: server.URL, server: server} +} + +// BaseURL is the base URL a module's generated client takes for the consumed +// module `as`: /modules/. +func (s *Solution) BaseURL(as string) string { + return s.URL + solution.PassthroughPathPrefix + as +} + +// Client is an HTTP client that calls as DefaultViewer. +func (s *Solution) Client() *http.Client { return s.ClientAs(DefaultViewer) } + +// ClientAs is an HTTP client that calls as viewer: the bearer, and the +// identity headers the gateway would stamp, replacing any the request sets. +// An empty field sends no header, as for a viewer the gateway would stamp none +// for. +func (s *Solution) ClientAs(viewer Viewer) *http.Client { + return &http.Client{Transport: viewerTransport{viewer: viewer, base: s.server.Client().Transport}} +} + +type viewerTransport struct { + viewer Viewer + base http.RoundTripper +} + +func (t viewerTransport) RoundTrip(r *http.Request) (*http.Response, error) { + r = r.Clone(r.Context()) + for header, value := range map[string]string{ + "authorization": t.viewer.Bearer, + "x-org-id": t.viewer.OrgID, + "x-session-id": t.viewer.SessionID, + } { + r.Header.Del(header) + if value != "" { + r.Header.Set(header, value) + } + } + return t.base.RoundTrip(r) +} diff --git a/passthroughtest/passthroughtest_test.go b/passthroughtest/passthroughtest_test.go new file mode 100644 index 0000000..ed6870b --- /dev/null +++ b/passthroughtest/passthroughtest_test.go @@ -0,0 +1,372 @@ +package passthroughtest_test + +import ( + "context" + "encoding/json" + "fmt" + "io" + "net/http" + "net/http/httptest" + "strings" + "sync" + "testing" + "time" + + "connectrpc.com/connect" + codefly "github.com/codefly-dev/sdk-go" + solution "github.com/codefly-dev/solution-runtime-go" + "github.com/codefly-dev/solution-runtime-go/passthroughtest" + "google.golang.org/genproto/googleapis/api/annotations" + "google.golang.org/protobuf/proto" + "google.golang.org/protobuf/reflect/protodesc" + "google.golang.org/protobuf/reflect/protoreflect" + "google.golang.org/protobuf/reflect/protoregistry" + "google.golang.org/protobuf/types/descriptorpb" + "google.golang.org/protobuf/types/dynamicpb" +) + +// widgetsFile is a composed module's API as its generated package registers +// it: a unary Search and a server-streaming Watch, each with a google.api.http +// binding under /v1/widgets. +var widgetsFile = func() protoreflect.FileDescriptor { + str := descriptorpb.FieldDescriptorProto_TYPE_STRING.Enum() + opt := descriptorpb.FieldDescriptorProto_LABEL_OPTIONAL.Enum() + field := func(name, json string, number int32) *descriptorpb.FieldDescriptorProto { + return &descriptorpb.FieldDescriptorProto{Name: proto.String(name), JsonName: proto.String(json), Number: proto.Int32(number), Label: opt, Type: str} + } + binding := func(rule *annotations.HttpRule) *descriptorpb.MethodOptions { + options := &descriptorpb.MethodOptions{} + proto.SetExtension(options, annotations.E_Http, rule) + return options + } + file := &descriptorpb.FileDescriptorProto{ + Name: proto.String("passthroughtest_test/widgets.proto"), + Package: proto.String("acme.widgets.v1"), + Syntax: proto.String("proto3"), + MessageType: []*descriptorpb.DescriptorProto{ + {Name: proto.String("WidgetRequest"), Field: []*descriptorpb.FieldDescriptorProto{field("widget_id", "widgetId", 1)}}, + {Name: proto.String("Widget"), Field: []*descriptorpb.FieldDescriptorProto{ + field("widget_id", "widgetId", 1), field("name", "name", 2), field("secret", "secret", 3), + }}, + }, + Service: []*descriptorpb.ServiceDescriptorProto{{ + Name: proto.String("Widgets"), + Method: []*descriptorpb.MethodDescriptorProto{ + {Name: proto.String("Search"), InputType: proto.String(".acme.widgets.v1.WidgetRequest"), OutputType: proto.String(".acme.widgets.v1.Widget"), + Options: binding(&annotations.HttpRule{Pattern: &annotations.HttpRule_Post{Post: "/v1/widgets/search"}, Body: "*"})}, + {Name: proto.String("Watch"), InputType: proto.String(".acme.widgets.v1.WidgetRequest"), OutputType: proto.String(".acme.widgets.v1.Widget"), + ServerStreaming: proto.Bool(true), + Options: binding(&annotations.HttpRule{Pattern: &annotations.HttpRule_Get{Get: "/v1/widgets/{widget_id}/watch"}})}, + }, + }}, + } + fd, err := protodesc.NewFile(file, protoregistry.GlobalFiles) + if err != nil { + panic(err) + } + if err := protoregistry.GlobalFiles.RegisterFile(fd); err != nil { + panic(err) + } + return fd +}() + +var ( + searchMethod = widgetsFile.Services().ByName("Widgets").Methods().ByName("Search") + watchMethod = widgetsFile.Services().ByName("Widgets").Methods().ByName("Watch") + readScope = solution.Scope{ResourceKind: "widgets", Actions: []string{"read"}} + watchScope = solution.Scope{ResourceKind: "widgets", Actions: []string{"watch"}} +) + +// declaration is a consuming solution's Consumes: Search under the module's +// read scope, Watch under its own watch scope, each returning only widget_id +// and name. +func declaration() solution.ConsumedModule { + mask := solution.MustFieldMask(dynamicpb.NewMessage(searchMethod.Output()), "widget_id", "name") + return solution.ConsumedModule{ + As: "widgets", + Scopes: []solution.Scope{readScope}, + Methods: []solution.ConsumedMethod{ + {Name: "/acme.widgets.v1.Widgets/Search", Response: mask}, + {Name: "/acme.widgets.v1.Widgets/Watch", Response: mask, Scopes: []solution.Scope{watchScope}}, + }, + } +} + +// module is the test's stand-in for the composed module: it answers Search +// with one widget and Watch with the lines the test feeds it, flushed one by +// one, and reports when a Watch request is cancelled. +type module struct { + *httptest.Server + lines chan string + canceled chan struct{} + // done ends every open stream when the test ends, so a stream the + // passthrough failed to cancel fails the test instead of hanging it. + done chan struct{} + + mu sync.Mutex + headers []http.Header +} + +func newModule(t *testing.T) *module { + t.Helper() + m := &module{lines: make(chan string, 16), canceled: make(chan struct{}, 1), done: make(chan struct{})} + m.Server = httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + m.mu.Lock() + m.headers = append(m.headers, r.Header.Clone()) + m.mu.Unlock() + switch { + case r.Method == http.MethodPost && r.URL.Path == "/v1/widgets/search": + var req struct { + WidgetID string `json:"widgetId"` + } + _ = json.NewDecoder(r.Body).Decode(&req) + w.Header().Set("content-type", "application/json") + _, _ = fmt.Fprintf(w, `{"widgetId":%q,"name":"Acme widget","secret":"s3cret"}`, req.WidgetID) + case r.Method == http.MethodGet && strings.HasSuffix(r.URL.Path, "/watch"): + w.Header().Set("content-type", "application/x-ndjson") + w.WriteHeader(http.StatusOK) + w.(http.Flusher).Flush() + for { + select { + case line, ok := <-m.lines: + if !ok { + return + } + _, _ = io.WriteString(w, line+"\n") + w.(http.Flusher).Flush() + case <-r.Context().Done(): + m.canceled <- struct{}{} + return + case <-m.done: + return + } + } + default: + http.NotFound(w, r) + } + })) + t.Cleanup(m.Close) + return m +} + +func widget(id string) string { + return fmt.Sprintf(`{"result":{"widgetId":%q,"name":"Acme widget","secret":"s3cret"}}`, id) +} + +// clients are the page: the module's Connect clients, over JSON as connect-es +// speaks, against the passthrough's base URL. +func clients(page *passthroughtest.Solution) (search, watch *connect.Client[dynamicpb.Message, dynamicpb.Message]) { + client := func(md protoreflect.MethodDescriptor) *connect.Client[dynamicpb.Message, dynamicpb.Message] { + return connect.NewClient[dynamicpb.Message, dynamicpb.Message](page.Client(), + page.BaseURL("widgets")+"/"+string(md.Parent().FullName())+"/"+string(md.Name()), + connect.WithSchema(md), connect.WithProtoJSON(), + connect.WithResponseInitializer(func(_ connect.Spec, msg any) error { + *msg.(*dynamicpb.Message) = *dynamicpb.NewMessage(md.Output()) + return nil + })) + } + return client(searchMethod), client(watchMethod) +} + +func request(id string) *connect.Request[dynamicpb.Message] { + msg := dynamicpb.NewMessage(searchMethod.Input()) + msg.Set(msg.Descriptor().Fields().ByName("widget_id"), protoreflect.ValueOfString(id)) + return connect.NewRequest(msg) +} + +func get(m *dynamicpb.Message, name string) string { + return m.Get(m.Descriptor().Fields().ByName(protoreflect.Name(name))).String() +} + +func has(m *dynamicpb.Message, name string) bool { + return m.Has(m.Descriptor().Fields().ByName(protoreflect.Name(name))) +} + +func start(t *testing.T) (*passthroughtest.Host, *module, *connect.Client[dynamicpb.Message, dynamicpb.Message], *connect.Client[dynamicpb.Message, dynamicpb.Message]) { + t.Helper() + upstream := newModule(t) + host := passthroughtest.NewHost(t).Module("widgets", upstream.URL) + search, watch := clients(passthroughtest.Start(t, host, declaration())) + // Registered last, so it runs first (cleanups run last-in, first-out): + // before any server's Close waits on a stream still open. + t.Cleanup(func() { close(upstream.done) }) + return host, upstream, search, watch +} + +func TestAUnaryCallIsAnsweredByTheModuleAsTheViewer(t *testing.T) { + host, upstream, search, _ := start(t) + + resp, err := search.CallUnary(context.Background(), request("w1")) + if err != nil { + t.Fatal(err) + } + if get(resp.Msg, "widget_id") != "w1" || get(resp.Msg, "name") != "Acme widget" { + t.Fatalf("response = %v", resp.Msg) + } + calls := host.Calls() + if len(calls) != 1 || calls[0].Method != http.MethodPost || calls[0].Path != "/v1/widgets/search" || calls[0].Mint == nil { + t.Fatalf("calls = %+v, want Search over its binding, presenting a minted capability", calls) + } + if calls[0].Bearer != passthroughtest.DefaultViewer.Bearer { + t.Fatalf("bearer = %q, want the viewer's", calls[0].Bearer) + } + mint := calls[0].Mint + if mint.Audience != "widgets" || mint.OrgID != passthroughtest.DefaultViewer.OrgID || mint.SessionID != passthroughtest.DefaultViewer.SessionID { + t.Fatalf("mint = %+v, want the module's audience in the viewer's org and session", mint) + } + upstream.mu.Lock() + presented := upstream.headers[0].Get(codefly.WorkContextHeaderName) + upstream.mu.Unlock() + if got := presented; got != mint.Token { + t.Fatalf("the module was presented %q, want the minted %q", got, mint.Token) + } +} + +func TestAStreamedCallYieldsEveryMessage(t *testing.T) { + _, upstream, _, watch := start(t) + for _, id := range []string{"w1", "w2", "w3"} { + upstream.lines <- widget(id) + } + close(upstream.lines) + + stream, err := watch.CallServerStream(context.Background(), request("w1")) + if err != nil { + t.Fatal(err) + } + defer func() { _ = stream.Close() }() + var ids []string + for stream.Receive() { + ids = append(ids, get(stream.Msg(), "widget_id")) + } + if err := stream.Err(); err != nil { + t.Fatal(err) + } + if strings.Join(ids, ",") != "w1,w2,w3" { + t.Fatalf("received %v, want w1,w2,w3 in order", ids) + } +} + +func TestEachMethodMintsItsOwnAuthority(t *testing.T) { + host, upstream, search, watch := start(t) + close(upstream.lines) + + if _, err := search.CallUnary(context.Background(), request("w1")); err != nil { + t.Fatal(err) + } + stream, err := watch.CallServerStream(context.Background(), request("w1")) + if err != nil { + t.Fatal(err) + } + for stream.Receive() { + } + if err := stream.Err(); err != nil { + t.Fatal(err) + } + _ = stream.Close() + + mints := host.Mints() + if len(mints) != 2 { + t.Fatalf("mints = %+v, want one per call", mints) + } + authority := map[string]string{} + for _, call := range host.Calls() { + if call.Mint == nil || len(call.Mint.Scopes) != 1 { + t.Fatalf("call %s %s presented %+v", call.Method, call.Path, call.Mint) + } + scope := call.Mint.Scopes[0] + authority[call.Path] = scope.ResourceKind + ":" + strings.Join(scope.Actions, ",") + } + if authority["/v1/widgets/search"] != "widgets:read" || authority["/v1/widgets/w1/watch"] != "widgets:watch" { + t.Fatalf("authority per call = %v, want Search under widgets:read and Watch under its own widgets:watch", authority) + } +} + +func TestOnlyTheDeclaredFieldsReachThePage(t *testing.T) { + _, upstream, search, watch := start(t) + upstream.lines <- widget("w1") + close(upstream.lines) + + resp, err := search.CallUnary(context.Background(), request("w1")) + if err != nil { + t.Fatal(err) + } + if has(resp.Msg, "secret") || !has(resp.Msg, "name") { + t.Fatalf("unary response = %v, want name and no secret", resp.Msg) + } + stream, err := watch.CallServerStream(context.Background(), request("w1")) + if err != nil { + t.Fatal(err) + } + defer func() { _ = stream.Close() }() + if !stream.Receive() { + t.Fatalf("no streamed message: %v", stream.Err()) + } + if has(stream.Msg(), "secret") || !has(stream.Msg(), "name") { + t.Fatalf("streamed message = %v, want name and no secret", stream.Msg()) + } +} + +func TestThePageDisconnectingCancelsTheModulesStream(t *testing.T) { + _, upstream, _, watch := start(t) + upstream.lines <- widget("w1") + + ctx, cancel := context.WithCancel(context.Background()) + defer cancel() + stream, err := watch.CallServerStream(ctx, request("w1")) + if err != nil { + t.Fatal(err) + } + if !stream.Receive() { + t.Fatalf("no streamed message: %v", stream.Err()) + } + // The page goes away mid-stream; the module never ends its stream itself. + cancel() + _ = stream.Close() + select { + case <-upstream.canceled: + case <-time.After(5 * time.Second): + t.Fatal("the module's stream was not cancelled when the page disconnected") + } +} + +func TestARefusedMintRefusesOnlyThatMethod(t *testing.T) { + host, upstream, search, watch := start(t) + close(upstream.lines) + host.RefuseMints(func(mint passthroughtest.Mint) *passthroughtest.Refusal { + for _, scope := range mint.Scopes { + for _, action := range scope.Actions { + if action == "watch" { + return &passthroughtest.Refusal{Code: "permission_denied", Message: "viewer is not allowed widgets:watch"} + } + } + } + return nil + }) + + stream, err := watch.CallServerStream(context.Background(), request("w1")) + if err == nil { + for stream.Receive() { + } + err = stream.Err() + _ = stream.Close() + } + if connect.CodeOf(err) != connect.CodePermissionDenied || !strings.Contains(err.Error(), "widgets:watch") { + t.Fatalf("watch = %v, want permission_denied naming the missing permission", err) + } + if _, err := search.CallUnary(context.Background(), request("w1")); err != nil { + t.Fatalf("search, whose authority the viewer holds, was refused: %v", err) + } + for _, call := range host.Calls() { + if strings.HasSuffix(call.Path, "/watch") { + t.Fatal("a refused method reached the module") + } + } +} + +func TestADeclarationServeWouldRefuseIsRefused(t *testing.T) { + host := passthroughtest.NewHost(t) // routes no module, so api.consumes lists none + _, err := passthroughtest.Handler(host, declaration()) + if err == nil || !strings.Contains(err.Error(), "api.consumes lists no module") { + t.Fatalf("err = %v, want the boot refusal of a module the solution does not consume", err) + } +} diff --git a/solution.go b/solution.go index 8f532ce..5e87da1 100644 --- a/solution.go +++ b/solution.go @@ -793,15 +793,8 @@ func (s *Server) serve(ctx context.Context, ln net.Listener) error { mux.HandleFunc(path, withCORS(s.wrapRequest(handler))) } mux.Handle("/assets/", http.StripPrefix("/assets/", withCORSHandler(s.assetsHandler()))) - if len(s.consumed) > 0 { - routes := s.passthrough - if routes == nil { - var err error - if routes, err = resolvePassthrough(s.consumed); err != nil { - return err - } - } - mux.Handle(PassthroughPathPrefix, s.passthroughHandler(routes)) + if err := s.mountPassthrough(mux); err != nil { + return err } manifestBody, _ := json.Marshal(s.manifestMap())