From 2ea5be503663a9fe2fdc1cda61da13a5acc09d11 Mon Sep 17 00:00:00 2001 From: youfak Date: Wed, 29 Jul 2026 10:14:56 +0800 Subject: [PATCH] feat: add controller HTTP adapters --- api/openapi/admin.yaml | 91 ++++ api/openapi/openapi_test.go | 23 +- api/openapi/proxy-pool.yaml | 57 +++ docs/api/admin.md | 21 + docs/api/distribution.md | 14 + docs/development/implementation-plan.md | 6 + docs/requirements/completion-audit.md | 12 +- internal/controller/admin/handler.go | 265 ++++++++++ internal/controller/admin/handler_test.go | 238 +++++++++ internal/controller/distribution/handler.go | 369 ++++++++++++++ .../controller/distribution/handler_test.go | 472 ++++++++++++++++++ internal/controller/extraction/service.go | 1 + internal/platform/httpapi/httpapi.go | 140 ++++++ internal/platform/httpapi/httpapi_test.go | 151 ++++++ 14 files changed, 1856 insertions(+), 4 deletions(-) create mode 100644 internal/controller/admin/handler.go create mode 100644 internal/controller/admin/handler_test.go create mode 100644 internal/controller/distribution/handler.go create mode 100644 internal/controller/distribution/handler_test.go create mode 100644 internal/platform/httpapi/httpapi.go create mode 100644 internal/platform/httpapi/httpapi_test.go diff --git a/api/openapi/admin.yaml b/api/openapi/admin.yaml index 283215b..b03c2e5 100644 --- a/api/openapi/admin.yaml +++ b/api/openapi/admin.yaml @@ -23,12 +23,18 @@ paths: responses: '200': description: 不含 Proxy 地址、Client 标识或 Secret 的聚合状态 + headers: + X-Request-ID: {$ref: '#/components/headers/RequestID'} content: application/json: schema: $ref: '#/components/schemas/Status' '401': {$ref: '#/components/responses/Unauthorized'} '403': {$ref: '#/components/responses/Forbidden'} + '400': {$ref: '#/components/responses/BadRequest'} + '405': {$ref: '#/components/responses/MethodNotAllowed'} + '500': {$ref: '#/components/responses/InternalServerError'} + '503': {$ref: '#/components/responses/ServiceUnavailable'} /api/v1/upstreams/{name}/enable: post: tags: [Upstreams] @@ -43,6 +49,10 @@ paths: '403': {$ref: '#/components/responses/Forbidden'} '404': {$ref: '#/components/responses/NotFound'} '409': {$ref: '#/components/responses/Conflict'} + '400': {$ref: '#/components/responses/BadRequest'} + '405': {$ref: '#/components/responses/MethodNotAllowed'} + '500': {$ref: '#/components/responses/InternalServerError'} + '503': {$ref: '#/components/responses/ServiceUnavailable'} /api/v1/upstreams/{name}/disable: post: tags: [Upstreams] @@ -57,6 +67,10 @@ paths: '403': {$ref: '#/components/responses/Forbidden'} '404': {$ref: '#/components/responses/NotFound'} '409': {$ref: '#/components/responses/Conflict'} + '400': {$ref: '#/components/responses/BadRequest'} + '405': {$ref: '#/components/responses/MethodNotAllowed'} + '500': {$ref: '#/components/responses/InternalServerError'} + '503': {$ref: '#/components/responses/ServiceUnavailable'} /api/v1/routing/{name}/switch: post: tags: [Routing] @@ -86,6 +100,13 @@ paths: '403': {$ref: '#/components/responses/Forbidden'} '404': {$ref: '#/components/responses/NotFound'} '409': {$ref: '#/components/responses/Conflict'} + '400': {$ref: '#/components/responses/BadRequest'} + '405': {$ref: '#/components/responses/MethodNotAllowed'} + '413': {$ref: '#/components/responses/RequestEntityTooLarge'} + '415': {$ref: '#/components/responses/UnsupportedMediaType'} + '422': {$ref: '#/components/responses/UnprocessableEntity'} + '500': {$ref: '#/components/responses/InternalServerError'} + '503': {$ref: '#/components/responses/ServiceUnavailable'} /api/v1/config/reload: post: tags: [Configuration] @@ -98,11 +119,17 @@ paths: '401': {$ref: '#/components/responses/Unauthorized'} '403': {$ref: '#/components/responses/Forbidden'} '409': {$ref: '#/components/responses/Conflict'} + '400': {$ref: '#/components/responses/BadRequest'} + '405': {$ref: '#/components/responses/MethodNotAllowed'} '422': description: 新配置无效，旧配置继续运行 + headers: + X-Request-ID: {$ref: '#/components/headers/RequestID'} content: application/problem+json: schema: {$ref: '#/components/schemas/Problem'} + '500': {$ref: '#/components/responses/InternalServerError'} + '503': {$ref: '#/components/responses/ServiceUnavailable'} components: securitySchemes: AdminApiKey: {type: apiKey, in: header, name: X-Admin-Key} @@ -119,6 +146,10 @@ components: in: header required: false schema: {type: string, maxLength: 128} + headers: + RequestID: + description: 服务端最终使用的请求标识。 + schema: {type: string, maxLength: 128} schemas: Status: type: object @@ -177,26 +208,86 @@ components: responses: MutationResult: description: 操作已提交或目标状态原本已满足 + headers: + X-Request-ID: {$ref: '#/components/headers/RequestID'} content: application/json: schema: {$ref: '#/components/schemas/MutationResult'} Unauthorized: description: 管理入口认证失败 + headers: + X-Request-ID: {$ref: '#/components/headers/RequestID'} content: application/problem+json: schema: {$ref: '#/components/schemas/Problem'} Forbidden: description: 调用主体无该管理权限 + headers: + X-Request-ID: {$ref: '#/components/headers/RequestID'} content: application/problem+json: schema: {$ref: '#/components/schemas/Problem'} NotFound: description: Upstream 或 Routing 不存在 + headers: + X-Request-ID: {$ref: '#/components/headers/RequestID'} content: application/problem+json: schema: {$ref: '#/components/schemas/Problem'} Conflict: description: 预期版本或 expectedCurrent 与权威状态不一致 + headers: + X-Request-ID: {$ref: '#/components/headers/RequestID'} + content: + application/problem+json: + schema: {$ref: '#/components/schemas/Problem'} + BadRequest: + description: 请求标识或请求体无效 + headers: + X-Request-ID: {$ref: '#/components/headers/RequestID'} + content: + application/problem+json: + schema: {$ref: '#/components/schemas/Problem'} + MethodNotAllowed: + description: 端点不支持该 HTTP 方法 + headers: + X-Request-ID: {$ref: '#/components/headers/RequestID'} + Allow: {schema: {type: string}} + content: + application/problem+json: + schema: {$ref: '#/components/schemas/Problem'} + RequestEntityTooLarge: + description: 请求体超过管理入口限制 + headers: + X-Request-ID: {$ref: '#/components/headers/RequestID'} + content: + application/problem+json: + schema: {$ref: '#/components/schemas/Problem'} + UnsupportedMediaType: + description: Content-Type 不是 application/json + headers: + X-Request-ID: {$ref: '#/components/headers/RequestID'} + content: + application/problem+json: + schema: {$ref: '#/components/schemas/Problem'} + UnprocessableEntity: + description: 命令字段或配置内容违反业务约束 + headers: + X-Request-ID: {$ref: '#/components/headers/RequestID'} + content: + application/problem+json: + schema: {$ref: '#/components/schemas/Problem'} + InternalServerError: + description: 未分类的控制面错误 + headers: + X-Request-ID: {$ref: '#/components/headers/RequestID'} + content: + application/problem+json: + schema: {$ref: '#/components/schemas/Problem'} + ServiceUnavailable: + description: 权威控制面暂时不可用 + headers: + X-Request-ID: {$ref: '#/components/headers/RequestID'} content: application/problem+json: schema: {$ref: '#/components/schemas/Problem'} diff --git a/api/openapi/openapi_test.go b/api/openapi/openapi_test.go index d558bdb..26208a1 100644 --- a/api/openapi/openapi_test.go +++ b/api/openapi/openapi_test.go @@ -21,9 +21,11 @@ func TestDistributionContract(t *testing.T) { if !ok { t.Fatal("exclusive extraction path is missing") } - if _, ok := extraction["post"]; !ok { + post, ok := extraction["post"] + if !ok { t.Fatal("exclusive extraction must use POST") } + requireResponses(t, post, "200", "400", "409", "413", "415", "422", "429", "500", "503") for path := range spec.Paths { if path == "/api/v1/leases" || path == "/api/v1/proxies/release" || path == "/api/v1/proxies/renew" { t.Fatalf("lease/release path is forbidden: %s", path) @@ -44,6 +46,25 @@ func TestAdminContract(t *testing.T) { t.Errorf("admin path is missing: %s", path) } } + requireResponses(t, spec.Paths["/api/v1/routing/{name}/switch"]["post"], + "200", "400", "401", "403", "404", "405", "409", "413", "415", "422", "500", "503") +} + +func requireResponses(t *testing.T, operation any, codes ...string) { + t.Helper() + operationMap, ok := operation.(map[string]any) + if !ok { + t.Fatalf("operation has type %T, want map", operation) + } + responses, ok := operationMap["responses"].(map[string]any) + if !ok { + t.Fatalf("responses has type %T, want map", operationMap["responses"]) + } + for _, code := range codes { + if _, ok := responses[code]; !ok { + t.Errorf("response %s is missing", code) + } + } } func readDocument(t *testing.T, path string) document { diff --git a/api/openapi/proxy-pool.yaml b/api/openapi/proxy-pool.yaml index 1553645..cdbddbe 100644 --- a/api/openapi/proxy-pool.yaml +++ b/api/openapi/proxy-pool.yaml @@ -73,6 +73,9 @@ paths: $ref: '#/components/responses/Forbidden' '409': description: allOrNothing 模式下符合条件的库存不足，未提取任何代理 + headers: + X-Request-ID: + $ref: '#/components/headers/RequestID' content: application/problem+json: schema: @@ -86,10 +89,16 @@ paths: requestId: req_01J4EXAMPLE '422': $ref: '#/components/responses/UnprocessableEntity' + '413': + $ref: '#/components/responses/RequestEntityTooLarge' + '415': + $ref: '#/components/responses/UnsupportedMediaType' '429': $ref: '#/components/responses/TooManyRequests' '503': $ref: '#/components/responses/ServiceUnavailable' + '500': + $ref: '#/components/responses/InternalServerError' /health/live: get: tags: [Health] @@ -179,22 +188,26 @@ components: properties: protocols: type: array + maxItems: 64 uniqueItems: true items: type: string enum: [http, https, socks5] regions: type: array + maxItems: 64 uniqueItems: true items: type: string carriers: type: array + maxItems: 64 uniqueItems: true items: type: string allowedUpstreams: type: array + maxItems: 64 uniqueItems: true items: type: string @@ -311,24 +324,36 @@ components: responses: BadRequest: description: 请求体、Header 或 JSON 格式无效 + headers: + X-Request-ID: + $ref: '#/components/headers/RequestID' content: application/problem+json: schema: $ref: '#/components/schemas/Problem' Unauthorized: description: 所配置的认证方法未通过 + headers: + X-Request-ID: + $ref: '#/components/headers/RequestID' content: application/problem+json: schema: $ref: '#/components/schemas/Problem' Forbidden: description: 来源访问控制或客户端权限拒绝 + headers: + X-Request-ID: + $ref: '#/components/headers/RequestID' content: application/problem+json: schema: $ref: '#/components/schemas/Problem' UnprocessableEntity: description: 参数语法有效但违反业务约束 + headers: + X-Request-ID: + $ref: '#/components/headers/RequestID' content: application/problem+json: schema: @@ -336,6 +361,8 @@ components: TooManyRequests: description: 超过全局或客户端速率限制 headers: + X-Request-ID: + $ref: '#/components/headers/RequestID' Retry-After: schema: type: integer @@ -345,6 +372,36 @@ components: $ref: '#/components/schemas/Problem' ServiceUnavailable: description: 权威存储不可用或服务正在排空 + headers: + X-Request-ID: + $ref: '#/components/headers/RequestID' + content: + application/problem+json: + schema: + $ref: '#/components/schemas/Problem' + RequestEntityTooLarge: + description: 请求体超过服务端限制 + headers: + X-Request-ID: + $ref: '#/components/headers/RequestID' + content: + application/problem+json: + schema: + $ref: '#/components/schemas/Problem' + UnsupportedMediaType: + description: Content-Type 不是 application/json + headers: + X-Request-ID: + $ref: '#/components/headers/RequestID' + content: + application/problem+json: + schema: + $ref: '#/components/schemas/Problem' + InternalServerError: + description: 未分类的服务端错误 + headers: + X-Request-ID: + $ref: '#/components/headers/RequestID' content: application/problem+json: schema: diff --git a/docs/api/admin.md b/docs/api/admin.md index 3e990aa..376416a 100644 --- a/docs/api/admin.md +++ b/docs/api/admin.md @@ -16,3 +16,24 @@ Admin API 使用独立监听器与权限，契约位于 `api/openapi/admin.yaml` 配置重载校验失败返回 422，旧配置继续运行。Status 只返回低基数聚合信息， 不得返回 Proxy 地址、凭据、Client 标识或完整 Provider URL。 + +## 运行时实现边界 + +`admin.Handler` 只依赖 `Service` 控制面接口，不直接操作数据库、路由游标或配置 +文件。Service 必须保证 Status 来自同一修订快照，并将状态变更、审计和 Outbox +放在同一权威提交边界中。`MutationResult.version` 表示已提交的全局控制面修订， +不能混用配置格式版本或单 Worker Snapshot 版本。 + +严格 JSON、请求体上限、Request ID、JSON/Problem 响应由 +`platform/httpapi` 公用实现提供。Admin Handler 必须部署在独立监听器，并由 +外层认证与授权中间件保护；网关使用的 `Proxy-Authorization`/407 语义不得复用 +到 Admin 的 `Authorization`/401 语义。 + +除契约中的 401/403/404/409/422 外，运行时还明确返回： + +- `400`：Request ID 或 JSON 无效。 +- `405`：方法不匹配，并返回 `Allow`。 +- `413`：请求体超过管理入口上限。 +- `415`：请求体不是 `application/json`。 +- `500`：未分类内部错误，隐藏底层错误文本。 +- `503`：权威控制面暂时不可用。 diff --git a/docs/api/distribution.md b/docs/api/distribution.md index 09f8649..d2b3b63 100644 --- a/docs/api/distribution.md +++ b/docs/api/distribution.md @@ -64,6 +64,7 @@ Invoke-RestMethod ` - `count` 至少为 1，且不超过服务端 `maxCountPerRequest`。 - `fulfillment` 省略时使用服务端配置，默认 `partial`。 - 所有过滤数组执行“数组内 OR、不同维度 AND”。空数组等同不限制。 +- 每个过滤维度最多包含 64 个值，且数组内不得重复。 - `allowedUpstreams` 只能缩小 Client 可访问的 Upstream 集，不能扩大权限。 ## 3. 成功响应 @@ -181,9 +182,12 @@ Extraction Record 必须与状态更新处在相同事务边界或由同一权 - `401`：认证失败。 - `403`：来源控制、权限或 Upstream 访问被拒绝。 - `409`：allOrNothing 库存不足，或幂等 Key 冲突。 +- `413`：请求体超过 Distribution 配置上限。 +- `415`：请求体不是 `application/json`。 - `422`：数量、枚举或过滤组合违反业务约束。 - `429`：全局或 Client 速率限制，响应 `Retry-After`。 - `503`：PostgreSQL 不可写、服务排空或权威状态不可用。 +- `500`：未分类的内部错误；响应不包含底层错误文本。 错误响应不得包含 Provider Secret、Proxy 凭据、SQL 或内部拓扑。 @@ -198,3 +202,13 @@ proxyId, clientId, sourceIP, requestId, upstream, extractedAt, expiresAt 无认证时 `clientId` 使用 `anonymous` 或稳定匿名标识并保留 `sourceIP`。记录只 用于审计、排错和计费事实，不承担资源归还语义。Proxy 到期后可以清理运行 记录，但 Extraction Record 按审计保留策略归档。 + +## 11. 运行时实现边界 + +`distribution.Handler` 是薄 HTTP Adapter，只负责严格解码、Header/DTO 校验、 +身份结果注入、错误映射和健康探针。独占提取、TTL、Gateway 预留及幂等事务 +继续由 `extraction.Service` 和持久化 Store 承担。 + +请求体解码、Request ID 与 Problem JSON 统一复用 `platform/httpapi`。身份解析 +通过 `IdentityResolver` 注入；进程装配必须在 Handler 外层完成认证、可信代理 +来源解析与权限控制，且解析结果至少包含稳定 Client ID 或 Source IP。 diff --git a/docs/development/implementation-plan.md b/docs/development/implementation-plan.md index 5fc5a5f..f7f1e48 100644 --- a/docs/development/implementation-plan.md +++ b/docs/development/implementation-plan.md @@ -163,6 +163,12 @@ test/{fixtures,integration,e2e,load}/ - [ ] Expose Distribution extraction/status and Admin status/enable/disable/switch/reload. - [ ] Add integration tests using Compose-backed PostgreSQL/Redis. +当前进度（2026-07-29）：已实现共享 `platform/httpapi`、Distribution +extract/live/ready Handler 与 Admin status/enable/disable/switch/reload Handler； +定向契约测试已覆盖严格 JSON、Body 上限、Request ID、幂等 Header、DTO 映射、 +404/405 及业务错误映射。端点正式勾选仍等待独立监听器装配、认证/授权中间件、 +PostgreSQL/Redis Adapter 与 Compose 集成测试。 + ## Task 11: Checker and Health Reducer **Files:** `internal/controller/health/*.go`, `cmd/proxy-checker/main.go`, tests diff --git a/docs/requirements/completion-audit.md b/docs/requirements/completion-audit.md index 47bc77f..be9b501 100644 --- a/docs/requirements/completion-audit.md +++ b/docs/requirements/completion-audit.md @@ -33,6 +33,12 @@ 健康时效与 Gateway 保留量；1,000 并发不重复。 - `OPS-001`：完整 Snapshot 目标、epoch/version、校验和验证及原子替换。 - `CAP-001 / GW 热路径边界`：本地 Dispatch 条件过滤与原子容量预留。 +- `GW-*`：HTTP 正向代理、HTTPS CONNECT、双向 Tunnel、超时、重试、保护链与 + 优雅停机 Handler 已实现并通过定向测试。 +- `PROVIDER-*`：Provider HTTP Client、严格响应上限、模板解析安全边界、凭据 + 引用 Store 与 Reconciler Adapter 已实现。 +- `DIST/Admin HTTP`：严格 JSON、Request ID、Problem 响应及 Distribution/Admin + Handler 已实现；独立进程装配与认证授权仍在后续范围。 ## 2. 已执行验证 @@ -56,13 +62,13 @@ CI 已配置 Linux race job。Docker/Kubernetes 仅完成静态验证，没有 以下已有设计、接口或部署位置，但尚无端到端生产实现： 1. `cmd/proxy-gateway/controller/checker/loadgen` 进程装配。 -2. HTTP 正向代理、HTTPS CONNECT、连接池、安全重试与隧道转发。 -3. Provider Adapter、模板沙箱、singleflight、Leader、退避和累计额度执行器。 +2. Gateway 进程装配、生产连接池调优与代表性流量压测。 +3. Provider 分布式 singleflight/Leader、长期凭据回收和累计额度执行器。 4. PostgreSQL repository、Extraction 行锁事务、Outbox 和迁移。 5. Redis Leader、速率限制、心跳与可重建协调适配器。 6. Worker ownership drain/ACK/过期回收和网络快照流。 7. Checker 调度、探测器和健康 reducer。 -8. Admin/Distribution handler、鉴权、限流和审计查询。 +8. Admin/Distribution 独立监听器装配、鉴权授权、分布式限流和审计查询。 9. 真实 Compose/Kubernetes 集成、故障演练和代表性集群负载测试。 ## 4. 容量结论 diff --git a/internal/controller/admin/handler.go b/internal/controller/admin/handler.go new file mode 100644 index 0000000..d4187c6 --- /dev/null +++ b/internal/controller/admin/handler.go @@ -0,0 +1,265 @@ +package admin + +import ( + "context" + "errors" + "fmt" + "net/http" + "strings" + + "github.com/proxy-pool/proxy-pool/internal/platform/httpapi" +) + +const ( + statusPath = "/api/v1/status" + reloadPath = "/api/v1/config/reload" + upstreamPrefix = "/api/v1/upstreams/" + routingPrefix = "/api/v1/routing/" + maxResourceNameBytes = 128 +) + +var ( + ErrInvalidHandler = errors.New("invalid admin HTTP handler") + ErrNotFound = errors.New("admin resource not found") + ErrConflict = errors.New("admin mutation conflict") + ErrInvalidConfiguration = errors.New("invalid configuration") + ErrUnavailable = errors.New("admin service unavailable") +) + +type Service interface { + Status(context.Context) (Status, error) + SetUpstreamEnabled(context.Context, SetUpstreamCommand) (MutationResult, error) + SwitchRouting(context.Context, SwitchCommand) (MutationResult, error) + ReloadConfiguration(context.Context, ReloadCommand) (MutationResult, error) +} + +type Options struct { + MaxBodyBytes int64 +} + +type Status struct { + ConfigVersion string `json:"configVersion"` + SnapshotVersion uint64 `json:"snapshotVersion"` + Upstreams []UpstreamStatus `json:"upstreams"` + Workers []WorkerStatus `json:"workers"` +} + +type UpstreamStatus struct { + Name string `json:"name"` + Enabled bool `json:"enabled"` + Available int64 `json:"available"` + Checking int64 `json:"checking"` + Suspect int64 `json:"suspect"` + Draining int64 `json:"draining"` + Extracted int64 `json:"extracted"` + ConsecutiveEmptyFetch int64 `json:"consecutiveEmptyFetch,omitempty"` + FetchErrorCount int64 `json:"fetchErrorCount,omitempty"` +} + +type WorkerStatus struct { + ID string `json:"id"` + Zone string `json:"zone"` + Connected bool `json:"connected"` + SnapshotVersion uint64 `json:"snapshotVersion"` + StaleSeconds int64 `json:"staleSeconds,omitempty"` +} + +type MutationResult struct { + RequestID string `json:"requestId"` + Changed bool `json:"changed"` + Version uint64 `json:"version"` + Message string `json:"message,omitempty"` +} + +type SetUpstreamCommand struct { + RequestID string + Name string + Enabled bool +} + +type SwitchCommand struct { + RequestID string `json:"-"` + Name string `json:"-"` + ExpectedCurrent string `json:"expectedCurrent"` + Target string `json:"target"` + Reason string `json:"reason,omitempty"` +} + +type ReloadCommand struct { + RequestID string +} + +type Handler struct { + service Service + maxBodyBytes int64 +} + +func NewHandler(service Service, options Options) (*Handler, error) { + if service == nil || options.MaxBodyBytes <= 0 { + return nil, ErrInvalidHandler + } + return &Handler{service: service, maxBodyBytes: options.MaxBodyBytes}, nil +} + +func (handler *Handler) ServeHTTP(writer http.ResponseWriter, request *http.Request) { + requestID, err := httpapi.ResolveRequestID(request) + if err != nil { + writeTransportProblem(writer, http.StatusBadRequest, "INVALID_REQUEST_ID", "Invalid request ID", "X-Request-ID is invalid", requestID) + return + } + + switch request.URL.Path { + case statusPath: + if !requireMethod(writer, request, http.MethodGet, requestID) { + return + } + handler.getStatus(writer, request, requestID) + return + case reloadPath: + if !requireMethod(writer, request, http.MethodPost, requestID) { + return + } + handler.reload(writer, request, requestID) + return + } + + if name, action, ok := matchNamedAction(request.URL.Path, upstreamPrefix, "enable", "disable"); ok { + if !requireMethod(writer, request, http.MethodPost, requestID) { + return + } + handler.setUpstreamEnabled(writer, request, name, action == "enable", requestID) + return + } + if name, _, ok := matchNamedAction(request.URL.Path, routingPrefix, "switch"); ok { + if !requireMethod(writer, request, http.MethodPost, requestID) { + return + } + handler.switchRouting(writer, request, name, requestID) + return + } + + writeTransportProblem(writer, http.StatusNotFound, "NOT_FOUND", "Not found", "the requested endpoint does not exist", requestID) +} + +func (handler *Handler) getStatus(writer http.ResponseWriter, request *http.Request, requestID string) { + status, err := handler.service.Status(request.Context()) + if err != nil { + writeServiceProblem(writer, err, requestID) + return + } + if status.Upstreams == nil { + status.Upstreams = []UpstreamStatus{} + } + if status.Workers == nil { + status.Workers = []WorkerStatus{} + } + writer.Header().Set(httpapi.HeaderRequestID, requestID) + _ = httpapi.WriteJSON(writer, http.StatusOK, status) +} + +func (handler *Handler) setUpstreamEnabled(writer http.ResponseWriter, request *http.Request, name string, enabled bool, requestID string) { + result, err := handler.service.SetUpstreamEnabled(request.Context(), SetUpstreamCommand{ + RequestID: requestID, + Name: name, + Enabled: enabled, + }) + if err != nil { + writeServiceProblem(writer, err, requestID) + return + } + writeMutation(writer, result, requestID) +} + +func (handler *Handler) switchRouting(writer http.ResponseWriter, request *http.Request, name, requestID string) { + var command SwitchCommand + if err := httpapi.DecodeJSON(writer, request, handler.maxBodyBytes, &command); err != nil { + writeDecodeProblem(writer, err, requestID) + return + } + if command.ExpectedCurrent == "" || command.Target == "" || + len(command.ExpectedCurrent) > maxResourceNameBytes || len(command.Target) > maxResourceNameBytes || + len(command.Reason) > 512 { + writeTransportProblem(writer, http.StatusUnprocessableEntity, "INVALID_SWITCH", "Invalid routing switch", "routing switch fields violate the API contract", requestID) + return + } + command.RequestID = requestID + command.Name = name + result, err := handler.service.SwitchRouting(request.Context(), command) + if err != nil { + writeServiceProblem(writer, err, requestID) + return + } + writeMutation(writer, result, requestID) +} + +func (handler *Handler) reload(writer http.ResponseWriter, request *http.Request, requestID string) { + result, err := handler.service.ReloadConfiguration(request.Context(), ReloadCommand{RequestID: requestID}) + if err != nil { + writeServiceProblem(writer, err, requestID) + return + } + writeMutation(writer, result, requestID) +} + +func writeMutation(writer http.ResponseWriter, result MutationResult, requestID string) { + result.RequestID = requestID + writer.Header().Set(httpapi.HeaderRequestID, requestID) + _ = httpapi.WriteJSON(writer, http.StatusOK, result) +} + +func matchNamedAction(path, prefix string, actions ...string) (string, string, bool) { + if !strings.HasPrefix(path, prefix) { + return "", "", false + } + remainder := strings.TrimPrefix(path, prefix) + name, action, ok := strings.Cut(remainder, "/") + if !ok || name == "" || len(name) > maxResourceNameBytes || strings.Contains(action, "/") { + return "", "", false + } + for _, allowed := range actions { + if action == allowed { + return name, action, true + } + } + return "", "", false +} + +func requireMethod(writer http.ResponseWriter, request *http.Request, allowed, requestID string) bool { + if request.Method == allowed { + return true + } + writer.Header().Set("Allow", allowed) + writeTransportProblem(writer, http.StatusMethodNotAllowed, "METHOD_NOT_ALLOWED", "Method not allowed", fmt.Sprintf("use %s for this endpoint", allowed), requestID) + return false +} + +func writeDecodeProblem(writer http.ResponseWriter, err error, requestID string) { + if errors.Is(err, httpapi.ErrUnsupportedMediaType) { + writeTransportProblem(writer, http.StatusUnsupportedMediaType, "UNSUPPORTED_MEDIA_TYPE", "Unsupported media type", "Content-Type must be application/json", requestID) + return + } + if errors.Is(err, httpapi.ErrBodyTooLarge) { + writeTransportProblem(writer, http.StatusRequestEntityTooLarge, "REQUEST_BODY_TOO_LARGE", "Request body too large", "request body exceeds the configured limit", requestID) + return + } + writeTransportProblem(writer, http.StatusBadRequest, "INVALID_JSON", "Invalid JSON", "request body must be one valid JSON document with no unknown fields", requestID) +} + +func writeServiceProblem(writer http.ResponseWriter, err error, requestID string) { + switch { + case errors.Is(err, ErrNotFound): + writeTransportProblem(writer, http.StatusNotFound, "NOT_FOUND", "Not found", "the requested resource does not exist", requestID) + case errors.Is(err, ErrConflict): + writeTransportProblem(writer, http.StatusConflict, "CONFLICT", "Mutation conflict", "the authoritative state changed before the mutation committed", requestID) + case errors.Is(err, ErrInvalidConfiguration): + writeTransportProblem(writer, http.StatusUnprocessableEntity, "INVALID_CONFIGURATION", "Invalid configuration", "the new configuration did not pass validation", requestID) + case errors.Is(err, ErrUnavailable): + writeTransportProblem(writer, http.StatusServiceUnavailable, "SERVICE_UNAVAILABLE", "Service unavailable", "the authoritative service is temporarily unavailable", requestID) + default: + writeTransportProblem(writer, http.StatusInternalServerError, "INTERNAL_ERROR", "Internal server error", "the request could not be completed", requestID) + } +} + +func writeTransportProblem(writer http.ResponseWriter, status int, code, title, detail, requestID string) { + httpapi.WriteProblem(writer, httpapi.NewProblem(status, code, title, detail, requestID)) +} diff --git a/internal/controller/admin/handler_test.go b/internal/controller/admin/handler_test.go new file mode 100644 index 0000000..c8ec4e7 --- /dev/null +++ b/internal/controller/admin/handler_test.go @@ -0,0 +1,238 @@ +package admin + +import ( + "context" + "encoding/json" + "errors" + "net/http" + "net/http/httptest" + "strings" + "testing" + + "github.com/proxy-pool/proxy-pool/internal/platform/httpapi" +) + +func TestHandlerReturnsStatusWithoutSensitiveDetails(t *testing.T) { + t.Parallel() + service := &stubService{status: Status{ + ConfigVersion: "cfg-2", + SnapshotVersion: 7, + Upstreams: []UpstreamStatus{{Name: "provider-a", Enabled: true, Available: 11}}, + Workers: []WorkerStatus{{ID: "worker-a", Zone: "cn-east", Connected: true, SnapshotVersion: 7}}, + }} + handler := mustHandler(t, service) + recorder := httptest.NewRecorder() + + handler.ServeHTTP(recorder, httptest.NewRequest(http.MethodGet, "/api/v1/status", nil)) + + if recorder.Code != http.StatusOK { + t.Fatalf("status = %d, want %d; body=%s", recorder.Code, http.StatusOK, recorder.Body.String()) + } + var response Status + if err := json.Unmarshal(recorder.Body.Bytes(), &response); err != nil { + t.Fatalf("decode response: %v", err) + } + if response.ConfigVersion != "cfg-2" || response.SnapshotVersion != 7 || len(response.Upstreams) != 1 { + t.Fatalf("unexpected status response: %+v", response) + } +} + +func TestNewHandlerRejectsMissingDependenciesAndInvalidLimit(t *testing.T) { + t.Parallel() + if _, err := NewHandler(nil, Options{MaxBodyBytes: 1024}); !errors.Is(err, ErrInvalidHandler) { + t.Fatalf("NewHandler(nil) error = %v, want %v", err, ErrInvalidHandler) + } + if _, err := NewHandler(&stubService{}, Options{}); !errors.Is(err, ErrInvalidHandler) { + t.Fatalf("NewHandler(zero limit) error = %v, want %v", err, ErrInvalidHandler) + } +} + +func TestHandlerEnablesAndDisablesUpstream(t *testing.T) { + t.Parallel() + service := &stubService{mutation: MutationResult{Changed: true, Version: 8}} + handler := mustHandler(t, service) + + for _, test := range []struct { + path string + enabled bool + }{ + {path: "/api/v1/upstreams/provider-a/enable", enabled: true}, + {path: "/api/v1/upstreams/provider-a/disable", enabled: false}, + } { + recorder := httptest.NewRecorder() + request := httptest.NewRequest(http.MethodPost, test.path, nil) + request.Header.Set(httpapi.HeaderRequestID, "req-admin") + + handler.ServeHTTP(recorder, request) + + if recorder.Code != http.StatusOK { + t.Fatalf("%s status = %d, want %d; body=%s", test.path, recorder.Code, http.StatusOK, recorder.Body.String()) + } + if service.lastUpstream.Name != "provider-a" || service.lastUpstream.Enabled != test.enabled || service.lastUpstream.RequestID != "req-admin" { + t.Fatalf("unexpected service call: %+v", service.lastUpstream) + } + if requestID := recorder.Header().Get(httpapi.HeaderRequestID); requestID != "req-admin" { + t.Fatalf("response request ID = %q, want req-admin", requestID) + } + } +} + +func TestHandlerSwitchesRoutingWithStrictJSON(t *testing.T) { + t.Parallel() + service := &stubService{mutation: MutationResult{Changed: true, Version: 9}} + handler := mustHandler(t, service) + request := httptest.NewRequest(http.MethodPost, "/api/v1/routing/checkout/switch", strings.NewReader( + `{"expectedCurrent":"provider-a","target":"provider-b","reason":"capacity"}`, + )) + request.Header.Set("Content-Type", httpapi.JSONContentType) + recorder := httptest.NewRecorder() + + handler.ServeHTTP(recorder, request) + + if recorder.Code != http.StatusOK { + t.Fatalf("status = %d, want %d; body=%s", recorder.Code, http.StatusOK, recorder.Body.String()) + } + if service.lastSwitch.Name != "checkout" || service.lastSwitch.ExpectedCurrent != "provider-a" || service.lastSwitch.Target != "provider-b" { + t.Fatalf("unexpected switch call: command=%+v", service.lastSwitch) + } +} + +func TestHandlerReloadsConfigurationWithCommandRequestID(t *testing.T) { + t.Parallel() + service := &stubService{mutation: MutationResult{Changed: true, Version: 10}} + handler := mustHandler(t, service) + request := httptest.NewRequest(http.MethodPost, "/api/v1/config/reload", nil) + request.Header.Set(httpapi.HeaderRequestID, "req-reload") + recorder := httptest.NewRecorder() + + handler.ServeHTTP(recorder, request) + + if recorder.Code != http.StatusOK { + t.Fatalf("status = %d, want %d; body=%s", recorder.Code, http.StatusOK, recorder.Body.String()) + } + if service.lastReload.RequestID != "req-reload" { + t.Fatalf("reload command = %+v", service.lastReload) + } + if !strings.Contains(recorder.Body.String(), `"requestId":"req-reload"`) { + t.Fatalf("unexpected reload response %q", recorder.Body.String()) + } +} + +func TestHandlerMapsServiceErrorsToProblemContract(t *testing.T) { + t.Parallel() + tests := []struct { + name string + serviceErr error + wantStatus int + wantCode string + }{ + {name: "not found", serviceErr: ErrNotFound, wantStatus: http.StatusNotFound, wantCode: "NOT_FOUND"}, + {name: "conflict", serviceErr: ErrConflict, wantStatus: http.StatusConflict, wantCode: "CONFLICT"}, + {name: "invalid configuration", serviceErr: ErrInvalidConfiguration, wantStatus: http.StatusUnprocessableEntity, wantCode: "INVALID_CONFIGURATION"}, + {name: "unavailable", serviceErr: ErrUnavailable, wantStatus: http.StatusServiceUnavailable, wantCode: "SERVICE_UNAVAILABLE"}, + {name: "internal", serviceErr: errors.New("database password=secret"), wantStatus: http.StatusInternalServerError, wantCode: "INTERNAL_ERROR"}, + } + + for _, test := range tests { + test := test + t.Run(test.name, func(t *testing.T) { + t.Parallel() + handler := mustHandler(t, &stubService{err: test.serviceErr}) + recorder := httptest.NewRecorder() + request := httptest.NewRequest(http.MethodPost, "/api/v1/config/reload", nil) + request.Header.Set(httpapi.HeaderRequestID, "req-error") + + handler.ServeHTTP(recorder, request) + + if recorder.Code != test.wantStatus { + t.Fatalf("status = %d, want %d; body=%s", recorder.Code, test.wantStatus, recorder.Body.String()) + } + if body := recorder.Body.String(); !strings.Contains(body, `"code":"`+test.wantCode+`"`) || strings.Contains(body, "password") { + t.Fatalf("unexpected problem body %q", body) + } + }) + } +} + +func TestHandlerRejectsInvalidTransportRequests(t *testing.T) { + t.Parallel() + tests := []struct { + name string + method string + path string + body string + content string + requestID string + wantStatus int + }{ + {name: "method", method: http.MethodPut, path: "/api/v1/config/reload", wantStatus: http.StatusMethodNotAllowed}, + {name: "unknown route", method: http.MethodGet, path: "/missing", wantStatus: http.StatusNotFound}, + {name: "invalid name", method: http.MethodPost, path: "/api/v1/upstreams//enable", wantStatus: http.StatusNotFound}, + {name: "unknown JSON field", method: http.MethodPost, path: "/api/v1/routing/r/switch", body: `{"expectedCurrent":"a","target":"b","extra":1}`, content: httpapi.JSONContentType, wantStatus: http.StatusBadRequest}, + {name: "unsupported media type", method: http.MethodPost, path: "/api/v1/routing/r/switch", body: `{}`, content: "text/plain", wantStatus: http.StatusUnsupportedMediaType}, + {name: "oversized body", method: http.MethodPost, path: "/api/v1/routing/r/switch", body: strings.Repeat(" ", 1100) + `{}`, content: httpapi.JSONContentType, wantStatus: http.StatusRequestEntityTooLarge}, + {name: "invalid switch fields", method: http.MethodPost, path: "/api/v1/routing/r/switch", body: `{"expectedCurrent":"a"}`, content: httpapi.JSONContentType, wantStatus: http.StatusUnprocessableEntity}, + {name: "invalid request ID", method: http.MethodPost, path: "/api/v1/config/reload", requestID: strings.Repeat("x", 129), wantStatus: http.StatusBadRequest}, + } + + for _, test := range tests { + test := test + t.Run(test.name, func(t *testing.T) { + t.Parallel() + handler := mustHandler(t, &stubService{}) + recorder := httptest.NewRecorder() + request := httptest.NewRequest(test.method, test.path, strings.NewReader(test.body)) + request.Header.Set("Content-Type", test.content) + request.Header.Set(httpapi.HeaderRequestID, test.requestID) + + handler.ServeHTTP(recorder, request) + + if recorder.Code != test.wantStatus { + t.Fatalf("status = %d, want %d; body=%s", recorder.Code, test.wantStatus, recorder.Body.String()) + } + if contentType := recorder.Header().Get("Content-Type"); contentType != httpapi.ProblemContentType { + t.Fatalf("Content-Type = %q, want %q", contentType, httpapi.ProblemContentType) + } + if requestID := recorder.Header().Get(httpapi.HeaderRequestID); requestID == "" { + t.Fatal("X-Request-ID response header is empty") + } + }) + } +} + +func mustHandler(t *testing.T, service Service) *Handler { + t.Helper() + handler, err := NewHandler(service, Options{MaxBodyBytes: 1024}) + if err != nil { + t.Fatalf("NewHandler() error = %v", err) + } + return handler +} + +type stubService struct { + status Status + mutation MutationResult + err error + lastUpstream SetUpstreamCommand + lastSwitch SwitchCommand + lastReload ReloadCommand +} + +func (service *stubService) Status(context.Context) (Status, error) { + return service.status, service.err +} + +func (service *stubService) SetUpstreamEnabled(_ context.Context, command SetUpstreamCommand) (MutationResult, error) { + service.lastUpstream = command + return service.mutation, service.err +} + +func (service *stubService) SwitchRouting(_ context.Context, command SwitchCommand) (MutationResult, error) { + service.lastSwitch = command + return service.mutation, service.err +} + +func (service *stubService) ReloadConfiguration(_ context.Context, command ReloadCommand) (MutationResult, error) { + service.lastReload = command + return service.mutation, service.err +} diff --git a/internal/controller/distribution/handler.go b/internal/controller/distribution/handler.go new file mode 100644 index 0000000..bb4cf0f --- /dev/null +++ b/internal/controller/distribution/handler.go @@ -0,0 +1,369 @@ +package distribution + +import ( + "context" + "errors" + "net/http" + "strings" + "time" + + controllerExtraction "github.com/proxy-pool/proxy-pool/internal/controller/extraction" + domainExtraction "github.com/proxy-pool/proxy-pool/internal/domain/extraction" + "github.com/proxy-pool/proxy-pool/internal/platform/httpapi" +) + +const ( + pathExtract = "/api/v1/proxies/extract" + pathLive = "/health/live" + pathReady = "/health/ready" + + maxExtractCount = 1000 + maxFilterValues = 64 + minIdempotencyKeySize = 8 + maxIdempotencyKeySize = 128 + headerIdempotencyKey = "Idempotency-Key" +) + +type Config struct { + BodyLimitBytes int64 +} + +type Dependencies struct { + Extractor Extractor + Identity IdentityResolver + Readiness ReadinessChecker +} + +type Extractor interface { + Extract(context.Context, controllerExtraction.Request) (controllerExtraction.Response, error) +} + +type Identity struct { + ClientID string + SourceIP string +} + +type IdentityResolver interface { + Resolve(*http.Request) (Identity, error) +} + +type ReadinessChecker interface { + Ready(context.Context) error +} + +type Handler struct { + bodyLimitBytes int64 + extractor Extractor + identity IdentityResolver + readiness ReadinessChecker +} + +type extractRequestDTO struct { + Count int `json:"count"` + Fulfillment string `json:"fulfillment,omitempty"` + Filters *extractFiltersDTO `json:"filters,omitempty"` +} + +type extractFiltersDTO struct { + Protocols []string `json:"protocols,omitempty"` + Regions []string `json:"regions,omitempty"` + Carriers []string `json:"carriers,omitempty"` + AllowedUpstreams []string `json:"allowedUpstreams,omitempty"` +} + +type extractResponseDTO struct { + RequestID string `json:"requestId"` + Requested int `json:"requested"` + Returned int `json:"returned"` + Proxies []extractedProxyDTO `json:"proxies"` +} + +type extractedProxyDTO struct { + ID string `json:"id"` + Protocol string `json:"protocol"` + Host string `json:"host"` + Port uint16 `json:"port"` + Username string `json:"username,omitempty"` + Password string `json:"password,omitempty"` + URL string `json:"url"` + Region string `json:"region,omitempty"` + Carrier string `json:"carrier,omitempty"` + Upstream string `json:"upstream"` + ExpiresAt string `json:"expiresAt"` + RemainingTTLSeconds int64 `json:"remainingTtlSeconds"` + ExtractedAt string `json:"extractedAt"` +} + +type healthDTO struct { + Status string `json:"status"` +} + +func NewHandler(config Config, deps Dependencies) (*Handler, error) { + switch { + case deps.Extractor == nil: + return nil, errors.New("create distribution handler: extractor is required") + case deps.Identity == nil: + return nil, errors.New("create distribution handler: identity resolver is required") + case deps.Readiness == nil: + return nil, errors.New("create distribution handler: readiness checker is required") + case config.BodyLimitBytes <= 0: + return nil, errors.New("create distribution handler: body limit must be greater than zero") + } + return &Handler{ + bodyLimitBytes: config.BodyLimitBytes, + extractor: deps.Extractor, + identity: deps.Identity, + readiness: deps.Readiness, + }, nil +} + +func (h *Handler) ServeHTTP(writer http.ResponseWriter, request *http.Request) { + requestID, requestIDErr := httpapi.ResolveRequestID(request) + if requestIDErr != nil { + h.writeProblem(writer, problemBadRequest(requestID, "INVALID_HEADER", "Invalid request header", "", nil)) + return + } + + switch request.URL.Path { + case pathExtract: + if request.Method != http.MethodPost { + writer.Header().Set("Allow", http.MethodPost) + h.writeProblem(writer, httpapi.NewProblem(http.StatusMethodNotAllowed, "METHOD_NOT_ALLOWED", "Method not allowed", "", requestID)) + return + } + h.handleExtract(writer, request, requestID) + case pathLive: + if request.Method != http.MethodGet { + writer.Header().Set("Allow", http.MethodGet) + h.writeProblem(writer, httpapi.NewProblem(http.StatusMethodNotAllowed, "METHOD_NOT_ALLOWED", "Method not allowed", "", requestID)) + return + } + h.writeJSON(writer, requestID, http.StatusOK, healthDTO{Status: "ok"}) + case pathReady: + if request.Method != http.MethodGet { + writer.Header().Set("Allow", http.MethodGet) + h.writeProblem(writer, httpapi.NewProblem(http.StatusMethodNotAllowed, "METHOD_NOT_ALLOWED", "Method not allowed", "", requestID)) + return + } + if err := h.readiness.Ready(request.Context()); err != nil { + h.writeProblem(writer, httpapi.NewProblem(http.StatusServiceUnavailable, "SERVICE_UNAVAILABLE", "Service unavailable", "", requestID)) + return + } + h.writeJSON(writer, requestID, http.StatusOK, healthDTO{Status: "ok"}) + default: + h.writeProblem(writer, httpapi.NewProblem(http.StatusNotFound, "NOT_FOUND", "Not found", "", requestID)) + } +} + +func (h *Handler) handleExtract(writer http.ResponseWriter, request *http.Request, requestID string) { + idempotencyKey, err := validateIdempotencyKey(request.Header.Values(headerIdempotencyKey)) + if err != nil { + h.writeProblem(writer, problemBadRequest(requestID, "INVALID_HEADER", "Invalid request header", "", []httpapi.InvalidParam{{ + Name: "Idempotency-Key", Reason: "must be 8..128 characters when present", + }})) + return + } + + var payload extractRequestDTO + if err := httpapi.DecodeJSON(writer, request, h.bodyLimitBytes, &payload); err != nil { + h.writeProblem(writer, problemFromDecodeError(requestID, err)) + return + } + + invalidParams := validateExtractRequest(payload) + if len(invalidParams) > 0 { + h.writeProblem(writer, httpapi.Problem{ + Type: "https://proxy-pool.local/problems/invalid-request", + Title: "Invalid request", + Status: http.StatusUnprocessableEntity, + Code: "INVALID_REQUEST", + RequestID: requestID, + InvalidParams: invalidParams, + }) + return + } + + identity, err := h.identity.Resolve(request) + if err != nil || (strings.TrimSpace(identity.ClientID) == "" && strings.TrimSpace(identity.SourceIP) == "") { + h.writeProblem(writer, problemBadRequest(requestID, "INVALID_REQUEST", "Invalid request", "", nil)) + return + } + + filters := payload.filtersOrZero() + serviceResponse, err := h.extractor.Extract(request.Context(), controllerExtraction.Request{ + RequestID: requestID, + ClientID: identity.ClientID, + SourceIP: identity.SourceIP, + IdempotencyKey: idempotencyKey, + Count: payload.Count, + Fulfillment: domainExtraction.Fulfillment(payload.Fulfillment), + Filters: controllerExtraction.Filters{ + Protocols: cloneStrings(filters.Protocols), + Regions: cloneStrings(filters.Regions), + Carriers: cloneStrings(filters.Carriers), + Upstreams: cloneStrings(filters.AllowedUpstreams), + }, + }) + if err != nil { + h.writeProblem(writer, problemFromExtractError(requestID, err)) + return + } + + response := extractResponseDTO{ + RequestID: requestID, + Requested: serviceResponse.Requested, + Returned: serviceResponse.Returned, + Proxies: make([]extractedProxyDTO, 0, len(serviceResponse.Proxies)), + } + for _, extracted := range serviceResponse.Proxies { + response.Proxies = append(response.Proxies, extractedProxyDTO{ + ID: extracted.ID, + Protocol: extracted.Protocol, + Host: extracted.Host, + Port: extracted.Port, + Username: extracted.Username, + Password: extracted.Password, + URL: extracted.URL, + Region: extracted.Region, + Carrier: extracted.Carrier, + Upstream: extracted.Upstream, + ExpiresAt: extracted.ExpiresAt.UTC().Format(time.RFC3339), + RemainingTTLSeconds: extracted.RemainingTTLSeconds, + ExtractedAt: extracted.ExtractedAt.UTC().Format(time.RFC3339), + }) + } + h.writeJSON(writer, requestID, http.StatusOK, response) +} + +func validateIdempotencyKey(values []string) (string, error) { + if len(values) == 0 || (len(values) == 1 && values[0] == "") { + return "", nil + } + if len(values) != 1 { + return "", errors.New("invalid idempotency key") + } + value := values[0] + if len(value) < minIdempotencyKeySize || len(value) > maxIdempotencyKeySize || strings.TrimSpace(value) != value { + return "", errors.New("invalid idempotency key") + } + for _, character := range value { + if character < 0x20 || character == 0x7f { + return "", errors.New("invalid idempotency key") + } + } + return value, nil +} + +func validateExtractRequest(payload extractRequestDTO) []httpapi.InvalidParam { + var invalid []httpapi.InvalidParam + if payload.Count < 1 || payload.Count > maxExtractCount { + invalid = append(invalid, httpapi.InvalidParam{Name: "count", Reason: "must be between 1 and 1000"}) + } + if payload.Fulfillment != "" && payload.Fulfillment != string(domainExtraction.Partial) && + payload.Fulfillment != string(domainExtraction.AllOrNothing) { + invalid = append(invalid, httpapi.InvalidParam{Name: "fulfillment", Reason: "must be partial or allOrNothing"}) + } + if payload.Filters == nil { + return invalid + } + invalid = append(invalid, validateUniqueStrings("filters.protocols", payload.Filters.Protocols, validProtocol)...) + invalid = append(invalid, validateUniqueStrings("filters.regions", payload.Filters.Regions, nil)...) + invalid = append(invalid, validateUniqueStrings("filters.carriers", payload.Filters.Carriers, nil)...) + invalid = append(invalid, validateUniqueStrings("filters.allowedUpstreams", payload.Filters.AllowedUpstreams, nil)...) + return invalid +} + +func validateUniqueStrings(name string, values []string, allowed map[string]struct{}) []httpapi.InvalidParam { + if len(values) > maxFilterValues { + return []httpapi.InvalidParam{{Name: name, Reason: "must contain at most 64 values"}} + } + seen := make(map[string]struct{}, len(values)) + invalid := make([]httpapi.InvalidParam, 0, 1) + for _, value := range values { + if _, exists := seen[value]; exists { + invalid = append(invalid, httpapi.InvalidParam{Name: name, Reason: "must not contain duplicates"}) + return invalid + } + seen[value] = struct{}{} + if allowed != nil { + if _, ok := allowed[value]; !ok { + invalid = append(invalid, httpapi.InvalidParam{Name: name, Reason: "contains unsupported value"}) + return invalid + } + } + } + return invalid +} + +var validProtocol = map[string]struct{}{ + "http": {}, + "https": {}, + "socks5": {}, +} + +func problemFromDecodeError(requestID string, err error) httpapi.Problem { + switch { + case errors.Is(err, httpapi.ErrUnsupportedMediaType): + return httpapi.NewProblem(http.StatusUnsupportedMediaType, "UNSUPPORTED_MEDIA_TYPE", "Unsupported media type", "Content-Type must be application/json", requestID) + case errors.Is(err, httpapi.ErrBodyTooLarge): + return httpapi.NewProblem(http.StatusRequestEntityTooLarge, "REQUEST_BODY_TOO_LARGE", "Request body too large", "request body exceeds the configured limit", requestID) + default: + return problemBadRequest(requestID, "INVALID_JSON", "Invalid JSON request body", "", nil) + } +} + +func problemFromExtractError(requestID string, err error) httpapi.Problem { + switch { + case errors.Is(err, domainExtraction.ErrInsufficientProxies): + return httpapi.NewProblem(http.StatusConflict, "INSUFFICIENT_PROXIES", "Insufficient proxies", "", requestID) + case errors.Is(err, domainExtraction.ErrIdempotencyConflict): + return httpapi.NewProblem(http.StatusConflict, "IDEMPOTENCY_CONFLICT", "Idempotency conflict", "", requestID) + case errors.Is(err, controllerExtraction.ErrCountExceeded): + return httpapi.NewProblem(http.StatusUnprocessableEntity, "COUNT_EXCEEDED", "Invalid request", "", requestID) + case errors.Is(err, controllerExtraction.ErrInvalidFulfillment): + return httpapi.NewProblem(http.StatusUnprocessableEntity, "INVALID_FULFILLMENT", "Invalid request", "", requestID) + case errors.Is(err, controllerExtraction.ErrInvalidRequest): + return httpapi.NewProblem(http.StatusUnprocessableEntity, "INVALID_REQUEST", "Invalid request", "", requestID) + case errors.Is(err, controllerExtraction.ErrAdmissionRejected): + return httpapi.NewProblem(http.StatusTooManyRequests, "RATE_LIMITED", "Too many requests", "", requestID) + case errors.Is(err, controllerExtraction.ErrUnavailable): + return httpapi.NewProblem(http.StatusServiceUnavailable, "SERVICE_UNAVAILABLE", "Service unavailable", "", requestID) + default: + return httpapi.NewProblem(http.StatusInternalServerError, "INTERNAL_ERROR", "Internal server error", "", requestID) + } +} + +func problemBadRequest(requestID, code, title, detail string, invalid []httpapi.InvalidParam) httpapi.Problem { + return httpapi.Problem{ + Type: "https://proxy-pool.local/problems/invalid-request", + Title: title, + Status: http.StatusBadRequest, + Code: code, + Detail: detail, + RequestID: requestID, + InvalidParams: invalid, + } +} + +func (h *Handler) writeJSON(writer http.ResponseWriter, requestID string, status int, value any) { + writer.Header().Set(httpapi.HeaderRequestID, requestID) + _ = httpapi.WriteJSON(writer, status, value) +} + +func (h *Handler) writeProblem(writer http.ResponseWriter, problem httpapi.Problem) { + if problem.RequestID == "" { + problem.RequestID = writer.Header().Get(httpapi.HeaderRequestID) + } + httpapi.WriteProblem(writer, problem) +} + +func cloneStrings(values []string) []string { + return append([]string(nil), values...) +} + +func (payload extractRequestDTO) filtersOrZero() extractFiltersDTO { + if payload.Filters == nil { + return extractFiltersDTO{} + } + return *payload.Filters +} diff --git a/internal/controller/distribution/handler_test.go b/internal/controller/distribution/handler_test.go new file mode 100644 index 0000000..22a60cf --- /dev/null +++ b/internal/controller/distribution/handler_test.go @@ -0,0 +1,472 @@ +package distribution + +import ( + "context" + "encoding/json" + "errors" + "fmt" + "net/http" + "net/http/httptest" + "strings" + "testing" + "time" + + controllerExtraction "github.com/proxy-pool/proxy-pool/internal/controller/extraction" + domainExtraction "github.com/proxy-pool/proxy-pool/internal/domain/extraction" + "github.com/proxy-pool/proxy-pool/internal/platform/httpapi" +) + +func TestNewHandlerValidatesDependenciesAndBodyLimit(t *testing.T) { + t.Parallel() + + identity := fakeIdentityResolver{identity: Identity{ClientID: "client-1", SourceIP: "198.51.100.8"}} + readiness := fakeReadinessChecker{} + extractor := &fakeExtractor{} + + tests := []struct { + name string + config Config + deps Dependencies + wantErr string + }{ + { + name: "missing extractor", + config: Config{BodyLimitBytes: 1024}, + deps: Dependencies{Identity: identity, Readiness: readiness}, + wantErr: "extractor", + }, + { + name: "missing identity", + config: Config{BodyLimitBytes: 1024}, + deps: Dependencies{Extractor: extractor, Readiness: readiness}, + wantErr: "identity", + }, + { + name: "missing readiness", + config: Config{BodyLimitBytes: 1024}, + deps: Dependencies{Extractor: extractor, Identity: identity}, + wantErr: "readiness", + }, + { + name: "non-positive body limit", + config: Config{BodyLimitBytes: 0}, + deps: Dependencies{Extractor: extractor, Identity: identity, Readiness: readiness}, + wantErr: "body limit", + }, + } + + for _, test := range tests { + test := test + t.Run(test.name, func(t *testing.T) { + t.Parallel() + handler, err := NewHandler(test.config, test.deps) + if err == nil || !strings.Contains(strings.ToLower(err.Error()), test.wantErr) { + t.Fatalf("NewHandler() = (%v, %v), want error containing %q", handler, err, test.wantErr) + } + }) + } +} + +func TestHandlerExtractSuccessMapsOpenAPIDTOAndReturnsRequestID(t *testing.T) { + t.Parallel() + + extractor := &fakeExtractor{response: controllerExtraction.Response{ + RequestID: "unexpected-service-request-id", + Requested: 2, + Returned: 1, + Proxies: []controllerExtraction.ExtractedProxy{{ + ID: "px-1", + Protocol: "http", + Host: "192.0.2.10", + Port: 8080, + Username: "user", + Password: "pass", + URL: "http://user:pass@192.0.2.10:8080", + Region: "shanghai", + Carrier: "ct", + Upstream: "provider-a", + ExpiresAt: time.Date(2026, 7, 28, 12, 5, 0, 0, time.UTC), + RemainingTTLSeconds: 300, + ExtractedAt: time.Date(2026, 7, 28, 12, 0, 0, 0, time.UTC), + }}, + }} + handler := mustNewHandler(t, Config{BodyLimitBytes: 4096}, Dependencies{ + Extractor: extractor, + Identity: fakeIdentityResolver{identity: Identity{ClientID: "tenant-a", SourceIP: "198.51.100.8"}}, + Readiness: fakeReadinessChecker{}, + }) + request := httptest.NewRequest(http.MethodPost, "/api/v1/proxies/extract", strings.NewReader(`{ + "count": 2, + "fulfillment": "partial", + "filters": { + "protocols": ["http"], + "regions": ["shanghai"], + "carriers": ["ct"], + "allowedUpstreams": ["provider-a"] + } + }`)) + request.Header.Set("Content-Type", "application/json") + request.Header.Set(httpapi.HeaderRequestID, "req-caller") + request.Header.Set("Idempotency-Key", "idem-12345678") + + response := httptest.NewRecorder() + handler.ServeHTTP(response, request) + + if response.Code != http.StatusOK { + t.Fatalf("status = %d, want 200", response.Code) + } + if got := response.Header().Get(httpapi.HeaderRequestID); got != "req-caller" { + t.Fatalf("X-Request-ID = %q, want req-caller", got) + } + if got := response.Header().Get("Content-Type"); got != httpapi.JSONContentType { + t.Fatalf("Content-Type = %q, want %q", got, httpapi.JSONContentType) + } + if extractor.calls != 1 { + t.Fatalf("extractor calls = %d, want 1", extractor.calls) + } + if extractor.request.ClientID != "tenant-a" || extractor.request.SourceIP != "198.51.100.8" { + t.Fatalf("identity request = %+v", extractor.request) + } + if extractor.request.IdempotencyKey != "idem-12345678" { + t.Fatalf("idempotency key = %q", extractor.request.IdempotencyKey) + } + if got := extractor.request.Filters.Upstreams; len(got) != 1 || got[0] != "provider-a" { + t.Fatalf("allowedUpstreams mapping = %v", got) + } + + var payload struct { + RequestID string `json:"requestId"` + Requested int `json:"requested"` + Returned int `json:"returned"` + Proxies []struct { + ID string `json:"id"` + Password string `json:"password"` + URL string `json:"url"` + Upstream string `json:"upstream"` + } `json:"proxies"` + } + if err := json.Unmarshal(response.Body.Bytes(), &payload); err != nil { + t.Fatalf("decode response: %v", err) + } + if payload.RequestID != "req-caller" || payload.Requested != 2 || payload.Returned != 1 { + t.Fatalf("payload = %+v", payload) + } + if len(payload.Proxies) != 1 || payload.Proxies[0].Password != "pass" || payload.Proxies[0].URL == "" { + t.Fatalf("proxies payload = %+v", payload.Proxies) + } +} + +func TestHandlerRejectsEmptyResolvedIdentity(t *testing.T) { + t.Parallel() + extractor := &fakeExtractor{} + handler := mustNewHandler(t, Config{BodyLimitBytes: 1024}, Dependencies{ + Extractor: extractor, + Identity: fakeIdentityResolver{}, + Readiness: fakeReadinessChecker{}, + }) + request := httptest.NewRequest(http.MethodPost, "/api/v1/proxies/extract", strings.NewReader(`{"count":1}`)) + request.Header.Set("Content-Type", httpapi.JSONContentType) + response := httptest.NewRecorder() + + handler.ServeHTTP(response, request) + + if response.Code != http.StatusBadRequest { + t.Fatalf("status = %d, want %d; body=%s", response.Code, http.StatusBadRequest, response.Body.String()) + } + if extractor.calls != 0 { + t.Fatalf("extractor calls = %d, want 0", extractor.calls) + } +} + +func TestHandlerRejectsDuplicateIdempotencyHeader(t *testing.T) { + t.Parallel() + extractor := &fakeExtractor{} + handler := mustNewHandler(t, Config{BodyLimitBytes: 1024}, Dependencies{ + Extractor: extractor, + Identity: fakeIdentityResolver{identity: Identity{ClientID: "client-1"}}, + Readiness: fakeReadinessChecker{}, + }) + request := httptest.NewRequest(http.MethodPost, "/api/v1/proxies/extract", strings.NewReader(`{"count":1}`)) + request.Header.Set("Content-Type", httpapi.JSONContentType) + request.Header.Add("Idempotency-Key", "idempotency-one") + request.Header.Add("Idempotency-Key", "idempotency-two") + response := httptest.NewRecorder() + + handler.ServeHTTP(response, request) + + if response.Code != http.StatusBadRequest { + t.Fatalf("status = %d, want %d; body=%s", response.Code, http.StatusBadRequest, response.Body.String()) + } + if extractor.calls != 0 { + t.Fatalf("extractor calls = %d, want 0", extractor.calls) + } +} + +func TestHandlerExtractMapsErrorsToProblemResponsesWithoutSensitiveLeakage(t *testing.T) { + t.Parallel() + tooManyRegions := `{"count":1,"filters":{"regions":["` + strings.Join(makeUniqueValues(65), `","`) + `"]}}` + + tests := []struct { + name string + requestBody string + contentType string + idempotencyKey string + extractErr error + bodyLimit int64 + wantStatus int + wantCode string + }{ + { + name: "invalid json unknown field", + requestBody: `{"count":1,"unexpected":true}`, + contentType: "application/json", + wantStatus: http.StatusBadRequest, + wantCode: "INVALID_JSON", + }, + { + name: "unsupported media type", + requestBody: `{"count":1}`, + contentType: "text/plain", + wantStatus: http.StatusUnsupportedMediaType, + wantCode: "UNSUPPORTED_MEDIA_TYPE", + }, + { + name: "request body too large", + requestBody: strings.Repeat(" ", 300) + `{"count":1}`, + contentType: "application/json", + wantStatus: http.StatusRequestEntityTooLarge, + wantCode: "REQUEST_BODY_TOO_LARGE", + }, + { + name: "invalid idempotency key", + requestBody: `{"count":1}`, + contentType: "application/json", + idempotencyKey: "short", + wantStatus: http.StatusBadRequest, + wantCode: "INVALID_HEADER", + }, + { + name: "invalid dto", + requestBody: `{"count":1001,"filters":{"protocols":["http","http"]}}`, + contentType: "application/json", + wantStatus: http.StatusUnprocessableEntity, + wantCode: "INVALID_REQUEST", + }, + { + name: "too many filter values", + requestBody: tooManyRegions, + contentType: "application/json", + bodyLimit: 4096, + wantStatus: http.StatusUnprocessableEntity, + wantCode: "INVALID_REQUEST", + }, + { + name: "insufficient proxies", + requestBody: `{"count":1,"fulfillment":"allOrNothing"}`, + contentType: "application/json", + extractErr: domainExtraction.ErrInsufficientProxies, + wantStatus: http.StatusConflict, + wantCode: "INSUFFICIENT_PROXIES", + }, + { + name: "idempotency conflict", + requestBody: `{"count":1}`, + contentType: "application/json", + extractErr: domainExtraction.ErrIdempotencyConflict, + wantStatus: http.StatusConflict, + wantCode: "IDEMPOTENCY_CONFLICT", + }, + { + name: "admission rejected", + requestBody: `{"count":1}`, + contentType: "application/json", + extractErr: controllerExtraction.ErrAdmissionRejected, + wantStatus: http.StatusTooManyRequests, + wantCode: "RATE_LIMITED", + }, + { + name: "count exceeded", + requestBody: `{"count":1}`, + contentType: "application/json", + extractErr: controllerExtraction.ErrCountExceeded, + wantStatus: http.StatusUnprocessableEntity, + wantCode: "COUNT_EXCEEDED", + }, + { + name: "unexpected error", + requestBody: `{"count":1}`, + contentType: "application/json", + extractErr: errors.New("backend secret password leaked"), + wantStatus: http.StatusInternalServerError, + wantCode: "INTERNAL_ERROR", + }, + { + name: "extraction unavailable", + requestBody: `{"count":1}`, + contentType: "application/json", + extractErr: controllerExtraction.ErrUnavailable, + wantStatus: http.StatusServiceUnavailable, + wantCode: "SERVICE_UNAVAILABLE", + }, + } + + for _, test := range tests { + test := test + t.Run(test.name, func(t *testing.T) { + t.Parallel() + extractor := &fakeExtractor{err: test.extractErr} + bodyLimit := test.bodyLimit + if bodyLimit == 0 { + bodyLimit = 256 + } + handler := mustNewHandler(t, Config{BodyLimitBytes: bodyLimit}, Dependencies{ + Extractor: extractor, + Identity: fakeIdentityResolver{identity: Identity{ClientID: "tenant-a", SourceIP: "198.51.100.8"}}, + Readiness: fakeReadinessChecker{}, + }) + request := httptest.NewRequest(http.MethodPost, "/api/v1/proxies/extract", strings.NewReader(test.requestBody)) + request.Header.Set("Content-Type", test.contentType) + request.Header.Set(httpapi.HeaderRequestID, "req-err") + if test.idempotencyKey != "" { + request.Header.Set("Idempotency-Key", test.idempotencyKey) + } + + response := httptest.NewRecorder() + handler.ServeHTTP(response, request) + + if response.Code != test.wantStatus { + t.Fatalf("status = %d, want %d", response.Code, test.wantStatus) + } + if got := response.Header().Get(httpapi.HeaderRequestID); got == "" { + t.Fatal("X-Request-ID header is empty") + } + if got := response.Header().Get("Content-Type"); got != httpapi.ProblemContentType { + t.Fatalf("Content-Type = %q, want %q", got, httpapi.ProblemContentType) + } + var problem httpapi.Problem + if err := json.Unmarshal(response.Body.Bytes(), &problem); err != nil { + t.Fatalf("decode problem: %v", err) + } + if problem.Status != test.wantStatus || problem.Code != test.wantCode { + t.Fatalf("problem = %+v", problem) + } + if body := response.Body.String(); strings.Contains(body, `"password"`) || strings.Contains(body, "secret") { + t.Fatalf("error body leaked sensitive data: %s", body) + } + }) + } +} + +func makeUniqueValues(count int) []string { + values := make([]string, count) + for index := range values { + values[index] = fmt.Sprintf("region-%d", index) + } + return values +} + +func TestHandlerHealthRoutesAndRoutingEdges(t *testing.T) { + t.Parallel() + + handler := mustNewHandler(t, Config{BodyLimitBytes: 1024}, Dependencies{ + Extractor: &fakeExtractor{}, + Identity: fakeIdentityResolver{identity: Identity{ClientID: "tenant-a", SourceIP: "198.51.100.8"}}, + Readiness: fakeReadinessChecker{}, + }) + + tests := []struct { + name string + method string + target string + wantStatus int + wantCT string + }{ + {name: "live ok", method: http.MethodGet, target: "/health/live", wantStatus: http.StatusOK, wantCT: httpapi.JSONContentType}, + {name: "ready ok", method: http.MethodGet, target: "/health/ready", wantStatus: http.StatusOK, wantCT: httpapi.JSONContentType}, + {name: "extract wrong method", method: http.MethodGet, target: "/api/v1/proxies/extract", wantStatus: http.StatusMethodNotAllowed, wantCT: httpapi.ProblemContentType}, + {name: "unknown route", method: http.MethodGet, target: "/missing", wantStatus: http.StatusNotFound, wantCT: httpapi.ProblemContentType}, + } + + for _, test := range tests { + test := test + t.Run(test.name, func(t *testing.T) { + t.Parallel() + request := httptest.NewRequest(test.method, test.target, nil) + response := httptest.NewRecorder() + handler.ServeHTTP(response, request) + if response.Code != test.wantStatus { + t.Fatalf("status = %d, want %d", response.Code, test.wantStatus) + } + if got := response.Header().Get(httpapi.HeaderRequestID); got == "" { + t.Fatal("X-Request-ID header is empty") + } + if got := response.Header().Get("Content-Type"); got != test.wantCT { + t.Fatalf("Content-Type = %q, want %q", got, test.wantCT) + } + }) + } +} + +func TestHandlerReadyMapsDependencyFailureTo503Problem(t *testing.T) { + t.Parallel() + + handler := mustNewHandler(t, Config{BodyLimitBytes: 1024}, Dependencies{ + Extractor: &fakeExtractor{}, + Identity: fakeIdentityResolver{identity: Identity{ClientID: "tenant-a", SourceIP: "198.51.100.8"}}, + Readiness: fakeReadinessChecker{err: errors.New("storage unavailable")}, + }) + request := httptest.NewRequest(http.MethodGet, "/health/ready", nil) + response := httptest.NewRecorder() + + handler.ServeHTTP(response, request) + + if response.Code != http.StatusServiceUnavailable { + t.Fatalf("status = %d, want 503", response.Code) + } + var problem httpapi.Problem + if err := json.Unmarshal(response.Body.Bytes(), &problem); err != nil { + t.Fatalf("decode problem: %v", err) + } + if problem.Code != "SERVICE_UNAVAILABLE" { + t.Fatalf("problem code = %q, want SERVICE_UNAVAILABLE", problem.Code) + } +} + +func mustNewHandler(t *testing.T, config Config, deps Dependencies) *Handler { + t.Helper() + handler, err := NewHandler(config, deps) + if err != nil { + t.Fatalf("NewHandler() error = %v", err) + } + return handler +} + +type fakeExtractor struct { + request controllerExtraction.Request + response controllerExtraction.Response + err error + calls int +} + +func (f *fakeExtractor) Extract(_ context.Context, request controllerExtraction.Request) (controllerExtraction.Response, error) { + f.calls++ + f.request = request + return f.response, f.err +} + +type fakeIdentityResolver struct { + identity Identity + err error +} + +func (f fakeIdentityResolver) Resolve(*http.Request) (Identity, error) { + return f.identity, f.err +} + +type fakeReadinessChecker struct { + err error +} + +func (f fakeReadinessChecker) Ready(context.Context) error { + return f.err +} diff --git a/internal/controller/extraction/service.go b/internal/controller/extraction/service.go index 3e2cf0f..a762b28 100644 --- a/internal/controller/extraction/service.go +++ b/internal/controller/extraction/service.go @@ -15,6 +15,7 @@ var ( ErrInvalidFulfillment = errors.New("invalid extraction fulfillment") ErrInvalidServicePolicy = errors.New("invalid extraction service policy") ErrAdmissionRejected = errors.New("extraction admission rejected") + ErrUnavailable = errors.New("extraction service unavailable") ) type Policy struct { diff --git a/internal/platform/httpapi/httpapi.go b/internal/platform/httpapi/httpapi.go new file mode 100644 index 0000000..1c833db --- /dev/null +++ b/internal/platform/httpapi/httpapi.go @@ -0,0 +1,140 @@ +package httpapi + +import ( + "crypto/rand" + "encoding/hex" + "encoding/json" + "errors" + "fmt" + "io" + "mime" + "net/http" + "strings" +) + +const ( + HeaderRequestID = "X-Request-ID" + JSONContentType = "application/json" + ProblemContentType = "application/problem+json" + maxRequestIDLength = 128 +) + +var ( + ErrUnsupportedMediaType = errors.New("unsupported media type") + ErrInvalidJSON = errors.New("invalid JSON request body") + ErrBodyTooLarge = errors.New("request body is too large") + ErrInvalidRequestID = errors.New("invalid request ID") +) + +type InvalidParam struct { + Name string `json:"name"` + Reason string `json:"reason"` +} + +type Problem struct { + Type string `json:"type"` + Title string `json:"title"` + Status int `json:"status"` + Code string `json:"code"` + Detail string `json:"detail,omitempty"` + RequestID string `json:"requestId,omitempty"` + InvalidParams []InvalidParam `json:"invalidParams,omitempty"` +} + +func NewProblem(status int, code, title, detail, requestID string) Problem { + return Problem{ + Type: "https://proxy-pool.local/problems/" + strings.ToLower(strings.ReplaceAll(code, "_", "-")), + Title: title, + Status: status, + Code: code, + Detail: detail, + RequestID: requestID, + } +} + +func DecodeJSON(writer http.ResponseWriter, request *http.Request, maxBytes int64, target any) error { + if request == nil || request.Body == nil || target == nil || maxBytes <= 0 { + return ErrInvalidJSON + } + mediaType, _, err := mime.ParseMediaType(request.Header.Get("Content-Type")) + if err != nil || !strings.EqualFold(mediaType, JSONContentType) { + return ErrUnsupportedMediaType + } + + request.Body = http.MaxBytesReader(writer, request.Body, maxBytes) + decoder := json.NewDecoder(request.Body) + decoder.DisallowUnknownFields() + if err := decoder.Decode(target); err != nil { + return classifyDecodeError(err) + } + var trailing json.RawMessage + if err := decoder.Decode(&trailing); !errors.Is(err, io.EOF) { + if err == nil { + return ErrInvalidJSON + } + return classifyDecodeError(err) + } + return nil +} + +func ResolveRequestID(request *http.Request) (string, error) { + if request == nil { + return fallbackRequestID(ErrInvalidRequestID) + } + values := request.Header.Values(HeaderRequestID) + if len(values) == 0 || (len(values) == 1 && values[0] == "") { + return generateRequestID() + } + if len(values) != 1 { + return fallbackRequestID(ErrInvalidRequestID) + } + requestID := values[0] + if len(requestID) > maxRequestIDLength || strings.TrimSpace(requestID) != requestID { + return fallbackRequestID(ErrInvalidRequestID) + } + for _, character := range requestID { + if character < 0x20 || character == 0x7f { + return fallbackRequestID(ErrInvalidRequestID) + } + } + return requestID, nil +} + +func WriteJSON(writer http.ResponseWriter, status int, value any) error { + writer.Header().Set("Content-Type", JSONContentType) + writer.WriteHeader(status) + return json.NewEncoder(writer).Encode(value) +} + +func WriteProblem(writer http.ResponseWriter, problem Problem) { + writer.Header().Set("Content-Type", ProblemContentType) + if problem.RequestID != "" { + writer.Header().Set(HeaderRequestID, problem.RequestID) + } + writer.WriteHeader(problem.Status) + _ = json.NewEncoder(writer).Encode(problem) +} + +func classifyDecodeError(err error) error { + var tooLarge *http.MaxBytesError + if errors.As(err, &tooLarge) { + return fmt.Errorf("%w: limit is %d bytes", ErrBodyTooLarge, tooLarge.Limit) + } + return ErrInvalidJSON +} + +func generateRequestID() (string, error) { + random := make([]byte, 16) + if _, err := rand.Read(random); err != nil { + return "", fmt.Errorf("generate request ID: %w", err) + } + return "req_" + hex.EncodeToString(random), nil +} + +func fallbackRequestID(reason error) (string, error) { + requestID, err := generateRequestID() + if err != nil { + return "", errors.Join(reason, err) + } + return requestID, reason +} diff --git a/internal/platform/httpapi/httpapi_test.go b/internal/platform/httpapi/httpapi_test.go new file mode 100644 index 0000000..dfdeb15 --- /dev/null +++ b/internal/platform/httpapi/httpapi_test.go @@ -0,0 +1,151 @@ +package httpapi + +import ( + "errors" + "net/http" + "net/http/httptest" + "strings" + "testing" +) + +func TestDecodeJSONAcceptsSingleStrictDocument(t *testing.T) { + t.Parallel() + type payload struct { + Count int `json:"count"` + } + + request := httptest.NewRequest(http.MethodPost, "/", strings.NewReader(`{"count":2}`)) + request.Header.Set("Content-Type", "application/json; charset=utf-8") + recorder := httptest.NewRecorder() + var decoded payload + + if err := DecodeJSON(recorder, request, 64, &decoded); err != nil { + t.Fatalf("DecodeJSON() error = %v", err) + } + if decoded.Count != 2 { + t.Fatalf("decoded count = %d, want 2", decoded.Count) + } +} + +func TestDecodeJSONRejectsUnsafeInput(t *testing.T) { + t.Parallel() + tests := []struct { + name string + contentType string + body string + maxBytes int64 + wantErr error + }{ + {name: "missing content type", body: `{}`, maxBytes: 64, wantErr: ErrUnsupportedMediaType}, + {name: "wrong content type", contentType: "text/plain", body: `{}`, maxBytes: 64, wantErr: ErrUnsupportedMediaType}, + {name: "unknown field", contentType: "application/json", body: `{"extra":true}`, maxBytes: 64, wantErr: ErrInvalidJSON}, + {name: "multiple documents", contentType: "application/json", body: `{} {}`, maxBytes: 64, wantErr: ErrInvalidJSON}, + {name: "oversized", contentType: "application/json", body: `{"value":"0123456789"}`, maxBytes: 8, wantErr: ErrBodyTooLarge}, + } + + for _, test := range tests { + test := test + t.Run(test.name, func(t *testing.T) { + t.Parallel() + request := httptest.NewRequest(http.MethodPost, "/", strings.NewReader(test.body)) + request.Header.Set("Content-Type", test.contentType) + recorder := httptest.NewRecorder() + var decoded struct { + Value string `json:"value"` + } + + err := DecodeJSON(recorder, request, test.maxBytes, &decoded) + if !errors.Is(err, test.wantErr) { + t.Fatalf("DecodeJSON() error = %v, want %v", err, test.wantErr) + } + }) + } +} + +func TestResolveRequestID(t *testing.T) { + t.Parallel() + + request := httptest.NewRequest(http.MethodGet, "/", nil) + request.Header.Set(HeaderRequestID, "caller-request") + requestID, err := ResolveRequestID(request) + if err != nil { + t.Fatalf("ResolveRequestID() error = %v", err) + } + if requestID != "caller-request" { + t.Fatalf("request ID = %q, want caller-request", requestID) + } + + generated, err := ResolveRequestID(httptest.NewRequest(http.MethodGet, "/", nil)) + if err != nil { + t.Fatalf("ResolveRequestID() generated error = %v", err) + } + if !strings.HasPrefix(generated, "req_") || len(generated) != 36 { + t.Fatalf("generated request ID = %q", generated) + } +} + +func TestResolveRequestIDRejectsInvalidValues(t *testing.T) { + t.Parallel() + for _, value := range []string{" request", strings.Repeat("a", 129), "request\x7f"} { + request := httptest.NewRequest(http.MethodGet, "/", nil) + request.Header.Set(HeaderRequestID, value) + requestID, err := ResolveRequestID(request) + if !errors.Is(err, ErrInvalidRequestID) { + t.Fatalf("ResolveRequestID(%q) error = %v, want %v", value, err, ErrInvalidRequestID) + } + if !strings.HasPrefix(requestID, "req_") { + t.Fatalf("ResolveRequestID(%q) fallback = %q, want generated ID", value, requestID) + } + } +} + +func TestResolveRequestIDRejectsDuplicateHeader(t *testing.T) { + t.Parallel() + request := httptest.NewRequest(http.MethodGet, "/", nil) + request.Header.Add(HeaderRequestID, "req-one") + request.Header.Add(HeaderRequestID, "req-two") + + requestID, err := ResolveRequestID(request) + if !errors.Is(err, ErrInvalidRequestID) { + t.Fatalf("ResolveRequestID() error = %v, want %v", err, ErrInvalidRequestID) + } + if !strings.HasPrefix(requestID, "req_") { + t.Fatalf("ResolveRequestID() fallback = %q, want generated ID", requestID) + } +} + +func TestWriteProblemUsesStableContract(t *testing.T) { + t.Parallel() + recorder := httptest.NewRecorder() + + WriteProblem(recorder, Problem{ + Type: "https://proxy-pool.local/problems/invalid-request", + Title: "Invalid request", + Status: http.StatusBadRequest, + Code: "INVALID_REQUEST", + Detail: "request payload is invalid", + RequestID: "req-1", + }) + + if recorder.Code != http.StatusBadRequest { + t.Fatalf("status = %d, want %d", recorder.Code, http.StatusBadRequest) + } + if contentType := recorder.Header().Get("Content-Type"); contentType != ProblemContentType { + t.Fatalf("Content-Type = %q, want %q", contentType, ProblemContentType) + } + if requestID := recorder.Header().Get(HeaderRequestID); requestID != "req-1" { + t.Fatalf("X-Request-ID = %q, want req-1", requestID) + } + if body := recorder.Body.String(); !strings.Contains(body, `"code":"INVALID_REQUEST"`) || strings.Contains(body, "\n ") { + t.Fatalf("unexpected problem body %q", body) + } +} + +func TestNewProblemBuildsCanonicalType(t *testing.T) { + t.Parallel() + problem := NewProblem(http.StatusConflict, "IDEMPOTENCY_CONFLICT", "Idempotency conflict", "request changed", "req-1") + if problem.Type != "https://proxy-pool.local/problems/idempotency-conflict" || + problem.Status != http.StatusConflict || problem.Code != "IDEMPOTENCY_CONFLICT" || problem.RequestID != "req-1" { + t.Fatalf("NewProblem() = %+v", problem) + } +}