youfak/proxy-pool
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat: restrict gateway routes by credential

Browse Source
This commit is contained in:
youfak 2026-08-02 16:21:10 +08:00
parent f644737435
commit 5389d6f886
14 changed files with 247 additions and 24 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

3
README.md
View File

@ -80,7 +80,8 @@ Proxy Pool 用 Controller 协调这些变化，并让 Gateway 数据面只消费
- **安全边界**：Gateway、Distribution 与 Admin 使用各自的认证语义，并支持
CIDR、可信代理、严格请求解析和敏感信息最小化；Admin 的读写权限与
Distribution 提取权限可按命中凭据分别收敛。Distribution 凭据还可限制单次
提取数量、可访问 Upstream 与地区；未配置时保持既有全范围行为。
提取数量、可访问 Upstream 与地区，Gateway 凭据可限制可访问 Routing；未配置时
保持既有全范围行为。
## 架构概览

29
docs/configuration/reference.md
View File

@ -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 是独立认证边界。改变其中一套
不得连带改变其他入口。

3
docs/development/implementation-plan.md
View File

@ -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.

5
docs/requirements/completion-audit.md
View File

@ -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
运行同一公用契约。

13
internal/config/config_test.go
View File

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

53
internal/config/validate.go
View File

@ -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 {

24
internal/domain/clientpolicy/policy.go
View File

@ -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

17
internal/domain/clientpolicy/policy_test.go
View File

@ -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")
}
}

8
internal/gateway/server/handler.go
View File

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

44
internal/gateway/server/handler_test.go
View File

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

10
internal/gateway/server/routing.go
View File

@ -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}
}

8
internal/platform/httpsecurity/protection.go
View File

@ -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) {

28
internal/platform/httpsecurity/security_test.go
View File

@ -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 {

26
internal/platform/httpsecurity/types.go
View File

@ -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
}