diff --git a/README.md b/README.md index 733ade2..23b0080 100644 --- a/README.md +++ b/README.md @@ -86,7 +86,8 @@ Proxy Pool 用 Controller 协调这些变化,并让 Gateway 数据面只消费 Distribution 提取权限可按命中凭据分别收敛。Distribution 凭据还可限制单次 提取数量、可访问 Upstream 与地区,Gateway 凭据可限制可访问 Routing;未配置时 保持既有全范围行为。Gateway 凭据还可在每个 Worker 内限制每分钟请求数, - 不把该热路径计数写入 Redis 或 PostgreSQL。 + 同时限制 HTTP 请求和 CONNECT 隧道的并发数;热路径计数不写入 Redis 或 + PostgreSQL。 ## 架构概览 diff --git a/docs/configuration/reference.md b/docs/configuration/reference.md index bbf7689..8a6b2c1 100644 --- a/docs/configuration/reference.md +++ b/docs/configuration/reference.md @@ -181,11 +181,12 @@ distribution: ### 3.4 Gateway Client 路由约束 -Gateway 的 auth.client 支持 allowedRoutings 和 requestsPerMinute。认证成功后, +Gateway 的 auth.client 支持 allowedRoutings、requestsPerMinute 和 +maxConcurrentConnections。认证成功后, Gateway 先完成目标地址策略和本地 Routing 匹配,再于本地 Dispatcher 前检查 -Routing 集合;不匹配返回 403,不会尝试选择或预留 Proxy。速率限制以认证 Client -为计数键,在每个 Gateway Worker 内执行,超过限制返回 429。两项检查都只读取 -请求上下文和当前本地状态,不访问 Redis、PostgreSQL 或 Provider。 +Routing 集合;不匹配返回 403,不会尝试选择或预留 Proxy。速率和并发限制均以 +认证 Client 为计数键,在每个 Gateway Worker 内执行,超过限制返回 429。三项 +检查都只读取请求上下文和当前本地状态,不访问 Redis、PostgreSQL 或 Provider。 allowedRoutings 中的每个名称必须对应一个启用的 Gateway Routing。Gateway 上的 maxExtractCount、allowedUpstreams 和 allowedRegions,以及 Distribution 上的 @@ -202,6 +203,7 @@ gateway: client: allowedRoutings: [checkout] requestsPerMinute: 600 + maxConcurrentConnections: 20 ~~~ Gateway、Distribution、Admin 与 Provider API 是独立认证边界。改变其中一套 diff --git a/docs/security/security-model.md b/docs/security/security-model.md index d13918c..f8eb3fb 100644 --- a/docs/security/security-model.md +++ b/docs/security/security-model.md @@ -28,6 +28,8 @@ 代理可用期或转发失败时失效。 - Gateway 凭据的 requestsPerMinute 限制按认证 Client 在本 Worker 内执行,不把 认证身份、会话或请求计数写入 PostgreSQL、Redis 或 Prometheus 标签。 +- Gateway 凭据的 maxConcurrentConnections 在请求或 CONNECT 隧道的完整生命周期 + 内保留一个本地租约;无论转发成功、失败还是服务排空,租约都会释放。 ## 3. 目标地址策略 diff --git a/internal/config/config_test.go b/internal/config/config_test.go index 298f678..db1573e 100644 --- a/internal/config/config_test.go +++ b/internal/config/config_test.go @@ -413,10 +413,15 @@ func TestValidateDistributionCredentialClientPolicy(t *testing.T) { t.Fatalf("Validate(gateway request rate policy) error = %v", err) } + cfg.Gateway.Auth.ClientPolicy = clientpolicy.Policy{MaxConcurrentConnections: 3} + if err := Validate(cfg); err != nil { + t.Fatalf("Validate(gateway concurrency policy) error = %v", err) + } + cfg.Gateway.Auth.ClientPolicy = clientpolicy.Policy{} - cfg.Distribution.Auth.ClientPolicy = clientpolicy.Policy{RequestsPerMinute: 60} + cfg.Distribution.Auth.ClientPolicy = clientpolicy.Policy{MaxConcurrentConnections: 3} if err := Validate(cfg); err == nil || !strings.Contains(err.Error(), "only supported on gateway") { - t.Fatalf("Validate(distribution request rate policy) error = %v", err) + t.Fatalf("Validate(distribution gateway limit policy) error = %v", err) } } diff --git a/internal/config/validate.go b/internal/config/validate.go index 2c67dae..5ef8763 100644 --- a/internal/config/validate.go +++ b/internal/config/validate.go @@ -552,8 +552,8 @@ func validateDistributionClientPolicies(auth Auth, upstreams map[string]Upstream } func validateDistributionClientPolicy(scope string, policy clientpolicy.Policy, upstreams map[string]Upstream) error { - if len(policy.AllowedRoutings) != 0 || policy.RequestsPerMinute != 0 { - return fmt.Errorf("validate %s: routing access and request rate limits are only supported on gateway", scope) + if len(policy.AllowedRoutings) != 0 || policy.RequestsPerMinute != 0 || policy.MaxConcurrentConnections != 0 { + return fmt.Errorf("validate %s: routing access and Gateway limits are only supported on gateway", scope) } for _, upstream := range policy.AllowedUpstreams { if _, exists := upstreams[upstream]; !exists { @@ -584,6 +584,9 @@ func validateGatewayClientPolicy(scope string, policy clientpolicy.Policy, routi if int64(policy.RequestsPerMinute) > MaximumExactCounter { return fmt.Errorf("validate %s.requestsPerMinute: exceeds exact counter range", scope) } + if int64(policy.MaxConcurrentConnections) > MaximumExactCounter { + return fmt.Errorf("validate %s.maxConcurrentConnections: exceeds exact counter range", 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) diff --git a/internal/domain/clientpolicy/policy.go b/internal/domain/clientpolicy/policy.go index 2a448c8..3324ba9 100644 --- a/internal/domain/clientpolicy/policy.go +++ b/internal/domain/clientpolicy/policy.go @@ -12,20 +12,21 @@ var ErrInvalidPolicy = errors.New("invalid client policy") // 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"` - RequestsPerMinute int `yaml:"requestsPerMinute"` - AllowedUpstreams []string `yaml:"allowedUpstreams"` - AllowedRegions []string `yaml:"allowedRegions"` - AllowedRoutings []string `yaml:"allowedRoutings"` + MaxExtractCount int `yaml:"maxExtractCount"` + RequestsPerMinute int `yaml:"requestsPerMinute"` + MaxConcurrentConnections int `yaml:"maxConcurrentConnections"` + AllowedUpstreams []string `yaml:"allowedUpstreams"` + AllowedRegions []string `yaml:"allowedRegions"` + AllowedRoutings []string `yaml:"allowedRoutings"` } func (policy Policy) IsZero() bool { - return policy.MaxExtractCount == 0 && policy.RequestsPerMinute == 0 && len(policy.AllowedUpstreams) == 0 && + return policy.MaxExtractCount == 0 && policy.RequestsPerMinute == 0 && policy.MaxConcurrentConnections == 0 && len(policy.AllowedUpstreams) == 0 && len(policy.AllowedRegions) == 0 && len(policy.AllowedRoutings) == 0 } func (policy Policy) Validate() error { - if policy.MaxExtractCount < 0 || policy.RequestsPerMinute < 0 || + if policy.MaxExtractCount < 0 || policy.RequestsPerMinute < 0 || policy.MaxConcurrentConnections < 0 || !validUniqueValues(policy.AllowedUpstreams) || !validUniqueValues(policy.AllowedRegions) || !validUniqueValues(policy.AllowedRoutings) { diff --git a/internal/domain/clientpolicy/policy_test.go b/internal/domain/clientpolicy/policy_test.go index d776abb..0516f4f 100644 --- a/internal/domain/clientpolicy/policy_test.go +++ b/internal/domain/clientpolicy/policy_test.go @@ -16,14 +16,16 @@ func TestValidate(t *testing.T) { }{ {name: "empty policy", policy: Policy{}}, {name: "bounded policy", policy: Policy{ - MaxExtractCount: 10, - RequestsPerMinute: 600, - AllowedUpstreams: []string{"provider-a", "provider-b"}, - AllowedRegions: []string{"shanghai", "beijing"}, - AllowedRoutings: []string{"checkout", "catalog"}, + MaxExtractCount: 10, + RequestsPerMinute: 600, + MaxConcurrentConnections: 3, + 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: "negative request rate", policy: Policy{RequestsPerMinute: -1}, wantErr: true}, + {name: "negative concurrency", policy: Policy{MaxConcurrentConnections: -1}, wantErr: true}, {name: "empty upstream", policy: Policy{AllowedUpstreams: []string{""}}, wantErr: true}, {name: "whitespace upstream", policy: Policy{AllowedUpstreams: []string{" provider-a"}}, wantErr: true}, {name: "duplicate upstream", policy: Policy{AllowedUpstreams: []string{"provider-a", "provider-a"}}, wantErr: true}, diff --git a/internal/gateway/server/bootstrap.go b/internal/gateway/server/bootstrap.go index d7e6126..0361b16 100644 --- a/internal/gateway/server/bootstrap.go +++ b/internal/gateway/server/bootstrap.go @@ -52,14 +52,18 @@ func BuildProtection(listener config.Listener) (Protection, error) { return Protection{Auth: auth, Access: access, Admission: chainGuards(admissionGuards...), ClientIPs: clientIPs}, nil } -func buildCredentialAdmission(authentication config.Auth) (*CredentialAdmissionGuard, error) { - limits := make(map[int]Admitter) +func buildCredentialAdmission(authentication config.Auth) (Guard, error) { + rateLimits := make(map[int]Admitter) + concurrencyLimits := make(map[int]struct{}) for _, policy := range credentialPolicies(authentication) { + if policy.MaxConcurrentConnections > 0 { + concurrencyLimits[policy.MaxConcurrentConnections] = struct{}{} + } limit := policy.RequestsPerMinute if limit == 0 { continue } - if _, exists := limits[limit]; exists { + if _, exists := rateLimits[limit]; exists { continue } limiter, err := platformAdmission.NewFixedWindow(platformAdmission.FixedWindowConfig{ @@ -69,9 +73,12 @@ func buildCredentialAdmission(authentication config.Auth) (*CredentialAdmissionG if err != nil { return nil, fmt.Errorf("build gateway credential admission: %w", err) } - limits[limit] = limiter + rateLimits[limit] = limiter } - return NewCredentialAdmissionGuard(limits), nil + return chainGuards( + NewCredentialAdmissionGuard(rateLimits), + NewCredentialConcurrencyGuard(concurrencyLimits), + ), nil } func credentialPolicies(authentication config.Auth) []clientpolicy.Policy { diff --git a/internal/gateway/server/bootstrap_test.go b/internal/gateway/server/bootstrap_test.go index ddc5c12..fbe8073 100644 --- a/internal/gateway/server/bootstrap_test.go +++ b/internal/gateway/server/bootstrap_test.go @@ -99,6 +99,46 @@ func TestBuildProtectionAppliesCredentialRequestRateLimit(t *testing.T) { } } +func TestBuildProtectionAppliesCredentialConcurrentConnectionLimit(t *testing.T) { + t.Parallel() + + protection, err := BuildProtection(config.Listener{ + Auth: config.Auth{ + Mode: "bearer", Token: "gateway-token", + ClientPolicy: clientpolicy.Policy{MaxConcurrentConnections: 1}, + }, + }) + if err != nil { + t.Fatalf("BuildProtection() error = %v", err) + } + newRequest := func() *http.Request { + request := httptest.NewRequest(http.MethodGet, "http://example.test", nil) + request.RemoteAddr = "198.51.100.8:1234" + request.Header.Set("Proxy-Authorization", "Bearer gateway-token") + if err := protection.Auth.Check(context.Background(), request); err != nil { + t.Fatalf("Auth.Check() error = %v", err) + } + return request + } + first := newRequest() + if err := protection.Admission.Check(context.Background(), first); err != nil { + t.Fatalf("Admission.Check(first) error = %v", err) + } + second := newRequest() + err = protection.Admission.Check(context.Background(), second) + var admissionError *HTTPError + if !errors.As(err, &admissionError) || admissionError.StatusCode != http.StatusTooManyRequests { + t.Fatalf("Admission.Check(second) error = %T %v, want 429", err, err) + } + + releaseCredentialReservation(first) + third := newRequest() + if err := protection.Admission.Check(context.Background(), third); err != nil { + t.Fatalf("Admission.Check(after release) error = %v", err) + } + releaseCredentialReservation(third) +} + func TestBuildProtectionAnyPreservesIPWhitelistRejectionIndependentOfOrder(t *testing.T) { t.Parallel() methods := [][]config.AuthMethod{ diff --git a/internal/gateway/server/handler.go b/internal/gateway/server/handler.go index 411f14c..0900a4d 100644 --- a/internal/gateway/server/handler.go +++ b/internal/gateway/server/handler.go @@ -187,6 +187,7 @@ func (handler *Handler) ServeHTTP(writer http.ResponseWriter, request *http.Requ return } } + defer releaseCredentialReservation(request) if request.Method == http.MethodConnect { target, err := handler.targets.EvaluateConnectAuthority(request.Context(), request.Host) diff --git a/internal/gateway/server/handler_test.go b/internal/gateway/server/handler_test.go index 9a9e90e..f424579 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/config" "proxy-pool/internal/domain/clientpolicy" outcomeDomain "proxy-pool/internal/domain/outcome" proxyDomain "proxy-pool/internal/domain/proxy" @@ -111,6 +112,71 @@ func TestHandlerRejectsGatewayRoutingOutsideCredentialPolicy(t *testing.T) { } } +func TestHandlerReleasesCredentialConcurrencyAfterRequest(t *testing.T) { + t.Parallel() + + protection, err := BuildProtection(config.Listener{ + Auth: config.Auth{Mode: "bearer", Token: "gateway-token", + ClientPolicy: clientpolicy.Policy{MaxConcurrentConnections: 1}}, + }) + if err != nil { + t.Fatalf("BuildProtection() error = %v", err) + } + dispatcher, view := dispatcherWithProxies(t, "proxy-a") + started := make(chan struct{}) + release := make(chan struct{}) + transport := &fakeTransport{roundTrip: func( + _ context.Context, _ proxyDomain.Proxy, _ *http.Request, commit ...func() error, + ) (*http.Response, error) { + if err := commit[0](); err != nil { + return nil, err + } + select { + case started <- struct{}{}: + default: + } + <-release + return &http.Response{StatusCode: http.StatusNoContent, Header: make(http.Header), Body: http.NoBody}, nil + }} + handler, err := New(Config{}, Dependencies{ + Auth: protection.Auth, Admission: protection.Admission, Targets: fakeTargets{}, + Router: RouteFunc(func(*http.Request) (dispatch.Request, error) { + return dispatch.Request{Upstreams: []string{"provider-a"}}, nil + }), + Dispatcher: dispatcher, Transport: transport, + }) + if err != nil { + t.Fatalf("New() error = %v", err) + } + request := func() *http.Request { + result := httptest.NewRequest(http.MethodGet, "http://example.test/resource", nil) + result.Header.Set("Proxy-Authorization", "Bearer gateway-token") + return result + } + first := httptest.NewRecorder() + done := make(chan struct{}) + go func() { + handler.ServeHTTP(first, request()) + close(done) + }() + <-started + + second := httptest.NewRecorder() + handler.ServeHTTP(second, request()) + if second.Code != http.StatusTooManyRequests { + t.Fatalf("second status = %d, want 429", second.Code) + } + close(release) + <-done + + third := httptest.NewRecorder() + handler.ServeHTTP(third, request()) + if third.Code != http.StatusNoContent { + t.Fatalf("third status = %d, want 204", third.Code) + } + assertNoLeakedCapacity(t, view) +} + func TestHandlerPinsAuthenticatedSessionToCommittedProxyAndStripsHeader(t *testing.T) { t.Parallel() diff --git a/internal/gateway/server/protection.go b/internal/gateway/server/protection.go index 421cf5d..5a8fc8c 100644 --- a/internal/gateway/server/protection.go +++ b/internal/gateway/server/protection.go @@ -5,6 +5,7 @@ import ( "errors" "fmt" "net/http" + "sync" "proxy-pool/internal/gateway/policy" "proxy-pool/internal/platform/httpsecurity" @@ -91,6 +92,85 @@ type CredentialAdmissionGuard struct { limiters map[int]Admitter } +type credentialReservationContextKey struct{} + +type clientConcurrency struct { + max int + mu sync.Mutex + use map[string]int +} + +func (limiter *clientConcurrency) acquire(clientID string) (func(), bool) { + limiter.mu.Lock() + defer limiter.mu.Unlock() + if limiter.use[clientID] >= limiter.max { + return nil, false + } + limiter.use[clientID]++ + var once sync.Once + return func() { + once.Do(func() { + limiter.mu.Lock() + defer limiter.mu.Unlock() + if limiter.use[clientID] <= 1 { + delete(limiter.use, clientID) + return + } + limiter.use[clientID]-- + }) + }, true +} + +type CredentialConcurrencyGuard struct { + limiters map[int]*clientConcurrency +} + +func NewCredentialConcurrencyGuard(limits map[int]struct{}) *CredentialConcurrencyGuard { + if len(limits) == 0 { + return nil + } + limiters := make(map[int]*clientConcurrency, len(limits)) + for limit := range limits { + if limit > 0 { + limiters[limit] = &clientConcurrency{max: limit, use: make(map[string]int)} + } + } + if len(limiters) == 0 { + return nil + } + return &CredentialConcurrencyGuard{limiters: limiters} +} + +func (guard *CredentialConcurrencyGuard) Check(_ context.Context, request *http.Request) error { + if guard == nil { + return nil + } + identity, authenticated := httpsecurity.IdentityFromRequest(request) + if !authenticated || identity.ClientPolicy.MaxConcurrentConnections == 0 { + return nil + } + limiter := guard.limiters[identity.ClientPolicy.MaxConcurrentConnections] + if limiter == nil || identity.ClientID == "" { + return &HTTPError{StatusCode: http.StatusInternalServerError, Cause: errors.New("gateway credential concurrency is not configured")} + } + release, acquired := limiter.acquire(identity.ClientID) + if !acquired { + return &HTTPError{StatusCode: http.StatusTooManyRequests, Cause: errors.New("gateway credential concurrency limit exceeded")} + } + *request = *request.WithContext(context.WithValue(request.Context(), credentialReservationContextKey{}, release)) + return nil +} + +func releaseCredentialReservation(request *http.Request) { + if request == nil { + return + } + release, found := request.Context().Value(credentialReservationContextKey{}).(func()) + if found && release != nil { + release() + } +} + func NewCredentialAdmissionGuard(limiters map[int]Admitter) *CredentialAdmissionGuard { if len(limiters) == 0 { return nil