diff --git a/README.md b/README.md index 390272f..33fc41e 100644 --- a/README.md +++ b/README.md @@ -80,7 +80,8 @@ Proxy Pool 用 Controller 协调这些变化，并让 Gateway 数据面只消费 - **安全边界**：Gateway、Distribution 与 Admin 使用各自的认证语义，并支持 CIDR、可信代理、严格请求解析和敏感信息最小化；Admin 的读写权限与 Distribution 提取权限可按命中凭据分别收敛。Distribution 凭据还可限制单次 - 提取数量、可访问 Upstream 与地区；未配置时保持既有全范围行为。 + 提取数量、可访问 Upstream 与地区，Gateway 凭据可限制可访问 Routing；未配置时 + 保持既有全范围行为。 ## 架构概览 diff --git a/docs/configuration/reference.md b/docs/configuration/reference.md index 905d68f..d736233 100644 --- a/docs/configuration/reference.md +++ b/docs/configuration/reference.md @@ -152,12 +152,11 @@ admin: ### 3.3 Distribution Client 提取约束 -Distribution 的已认证凭据可在 auth.client 中设置固定的提取约束；mode: any -则必须在实际命中的 auth.methods 项下配置。当前该字段只允许出现在 -Distribution 监听器，Gateway 与 Admin 配置会被启动校验拒绝，防止出现只声明 +认证凭据可在 auth.client 中设置固定的 Client 约束；mode: any 则必须在实际 +命中的 auth.methods 项下配置。当前 Client 约束按入口明确分工，避免出现只声明 不执行的访问控制。 -- maxExtractCount：单次提取的额外上限；0 表示不追加上限。 +- maxExtractCount：Distribution 单次提取的额外上限；0 表示不追加上限。 - allowedUpstreams：允许访问的 Upstream 名称集合。省略请求过滤条件时，服务端 自动使用这个集合；请求携带集合外名称会返回 403。 - allowedRegions：允许访问的地区集合，行为与 allowedUpstreams 相同。 @@ -180,6 +179,28 @@ distribution: allowedRegions: [shanghai, beijing] ~~~ +### 3.4 Gateway Client 路由约束 + +Gateway 的 auth.client 只接受 allowedRoutings。认证成功后，Gateway 先完成目标 +地址策略和本地 Routing 匹配，再于本地 Dispatcher 前检查该集合；不匹配返回 403， +不会尝试选择或预留 Proxy。该检查只读取请求上下文和当前本地 Snapshot，不访问 +Redis、PostgreSQL 或 Provider。 + +allowedRoutings 中的每个名称必须对应一个启用的 Gateway Routing。Gateway 上的 +maxExtractCount、allowedUpstreams 和 allowedRegions，以及 Distribution 上的 +allowedRoutings 均会被配置校验拒绝。Admin 不支持任何 client 约束。 + +~~~yaml +gateway: + auth: + mode: any + methods: + - mode: bearer + valueFile: /run/secrets/checkout-gateway-token + client: + allowedRoutings: [checkout] +~~~ + Gateway、Distribution、Admin 与 Provider API 是独立认证边界。改变其中一套 不得连带改变其他入口。 diff --git a/docs/development/implementation-plan.md b/docs/development/implementation-plan.md index bb385f1..be2f79d 100644 --- a/docs/development/implementation-plan.md +++ b/docs/development/implementation-plan.md @@ -192,7 +192,8 @@ reject/wait/direct，并会在快照刷新后看到停用状态。 Gateway request path on immutable local snapshots with no Redis/PostgreSQL calls. - [x] Expose Distribution extraction/status and Admin status/audit/enable/disable/switch/reload HTTP handlers and contracts with credential-level endpoint permissions and - Distribution credential-level extraction boundaries. + Distribution credential-level extraction boundaries; enforce matched + Gateway credential routing boundaries before local dispatch. - [x] Add Compose-backed Redis 8.2 integration and shared Adapter contract tests. - [x] Add PostgreSQL management Adapter and Compose-backed integration tests. diff --git a/docs/requirements/completion-audit.md b/docs/requirements/completion-audit.md index 0ee7f5b..f13ed0b 100644 --- a/docs/requirements/completion-audit.md +++ b/docs/requirements/completion-audit.md @@ -47,8 +47,9 @@ - `DIST/Admin HTTP`：严格 JSON、Request ID、Problem 响应及 Distribution/Admin Handler 已实现；Admin 审计查询以有界 `afterId` 游标读取 PostgreSQL 权威记录， 不读取 Proxy 或 Redis 活动池。共享认证、CIDR、可信代理、Client ID、凭据级 - 固定权限、Distribution 凭据级提取数量/Upstream/地区限制与本地准入保护链已接入， - Controller Runtime 已将二者装配到独立监听器并支持联动优雅停机。 + 固定权限、Distribution 凭据级提取数量/Upstream/地区限制、Gateway 凭据级 + Routing 限制与本地准入保护链已接入，Controller Runtime 已将二者装配到独立 + 监听器并支持联动优雅停机。 - `Redis Activity Adapter`：真实 Redis 8.2 已覆盖 Provider Upsert、健康更新、 原子独占提取、短期幂等、Worker ownership、库存和有界过期清理，Memory/Redis 运行同一公用契约。 diff --git a/internal/config/config_test.go b/internal/config/config_test.go index acb5427..a35fd8d 100644 --- a/internal/config/config_test.go +++ b/internal/config/config_test.go @@ -393,10 +393,19 @@ func TestValidateDistributionCredentialClientPolicy(t *testing.T) { cfg.Distribution.Auth.ClientPolicy = clientpolicy.Policy{} cfg.Gateway.Auth = Auth{ Mode: "bearer", Token: "gateway-token", - ClientPolicy: clientpolicy.Policy{MaxExtractCount: 1}, + ClientPolicy: clientpolicy.Policy{AllowedRoutings: []string{"extract"}}, } + cfg.Routing[0].Purpose = "gateway" + if err := Validate(cfg); err != nil { + t.Fatalf("Validate(gateway routing policy) error = %v", err) + } + cfg.Gateway.Auth.ClientPolicy.AllowedRoutings = []string{"missing"} + if err := Validate(cfg); err == nil || !strings.Contains(err.Error(), "missing") { + t.Fatalf("Validate(unknown gateway routing) error = %v", err) + } + cfg.Gateway.Auth.ClientPolicy = clientpolicy.Policy{MaxExtractCount: 1} if err := Validate(cfg); err == nil || !strings.Contains(err.Error(), "only supported on distribution") { - t.Fatalf("Validate(gateway client policy) error = %v", err) + t.Fatalf("Validate(gateway extraction policy) error = %v", err) } } diff --git a/internal/config/validate.go b/internal/config/validate.go index 9ff4d63..aaf4eaf 100644 --- a/internal/config/validate.go +++ b/internal/config/validate.go @@ -38,11 +38,10 @@ func Validate(cfg *Config) error { return err } } - if err := validateNoClientPolicy("gateway", cfg.Gateway.Auth); err != nil { - return err - } - if err := validateNoClientPolicy("admin", cfg.Admin.Auth); err != nil { - return err + if cfg.Admin.Enabled { + if err := validateNoClientPolicy("admin", cfg.Admin.Auth); err != nil { + return err + } } if cfg.Metrics.Enabled { if cfg.Metrics.Listen == "" { @@ -83,11 +82,15 @@ func Validate(cfg *Config) error { return fmt.Errorf("validate configuration: at least one enabled upstream is required") } seen := make(map[string]struct{}, len(cfg.Routing)) + gatewayRoutings := make(map[string]struct{}, len(cfg.Routing)) targetProfilesByUpstream := make(map[string]int, len(cfg.Upstreams)) for index, route := range cfg.Routing { if err := validateRouting(index, route, cfg.Upstreams, seen); err != nil { return err } + if route.Enabled && route.Purpose == "gateway" { + gatewayRoutings[route.Name] = struct{}{} + } if !route.Enabled || len(route.Check.Targets) == 0 { continue } @@ -102,6 +105,11 @@ func Validate(cfg *Config) error { } } } + if cfg.Gateway.Enabled { + if err := validateGatewayClientPolicies(cfg.Gateway.Auth, gatewayRoutings); err != nil { + return err + } + } if cfg.Distribution.Enabled { if err := validateDistributionClientPolicies(cfg.Distribution.Auth, cfg.Upstreams); err != nil { return err @@ -485,11 +493,11 @@ func validateNoClientPolicy(listener string, auth Auth) error { } func validateDistributionClientPolicies(auth Auth, upstreams map[string]Upstream) error { - if err := validateClientPolicyUpstreams("distribution auth.client", auth.ClientPolicy, upstreams); err != nil { + if err := validateDistributionClientPolicy("distribution auth.client", auth.ClientPolicy, upstreams); err != nil { return err } for index, method := range auth.Methods { - if err := validateClientPolicyUpstreams( + if err := validateDistributionClientPolicy( fmt.Sprintf("distribution auth.methods[%d].client", index), method.ClientPolicy, upstreams, ); err != nil { return err @@ -498,7 +506,10 @@ func validateDistributionClientPolicies(auth Auth, upstreams map[string]Upstream return nil } -func validateClientPolicyUpstreams(scope string, policy clientpolicy.Policy, upstreams map[string]Upstream) error { +func validateDistributionClientPolicy(scope string, policy clientpolicy.Policy, upstreams map[string]Upstream) error { + if len(policy.AllowedRoutings) != 0 { + return fmt.Errorf("validate %s.allowedRoutings: routing access is only supported on gateway", scope) + } for _, upstream := range policy.AllowedUpstreams { if _, exists := upstreams[upstream]; !exists { return fmt.Errorf("validate %s.allowedUpstreams: upstream %q does not exist", scope, upstream) @@ -507,6 +518,32 @@ func validateClientPolicyUpstreams(scope string, policy clientpolicy.Policy, ups return nil } +func validateGatewayClientPolicies(auth Auth, routings map[string]struct{}) error { + if err := validateGatewayClientPolicy("gateway auth.client", auth.ClientPolicy, routings); err != nil { + return err + } + for index, method := range auth.Methods { + if err := validateGatewayClientPolicy( + fmt.Sprintf("gateway auth.methods[%d].client", index), method.ClientPolicy, routings, + ); err != nil { + return err + } + } + return nil +} + +func validateGatewayClientPolicy(scope string, policy clientpolicy.Policy, routings map[string]struct{}) error { + if policy.MaxExtractCount != 0 || len(policy.AllowedUpstreams) != 0 || len(policy.AllowedRegions) != 0 { + return fmt.Errorf("validate %s: extraction constraints are only supported on distribution", scope) + } + for _, routing := range policy.AllowedRoutings { + if _, exists := routings[routing]; !exists { + return fmt.Errorf("validate %s.allowedRoutings: routing %q does not exist or is not an enabled gateway routing", scope, routing) + } + } + return nil +} + func validateCIDRs(name string, cidrs []string) error { for _, cidr := range cidrs { if _, _, err := net.ParseCIDR(cidr); err != nil { diff --git a/internal/domain/clientpolicy/policy.go b/internal/domain/clientpolicy/policy.go index 9f3f6bd..5d2e46b 100644 --- a/internal/domain/clientpolicy/policy.go +++ b/internal/domain/clientpolicy/policy.go @@ -9,22 +9,25 @@ import ( var ErrInvalidPolicy = errors.New("invalid client policy") -// Policy limits an authenticated client's extraction request. Zero values -// preserve the existing unrestricted listener behavior. +// Policy limits an authenticated client's permitted Distribution extraction or +// Gateway routing. Zero values preserve the existing unrestricted behavior. type Policy struct { MaxExtractCount int `yaml:"maxExtractCount"` AllowedUpstreams []string `yaml:"allowedUpstreams"` AllowedRegions []string `yaml:"allowedRegions"` + AllowedRoutings []string `yaml:"allowedRoutings"` } func (policy Policy) IsZero() bool { - return policy.MaxExtractCount == 0 && len(policy.AllowedUpstreams) == 0 && len(policy.AllowedRegions) == 0 + return policy.MaxExtractCount == 0 && len(policy.AllowedUpstreams) == 0 && + len(policy.AllowedRegions) == 0 && len(policy.AllowedRoutings) == 0 } func (policy Policy) Validate() error { if policy.MaxExtractCount < 0 || !validUniqueValues(policy.AllowedUpstreams) || - !validUniqueValues(policy.AllowedRegions) { + !validUniqueValues(policy.AllowedRegions) || + !validUniqueValues(policy.AllowedRoutings) { return ErrInvalidPolicy } return nil @@ -33,6 +36,7 @@ func (policy Policy) Validate() error { func (policy Policy) Clone() Policy { policy.AllowedUpstreams = append([]string(nil), policy.AllowedUpstreams...) policy.AllowedRegions = append([]string(nil), policy.AllowedRegions...) + policy.AllowedRoutings = append([]string(nil), policy.AllowedRoutings...) return policy } @@ -48,6 +52,18 @@ func (policy Policy) RestrictRegions(requested []string) ([]string, bool) { return restrict(requested, policy.AllowedRegions) } +func (policy Policy) AllowsRouting(routing string) bool { + if len(policy.AllowedRoutings) == 0 { + return true + } + for _, allowed := range policy.AllowedRoutings { + if routing == allowed { + return true + } + } + return false +} + func restrict(requested, allowed []string) ([]string, bool) { if len(allowed) == 0 { return append([]string(nil), requested...), true diff --git a/internal/domain/clientpolicy/policy_test.go b/internal/domain/clientpolicy/policy_test.go index a3494fd..e7c3d99 100644 --- a/internal/domain/clientpolicy/policy_test.go +++ b/internal/domain/clientpolicy/policy_test.go @@ -19,6 +19,7 @@ func TestValidate(t *testing.T) { MaxExtractCount: 10, AllowedUpstreams: []string{"provider-a", "provider-b"}, AllowedRegions: []string{"shanghai", "beijing"}, + AllowedRoutings: []string{"checkout", "catalog"}, }}, {name: "negative max count", policy: Policy{MaxExtractCount: -1}, wantErr: true}, {name: "empty upstream", policy: Policy{AllowedUpstreams: []string{""}}, wantErr: true}, @@ -26,6 +27,7 @@ func TestValidate(t *testing.T) { {name: "duplicate upstream", policy: Policy{AllowedUpstreams: []string{"provider-a", "provider-a"}}, wantErr: true}, {name: "empty region", policy: Policy{AllowedRegions: []string{""}}, wantErr: true}, {name: "duplicate region", policy: Policy{AllowedRegions: []string{"shanghai", "shanghai"}}, wantErr: true}, + {name: "duplicate routing", policy: Policy{AllowedRoutings: []string{"checkout", "checkout"}}, wantErr: true}, } for _, test := range tests { @@ -90,3 +92,18 @@ func TestPolicyWithoutRestrictionsPreservesRequestAndClones(t *testing.T) { t.Fatal("RestrictUpstreams() aliases caller input") } } + +func TestPolicyRestrictsGatewayRoutings(t *testing.T) { + t.Parallel() + + policy := Policy{AllowedRoutings: []string{"checkout"}} + if !policy.AllowsRouting("checkout") { + t.Fatal("AllowsRouting() rejected the configured routing") + } + if policy.AllowsRouting("catalog") { + t.Fatal("AllowsRouting() accepted a routing outside the credential policy") + } + if !(Policy{}).AllowsRouting("catalog") { + t.Fatal("empty policy must preserve legacy gateway access") + } +} diff --git a/internal/gateway/server/handler.go b/internal/gateway/server/handler.go index 741002f..31f6eb8 100644 --- a/internal/gateway/server/handler.go +++ b/internal/gateway/server/handler.go @@ -192,6 +192,10 @@ func (handler *Handler) ServeHTTP(writer http.ResponseWriter, request *http.Requ writeGatewayError(writer, fmt.Errorf("route gateway CONNECT: %w", err)) return } + if err := authorizeClientRouting(request, route); err != nil { + writeGatewayError(writer, err) + return + } handler.connect(writer, request, target, route) return } @@ -209,6 +213,10 @@ func (handler *Handler) ServeHTTP(writer http.ResponseWriter, request *http.Requ writeGatewayError(writer, fmt.Errorf("route gateway request: %w", err)) return } + if err := authorizeClientRouting(request, route); err != nil { + writeGatewayError(writer, err) + return + } handler.forwardHTTP(writer, request, target, route) } diff --git a/internal/gateway/server/handler_test.go b/internal/gateway/server/handler_test.go index 3b6d184..5677ca9 100644 --- a/internal/gateway/server/handler_test.go +++ b/internal/gateway/server/handler_test.go @@ -15,6 +15,7 @@ import ( "testing" "time" + "proxy-pool/internal/domain/clientpolicy" outcomeDomain "proxy-pool/internal/domain/outcome" proxyDomain "proxy-pool/internal/domain/proxy" "proxy-pool/internal/domain/routing" @@ -22,6 +23,7 @@ import ( "proxy-pool/internal/gateway/policy" "proxy-pool/internal/gateway/snapshot" transportDomain "proxy-pool/internal/gateway/transport" + "proxy-pool/internal/platform/httpsecurity" ) func TestHandlerRunsProtectionAndTargetPolicyBeforeRouting(t *testing.T) { @@ -67,6 +69,48 @@ func TestHandlerRunsProtectionAndTargetPolicyBeforeRouting(t *testing.T) { } } +func TestHandlerRejectsGatewayRoutingOutsideCredentialPolicy(t *testing.T) { + t.Parallel() + + authentication, err := httpsecurity.New(httpsecurity.Config{ + Authentication: httpsecurity.Authentication{ + Mode: httpsecurity.ModeBearer, Token: "gateway-token", + ClientPolicy: clientpolicy.Policy{AllowedRoutings: []string{"checkout"}}, + }, + ClientIdentification: httpsecurity.ClientAuthenticated, + Semantics: httpsecurity.ProxySemantics, + }, nil) + if err != nil { + t.Fatalf("New HTTP protection: %v", err) + } + handler, err := New(Config{}, Dependencies{ + Auth: authentication, + Targets: fakeTargets{evaluateURL: func(context.Context, string) (policy.Authority, error) { + return policy.Authority{Host: "example.test", Port: 80}, nil + }}, + Router: RouteFunc(func(*http.Request) (dispatch.Request, error) { + return dispatch.Request{RoutingName: "catalog", Upstreams: []string{"provider-a"}}, nil + }), + Dispatcher: DispatcherFunc(func(dispatch.Request) (*dispatch.Lease, error) { + t.Fatal("dispatcher must not run for a route outside the credential policy") + return nil, nil + }), + Transport: &fakeTransport{}, + }) + if err != nil { + t.Fatalf("New gateway handler: %v", err) + } + request := httptest.NewRequest(http.MethodGet, "http://example.test/catalog", nil) + request.Header.Set("Proxy-Authorization", "Bearer gateway-token") + response := httptest.NewRecorder() + + handler.ServeHTTP(response, request) + + if response.Code != http.StatusForbidden { + t.Fatalf("status = %d, want 403", response.Code) + } +} + func TestHandlerRetriesGETWithAnotherProxyBeforeResponseCommit(t *testing.T) { t.Parallel() diff --git a/internal/gateway/server/routing.go b/internal/gateway/server/routing.go index a985704..1ddd729 100644 --- a/internal/gateway/server/routing.go +++ b/internal/gateway/server/routing.go @@ -11,11 +11,13 @@ import ( "proxy-pool/internal/domain/routing" "proxy-pool/internal/gateway/dispatch" "proxy-pool/internal/gateway/snapshot" + "proxy-pool/internal/platform/httpsecurity" ) var ( ErrRouteNotFound = errors.New("no gateway routing rule matched") ErrRouteRejected = errors.New("gateway routing rule rejected the request") + ErrRouteForbidden = errors.New("gateway routing is not permitted for the authenticated client") ErrDirectRouteUnsupported = errors.New("direct gateway routing is not implemented") ) @@ -23,6 +25,14 @@ type RulesRouter struct { rules *routing.RuleSet } +func authorizeClientRouting(request *http.Request, route dispatch.Request) error { + identity, authenticated := httpsecurity.IdentityFromRequest(request) + if !authenticated || identity.ClientPolicy.AllowsRouting(route.RoutingName) { + return nil + } + return &HTTPError{StatusCode: http.StatusForbidden, Cause: ErrRouteForbidden} +} + func NewRulesRouter(rules *routing.RuleSet) *RulesRouter { return &RulesRouter{rules: rules} } diff --git a/internal/platform/httpsecurity/protection.go b/internal/platform/httpsecurity/protection.go index e07a21e..cb8eac8 100644 --- a/internal/platform/httpsecurity/protection.go +++ b/internal/platform/httpsecurity/protection.go @@ -57,8 +57,12 @@ func (protection *Protection) Resolve(request *http.Request) (Identity, error) { } func (protection *Protection) Check(ctx context.Context, request *http.Request) error { - _, err := protection.evaluate(ctx, request) - return err + identity, err := protection.evaluate(ctx, request) + if err != nil { + return err + } + attachIdentity(request, identity) + return nil } func (protection *Protection) evaluate(ctx context.Context, request *http.Request) (Identity, error) { diff --git a/internal/platform/httpsecurity/security_test.go b/internal/platform/httpsecurity/security_test.go index b56eae8..1cb515d 100644 --- a/internal/platform/httpsecurity/security_test.go +++ b/internal/platform/httpsecurity/security_test.go @@ -177,6 +177,34 @@ func TestProtectionCarriesClientPolicyFromMatchedCredential(t *testing.T) { } } +func TestProtectionCheckAttachesDetachedIdentityToRequest(t *testing.T) { + t.Parallel() + + protection := mustProtection(t, Config{ + Authentication: Authentication{ + Mode: ModeBearer, Token: "gateway-token", + ClientPolicy: clientpolicy.Policy{AllowedRoutings: []string{"checkout"}}, + }, + ClientIdentification: ClientAuthenticated, + }, nil) + request := newRequest() + request.Header.Set("Authorization", "Bearer gateway-token") + + if err := protection.Check(request.Context(), request); err != nil { + t.Fatalf("Check() error = %v", err) + } + identity, ok := IdentityFromRequest(request) + if !ok || identity.ClientID != credentialSubject(ModeBearer, "gateway-token") || + !slices.Equal(identity.ClientPolicy.AllowedRoutings, []string{"checkout"}) { + t.Fatalf("IdentityFromRequest() = (%+v, %v)", identity, ok) + } + identity.ClientPolicy.AllowedRoutings[0] = "mutated" + next, ok := IdentityFromRequest(request) + if !ok || next.ClientPolicy.AllowedRoutings[0] != "checkout" { + t.Fatalf("IdentityFromRequest() returned aliased policy: %+v", next.ClientPolicy) + } +} + func TestProtectionAnyPreservesSourceRejection(t *testing.T) { t.Parallel() tests := []struct { diff --git a/internal/platform/httpsecurity/types.go b/internal/platform/httpsecurity/types.go index c2d563c..a392103 100644 --- a/internal/platform/httpsecurity/types.go +++ b/internal/platform/httpsecurity/types.go @@ -29,6 +29,8 @@ const ( var ErrInvalidConfig = errors.New("invalid HTTP security configuration") +type identityContextKey struct{} + type Semantics uint8 type Authentication struct { @@ -76,6 +78,30 @@ func (identity Identity) Allows(required string) bool { return authorization.Allows(identity.Permissions, required) } +// IdentityFromRequest returns the authenticated identity attached by +// Protection.Check. The returned value is detached from request state. +func IdentityFromRequest(request *http.Request) (Identity, bool) { + if request == nil { + return Identity{}, false + } + identity, ok := request.Context().Value(identityContextKey{}).(Identity) + if !ok { + return Identity{}, false + } + identity.Permissions = append([]string(nil), identity.Permissions...) + identity.ClientPolicy = identity.ClientPolicy.Clone() + return identity, true +} + +func attachIdentity(request *http.Request, identity Identity) { + if request == nil { + return + } + identity.Permissions = append([]string(nil), identity.Permissions...) + identity.ClientPolicy = identity.ClientPolicy.Clone() + *request = *request.WithContext(context.WithValue(request.Context(), identityContextKey{}, identity)) +} + type Admitter interface { Admit(context.Context, string) error }