diff --git a/spinifex/gateway/auth.go b/spinifex/gateway/auth.go index 9a7fa156a..f99706b20 100644 --- a/spinifex/gateway/auth.go +++ b/spinifex/gateway/auth.go @@ -4,7 +4,6 @@ import ( "bytes" "context" "errors" - "fmt" authlimit "github.com/mulgadc/spinifex/spinifex/ingress/aws/ratelimit" "io" "log/slog" @@ -19,6 +18,7 @@ import ( gateway_eks "github.com/mulgadc/spinifex/spinifex/gateway/eks" handlers_iam "github.com/mulgadc/spinifex/spinifex/handlers/iam" handlers_sts "github.com/mulgadc/spinifex/spinifex/handlers/sts" + "github.com/mulgadc/spinifex/spinifex/ingress/aws/dispatch" ingresshttp "github.com/mulgadc/spinifex/spinifex/ingress/http" ) @@ -101,16 +101,19 @@ func (gw *GatewayConfig) SigV4AuthMiddleware() func(http.Handler) http.Handler { r = r.WithContext(context.WithValue(r.Context(), ctxService, sig.Credential.Service)) // Reject unknown services before crypto; otherwise Verify re-signs with - // the client-claimed service name and rubber-stamps the scope. - if !supportedServices[sig.Credential.Service] { + // the client-claimed service name and rubber-stamps the scope. A + // credential-scope service is served if the legacy table or the + // registration seam carries it. + _, isRegistered := gw.registered(sig.Credential.Service) + if !supportedServices[sig.Credential.Service] && !isRegistered { slog.Warn("Auth failure: unsupported service in credential scope", "accessKeyID", sig.Credential.AccessKeyID, "sourceIP", clientIP, "service", sig.Credential.Service) // Consistent with an unimplemented action on a served service: this // gate runs before Verify, so nothing was ever wrong with the // credentials, and no signature was ever checked. - gw.writeSigV4Error(w, r, awserrors.ErrorInvalidAction, - fmt.Sprintf("Service %q is not served by this gateway.", sig.Credential.Service)) + code, message := dispatch.UnknownService(sig.Credential.Service) + gw.writeSigV4Error(w, r, code, message) return } @@ -240,10 +243,14 @@ func (gw *GatewayConfig) SigV4AuthMiddleware() func(http.Handler) http.Handler { ctx = context.WithValue(ctx, ctxAction, method) } - // JSON-1.1 and path-routed REST-JSON services carry no query-protocol - // Action; without this every one of their requests shared the same - // "unknown" throttle bucket, so no per-action override could ever match. - if action := resolveNonQueryAction(r, sig.Credential.Service); action != "" { + // A registered service names its own action; everything else keeps + // resolving JSON-1.1/path-routed REST-JSON actions here, since + // those carry no query-protocol Action for the parse above to find. + if entry, ok := gw.registered(sig.Credential.Service); ok { + if action := entry.ResolveAction(r); action != "" { + ctx = context.WithValue(ctx, ctxAction, action) + } + } else if action := resolveNonQueryAction(r, sig.Credential.Service); action != "" { ctx = context.WithValue(ctx, ctxAction, action) } @@ -519,7 +526,7 @@ func (gw *GatewayConfig) writeSigV4Error(w http.ResponseWriter, r *http.Request, // the SDK chokes deserializing our XML into its shape. An unserved scope is // absent from jsonErrorService by definition, so its protocol is read off // the request itself instead. - if jsonErrorService(svc) || requestSignalsJSONProtocol(r) { + if gw.jsonErrorService(svc) || requestSignalsJSONProtocol(r) { w.Header().Set("Content-Type", eksJSONContentType) w.Header().Set("X-Amzn-Errortype", jsonErrorType(errorCode)) w.WriteHeader(errorMsg.HTTPCode) diff --git a/spinifex/gateway/dispatch_seam_test.go b/spinifex/gateway/dispatch_seam_test.go new file mode 100644 index 000000000..2a776f7d3 --- /dev/null +++ b/spinifex/gateway/dispatch_seam_test.go @@ -0,0 +1,467 @@ +// Tests for the generic gateway consuming the dispatch registration seam +// with fake services registered only here. No production service registers +// yet, so these are the only tests exercising the registered-dispatch path +// end to end. + +package gateway + +import ( + "bytes" + "context" + "encoding/xml" + "errors" + "io" + "net/http" + "net/http/httptest" + "testing" + "time" + + "github.com/go-chi/chi/v5" + "github.com/mulgadc/bluebottle/pkg/iampolicy" + "github.com/mulgadc/spinifex/internal/testkit" + "github.com/mulgadc/spinifex/spinifex/foundation/aws/errors" + "github.com/mulgadc/spinifex/spinifex/foundation/telemetry" + handlers_iam "github.com/mulgadc/spinifex/spinifex/handlers/iam" + "github.com/mulgadc/spinifex/spinifex/ingress/aws/dispatch" + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" + "go.opentelemetry.io/otel" + sdktrace "go.opentelemetry.io/otel/sdk/trace" + "go.opentelemetry.io/otel/sdk/trace/tracetest" +) + +// Fake service names/actions registered only in this file. +const ( + fakeJSONService = "testsvc" + fakeXMLService = "testsvcxml" + fakeNoResolverService = "testsvcnoresolver" + fakeAction = "FakeAction" + fakeXMLAction = "XMLAction" + fakeNoResolverAction = "NoResolverAction" +) + +// dispatchCapture records what a fake Dispatcher observed and controls what +// it does, so one Dispatcher body can drive every seam test's distinct need. +type dispatchCapture struct { + accountID string + region string + body []byte + bodyRead bool + + // authorizeResource, when non-empty, makes the dispatcher call + // inv.Authorize before succeeding, and return its error if any. + authorizeResource string + authorizeCalled bool + authorized bool + + // resolveActor makes the dispatcher call inv.Actor and record the result. + resolveActor bool + actor dispatch.Actor + actorErr error + + // dispatchErr, if set, is returned after the steps above run. + dispatchErr error +} + +func (c *dispatchCapture) dispatcher() dispatch.Dispatcher { + return func(w http.ResponseWriter, inv dispatch.Invocation) error { + body, err := io.ReadAll(inv.Request.Body) + if err != nil { + return err + } + c.accountID = inv.AccountID + c.region = inv.Region + c.body = body + c.bodyRead = true + + if c.authorizeResource != "" { + c.authorizeCalled = true + if err := inv.Authorize(fakeJSONService, fakeAction, []string{c.authorizeResource}, nil); err != nil { + return err + } + c.authorized = true + } + + if c.resolveActor { + actor, err := inv.Actor() + c.actor, c.actorErr = actor, err + if err != nil { + return err + } + } + + if c.dispatchErr != nil { + return c.dispatchErr + } + + w.Header().Set("Content-Type", "application/json") + w.WriteHeader(http.StatusOK) + _, _ = w.Write([]byte(`{"ok":true}`)) + return nil + } +} + +func fakeJSONRegistration(c *dispatchCapture) dispatch.Registration { + return dispatch.Registration{ + Service: fakeJSONService, + ResolveAction: func(r *http.Request) string { return fakeAction }, + Dispatch: c.dispatcher(), + Errors: dispatch.ErrorEnvelopeJSON, + Inventory: dispatch.Inventory{Registered: []string{fakeAction}}, + } +} + +func fakeXMLRegistration(c *dispatchCapture) dispatch.Registration { + return dispatch.Registration{ + Service: fakeXMLService, + ResolveAction: func(r *http.Request) string { return fakeXMLAction }, + Dispatch: c.dispatcher(), + Errors: dispatch.ErrorEnvelopeXML, + Inventory: dispatch.Inventory{Registered: []string{fakeXMLAction}}, + } +} + +// fakeNoResolverRegistration carries no ResolveAction, mirroring a legacy +// request no router matches: nothing names its action ahead of dispatch. +func fakeNoResolverRegistration(c *dispatchCapture) dispatch.Registration { + return dispatch.Registration{ + Service: fakeNoResolverService, + Dispatch: c.dispatcher(), + Errors: dispatch.ErrorEnvelopeJSON, + Inventory: dispatch.Inventory{Registered: []string{fakeNoResolverAction}}, + } +} + +// dispatchSeamApp wires SigV4AuthMiddleware -> traceActionEnricher -> Request +// on live test NATS, with regs registered through dispatch.NewBuilder, so a +// fake service's full pipeline mirrors ecrPipelineApp for a real one. +func dispatchSeamApp(t *testing.T, docs []handlers_iam.PolicyDocument, regs ...dispatch.Registration) (*GatewayConfig, http.Handler) { + t.Helper() + encryptedSecret, err := handlers_iam.EncryptSecret(testSecretKey, testMasterKey) + require.NoError(t, err) + _, nc := testutil.StartTestNATS(t) + + b := dispatch.NewBuilder() + for _, reg := range regs { + require.NoError(t, b.Register(reg)) + } + + gw := &GatewayConfig{ + DisableLogging: true, + Region: testRegion, + NATSConn: nc, + Services: b.Build(), + IAMService: &policyMockIAMService{ + mockIAMService: mockIAMService{ + masterKey: testMasterKey, + accessKeys: map[string]*handlers_iam.AccessKey{ + testAccessKey: { + AccessKeyID: testAccessKey, + SecretAccessKey: encryptedSecret, + UserName: "alice", + AccountID: ecrTestAccount, + Status: "Active", + }, + }, + }, + getUserPoliciesFn: func(_, _ string) ([]handlers_iam.PolicyDocument, error) { return docs, nil }, + }, + } + + r := chi.NewRouter() + r.Use(requestAuditMiddleware) + r.Use(gw.SigV4AuthMiddleware()) + r.Use(traceActionEnricher) + r.HandleFunc("/*", gw.Request) + return gw, r +} + +// signDispatchRequest builds a POST / signed for a fake service, the same +// shape a JSON 1.1 client sends. +func signDispatchRequest(t *testing.T, service string, body []byte) *http.Request { + t.Helper() + req := httptest.NewRequest(http.MethodPost, "/", bytes.NewReader(body)) + req.Host = "localhost:9999" + req.Header.Set("X-Amz-Target", "Whatever."+fakeAction) + signTestRequestAs(t, req, body, testAccessKey, testSecretKey, service, time.Now().UTC()) + return req +} + +// --- a. request reaches the registered dispatcher with full context ------- + +func TestDispatchSeam_RequestReachesDispatcherWithContext(t *testing.T) { + capt := &dispatchCapture{} + _, app := dispatchSeamApp(t, allowAllECRDocs(), fakeJSONRegistration(capt)) + + body := []byte(`{"hello":"world"}`) + resp := doRequest(app, signDispatchRequest(t, fakeJSONService, body)) + + require.Equal(t, http.StatusOK, resp.StatusCode) + assert.True(t, capt.bodyRead, "dispatcher must have read the request body") + assert.Equal(t, ecrTestAccount, capt.accountID) + assert.Equal(t, testRegion, capt.region) + assert.Equal(t, body, capt.body, "dispatcher must see the full body") +} + +// --- b. resolver's action reaches throttle/trace/audit --------------------- + +func TestDispatchSeam_ResolvedActionReachesThrottleTraceAndAudit(t *testing.T) { + sr := tracetest.NewSpanRecorder() + prev := otel.GetTracerProvider() + otel.SetTracerProvider(sdktrace.NewTracerProvider(sdktrace.WithSpanProcessor(sr))) + defer otel.SetTracerProvider(prev) + + capt := &dispatchCapture{} + gw, _ := dispatchSeamApp(t, allowAllECRDocs(), fakeJSONRegistration(capt)) + + var gotCtxAction string + var gotAudit *requestAudit + r := chi.NewRouter() + r.Use(otelsetup.HTTPMiddleware("awsgw")) + r.Use(requestAuditMiddleware) + r.Use(gw.SigV4AuthMiddleware()) + r.Use(traceActionEnricher) + r.Use(func(next http.Handler) http.Handler { + return http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + gotCtxAction, _ = r.Context().Value(ctxAction).(string) + next.ServeHTTP(w, r) + gotAudit = auditFrom(r.Context()) + }) + }) + r.HandleFunc("/*", gw.Request) + + doRequest(r, signDispatchRequest(t, fakeJSONService, []byte("{}"))) + + assert.Equal(t, fakeAction, gotCtxAction, "resolver's label must reach ctxAction for the throttle key") + + spans := sr.Ended() + require.Len(t, spans, 1) + span := spans[0] + assert.Equal(t, fakeJSONService+"."+fakeAction, span.Name()) + got := map[string]string{} + for _, kv := range span.Attributes() { + got[string(kv.Key)] = kv.Value.String() + } + assert.Equal(t, fakeAction, got["aws.action"]) + assert.Equal(t, fakeJSONService, got["aws.service"]) + + require.NotNil(t, gotAudit) + assert.Equal(t, fakeAction, gotAudit.action) +} + +// --- c. Authorize denies or allows ------------------------------------------ + +func TestDispatchSeam_AuthorizeDeniesOnMismatchAllowsOnMatch(t *testing.T) { + const allowedResource = "arn:aws:testsvc:" + testRegion + ":" + ecrTestAccount + ":widget/widget1" + const otherResource = "arn:aws:testsvc:" + testRegion + ":" + ecrTestAccount + ":widget/other" + + t.Run("scoped to a different resource is denied", func(t *testing.T) { + capt := &dispatchCapture{authorizeResource: allowedResource} + scopedDocs := []handlers_iam.PolicyDocument{{ + Version: "2012-10-17", + Statement: []handlers_iam.Statement{ + {Effect: "Allow", Action: handlers_iam.StringOrArr{"testsvc:FakeAction"}, Resource: handlers_iam.StringOrArr{otherResource}}, + }, + }} + _, app := dispatchSeamApp(t, scopedDocs, fakeJSONRegistration(capt)) + resp := doRequest(app, signDispatchRequest(t, fakeJSONService, []byte("{}"))) + + require.Equal(t, http.StatusForbidden, resp.StatusCode) + body := decodeJSONError(t, resp) + assert.Equal(t, "AccessDeniedException", body.Type) + assert.True(t, capt.authorizeCalled) + assert.False(t, capt.authorized) + }) + + t.Run("matching policy allows", func(t *testing.T) { + capt := &dispatchCapture{authorizeResource: allowedResource} + matchingDocs := []handlers_iam.PolicyDocument{{ + Version: "2012-10-17", + Statement: []handlers_iam.Statement{ + {Effect: "Allow", Action: handlers_iam.StringOrArr{"testsvc:FakeAction"}, Resource: handlers_iam.StringOrArr{allowedResource}}, + }, + }} + _, app := dispatchSeamApp(t, matchingDocs, fakeJSONRegistration(capt)) + resp := doRequest(app, signDispatchRequest(t, fakeJSONService, []byte("{}"))) + + require.Equal(t, http.StatusOK, resp.StatusCode) + assert.True(t, capt.authorized) + }) +} + +// --- d. Actor returns the verified caller, ignoring the body --------------- + +func TestDispatchSeam_ActorReturnsVerifiedCallerIgnoringBody(t *testing.T) { + capt := &dispatchCapture{resolveActor: true} + _, app := dispatchSeamApp(t, allowAllECRDocs(), fakeJSONRegistration(capt)) + + // A foreign account/ARN in the body must never reach the Actor: it comes + // from the verified SigV4 context, never from request content. + body := []byte(`{"accountId":"999999999999","callerArn":"arn:aws:iam::999999999999:user/evil"}`) + resp := doRequest(app, signDispatchRequest(t, fakeJSONService, body)) + + require.Equal(t, http.StatusOK, resp.StatusCode) + require.NoError(t, capt.actorErr) + assert.Equal(t, ecrTestAccount, capt.actor.AccountID) + assert.Equal(t, "arn:aws:iam::"+ecrTestAccount+":user/alice", capt.actor.CallerARN) + assert.Equal(t, principalTypeUser, capt.actor.PrincipalType) + assert.Equal(t, testAccessKey, capt.actor.AccessKeyID) +} + +// --- e. dispatcher error renders in the registration's declared envelope --- + +func TestDispatchSeam_DispatcherErrorRendersInDeclaredEnvelope(t *testing.T) { + t.Run("JSON envelope service", func(t *testing.T) { + capt := &dispatchCapture{dispatchErr: errors.New(awserrors.ErrorInternalError)} + _, app := dispatchSeamApp(t, allowAllECRDocs(), fakeJSONRegistration(capt)) + resp := doRequest(app, signDispatchRequest(t, fakeJSONService, []byte("{}"))) + + assert.Equal(t, eksJSONContentType, resp.Header.Get("Content-Type")) + assert.Equal(t, jsonErrorType(awserrors.ErrorInternalError), resp.Header.Get("X-Amzn-Errortype")) + body := decodeJSONError(t, resp) + assert.Equal(t, jsonErrorType(awserrors.ErrorInternalError), body.Type) + }) + + t.Run("XML envelope service", func(t *testing.T) { + capt := &dispatchCapture{dispatchErr: errors.New(awserrors.ErrorInternalError)} + _, app := dispatchSeamApp(t, allowAllECRDocs(), fakeXMLRegistration(capt)) + resp := doRequest(app, signDispatchRequest(t, fakeXMLService, []byte("{}"))) + + assert.Equal(t, "application/xml", resp.Header.Get("Content-Type")) + b, err := io.ReadAll(resp.Body) + require.NoError(t, err) + assert.Contains(t, string(b), awserrors.ErrorInternalError) + }) +} + +// --- f. NATS down gets the same 503 a legacy service gets ------------------ + +func TestDispatchSeam_NATSDownGetsSame503AsLegacy(t *testing.T) { + b := dispatch.NewBuilder() + require.NoError(t, b.Register(fakeJSONRegistration(&dispatchCapture{}))) + + gw := &GatewayConfig{DisableLogging: true, IAMService: allowAllIAMService(), Services: b.Build()} + + req := httptest.NewRequest(http.MethodPost, "/", nil) + ctx := context.WithValue(req.Context(), ctxService, fakeJSONService) + req = req.WithContext(ctx) + + w := httptest.NewRecorder() + gw.Request(w, req) + resp := w.Result() + + assert.Equal(t, http.StatusServiceUnavailable, resp.StatusCode) + assert.Equal(t, eksJSONContentType, resp.Header.Get("Content-Type")) + body := decodeJSONError(t, resp) + assert.Equal(t, "ServiceUnavailableException", body.Type) +} + +// --- g. an unregistered, unknown service still gets InvalidAction ---------- + +// A non-nil, populated registry must not change the answer for a scope +// nothing serves: the existing UnservedService characterization tests in +// auth_test.go pin the same behaviour with a nil registry. +func TestDispatchSeam_UnregisteredUnknownServiceStillInvalidAction(t *testing.T) { + b := dispatch.NewBuilder() + require.NoError(t, b.Register(fakeJSONRegistration(&dispatchCapture{}))) + + encryptedSecret, err := handlers_iam.EncryptSecret(testSecretKey, testMasterKey) + require.NoError(t, err) + gw := &GatewayConfig{ + DisableLogging: true, + Region: testRegion, + Services: b.Build(), + IAMService: &mockIAMService{ + masterKey: testMasterKey, + accessKeys: map[string]*handlers_iam.AccessKey{ + testAccessKey: {AccessKeyID: testAccessKey, SecretAccessKey: encryptedSecret, UserName: "root", Status: "Active"}, + }, + }, + } + + r := chi.NewRouter() + r.Use(gw.SigV4AuthMiddleware()) + r.HandleFunc("/*", gw.Request) + + req := httptest.NewRequest(http.MethodGet, "/", nil) + req.Host = "localhost:9999" + signTestRequestAs(t, req, nil, testAccessKey, testSecretKey, "route53", time.Now().UTC()) + + resp := doRequest(r, req) + require.Equal(t, http.StatusBadRequest, resp.StatusCode) + + body, err := io.ReadAll(resp.Body) + require.NoError(t, err) + var parsed struct { + Error struct { + Code string `xml:"Code"` + Message string `xml:"Message"` + } `xml:"Error"` + } + require.NoError(t, xml.Unmarshal(body, &parsed)) + assert.Equal(t, awserrors.ErrorInvalidAction, parsed.Error.Code) + assert.Contains(t, parsed.Error.Message, "route53") +} + +// --- h. ValidateServices overlap guard -------------------------------------- + +func TestValidateServices(t *testing.T) { + t.Run("nil registry passes", func(t *testing.T) { + gw := &GatewayConfig{} + assert.NoError(t, gw.ValidateServices()) + }) + + t.Run("empty registry passes", func(t *testing.T) { + gw := &GatewayConfig{Services: dispatch.NewBuilder().Build()} + assert.NoError(t, gw.ValidateServices()) + }) + + t.Run("claiming a legacy name fails", func(t *testing.T) { + b := dispatch.NewBuilder() + require.NoError(t, b.Register(dispatch.Registration{ + Service: "ecr", + Dispatch: func(http.ResponseWriter, dispatch.Invocation) error { return nil }, + Errors: dispatch.ErrorEnvelopeJSON, + Inventory: dispatch.Inventory{Registered: []string{fakeAction}}, + })) + gw := &GatewayConfig{Services: b.Build()} + assert.Error(t, gw.ValidateServices()) + }) +} + +// --- i. a registered service with no resolver still dispatches ------------- + +// The throttle key function falls back to the literal "unknown" for a +// request with no ctxAction (gateway.go's throttleKeyFuncs), exactly what an +// unresolved legacy request leaves it at today. +func TestDispatchSeam_NoResolverDispatchesWithLegacyFallback(t *testing.T) { + capt := &dispatchCapture{} + gw, _ := dispatchSeamApp(t, allowAllECRDocs(), fakeNoResolverRegistration(capt)) + + var ctxActionSet bool + r := chi.NewRouter() + r.Use(gw.SigV4AuthMiddleware()) + r.Use(func(next http.Handler) http.Handler { + return http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + _, ctxActionSet = r.Context().Value(ctxAction).(string) + next.ServeHTTP(w, r) + }) + }) + r.HandleFunc("/*", gw.Request) + + resp := doRequest(r, signDispatchRequest(t, fakeNoResolverService, []byte("{}"))) + + require.Equal(t, http.StatusOK, resp.StatusCode) + assert.True(t, capt.bodyRead, "dispatch must still run with no resolver") + assert.False(t, ctxActionSet, "no resolver means ctxAction is never set, same as an unresolved legacy request") + + key, err := gw.throttleKeyFuncs()[1](httptest.NewRequest(http.MethodGet, "/", nil)) + require.NoError(t, err) + assert.Equal(t, "unknown", key, "the documented fallback for a request with no ctxAction") +} + +// compile-time sanity: AuthorizeFunc/ActorFunc match the closures Request builds. +var ( + _ dispatch.AuthorizeFunc = (func(string, string, []string, iampolicy.ConditionKeys) error)(nil) + _ dispatch.ActorFunc = (func() (dispatch.Actor, error))(nil) +) diff --git a/spinifex/gateway/gateway.go b/spinifex/gateway/gateway.go index 8c8e5e66c..8fd62db20 100644 --- a/spinifex/gateway/gateway.go +++ b/spinifex/gateway/gateway.go @@ -44,6 +44,7 @@ import ( handlers_iam "github.com/mulgadc/spinifex/spinifex/handlers/iam" handlers_quota "github.com/mulgadc/spinifex/spinifex/handlers/quota" handlers_sts "github.com/mulgadc/spinifex/spinifex/handlers/sts" + "github.com/mulgadc/spinifex/spinifex/ingress/aws/dispatch" ingresshttp "github.com/mulgadc/spinifex/spinifex/ingress/http" "github.com/nats-io/nats.go" "github.com/nats-io/nats.go/jetstream" @@ -243,6 +244,61 @@ type GatewayConfig struct { // the concrete client, so a test can inject a fake without a live NATS // connection. BedrockAgentVector ochrevector.VectorService + + // Services holds the signed AWS services dispatched through the + // registration seam. Nil means nothing is registered, so every request + // falls through to the legacy switch in Request. + Services *dispatch.Registry +} + +// registered reports whether svc is served through the registration seam. A +// nil registry (no registrations) always reports false. +func (gw *GatewayConfig) registered(svc string) (*dispatch.Entry, bool) { + if gw.Services == nil { + return nil, false + } + return gw.Services.Lookup(svc) +} + +// ValidateServices fails if a name the registry serves is also served by the +// legacy supportedServices table, so no service is ever reachable down both +// dispatch paths at once. +func (gw *GatewayConfig) ValidateServices() error { + if gw.Services == nil { + return nil + } + for svc := range supportedServices { + if _, ok := gw.Services.Lookup(svc); ok { + return fmt.Errorf("gateway: service %q is registered and also served by the legacy dispatch table", svc) + } + } + return nil +} + +// registeredActorFunc returns the lazy Actor resolver a registered +// dispatcher's Invocation carries. It reads the same context values and +// shared buildCallerARN helper ecrAuthorizationPrincipal uses, so nothing is +// computed until a dispatcher calls it. +func (gw *GatewayConfig) registeredActorFunc(r *http.Request) dispatch.ActorFunc { + return func() (dispatch.Actor, error) { + ctx := r.Context() + accountID, _ := ctx.Value(ctxAccountID).(string) + identity, _ := ctx.Value(ctxIdentity).(string) + principalType, _ := ctx.Value(ctxPrincipalType).(string) + assumedRoleARN, _ := ctx.Value(ctxAssumedRoleARN).(string) + accessKey, _ := ctx.Value(ctxAccessKey).(string) + + callerARN, err := buildCallerARN(accountID, identity, principalType, assumedRoleARN) + if err != nil { + return dispatch.Actor{}, err + } + return dispatch.Actor{ + AccountID: accountID, + CallerARN: callerARN, + PrincipalType: principalType, + AccessKeyID: accessKey, + }, nil + } } var supportedServices = map[string]bool{ @@ -394,6 +450,18 @@ func jsonErrorService(svc string) bool { return false } +// jsonErrorService reports whether svc's errors render in the AWS JSON 1.1 +// envelope: a registered service's own declared envelope, or the legacy +// table for every service not yet registered through the seam. Every error +// emitter calls this method, never the free function, so the two can never +// disagree on a registered service. +func (gw *GatewayConfig) jsonErrorService(svc string) bool { + if entry, ok := gw.registered(svc); ok { + return entry.ErrorEnvelope() == dispatch.ErrorEnvelopeJSON + } + return jsonErrorService(svc) +} + // requestSignalsJSONProtocol reads r's own headers for the AWS JSON-1.x // tells, for a scope jsonErrorService has no entry for because the gateway // does not serve it. A JSON-1.1 action always carries X-Amz-Target, and a @@ -415,7 +483,7 @@ func (gw *GatewayConfig) writeClusterUnavailable(w http.ResponseWriter, r *http. requestID := uuid.NewV4().String() // AWS JSON 1.1 services (EKS/ECS/bedrock family, …) get a JSON body. - if jsonErrorService(svc) { + if gw.jsonErrorService(svc) { body := gateway_eks.GenerateEKSErrorResponse(awserrors.ErrorServiceUnavailable, clusterUnavailableMsg) w.Header().Set("Content-Type", eksJSONContentType) w.Header().Set("X-Amzn-Errortype", jsonErrorType(awserrors.ErrorServiceUnavailable)) @@ -447,7 +515,7 @@ func (gw *GatewayConfig) writeThrottleError(w http.ResponseWriter, r *http.Reque errorMsg := awserrors.ErrorLookup[errorCode] // AWS JSON 1.1 services (EKS/ECS/bedrock family, …) get a JSON body. - if jsonErrorService(svc) { + if gw.jsonErrorService(svc) { body := gateway_eks.GenerateEKSErrorResponse(errorCode, errorMsg.Message) w.Header().Set("Content-Type", eksJSONContentType) w.Header().Set("X-Amzn-Errortype", jsonErrorType(errorCode)) @@ -479,45 +547,60 @@ func (gw *GatewayConfig) Request(w http.ResponseWriter, r *http.Request) { } // Fail fast when NATS is down; every NATS-bound handler would otherwise hang - // until per-call timeout. + // until per-call timeout. Applies ahead of either dispatch path below. if gw.NATSConn == nil || !gw.NATSConn.IsConnected() { gw.writeClusterUnavailable(w, r, svc) return } - switch svc { - case "ec2": - err = gw.EC2_Request(w, r) - case "iam": - err = gw.IAM_Request(w, r) - case "sts": - err = gw.STS_Request(w, r) - case "elasticloadbalancing": - err = gw.ELBv2_Request(w, r) - case "eks": - err = gw.EKS_Request(w, r) - case "bedrock": - err = gw.Bedrock_Request(w, r) - case "bedrock-runtime": - err = gw.BedrockRuntime_Request(w, r) - case "bedrock-agent": - err = gw.BedrockAgent_Request(w, r) - case "bedrock-agent-runtime": - err = gw.BedrockAgentRuntime_Request(w, r) - case "ecs": - err = gw.ECS_Request(w, r) - case "ecr": - err = gw.ECR_Request(w, r) - case "acm": - err = gw.ACM_Request(w, r) - case "rds": - err = gw.RDS_Request(w, r) - case "tagging": - err = gw.Tagging_Request(w, r) - case "spinifex": - err = gw.Spinifex_Request(w, r) - default: - err = errors.New(awserrors.ErrorUnsupportedOperation) + if entry, ok := gw.registered(svc); ok { + inv := dispatch.Invocation{ + Request: r, + AccountID: mustCtxString(r, ctxAccountID), + Region: gw.Region, + Authorize: func(service, action string, resources []string, keys iampolicy.ConditionKeys) error { + return gw.checkPolicyResourcesWithKeys(r, service, action, resources, keys) + }, + Actor: gw.registeredActorFunc(r), + } + err = entry.Dispatch(w, inv) + } else { + // Legacy dispatch path for every service not yet registered through + // the seam. This switch must only shrink as services register. + switch svc { + case "ec2": + err = gw.EC2_Request(w, r) + case "iam": + err = gw.IAM_Request(w, r) + case "sts": + err = gw.STS_Request(w, r) + case "elasticloadbalancing": + err = gw.ELBv2_Request(w, r) + case "eks": + err = gw.EKS_Request(w, r) + case "bedrock": + err = gw.Bedrock_Request(w, r) + case "bedrock-runtime": + err = gw.BedrockRuntime_Request(w, r) + case "bedrock-agent": + err = gw.BedrockAgent_Request(w, r) + case "bedrock-agent-runtime": + err = gw.BedrockAgentRuntime_Request(w, r) + case "ecs": + err = gw.ECS_Request(w, r) + case "ecr": + err = gw.ECR_Request(w, r) + case "acm": + err = gw.ACM_Request(w, r) + case "rds": + err = gw.RDS_Request(w, r) + case "tagging": + err = gw.Tagging_Request(w, r) + case "spinifex": + err = gw.Spinifex_Request(w, r) + default: + err = errors.New(awserrors.ErrorUnsupportedOperation) + } } if err != nil { @@ -533,6 +616,11 @@ func (gw *GatewayConfig) GetService(r *http.Request) (string, error) { if !ok { return "", errors.New(awserrors.ErrorAuthFailure) } + // A registered service needs no bedrock sub-service rewriting: that + // rewrite exists only for the legacy "bedrock" signing-name split below. + if _, ok := gw.registered(svc); ok { + return svc, nil + } // The whole Bedrock family shares the SigV4 signing name "bedrock"; the // request path is the only discriminator, since the gateway serves one endpoint. if svc == "bedrock" { @@ -897,7 +985,7 @@ func (gw *GatewayConfig) ErrorHandler(w http.ResponseWriter, r *http.Request, er // EKS, ECR, ACM, ECS, tagging, and the bedrock family use AWS JSON 1.1; // query/XML services fall through. - if jsonErrorService(svc) { + if gw.jsonErrorService(svc) { body := gateway_eks.GenerateEKSErrorResponse(code, errorMsg.Message) slog.Debug("Generated JSON error response", "service", svc, "error", err, "code", code, "json", string(body), "requestId", requestId) w.Header().Set("Content-Type", eksJSONContentType) diff --git a/spinifex/runtime/roles/awsgw/awsgw.go b/spinifex/runtime/roles/awsgw/awsgw.go index f2032c4d7..b5cea33c7 100644 --- a/spinifex/runtime/roles/awsgw/awsgw.go +++ b/spinifex/runtime/roles/awsgw/awsgw.go @@ -41,6 +41,7 @@ import ( handlers_iam "github.com/mulgadc/spinifex/spinifex/handlers/iam" handlers_quota "github.com/mulgadc/spinifex/spinifex/handlers/quota" handlers_sts "github.com/mulgadc/spinifex/spinifex/handlers/sts" + "github.com/mulgadc/spinifex/spinifex/ingress/aws/dispatch" "github.com/mulgadc/spinifex/spinifex/providers/objectstore" "github.com/mulgadc/spinifex/spinifex/runtime/compute/cache" "github.com/mulgadc/spinifex/spinifex/runtime/compute/vm" @@ -123,6 +124,20 @@ func resolveSignupMaxAccounts(cfg signupConfig) int { return *cfg.MaxAccounts } +// wireServiceRegistry builds gw.Services from regs and validates it against +// the legacy dispatch table before the gateway ever serves a request. regs is +// empty in production: no service registers through the seam yet. +func wireServiceRegistry(gw *gateway.GatewayConfig, regs ...dispatch.Registration) error { + b := dispatch.NewBuilder() + for _, reg := range regs { + if err := b.Register(reg); err != nil { + return err + } + } + gw.Services = b.Build() + return gw.ValidateServices() +} + // loadAWSGWConfig reads and parses awsgw.toml once, returning the [ratelimit] and // [quota] sections together. Both sections default to their zero value (a // disabled no-op) when absent, so a config without either block stays valid. @@ -485,6 +500,12 @@ func launchService(config *config.ClusterConfig) error { BedrockAgentVector: bedrockAgentVector, } + // No service registers through the seam yet; wireServiceRegistry only + // proves the registry is wired and cannot overlap the legacy table. + if err := wireServiceRegistry(&gw); err != nil { + return fmt.Errorf("awsgw: %w", err) + } + // Rotate the ECR signing key on a 30-day cadence, retaining the previous keys // until their tokens expire. The rotator keeps the issuer/verifier current as // keys roll. Bound to the same lifetime context as the STS janitor. diff --git a/spinifex/runtime/roles/awsgw/awsgw_test.go b/spinifex/runtime/roles/awsgw/awsgw_test.go index cc3528f3c..4f5b8b956 100644 --- a/spinifex/runtime/roles/awsgw/awsgw_test.go +++ b/spinifex/runtime/roles/awsgw/awsgw_test.go @@ -1,14 +1,40 @@ package awsgw import ( + "net/http" "os" "path/filepath" "testing" + "github.com/mulgadc/spinifex/spinifex/gateway" + "github.com/mulgadc/spinifex/spinifex/ingress/aws/dispatch" "github.com/stretchr/testify/assert" "github.com/stretchr/testify/require" ) +// wireServiceRegistry is the exact call launchService makes to wire +// gw.Services before serving a request. An empty call (production's shape) +// must wire a non-nil, overlap-free registry; a registration claiming a +// legacy name must fail it, so the role refuses to start. +func TestWireServiceRegistry(t *testing.T) { + t.Run("empty registry wires non-nil Services and passes", func(t *testing.T) { + gw := &gateway.GatewayConfig{} + require.NoError(t, wireServiceRegistry(gw)) + assert.NotNil(t, gw.Services) + }) + + t.Run("a registration claiming a legacy name fails startup", func(t *testing.T) { + gw := &gateway.GatewayConfig{} + err := wireServiceRegistry(gw, dispatch.Registration{ + Service: "ecr", + Dispatch: func(http.ResponseWriter, dispatch.Invocation) error { return nil }, + Errors: dispatch.ErrorEnvelopeJSON, + Inventory: dispatch.Inventory{Registered: []string{"Whatever"}}, + }) + require.Error(t, err) + }) +} + func TestLoadThrottleConfig_Enabled(t *testing.T) { dir := t.TempDir() path := filepath.Join(dir, "awsgw.toml")