refactor: move AWS service selection into ingress dispatch

Selector answers served, action, error protocol, registered entry and
startup overlap. The gateway supplies its legacy tables through the
dispatch.Legacy interface; legacy dispatch and invocation building stay
in gateway.
This commit is contained in:
joshsiv-mulga
2026-10-07 23:43:32 +10:00
parent d70a5a43f7
commit b81ae4eb4e
4 changed files with 292 additions and 26 deletions
+4 -9
View File
@@ -104,8 +104,7 @@ func (gw *GatewayConfig) SigV4AuthMiddleware() func(http.Handler) http.Handler {
// 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 {
if !gw.selector().Served(sig.Credential.Service) {
slog.Warn("Auth failure: unsupported service in credential scope",
"accessKeyID", sig.Credential.AccessKeyID, "sourceIP", clientIP,
"service", sig.Credential.Service)
@@ -244,13 +243,9 @@ func (gw *GatewayConfig) SigV4AuthMiddleware() func(http.Handler) http.Handler {
}
// 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 != "" {
// resolving JSON-1.1/path-routed REST-JSON actions through the legacy
// resolver, since those carry no query-protocol Action to parse above.
if action := gw.selector().ResolveAction(r, sig.Credential.Service); action != "" {
ctx = context.WithValue(ctx, ctxAction, action)
}
+22 -17
View File
@@ -235,28 +235,36 @@ type GatewayConfig struct {
Services *dispatch.Registry
}
// selector answers service selection over Services and the legacy tables. It
// holds only the two pointers, so it is built per call and needs no cache.
func (gw *GatewayConfig) selector() *dispatch.Selector {
return dispatch.NewSelector(gw.Services, legacyServices{})
}
// legacyServices answers selection for services still on Request's switch.
type legacyServices struct{}
var _ dispatch.Legacy = legacyServices{}
func (legacyServices) Serves(svc string) bool { return supportedServices[svc] }
func (legacyServices) ResolveAction(r *http.Request, svc string) string {
return resolveNonQueryAction(r, svc)
}
func (legacyServices) JSONErrors(svc string) bool { return jsonErrorService(svc) }
// 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)
return gw.selector().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
return gw.selector().Validate()
}
// registeredActorFunc returns the lazy Actor resolver a registered
@@ -406,10 +414,7 @@ func jsonErrorService(svc string) bool {
// 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)
return gw.selector().JSONErrors(svc)
}
// clusterUnavailableMsg is the 503 body when NATS is disconnected. Points
+96
View File
@@ -0,0 +1,96 @@
package dispatch
import (
"fmt"
"net/http"
"slices"
)
// Legacy answers service selection for signed services not yet registered,
// so the selector can make one decision across both dispatch paths.
type Legacy interface {
// Serves reports whether the legacy path dispatches service.
Serves(service string) bool
// ResolveAction returns the advisory action label for r, or "" if unknown.
ResolveAction(r *http.Request, service string) string
// JSONErrors reports whether service's errors use the AWS JSON 1.1 envelope.
JSONErrors(service string) bool
}
// Selector picks the registration or legacy path serving a credential-scope
// service. A registered service always wins. Safe for concurrent use when
// legacy is, since neither it nor the registry is mutated.
type Selector struct {
registry *Registry
legacy Legacy
}
// NewSelector returns a selector over reg, which may be nil when nothing is
// registered, falling back to legacy, which may be nil when nothing is.
func NewSelector(reg *Registry, legacy Legacy) *Selector {
if legacy == nil {
legacy = noLegacy{}
}
return &Selector{registry: reg, legacy: legacy}
}
// Lookup returns the registration serving service, if any.
func (s *Selector) Lookup(service string) (*Entry, bool) {
if s.registry == nil {
return nil, false
}
return s.registry.Lookup(service)
}
// Served reports whether either dispatch path serves service.
func (s *Selector) Served(service string) bool {
if _, ok := s.Lookup(service); ok {
return true
}
return s.legacy.Serves(service)
}
// ResolveAction returns the advisory action label for r: the registration's
// own resolver when service is registered, otherwise the legacy resolver.
func (s *Selector) ResolveAction(r *http.Request, service string) string {
if e, ok := s.Lookup(service); ok {
return e.ResolveAction(r)
}
return s.legacy.ResolveAction(r, service)
}
// JSONErrors reports whether service's errors render in the AWS JSON 1.1
// envelope: a registration's declared envelope, else the legacy answer.
func (s *Selector) JSONErrors(service string) bool {
if e, ok := s.Lookup(service); ok {
return e.ErrorEnvelope() == ErrorEnvelopeJSON
}
return s.legacy.JSONErrors(service)
}
// Validate fails if any name the registry serves, alias included, is also
// served by the legacy path, so no service is reachable down both at once.
func (s *Selector) Validate() error {
if s.registry == nil {
return nil
}
names := make([]string, 0, len(s.registry.entries))
for n := range s.registry.entries {
names = append(names, n)
}
slices.Sort(names)
for _, n := range names {
if s.legacy.Serves(n) {
return fmt.Errorf("dispatch: service %q is registered and also served by the legacy dispatch table", n)
}
}
return nil
}
type noLegacy struct{}
func (noLegacy) Serves(string) bool { return false }
func (noLegacy) ResolveAction(*http.Request, string) string { return "" }
func (noLegacy) JSONErrors(string) bool { return false }
var _ Legacy = noLegacy{}
@@ -0,0 +1,170 @@
package dispatch
import (
"net/http"
"net/http/httptest"
"sync"
"testing"
"github.com/stretchr/testify/assert"
"github.com/stretchr/testify/require"
)
// fakeLegacy serves a fixed set of names, labels every request "LegacyAction"
// and renders JSON errors only for the names in json.
type fakeLegacy struct {
served map[string]bool
json map[string]bool
}
func (f fakeLegacy) Serves(svc string) bool { return f.served[svc] }
func (f fakeLegacy) ResolveAction(*http.Request, string) string { return "LegacyAction" }
func (f fakeLegacy) JSONErrors(svc string) bool { return f.json[svc] }
var _ Legacy = fakeLegacy{}
func newFakeLegacy(served ...string) fakeLegacy {
f := fakeLegacy{served: map[string]bool{}, json: map[string]bool{}}
for _, s := range served {
f.served[s] = true
}
return f
}
// selectorRegistry registers "ecr" (XML envelope, resolver "RegisteredAction",
// alias "ecr-alt") and "noresolver" (JSON envelope, no resolver).
func selectorRegistry(t *testing.T) *Registry {
t.Helper()
b := NewBuilder()
ecr := validReg("ecr")
ecr.Aliases = []string{"ecr-alt"}
ecr.InventoryName = "ecrinv"
ecr.Errors = ErrorEnvelopeXML
ecr.ResolveAction = func(*http.Request) string { return "RegisteredAction" }
require.NoError(t, b.Register(ecr))
require.NoError(t, b.Register(validReg("noresolver")))
return b.Build()
}
func TestSelectorRegisteredWins(t *testing.T) {
// The legacy side claims every name and answers the opposite envelope, so
// any fall-through to it on a registered name is visible.
legacy := newFakeLegacy()
legacy.json["ecr"] = true
legacy.json["ecr-alt"] = true
s := NewSelector(selectorRegistry(t), legacy)
req := httptest.NewRequest(http.MethodPost, "/", nil)
for _, name := range []string{"ecr", "ecr-alt"} {
t.Run(name, func(t *testing.T) {
assert.True(t, s.Served(name))
assert.Equal(t, "RegisteredAction", s.ResolveAction(req, name))
assert.False(t, s.JSONErrors(name), "registered XML envelope must beat legacy JSON")
e, ok := s.Lookup(name)
require.True(t, ok)
assert.Equal(t, "ecr", e.Service())
})
}
t.Run("registered without resolver stays unlabelled", func(t *testing.T) {
assert.Empty(t, s.ResolveAction(req, "noresolver"), "must not fall back to the legacy resolver")
assert.True(t, s.JSONErrors("noresolver"))
})
t.Run("inventory name is not a service", func(t *testing.T) {
assert.False(t, s.Served("ecrinv"))
_, ok := s.Lookup("ecrinv")
assert.False(t, ok)
})
}
func TestSelectorLegacyFallback(t *testing.T) {
legacy := newFakeLegacy("ec2", "acm")
legacy.json["acm"] = true
s := NewSelector(selectorRegistry(t), legacy)
req := httptest.NewRequest(http.MethodPost, "/", nil)
assert.True(t, s.Served("ec2"))
assert.True(t, s.Served("acm"))
assert.False(t, s.Served("route53"))
assert.Equal(t, "LegacyAction", s.ResolveAction(req, "ec2"))
assert.False(t, s.JSONErrors("ec2"))
assert.True(t, s.JSONErrors("acm"))
_, ok := s.Lookup("ec2")
assert.False(t, ok)
}
func TestSelectorNilRegistry(t *testing.T) {
legacy := newFakeLegacy("ec2")
legacy.json["ec2"] = true
s := NewSelector(nil, legacy)
req := httptest.NewRequest(http.MethodPost, "/", nil)
_, ok := s.Lookup("ecr")
assert.False(t, ok)
assert.False(t, s.Served("ecr"))
assert.True(t, s.Served("ec2"))
assert.Equal(t, "LegacyAction", s.ResolveAction(req, "ec2"))
assert.True(t, s.JSONErrors("ec2"))
assert.NoError(t, s.Validate())
}
func TestSelectorNilLegacyServesOnlyRegistered(t *testing.T) {
s := NewSelector(selectorRegistry(t), nil)
req := httptest.NewRequest(http.MethodPost, "/", nil)
assert.True(t, s.Served("ecr"))
assert.False(t, s.Served("ec2"))
assert.Empty(t, s.ResolveAction(req, "ec2"))
assert.False(t, s.JSONErrors("ec2"))
assert.NoError(t, s.Validate())
}
func TestSelectorValidate(t *testing.T) {
cases := []struct {
name string
legacy []string
wantErr bool
}{
{"disjoint", []string{"ec2", "acm"}, false},
{"primary name overlaps", []string{"ec2", "ecr"}, true},
{"alias overlaps", []string{"ecr-alt"}, true},
{"inventory name is not a served name", []string{"ecrinv"}, false},
}
for _, tc := range cases {
t.Run(tc.name, func(t *testing.T) {
err := NewSelector(selectorRegistry(t), newFakeLegacy(tc.legacy...)).Validate()
if tc.wantErr {
assert.Error(t, err)
} else {
assert.NoError(t, err)
}
})
}
t.Run("empty registry", func(t *testing.T) {
assert.NoError(t, NewSelector(NewBuilder().Build(), newFakeLegacy("ecr")).Validate())
})
}
func TestSelectorConcurrentUse(t *testing.T) {
legacy := newFakeLegacy("ec2")
s := NewSelector(selectorRegistry(t), legacy)
var wg sync.WaitGroup
for range 16 {
wg.Go(func() {
req := httptest.NewRequest(http.MethodPost, "/", nil)
for range 100 {
assert.True(t, s.Served("ecr"))
assert.True(t, s.Served("ec2"))
assert.Equal(t, "RegisteredAction", s.ResolveAction(req, "ecr"))
assert.Equal(t, "LegacyAction", s.ResolveAction(req, "ec2"))
assert.False(t, s.JSONErrors("ecr"))
assert.NoError(t, s.Validate())
}
})
}
wg.Wait()
}