From ec3ceb2c9ac900aa1f8d59f629fb611de0f30b05 Mon Sep 17 00:00:00 2001 From: youfak Date: Sun, 2 Aug 2026 15:29:15 +0800 Subject: [PATCH] feat: expose admin audit pages --- README.md | 2 +- api/openapi/admin.yaml | 62 ++++++++++++++ api/openapi/openapi_test.go | 67 ++++++++++++++- docs/api/admin.md | 14 ++++ docs/requirements/completion-audit.md | 10 ++- internal/controller/admin/handler.go | 93 +++++++++++++++++++++ internal/controller/admin/handler_test.go | 85 +++++++++++++++++++ internal/controller/admin/service.go | 31 +++++++ internal/controller/admin/service_test.go | 63 ++++++++++++-- internal/controller/runtime/runtime_test.go | 11 +++ 10 files changed, 423 insertions(+), 15 deletions(-) diff --git a/README.md b/README.md index d66e84a..4b6751f 100644 --- a/README.md +++ b/README.md @@ -111,7 +111,7 @@ flowchart LR - **已完成**：严格配置、Provider 获取与协调、Redis 活动池、Distribution 原子 提取与限流、Controller 的 Admin/Distribution/Metrics 监听，以及 PostgreSQL - 管理状态；WorkerControlPlane 的 Register、Snapshot ACK、Runtime 心跳接收和 + 管理状态与有界审计查询；WorkerControlPlane 的 Register、Snapshot ACK、Runtime 心跳接收和 Redis 会话栅栏，以及 Gateway Outcome 上报的有界队列、序列确认与重试； Controller 的 Redis 共享 BASIC/EGRESS/TARGET 检查任务、按上游的有界轮转调度、HTTP/HTTPS/SOCKS5 Checker 探测和 diff --git a/api/openapi/admin.yaml b/api/openapi/admin.yaml index 425fa1a..1f815e8 100644 --- a/api/openapi/admin.yaml +++ b/api/openapi/admin.yaml @@ -7,6 +7,7 @@ servers: - url: http://127.0.0.1:8082 tags: - name: Status + - name: Audit - name: Upstreams - name: Routing - name: Configuration @@ -36,6 +37,42 @@ paths: '405': {$ref: '#/components/responses/MethodNotAllowed'} '500': {$ref: '#/components/responses/InternalServerError'} '503': {$ref: '#/components/responses/ServiceUnavailable'} + /api/v1/audit: + get: + tags: [Audit] + operationId: listAuditRecords + summary: 按 ID 游标读取管理面审计记录 + description: | + 仅返回权威管理面的变更审计记录，按 `id` 升序排列。`afterId` 是排他游标： + 后续页面只包含 `id` 大于该值的记录。未传 `limit` 时返回 100 条，单页最多 + 1000 条。 + + 审计接口不返回 Proxy、Upstream URL、凭据或活动池内容。 + parameters: + - name: afterId + in: query + required: false + description: 排他游标；仅返回 ID 大于此值的记录。 + schema: {type: integer, minimum: 0, default: 0} + - name: limit + in: query + required: false + description: 单页记录数；默认 100，最大 1000。 + schema: {type: integer, minimum: 1, maximum: 1000, default: 100} + responses: + '200': + description: 管理面审计记录页 + headers: + X-Request-ID: {$ref: '#/components/headers/RequestID'} + content: + application/json: + schema: {$ref: '#/components/schemas/AuditPage'} + '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] @@ -152,6 +189,31 @@ components: description: 服务端最终使用的请求标识。 schema: {type: string, maxLength: 128} schemas: + AuditPage: + type: object + additionalProperties: false + required: [records] + properties: + records: + type: array + description: 按 ID 升序的审计记录；没有更多记录时为空数组。 + items: {$ref: '#/components/schemas/AuditRecord'} + AuditRecord: + type: object + additionalProperties: false + required: [id, requestId, actorId, action, resourceType, resourceName, changed, version, occurredAt] + properties: + id: {type: integer, minimum: 0} + requestId: {type: string} + actorId: {type: string} + sourceIp: {type: string} + action: {type: string} + resourceType: {type: string} + resourceName: {type: string} + changed: {type: boolean} + version: {type: integer, minimum: 0} + reason: {type: string} + occurredAt: {type: string, format: date-time} Status: type: object additionalProperties: false diff --git a/api/openapi/openapi_test.go b/api/openapi/openapi_test.go index 26208a1..0915e94 100644 --- a/api/openapi/openapi_test.go +++ b/api/openapi/openapi_test.go @@ -8,8 +8,9 @@ import ( ) type document struct { - OpenAPI string `yaml:"openapi"` - Paths map[string]map[string]any `yaml:"paths"` + OpenAPI string `yaml:"openapi"` + Paths map[string]map[string]any `yaml:"paths"` + Components map[string]any `yaml:"components"` } func TestDistributionContract(t *testing.T) { @@ -37,6 +38,7 @@ func TestAdminContract(t *testing.T) { spec := readDocument(t, "admin.yaml") for _, path := range []string{ "/api/v1/status", + "/api/v1/audit", "/api/v1/upstreams/{name}/enable", "/api/v1/upstreams/{name}/disable", "/api/v1/routing/{name}/switch", @@ -48,6 +50,67 @@ func TestAdminContract(t *testing.T) { } requireResponses(t, spec.Paths["/api/v1/routing/{name}/switch"]["post"], "200", "400", "401", "403", "404", "405", "409", "413", "415", "422", "500", "503") + requireResponses(t, spec.Paths["/api/v1/audit"]["get"], + "200", "400", "401", "403", "405", "500", "503") + requireAuditPaginationContract(t, spec.Paths["/api/v1/audit"]["get"]) + requireAuditSchemas(t, spec) +} + +func requireAuditPaginationContract(t *testing.T, operation any) { + t.Helper() + operationMap, ok := operation.(map[string]any) + if !ok { + t.Fatalf("audit operation has type %T, want map", operation) + } + parameters, ok := operationMap["parameters"].([]any) + if !ok { + t.Fatalf("audit parameters has type %T, want array", operationMap["parameters"]) + } + afterID := parameterSchema(t, parameters, "afterId") + if afterID["type"] != "integer" || afterID["minimum"] != 0 || afterID["default"] != 0 { + t.Errorf("afterId schema = %#v, want integer with minimum/default 0", afterID) + } + limit := parameterSchema(t, parameters, "limit") + if limit["type"] != "integer" || limit["minimum"] != 1 || limit["maximum"] != 1000 || limit["default"] != 100 { + t.Errorf("limit schema = %#v, want integer [1, 1000] with default 100", limit) + } +} + +func parameterSchema(t *testing.T, parameters []any, name string) map[string]any { + t.Helper() + for _, raw := range parameters { + parameter, ok := raw.(map[string]any) + if !ok || parameter["name"] != name || parameter["in"] != "query" { + continue + } + schema, ok := parameter["schema"].(map[string]any) + if !ok { + t.Fatalf("%s parameter schema has type %T, want map", name, parameter["schema"]) + } + return schema + } + t.Fatalf("query parameter %q is missing", name) + return nil +} + +func requireAuditSchemas(t *testing.T, spec document) { + t.Helper() + schemas, ok := spec.Components["schemas"].(map[string]any) + if !ok { + t.Fatal("components.schemas is missing") + } + for _, name := range []string{"AuditPage", "AuditRecord"} { + if _, ok := schemas[name].(map[string]any); !ok { + t.Errorf("schema %s is missing", name) + } + } + record, _ := schemas["AuditRecord"].(map[string]any) + properties, _ := record["properties"].(map[string]any) + for _, forbidden := range []string{"proxy", "upstreamUrl", "credential", "activePool"} { + if _, exists := properties[forbidden]; exists { + t.Errorf("AuditRecord must not expose %q", forbidden) + } + } } func requireResponses(t *testing.T, operation any, codes ...string) { diff --git a/docs/api/admin.md b/docs/api/admin.md index 4aef0f6..9967a23 100644 --- a/docs/api/admin.md +++ b/docs/api/admin.md @@ -6,6 +6,7 @@ Admin API 使用独立监听器与权限，契约位于 `api/openapi/admin.yaml` ## 端点 - `GET /api/v1/status`：返回配置/快照版本、Upstream 聚合计数和 Worker 状态。 +- `GET /api/v1/audit`：按审计记录 ID 升序读取权威管理面的变更记录。 - `POST /api/v1/upstreams/{name}/enable`：启用 Upstream。 - `POST /api/v1/upstreams/{name}/disable`：停止新 Fetch/分配并自然 Drain。 - `POST /api/v1/routing/{name}/switch`：用 expectedCurrent 做 CAS 手工切换。 @@ -17,6 +18,19 @@ Admin API 使用独立监听器与权限，契约位于 `api/openapi/admin.yaml` 配置重载校验失败返回 422，旧配置继续运行。Status 只返回低基数聚合信息， 不得返回 Proxy 地址、凭据、Client 标识或完整 Provider URL。 +## 审计分页与数据范围 + +`GET /api/v1/audit` 使用 `afterId` 排他游标分页：响应只包含 `id` 大于 +`afterId` 的记录，并按 `id` 升序排列；未传时 `afterId` 为 `0`。未传 `limit` +时服务端使用 `100`； +`limit` 必须为 `1..1000`，其中 `1000` 等于 `adminstate.MaxPageSize`。客户端应将 +本页最后一条记录的 `id` 作为下一次请求的 `afterId`；空 `records` 表示当前游标后 +没有记录。 + +该接口仅暴露管理面变更记录及操作者标识、可信来源 IP、操作类型、资源标识、变更 +状态、控制面版本、原因和发生时间。它不返回 Proxy、任何 Upstream/Provider URL、 +凭据或 Secret、Provider 响应载荷，也不读取或暴露 Redis 活动池及其提取状态。 + ## 运行时实现边界 `admin.Handler` 只依赖 `Service` 控制面接口，不直接操作数据库、路由游标或配置 diff --git a/docs/requirements/completion-audit.md b/docs/requirements/completion-audit.md index a1dbd9e..83fd082 100644 --- a/docs/requirements/completion-audit.md +++ b/docs/requirements/completion-audit.md @@ -16,7 +16,8 @@ - Distribution OpenAPI：一次性独占提取、partial/allOrNothing、幂等键、 Redis TTL 活动池原子语义、TTL/健康过滤结果与标准错误。 -- Admin OpenAPI：状态、Upstream 启停、Routing 切换和配置重载。 +- Admin OpenAPI：状态、按 ID 游标分页的权威审计查询、Upstream 启停、Routing + 切换和配置重载。 - 两份 OpenAPI 已进入 Go/CI 结构门禁，覆盖本地引用闭合、operationId、响应和 security scheme；完整标准工具验证仍待补齐。 - Protobuf：Worker 注册、全量/增量 Snapshot、`usable_until`、ACK、运行态/ @@ -44,8 +45,9 @@ - `PROVIDER-*`：Provider HTTP Client、严格响应上限、模板解析安全边界、凭据 引用 Store 与 Reconciler Adapter 已实现。 - `DIST/Admin HTTP`：严格 JSON、Request ID、Problem 响应及 Distribution/Admin - Handler 已实现；共享认证、CIDR、可信代理、Client ID 与本地准入保护链已接入， - Controller Runtime 已将二者装配到独立监听器并支持联动优雅停机。 + Handler 已实现；Admin 审计查询以有界 `afterId` 游标读取 PostgreSQL 权威记录， + 不读取 Proxy 或 Redis 活动池。共享认证、CIDR、可信代理、Client ID 与本地准入 + 保护链已接入，Controller Runtime 已将二者装配到独立监听器并支持联动优雅停机。 - `Redis Activity Adapter`：真实 Redis 8.2 已覆盖 Provider Upsert、健康更新、 原子独占提取、短期幂等、Worker ownership、库存和有界过期清理，Memory/Redis 运行同一公用契约。 @@ -123,7 +125,7 @@ Controller/Gateway 入口，完整 mTLS 运行时拓扑仍只有静态验证。 7. BASIC Checker 调度与 HTTP/HTTPS/SOCKS5 探测器、全局与 TARGET Profile 的 Memory/Redis 原子归并、Controller Reducer 和 Observation 上报 RPC 已完成；EGRESS、TARGET 生产任务调度与 REMOVE 编排仍待实现。 -8. Admin/Distribution 细粒度授权和审计查询；Distribution 分布式限流已完成。 +8. Admin/Distribution 细粒度授权；Distribution 分布式限流和 Admin 审计查询已完成。 9. 真实 Compose/Kubernetes 集成、故障演练和代表性集群负载测试。 10. 将 reject/wait/direct 接入 Distribution 运行链。 ## 4. 容量结论 diff --git a/internal/controller/admin/handler.go b/internal/controller/admin/handler.go index 5d86aca..66487a9 100644 --- a/internal/controller/admin/handler.go +++ b/internal/controller/admin/handler.go @@ -5,18 +5,23 @@ import ( "errors" "fmt" "net/http" + "strconv" "strings" + "time" + "proxy-pool/internal/domain/adminstate" "proxy-pool/internal/platform/httpapi" "proxy-pool/internal/platform/httpsecurity" ) const ( statusPath = "/api/v1/status" + auditPath = "/api/v1/audit" reloadPath = "/api/v1/config/reload" upstreamPrefix = "/api/v1/upstreams/" routingPrefix = "/api/v1/routing/" maxResourceNameBytes = 128 + defaultAuditPageSize = 100 ) var ( @@ -29,6 +34,7 @@ var ( type Service interface { Status(context.Context) (Status, error) + ReadAudit(context.Context, AuditQuery) (AuditPage, error) SetUpstreamEnabled(context.Context, SetUpstreamCommand) (MutationResult, error) SwitchRouting(context.Context, SwitchCommand) (MutationResult, error) ReloadConfiguration(context.Context, ReloadCommand) (MutationResult, error) @@ -78,6 +84,36 @@ type MutationResult struct { Message string `json:"message,omitempty"` } +type AuditQuery struct { + AfterID uint64 + Limit int +} + +func (query AuditQuery) Validate() error { + if query.Limit <= 0 || query.Limit > adminstate.MaxPageSize { + return ErrInvalidConfiguration + } + return nil +} + +type AuditRecord struct { + ID uint64 `json:"id"` + RequestID string `json:"requestId"` + ActorID string `json:"actorId"` + SourceIP string `json:"sourceIp,omitempty"` + Action string `json:"action"` + ResourceType string `json:"resourceType"` + ResourceName string `json:"resourceName"` + Changed bool `json:"changed"` + Version uint64 `json:"version"` + Reason string `json:"reason,omitempty"` + OccurredAt time.Time `json:"occurredAt"` +} + +type AuditPage struct { + Records []AuditRecord `json:"records"` +} + type SetUpstreamCommand struct { RequestID string ActorID string @@ -136,6 +172,12 @@ func (handler *Handler) ServeHTTP(writer http.ResponseWriter, request *http.Requ } handler.getStatus(writer, request, requestID) return + case auditPath: + if !requireMethod(writer, request, http.MethodGet, requestID) { + return + } + handler.readAudit(writer, request, requestID) + return case reloadPath: if !requireMethod(writer, request, http.MethodPost, requestID) { return @@ -178,6 +220,57 @@ func (handler *Handler) getStatus(writer http.ResponseWriter, request *http.Requ _ = httpapi.WriteJSON(writer, http.StatusOK, status) } +func (handler *Handler) readAudit(writer http.ResponseWriter, request *http.Request, requestID string) { + query, err := parseAuditQuery(request) + if err != nil { + writeTransportProblem(writer, http.StatusBadRequest, "INVALID_AUDIT_QUERY", "Invalid audit query", "audit pagination fields violate the API contract", requestID) + return + } + page, err := handler.service.ReadAudit(request.Context(), query) + if err != nil { + writeServiceProblem(writer, err, requestID) + return + } + if page.Records == nil { + page.Records = []AuditRecord{} + } + writer.Header().Set(httpapi.HeaderRequestID, requestID) + _ = httpapi.WriteJSON(writer, http.StatusOK, page) +} + +func parseAuditQuery(request *http.Request) (AuditQuery, error) { + if request == nil || request.URL == nil { + return AuditQuery{}, ErrInvalidConfiguration + } + values := request.URL.Query() + query := AuditQuery{Limit: defaultAuditPageSize} + for name, value := range values { + if len(value) != 1 { + return AuditQuery{}, ErrInvalidConfiguration + } + switch name { + case "afterId": + parsed, err := strconv.ParseUint(value[0], 10, 64) + if err != nil { + return AuditQuery{}, ErrInvalidConfiguration + } + query.AfterID = parsed + case "limit": + parsed, err := strconv.ParseUint(value[0], 10, 32) + if err != nil || parsed == 0 || parsed > adminstate.MaxPageSize { + return AuditQuery{}, ErrInvalidConfiguration + } + query.Limit = int(parsed) + default: + return AuditQuery{}, ErrInvalidConfiguration + } + } + if err := query.Validate(); err != nil { + return AuditQuery{}, err + } + return query, nil +} + func (handler *Handler) setUpstreamEnabled( writer http.ResponseWriter, request *http.Request, diff --git a/internal/controller/admin/handler_test.go b/internal/controller/admin/handler_test.go index fdefaf4..04f0174 100644 --- a/internal/controller/admin/handler_test.go +++ b/internal/controller/admin/handler_test.go @@ -8,6 +8,7 @@ import ( "net/http/httptest" "strings" "testing" + "time" "proxy-pool/internal/platform/httpapi" "proxy-pool/internal/platform/httpsecurity" @@ -38,6 +39,80 @@ func TestHandlerReturnsStatusWithoutSensitiveDetails(t *testing.T) { } } +func TestHandlerReadsBoundedAuditPage(t *testing.T) { + t.Parallel() + occurredAt := time.Date(2026, 8, 2, 9, 30, 0, 0, time.UTC) + service := &stubService{audit: AuditPage{Records: []AuditRecord{{ + ID: 8, RequestID: "req-8", ActorID: "admin:alice", Action: "switch_routing", + ResourceType: "routing", ResourceName: "checkout", Changed: true, Version: 12, + Reason: "capacity", OccurredAt: occurredAt, + }}}} + handler := mustHandler(t, service) + recorder := httptest.NewRecorder() + request := httptest.NewRequest(http.MethodGet, "/api/v1/audit?afterId=7&limit=2", nil) + request.Header.Set(httpapi.HeaderRequestID, "req-audit-page") + + 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.auditCalls != 1 || service.lastAudit != (AuditQuery{AfterID: 7, Limit: 2}) { + t.Fatalf("ReadAudit() calls=%d query=%+v", service.auditCalls, service.lastAudit) + } + if requestID := recorder.Header().Get(httpapi.HeaderRequestID); requestID != "req-audit-page" { + t.Fatalf("response request ID = %q, want req-audit-page", requestID) + } + var response AuditPage + if err := json.Unmarshal(recorder.Body.Bytes(), &response); err != nil { + t.Fatalf("decode response: %v", err) + } + if len(response.Records) != 1 || response.Records[0].ID != 8 || !response.Records[0].OccurredAt.Equal(occurredAt) { + t.Fatalf("unexpected audit response: %+v", response) + } +} + +func TestHandlerUsesDefaultAuditLimitAndRejectsInvalidAuditQueries(t *testing.T) { + t.Parallel() + tests := []struct { + name string + path string + wantStatus int + wantQuery AuditQuery + }{ + {name: "default", path: "/api/v1/audit", wantStatus: http.StatusOK, wantQuery: AuditQuery{Limit: defaultAuditPageSize}}, + {name: "unknown field", path: "/api/v1/audit?beforeId=1", wantStatus: http.StatusBadRequest}, + {name: "duplicate field", path: "/api/v1/audit?limit=1&limit=2", wantStatus: http.StatusBadRequest}, + {name: "zero limit", path: "/api/v1/audit?limit=0", wantStatus: http.StatusBadRequest}, + {name: "oversized limit", path: "/api/v1/audit?limit=1001", wantStatus: http.StatusBadRequest}, + {name: "invalid cursor", path: "/api/v1/audit?afterId=nope", wantStatus: http.StatusBadRequest}, + } + for _, test := range tests { + test := test + t.Run(test.name, func(t *testing.T) { + t.Parallel() + service := &stubService{} + handler := mustHandler(t, service) + recorder := httptest.NewRecorder() + + handler.ServeHTTP(recorder, httptest.NewRequest(http.MethodGet, test.path, nil)) + + if recorder.Code != test.wantStatus { + t.Fatalf("status = %d, want %d; body=%s", recorder.Code, test.wantStatus, recorder.Body.String()) + } + if test.wantStatus == http.StatusOK { + if service.auditCalls != 1 || service.lastAudit != test.wantQuery { + t.Fatalf("ReadAudit() calls=%d query=%+v, want %+v", service.auditCalls, service.lastAudit, test.wantQuery) + } + return + } + if service.auditCalls != 0 || !strings.Contains(recorder.Body.String(), `"code":"INVALID_AUDIT_QUERY"`) { + t.Fatalf("invalid audit query reached service or returned wrong problem: calls=%d body=%s", service.auditCalls, recorder.Body.String()) + } + }) + } +} + func TestNewHandlerRejectsMissingDependenciesAndInvalidLimit(t *testing.T) { t.Parallel() if _, err := NewHandler(nil, allowAuthorizer{}, Options{MaxBodyBytes: 1024}); !errors.Is(err, ErrInvalidHandler) { @@ -232,6 +307,7 @@ func TestHandlerRejectsInvalidTransportRequests(t *testing.T) { wantStatus int }{ {name: "method", method: http.MethodPut, path: "/api/v1/config/reload", wantStatus: http.StatusMethodNotAllowed}, + {name: "audit method", method: http.MethodPost, path: "/api/v1/audit", 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}, @@ -277,12 +353,15 @@ func mustHandler(t *testing.T, service Service) *Handler { type stubService struct { status Status + audit AuditPage mutation MutationResult err error + lastAudit AuditQuery lastUpstream SetUpstreamCommand lastSwitch SwitchCommand lastReload ReloadCommand statusCalls int + auditCalls int } func (service *stubService) Status(context.Context) (Status, error) { @@ -290,6 +369,12 @@ func (service *stubService) Status(context.Context) (Status, error) { return service.status, service.err } +func (service *stubService) ReadAudit(_ context.Context, query AuditQuery) (AuditPage, error) { + service.auditCalls++ + service.lastAudit = query + return service.audit, service.err +} + func (service *stubService) SetUpstreamEnabled(_ context.Context, command SetUpstreamCommand) (MutationResult, error) { service.lastUpstream = command return service.mutation, service.err diff --git a/internal/controller/admin/service.go b/internal/controller/admin/service.go index cf72a47..6a06347 100644 --- a/internal/controller/admin/service.go +++ b/internal/controller/admin/service.go @@ -16,6 +16,7 @@ var ErrInvalidApplicationService = errors.New("invalid admin application service type StateRepository interface { adminstate.Mutator adminstate.SnapshotReader + adminstate.AuditReader } type OperationalStatusReader interface { @@ -150,6 +151,36 @@ func (service *ApplicationService) SwitchRouting(ctx context.Context, command Sw return mutationResult(result), mapAdminStateError(err) } +// ReadAudit returns one bounded, stable page from the authoritative management +// audit log. It deliberately exposes only management-plane records and never +// reads Proxy activity, credentials, or Provider payloads. +func (service *ApplicationService) ReadAudit(ctx context.Context, query AuditQuery) (AuditPage, error) { + if service == nil || nilInterface(service.state) || query.Validate() != nil { + return AuditPage{}, ErrInvalidConfiguration + } + records, err := service.state.ReadAudit(ctx, adminstate.AuditQuery{AfterID: query.AfterID, Limit: query.Limit}) + if err != nil { + return AuditPage{}, mapAdminStateError(err) + } + page := AuditPage{Records: make([]AuditRecord, 0, len(records))} + for _, record := range records { + page.Records = append(page.Records, AuditRecord{ + ID: record.ID, + RequestID: record.RequestID, + ActorID: record.Actor.ID, + SourceIP: record.Actor.SourceIP, + Action: string(record.Action), + ResourceType: record.ResourceType, + ResourceName: record.ResourceName, + Changed: record.Changed, + Version: record.Revision, + Reason: record.Reason, + OccurredAt: record.OccurredAt.UTC(), + }) + } + return page, nil +} + func (service *ApplicationService) Status(ctx context.Context) (Status, error) { snapshot, err := service.state.Snapshot(ctx) if err != nil { diff --git a/internal/controller/admin/service_test.go b/internal/controller/admin/service_test.go index 380bcc7..007efbb 100644 --- a/internal/controller/admin/service_test.go +++ b/internal/controller/admin/service_test.go @@ -206,6 +206,42 @@ func TestApplicationServiceMapsRoutingSwitchAndDomainErrors(t *testing.T) { } } +func TestApplicationServiceReadsAuthoritativeAuditPage(t *testing.T) { + t.Parallel() + occurredAt := time.Date(2026, 8, 2, 10, 0, 0, 0, time.FixedZone("test", 8*60*60)) + state := &recordingAdminState{audits: []adminstate.AuditRecord{{ + ID: 42, RequestID: "req-switch", Actor: adminstate.Actor{ID: "admin:alice", SourceIP: "192.0.2.10"}, + Action: adminstate.ActionSwitchRoute, ResourceType: "routing", ResourceName: "checkout", + Changed: true, Revision: 9, Reason: "capacity", OccurredAt: occurredAt, + }}} + service := mustApplicationService(t, state, applicationTestOptions(time.Now)) + + page, err := service.ReadAudit(context.Background(), AuditQuery{AfterID: 41, Limit: 2}) + + if err != nil { + t.Fatalf("ReadAudit() error = %v", err) + } + if state.lastAuditQuery != (adminstate.AuditQuery{AfterID: 41, Limit: 2}) { + t.Fatalf("audit query = %+v", state.lastAuditQuery) + } + if len(page.Records) != 1 { + t.Fatalf("audit records = %+v", page.Records) + } + record := page.Records[0] + if record.ID != 42 || record.Action != string(adminstate.ActionSwitchRoute) || record.Version != 9 || + record.ActorID != "admin:alice" || record.SourceIP != "192.0.2.10" || !record.OccurredAt.Equal(occurredAt.UTC()) { + t.Fatalf("mapped audit record = %+v", record) + } + + state.err = adminstate.ErrUnavailable + if _, err := service.ReadAudit(context.Background(), AuditQuery{Limit: 1}); !errors.Is(err, ErrUnavailable) || !errors.Is(err, adminstate.ErrUnavailable) { + t.Fatalf("ReadAudit() unavailable error = %v", err) + } + if _, err := service.ReadAudit(context.Background(), AuditQuery{}); !errors.Is(err, ErrInvalidConfiguration) { + t.Fatalf("ReadAudit() invalid query error = %v", err) + } +} + func TestApplicationServiceBuildsStatusFromAuthoritativeAndOperationalSnapshots(t *testing.T) { t.Parallel() state := &recordingAdminState{snapshot: adminstate.Snapshot{ @@ -642,14 +678,16 @@ func mustApplicationService(t *testing.T, state StateRepository, options Applica } type recordingAdminState struct { - mutation adminstate.MutationResult - err error - snapshot adminstate.Snapshot - lastUpstream adminstate.SetUpstreamCommand - lastSwitch adminstate.SwitchRoutingCommand - lastDisable adminstate.DisableRoutingCommand - lastConfig adminstate.CommitConfigCommand - onCommit func(adminstate.CommitConfigCommand) + mutation adminstate.MutationResult + err error + snapshot adminstate.Snapshot + audits []adminstate.AuditRecord + lastAuditQuery adminstate.AuditQuery + lastUpstream adminstate.SetUpstreamCommand + lastSwitch adminstate.SwitchRoutingCommand + lastDisable adminstate.DisableRoutingCommand + lastConfig adminstate.CommitConfigCommand + onCommit func(adminstate.CommitConfigCommand) } type orderedCommitState struct { @@ -683,6 +721,10 @@ func (*orderedCommitState) Snapshot(context.Context) (adminstate.Snapshot, error return adminstate.Snapshot{}, nil } +func (*orderedCommitState) ReadAudit(context.Context, adminstate.AuditQuery) ([]adminstate.AuditRecord, error) { + return nil, nil +} + func (state *recordingAdminState) SetUpstreamEnabled(_ context.Context, command adminstate.SetUpstreamCommand) (adminstate.MutationResult, error) { state.lastUpstream = command return state.mutation, state.err @@ -710,6 +752,11 @@ func (state *recordingAdminState) Snapshot(context.Context) (adminstate.Snapshot return state.snapshot, state.err } +func (state *recordingAdminState) ReadAudit(_ context.Context, query adminstate.AuditQuery) ([]adminstate.AuditRecord, error) { + state.lastAuditQuery = query + return append([]adminstate.AuditRecord(nil), state.audits...), state.err +} + type staticOperationalStatusReader struct { status OperationalStatus err error diff --git a/internal/controller/runtime/runtime_test.go b/internal/controller/runtime/runtime_test.go index 25f7dde..461b863 100644 --- a/internal/controller/runtime/runtime_test.go +++ b/internal/controller/runtime/runtime_test.go @@ -57,12 +57,16 @@ func TestRuntimeServesDistributionAndAdminOnIndependentListeners(t *testing.T) { assertStatus(t, http.MethodGet, adminURL+"/api/v1/status", nil, http.StatusUnauthorized) adminHeaders := http.Header{"Authorization": []string{"Bearer admin-token"}} assertStatus(t, http.MethodGet, adminURL+"/api/v1/status", adminHeaders, http.StatusOK) + assertStatus(t, http.MethodGet, adminURL+"/api/v1/audit", adminHeaders, http.StatusOK) assertStatus(t, http.MethodGet, adminURL+"/health/live", adminHeaders, http.StatusNotFound) assertStatus(t, http.MethodGet, metricsURL+"/readyz", nil, http.StatusOK) assertStatus(t, http.MethodGet, metricsURL+"/api/v1/status", nil, http.StatusNotFound) if adminService.statusCalls.Load() != 1 { t.Fatalf("admin status calls = %d, want 1", adminService.statusCalls.Load()) } + if adminService.auditCalls.Load() != 1 { + t.Fatalf("admin audit calls = %d, want 1", adminService.auditCalls.Load()) + } cancel() select { @@ -229,7 +233,9 @@ func (stubReadiness) Ready(context.Context) error { return nil } type stubAdminService struct { status admin.Status + audit admin.AuditPage statusCalls atomic.Int64 + auditCalls atomic.Int64 } func (service *stubAdminService) Status(context.Context) (admin.Status, error) { @@ -237,6 +243,11 @@ func (service *stubAdminService) Status(context.Context) (admin.Status, error) { return service.status, nil } +func (service *stubAdminService) ReadAudit(context.Context, admin.AuditQuery) (admin.AuditPage, error) { + service.auditCalls.Add(1) + return service.audit, nil +} + func (*stubAdminService) SetUpstreamEnabled(context.Context, admin.SetUpstreamCommand) (admin.MutationResult, error) { return admin.MutationResult{}, nil }