From a71d1ac2d360087a9e059f90f0d79402d02ed5e0 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Alby=20Hern=C3=A1ndez?= Date: Tue, 14 Jul 2026 14:32:58 +0100 Subject: [PATCH 1/2] feat: unify access logging across both engines with stream correlation The logging block of the policy was only honored by the extAuthz HTTP server; the extProc gRPC engine never logged request or response data, leaving operators blind to what actually reaches it. - Extract the access-log rendering (exclude/redact/mask/logBody) from httpserver into a shared internal/accesslog package. RequestAttrs is the former accessLogAttrs, byte-for-byte; ResponseAttrs is new and applies the same rules to response status, headers and body. No behavior change for extAuthz. - The extProc engine now emits one 'extProc access' record at INFO for every phase message Envoy sends, with the data known at that point. What reaches the engine (which phases, whether bodies are streamed) remains a deployment concern of the Envoy processing_mode. - Correlate everything: a per-stream stream_id (Envoy opens one ext_proc stream per HTTP request) is attached to all access, mutation and overflow records of a request, and x-request-id is promoted to a top-level request_id field when present. Documented in POLICY_DSL.md and recorded as ADR D-024. --- .agents/DECISIONS.md | 26 + .agents/POLICY_DSL.md | 5 + .../access.go => accesslog/accesslog.go} | 61 +- .../accesslog_test.go} | 136 ++++- internal/grpcserver/access_test.go | 542 ++++++++++++++++++ internal/grpcserver/server.go | 73 ++- internal/httpserver/server.go | 5 +- 7 files changed, 831 insertions(+), 17 deletions(-) rename internal/{httpserver/access.go => accesslog/accesslog.go} (64%) rename internal/{httpserver/access_test.go => accesslog/accesslog_test.go} (56%) create mode 100644 internal/grpcserver/access_test.go diff --git a/.agents/DECISIONS.md b/.agents/DECISIONS.md index 456f558..ce4681f 100644 --- a/.agents/DECISIONS.md +++ b/.agents/DECISIONS.md @@ -591,3 +591,29 @@ just the expression `response.header`. being placed in the page (base64 in CEL, decoded by the button's JS) to avoid HTML injection. The example uses `base64.encode(...)` and a `__TARGET_B64__` placeholder. + + +## D-024 - Unified, engine-agnostic access logging + +**Context.** Originally, only the extAuthz engine supported access logging using the `logging` block configuration. The extProc engine had no structured access logging capability. + +**Decision.** Unify the access-logging capability across both engines using a new shared package `internal/accesslog`. The extProc engine will emit a structured access log record at INFO level ("extProc access") for every phase message Envoy sends, unconditionally. + +- `requestHeaders` phase logs `accesslog.RequestAttrs`. +- `requestBody` phase logs `accesslog.RequestAttrs` (containing the accumulated body). +- `responseHeaders` phase logs `accesslog.RequestAttrs` and `accesslog.ResponseAttrs`. +- `responseBody` phase logs `accesslog.ResponseAttrs` (containing the accumulated response body). + +The existing per-request custom headers, redactions, masking, and body log toggle configurations under the `logging` block apply symmetrically to both engines. + +**Rationale.** The request validator is a generic gatekeeping product. The deployment-side EnvoyFilter configuration controls which messages and metadata reach the extProc server. Delegating traffic volume and level details to Envoy's filters is standard practice; therefore, the validator does not need to duplicate these knobs internally, and can log every received message uniformly. + +**Alternative Rejected.** +- Logging only on a rule match: rejected because operators need visibility into all traffic flowing through the validator, not just match events. +- Configurable per-phase logging knobs: rejected to avoid redundant configuration overhead and keep the code path simple. + +**Consequences.** +- Extracted common header exclusion/redaction and body formatting logic into a unified `internal/accesslog` package. +- Solved the logging gap where extProc was entirely blind to structured, redacted request/response auditing. +- No behavior or schema changes for extAuthz logs. + diff --git a/.agents/POLICY_DSL.md b/.agents/POLICY_DSL.md index 7e4a550..0827a29 100644 --- a/.agents/POLICY_DSL.md +++ b/.agents/POLICY_DSL.md @@ -78,6 +78,11 @@ logging: redactQueryParams: [access_token, id_token, code] ``` +The logging block applies to BOTH engines: + +- extAuthz emits one access record per delegated request ("request decided"). +- extProc emits one access record per phase message at INFO ("extProc access"). The request and/or response body appears in extProc logs only when Envoy's processing_mode actually sends the body AND logBody is true. extProc records carry a stream_id shared by all phases of one HTTP request (one ext_proc stream = one request) and a request_id copied from the x-request-id header when present. + Behaviour: - Header keys are normalised to lowercase before exclude/redact checks. diff --git a/internal/httpserver/access.go b/internal/accesslog/accesslog.go similarity index 64% rename from internal/httpserver/access.go rename to internal/accesslog/accesslog.go index 2ece4cd..d133240 100644 --- a/internal/httpserver/access.go +++ b/internal/accesslog/accesslog.go @@ -1,7 +1,8 @@ // SPDX-FileCopyrightText: 2026 Alby Hernández // SPDX-License-Identifier: Apache-2.0 -package httpserver +// Package accesslog provides utilities for structured access logging across both engines. +package accesslog import ( "log/slog" @@ -11,14 +12,11 @@ import ( "request-validator/internal/policy" ) -// accessLogAttrs builds the slog group describing a single request. The -// caller adds higher-level fields (decision, rule, reason, dryRun, duration) -// around it; this function is concerned only with what came in. -// +// RequestAttrs builds the slog group describing a single request. // Header keys are always lowercase. Excluded headers are dropped, redacted // headers have their values masked. The body is included only when // logging.LogBody is true; the body size is included always. -func accessLogAttrs(req *policy.Request, lg policy.Logging) slog.Attr { +func RequestAttrs(req *policy.Request, lg policy.Logging) slog.Attr { exclude := lowerSet(lg.ExcludeHeaders) redact := lowerSet(lg.RedactHeaders) @@ -61,6 +59,57 @@ func accessLogAttrs(req *policy.Request, lg policy.Logging) slog.Attr { ) } +// ResponseAttrs builds the slog group describing a single response. +// Header keys are always lowercase. Excluded headers are dropped, redacted +// headers have their values masked. The body is included only when +// logging.LogBody is true; the body size is included always. +func ResponseAttrs(resp *policy.Response, lg policy.Logging) slog.Attr { + if resp == nil { + return slog.Group("response", + slog.Int("status", 0), + slog.Group("headers"), + slog.Group("body", + slog.Int("size", 0), + slog.String("content_type", ""), + ), + ) + } + + exclude := lowerSet(lg.ExcludeHeaders) + redact := lowerSet(lg.RedactHeaders) + + hdrs := make([]any, 0, len(resp.Headers)*2) + for k, vs := range resp.Headers { + lk := strings.ToLower(k) + if exclude[lk] { + continue + } + joined := strings.Join(vs, ", ") + if redact[lk] { + joined = mask(joined, lg.RedactReveal) + } + hdrs = append(hdrs, slog.String(lk, joined)) + } + + body := slog.Group("body", + slog.Int("size", len(resp.Body)), + slog.String("content_type", strings.ToLower(resp.Headers.Get("Content-Type"))), + ) + if lg.LogBody && len(resp.Body) > 0 { + body = slog.Group("body", + slog.Int("size", len(resp.Body)), + slog.String("content_type", strings.ToLower(resp.Headers.Get("Content-Type"))), + slog.String("raw", string(resp.Body)), + ) + } + + return slog.Group("response", + slog.Int("status", resp.Status), + slog.Group("headers", hdrs...), + body, + ) +} + // lowerSet builds a lowercase string set out of a slice. Empty entries are // ignored. func lowerSet(xs []string) map[string]bool { diff --git a/internal/httpserver/access_test.go b/internal/accesslog/accesslog_test.go similarity index 56% rename from internal/httpserver/access_test.go rename to internal/accesslog/accesslog_test.go index a8760cc..967f359 100644 --- a/internal/httpserver/access_test.go +++ b/internal/accesslog/accesslog_test.go @@ -1,7 +1,7 @@ // SPDX-FileCopyrightText: 2026 Alby Hernández // SPDX-License-Identifier: Apache-2.0 -package httpserver +package accesslog import ( "bytes" @@ -69,7 +69,7 @@ func TestAccessLogAttrsExcludeAndRedact(t *testing.T) { RedactQueryParams: []string{"code"}, } - log.Logger().Info("test", "decision", "allow", accessLogAttrs(req, lg)) + log.Logger().Info("test", "decision", "allow", RequestAttrs(req, lg)) out := buf.String() if !strings.Contains(out, `"decision":"allow"`) { @@ -115,7 +115,7 @@ func TestAccessLogAttrsLogBody(t *testing.T) { Body: []byte(`{"a":1}`), } lg := policy.Logging{LogBody: true} - log.Logger().Info("test", accessLogAttrs(req, lg)) + log.Logger().Info("test", RequestAttrs(req, lg)) out := buf.String() // Parse the JSON to assert structurally (less fragile than substring matching). @@ -145,7 +145,7 @@ func TestAccessLogHeaderKeysLowercased(t *testing.T) { hdrs := http.Header{} hdrs.Set("X-Custom-Header", "value") req := &policy.Request{Headers: hdrs} - log.Logger().Info("test", accessLogAttrs(req, policy.Logging{})) + log.Logger().Info("test", RequestAttrs(req, policy.Logging{})) if !strings.Contains(buf.String(), `"x-custom-header":"value"`) { t.Fatalf("expected lowercase key in: %s", buf.String()) @@ -167,3 +167,131 @@ func TestConsoleFormatProducesLine(t *testing.T) { t.Fatalf("console output unexpected: %q", out) } } + +func TestResponseAttrsExcludeAndRedact(t *testing.T) { + var buf bytes.Buffer + if err := log.Configure(log.Options{Level: "info", Format: log.FormatJSON, Writer: &buf}); err != nil { + t.Fatal(err) + } + t.Cleanup(func() { _ = log.Configure(log.Options{}) }) + + hdrs := http.Header{} + hdrs.Set("Content-Type", "application/json") + hdrs.Set("Set-Cookie", "session=verysecret") + hdrs.Set("Authorization", "Bearer 0123456789abcdef") + hdrs.Set("X-Custom", "intact-value") + + resp := &policy.Response{ + Status: 201, + Headers: hdrs, + Body: []byte(`{"status":"created"}`), + } + + lg := policy.Logging{ + ExcludeHeaders: []string{"set-cookie"}, + RedactHeaders: []string{"authorization"}, + RedactReveal: 6, + } + + log.Logger().Info("test_resp", ResponseAttrs(resp, lg)) + + out := buf.String() + + // Parse JSON to verify fields + var rec map[string]any + if err := json.Unmarshal(bytes.TrimSpace(buf.Bytes()), &rec); err != nil { + t.Fatalf("invalid JSON output: %v -- %s", err, out) + } + + respRec, ok := rec["response"].(map[string]any) + if !ok { + t.Fatalf("missing response group in log: %s", out) + } + + if int(respRec["status"].(float64)) != 201 { + t.Fatalf("status mismatch: %v", respRec["status"]) + } + + headers, ok := respRec["headers"].(map[string]any) + if !ok { + t.Fatalf("missing headers in response log: %s", out) + } + + if _, exists := headers["set-cookie"]; exists { + t.Fatalf("set-cookie should have been excluded: %v", headers) + } + + authVal, ok := headers["authorization"].(string) + if !ok { + t.Fatalf("authorization missing in headers: %v", headers) + } + // "Bearer 0123456789abcdef" is 21 chars. RedactReveal=6. + // 2*6 = 12 <= 21, so prefix "Bearer" (6 chars) is shown, rest is asterisks. + if !strings.HasPrefix(authVal, "Bearer") || strings.Contains(authVal, "0123456789") { + t.Fatalf("authorization not masked correctly: %q", authVal) + } + + customVal, ok := headers["x-custom"].(string) + if !ok || customVal != "intact-value" { + t.Fatalf("x-custom header mismatch: %v", headers) + } +} + +func TestResponseAttrsBodyOnlyWhenLogBody(t *testing.T) { + // First run: LogBody = false + { + var buf bytes.Buffer + if err := log.Configure(log.Options{Level: "info", Format: log.FormatJSON, Writer: &buf}); err != nil { + t.Fatal(err) + } + resp := &policy.Response{ + Status: 200, + Headers: http.Header{"Content-Type": []string{"text/plain"}}, + Body: []byte("my-body-content"), + } + lg := policy.Logging{LogBody: false} + log.Logger().Info("test_resp_no_body", ResponseAttrs(resp, lg)) + + var rec map[string]any + if err := json.Unmarshal(bytes.TrimSpace(buf.Bytes()), &rec); err != nil { + t.Fatalf("invalid JSON output: %v", err) + } + respRec := rec["response"].(map[string]any) + body := respRec["body"].(map[string]any) + if int(body["size"].(float64)) != len("my-body-content") { + t.Fatalf("wrong size when logBody is false: %v", body) + } + if _, exists := body["raw"]; exists { + t.Fatalf("body raw should be absent when LogBody is false: %v", body) + } + } + + // Second run: LogBody = true + { + var buf bytes.Buffer + if err := log.Configure(log.Options{Level: "info", Format: log.FormatJSON, Writer: &buf}); err != nil { + t.Fatal(err) + } + resp := &policy.Response{ + Status: 200, + Headers: http.Header{"Content-Type": []string{"text/plain"}}, + Body: []byte("my-body-content"), + } + lg := policy.Logging{LogBody: true} + log.Logger().Info("test_resp_with_body", ResponseAttrs(resp, lg)) + + var rec map[string]any + if err := json.Unmarshal(bytes.TrimSpace(buf.Bytes()), &rec); err != nil { + t.Fatalf("invalid JSON output: %v", err) + } + respRec := rec["response"].(map[string]any) + body := respRec["body"].(map[string]any) + if int(body["size"].(float64)) != len("my-body-content") { + t.Fatalf("wrong size when logBody is true: %v", body) + } + raw, ok := body["raw"].(string) + if !ok || raw != "my-body-content" { + t.Fatalf("body raw should equal the body string when LogBody is true: %v", body) + } + } +} diff --git a/internal/grpcserver/access_test.go b/internal/grpcserver/access_test.go new file mode 100644 index 0000000..802389e --- /dev/null +++ b/internal/grpcserver/access_test.go @@ -0,0 +1,542 @@ +// SPDX-FileCopyrightText: 2026 Alby Hernández +// SPDX-License-Identifier: Apache-2.0 + +package grpcserver + +import ( + "bytes" + "encoding/json" + "strings" + "testing" + + epb "github.com/envoyproxy/go-control-plane/envoy/service/ext_proc/v3" + + "request-validator/internal/log" +) + +func TestProcessLogsRequestHeadersAccess(t *testing.T) { + var buf bytes.Buffer + if err := log.Configure(log.Options{Level: "info", Format: log.FormatJSON, Writer: &buf}); err != nil { + t.Fatal(err) + } + t.Cleanup(func() { _ = log.Configure(log.Options{}) }) + + yamlStr := ` +defaults: + extProc: + maxBodyBytes: 1024 + onBodyOverflow: fail +logging: + excludeHeaders: + - cookie + redactHeaders: + - authorization + redactReveal: 6 +groups: + - name: test-group + parameters: + engine: extProc + mode: applyAll + phase: requestHeaders + match: "true" + rules: + - name: dummy + match: "true" +` + cfg := mustLoadConfig(t, yamlStr) + srv := New(cfg) + + stream := &fakeStream{ + incoming: []*epb.ProcessingRequest{ + { + Request: &epb.ProcessingRequest_RequestHeaders{ + RequestHeaders: &epb.HttpHeaders{ + Headers: makeHeaderMap(map[string]string{ + ":method": "GET", + ":path": "/hello", + ":scheme": "http", + ":authority": "localhost", + "x-pikaso-client": "plugin:test", + "Cookie": "session=secret", + "Authorization": "Bearer eyJ1234567890", + }), + }, + }, + }, + }, + } + + err := srv.Process(stream) + if err != nil { + t.Fatalf("unexpected error: %v", err) + } + + out := buf.String() + if !strings.Contains(out, `"extProc access"`) { + t.Fatalf("expected 'extProc access' log record, got: %s", out) + } + if !strings.Contains(out, `"phase":"requestHeaders"`) { + t.Fatalf("expected phase:requestHeaders, got: %s", out) + } + if !strings.Contains(out, `"x-pikaso-client":"plugin:test"`) { + t.Fatalf("expected x-pikaso-client in logs, got: %s", out) + } + if strings.Contains(strings.ToLower(out), "cookie") || strings.Contains(out, "session=secret") { + t.Fatalf("cookie or its value should have been excluded, got: %s", out) + } + if !strings.Contains(out, `"authorization":"Bearer*`) { + t.Fatalf("authorization should be redacted Bearer*, got: %s", out) + } +} + +func TestProcessLogsResponsePhases(t *testing.T) { + var buf bytes.Buffer + if err := log.Configure(log.Options{Level: "info", Format: log.FormatJSON, Writer: &buf}); err != nil { + t.Fatal(err) + } + t.Cleanup(func() { _ = log.Configure(log.Options{}) }) + + yamlStr := ` +defaults: + extProc: + maxBodyBytes: 1024 + onBodyOverflow: fail +logging: + logBody: true +groups: + - name: test-group + parameters: + engine: extProc + mode: applyAll + phase: responseHeaders + match: "true" + rules: + - name: dummy + match: "true" +` + cfg := mustLoadConfig(t, yamlStr) + srv := New(cfg) + + stream := &fakeStream{ + incoming: []*epb.ProcessingRequest{ + { + Request: &epb.ProcessingRequest_ResponseHeaders{ + ResponseHeaders: &epb.HttpHeaders{ + Headers: makeHeaderMap(map[string]string{ + ":status": "200", + "Content-Type": "application/json", + }), + }, + }, + }, + { + Request: &epb.ProcessingRequest_ResponseBody{ + ResponseBody: &epb.HttpBody{ + Body: []byte(`{"response_key":"response_val"}`), + }, + }, + }, + }, + } + + err := srv.Process(stream) + if err != nil { + t.Fatalf("unexpected error: %v", err) + } + + out := buf.String() + if !strings.Contains(out, `"extProc access"`) { + t.Fatalf("expected 'extProc access' log record, got: %s", out) + } + if !strings.Contains(out, `"status":200`) { + t.Fatalf("expected response.status:200, got: %s", out) + } + if !strings.Contains(out, `"phase":"responseHeaders"`) { + t.Fatalf("expected phase:responseHeaders log, got: %s", out) + } + if !strings.Contains(out, `"phase":"responseBody"`) { + t.Fatalf("expected phase:responseBody log, got: %s", out) + } + if !strings.Contains(out, `"raw":"{\"response_key\":\"response_val\"}"`) { + t.Fatalf("expected response body raw to be logged, got: %s", out) + } +} + +func TestStreamIDSharedAcrossPhases(t *testing.T) { + var buf bytes.Buffer + if err := log.Configure(log.Options{Level: "debug", Format: log.FormatJSON, Writer: &buf}); err != nil { + t.Fatal(err) + } + t.Cleanup(func() { _ = log.Configure(log.Options{}) }) + + yamlStr := ` +defaults: + extProc: + maxBodyBytes: 1024 + onBodyOverflow: fail +groups: + - name: headers-req + parameters: + engine: extProc + mode: applyAll + phase: requestHeaders + match: "true" + rules: + - name: r1 + match: "true" + - name: body-req + parameters: + engine: extProc + mode: applyAll + phase: requestBody + match: "true" + rules: + - name: r2 + match: "true" + - name: headers-resp + parameters: + engine: extProc + mode: applyAll + phase: responseHeaders + match: "true" + rules: + - name: r3 + match: "true" + - name: body-resp + parameters: + engine: extProc + mode: applyAll + phase: responseBody + match: "true" + rules: + - name: r4 + match: "true" +` + cfg := mustLoadConfig(t, yamlStr) + srv := New(cfg) + + stream := &fakeStream{ + incoming: []*epb.ProcessingRequest{ + { + Request: &epb.ProcessingRequest_RequestHeaders{ + RequestHeaders: &epb.HttpHeaders{ + Headers: makeHeaderMap(map[string]string{ + ":method": "GET", + ":path": "/hello", + ":scheme": "http", + ":authority": "localhost", + }), + }, + }, + }, + { + Request: &epb.ProcessingRequest_RequestBody{ + RequestBody: &epb.HttpBody{ + Body: []byte(`{"request_key":"request_val"}`), + }, + }, + }, + { + Request: &epb.ProcessingRequest_ResponseHeaders{ + ResponseHeaders: &epb.HttpHeaders{ + Headers: makeHeaderMap(map[string]string{ + ":status": "200", + "Content-Type": "application/json", + }), + }, + }, + }, + { + Request: &epb.ProcessingRequest_ResponseBody{ + ResponseBody: &epb.HttpBody{ + Body: []byte(`{"response_key":"response_val"}`), + }, + }, + }, + }, + } + + err := srv.Process(stream) + if err != nil { + t.Fatalf("unexpected error: %v", err) + } + + var records []map[string]any + lines := strings.Split(buf.String(), "\n") + for _, line := range lines { + line = strings.TrimSpace(line) + if line == "" { + continue + } + var record map[string]any + if err := json.Unmarshal([]byte(line), &record); err != nil { + t.Logf("failed to unmarshal log line %q: %v", line, err) + continue + } + records = append(records, record) + } + + var streamIDsFound []string + var phaseEvaluatedStreamIDs []string + for _, rec := range records { + msg, _ := rec["msg"].(string) + if msg == "extProc access" { + sid, _ := rec["stream_id"].(string) + if sid == "" { + t.Errorf("expected non-empty stream_id in extProc access log: %v", rec) + } else { + streamIDsFound = append(streamIDsFound, sid) + } + } else if msg == "extProc phase evaluated" { + sid, _ := rec["stream_id"].(string) + if sid == "" { + t.Errorf("expected non-empty stream_id in extProc phase evaluated log: %v", rec) + } else { + phaseEvaluatedStreamIDs = append(phaseEvaluatedStreamIDs, sid) + } + } + } + + if len(streamIDsFound) != 4 { + t.Fatalf("expected exactly 4 extProc access logs, got %d", len(streamIDsFound)) + } + if len(phaseEvaluatedStreamIDs) != 4 { + t.Fatalf("expected exactly 4 extProc phase evaluated logs, got %d", len(phaseEvaluatedStreamIDs)) + } + + // Assert all of them are equal to the first one + firstID := streamIDsFound[0] + for _, id := range streamIDsFound { + if id != firstID { + t.Errorf("expected all extProc access stream_ids to be equal (%s), but got %s", firstID, id) + } + } + for _, id := range phaseEvaluatedStreamIDs { + if id != firstID { + t.Errorf("expected all extProc phase evaluated stream_ids to match access stream_id (%s), but got %s", firstID, id) + } + } +} + +func TestStreamIDDiffersBetweenStreams(t *testing.T) { + yamlStr := ` +defaults: + extProc: + maxBodyBytes: 1024 + onBodyOverflow: fail +groups: + - name: test-group + parameters: + engine: extProc + mode: applyAll + phase: requestHeaders + match: "true" + rules: + - name: dummy + match: "true" +` + cfg := mustLoadConfig(t, yamlStr) + srv := New(cfg) + + // Stream 1 + var buf1 bytes.Buffer + if err := log.Configure(log.Options{Level: "info", Format: log.FormatJSON, Writer: &buf1}); err != nil { + t.Fatal(err) + } + stream1 := &fakeStream{ + incoming: []*epb.ProcessingRequest{ + { + Request: &epb.ProcessingRequest_RequestHeaders{ + RequestHeaders: &epb.HttpHeaders{ + Headers: makeHeaderMap(map[string]string{ + ":method": "GET", + ":path": "/hello", + ":scheme": "http", + ":authority": "localhost", + }), + }, + }, + }, + }, + } + if err := srv.Process(stream1); err != nil { + t.Fatal(err) + } + + // Stream 2 + var buf2 bytes.Buffer + if err := log.Configure(log.Options{Level: "info", Format: log.FormatJSON, Writer: &buf2}); err != nil { + t.Fatal(err) + } + stream2 := &fakeStream{ + incoming: []*epb.ProcessingRequest{ + { + Request: &epb.ProcessingRequest_RequestHeaders{ + RequestHeaders: &epb.HttpHeaders{ + Headers: makeHeaderMap(map[string]string{ + ":method": "GET", + ":path": "/hello", + ":scheme": "http", + ":authority": "localhost", + }), + }, + }, + }, + }, + } + if err := srv.Process(stream2); err != nil { + t.Fatal(err) + } + + // Restore logger + _ = log.Configure(log.Options{}) + + // Parse stream 1 ID + id1 := extractStreamID(t, buf1.String()) + // Parse stream 2 ID + id2 := extractStreamID(t, buf2.String()) + + if id1 == "" || id2 == "" { + t.Fatalf("expected non-empty stream IDs, got stream1=%q, stream2=%q", id1, id2) + } + if id1 == id2 { + t.Errorf("expected different stream IDs for different streams, but both were %s", id1) + } +} + +func extractStreamID(t *testing.T, logOutput string) string { + t.Helper() + lines := strings.Split(logOutput, "\n") + for _, line := range lines { + line = strings.TrimSpace(line) + if line == "" { + continue + } + var record map[string]any + if err := json.Unmarshal([]byte(line), &record); err != nil { + continue + } + if record["msg"] == "extProc access" { + if sid, ok := record["stream_id"].(string); ok { + return sid + } + } + } + return "" +} + +func TestRequestIDPromotedFromHeader(t *testing.T) { + yamlStr := ` +defaults: + extProc: + maxBodyBytes: 1024 + onBodyOverflow: fail +groups: + - name: test-group + parameters: + engine: extProc + mode: applyAll + phase: requestHeaders + match: "true" + rules: + - name: dummy + match: "true" +` + cfg := mustLoadConfig(t, yamlStr) + srv := New(cfg) + + // Case 1: with header "x-request-id" + var buf1 bytes.Buffer + if err := log.Configure(log.Options{Level: "info", Format: log.FormatJSON, Writer: &buf1}); err != nil { + t.Fatal(err) + } + stream1 := &fakeStream{ + incoming: []*epb.ProcessingRequest{ + { + Request: &epb.ProcessingRequest_RequestHeaders{ + RequestHeaders: &epb.HttpHeaders{ + Headers: makeHeaderMap(map[string]string{ + ":method": "GET", + ":path": "/hello", + ":scheme": "http", + ":authority": "localhost", + "x-request-id": "req-abc-123", + }), + }, + }, + }, + }, + } + if err := srv.Process(stream1); err != nil { + t.Fatal(err) + } + + // Case 2: without header "x-request-id" + var buf2 bytes.Buffer + if err := log.Configure(log.Options{Level: "info", Format: log.FormatJSON, Writer: &buf2}); err != nil { + t.Fatal(err) + } + stream2 := &fakeStream{ + incoming: []*epb.ProcessingRequest{ + { + Request: &epb.ProcessingRequest_RequestHeaders{ + RequestHeaders: &epb.HttpHeaders{ + Headers: makeHeaderMap(map[string]string{ + ":method": "GET", + ":path": "/hello", + ":scheme": "http", + ":authority": "localhost", + }), + }, + }, + }, + }, + } + if err := srv.Process(stream2); err != nil { + t.Fatal(err) + } + + _ = log.Configure(log.Options{}) + + // Check buf1 has "request_id":"req-abc-123" + foundReqID := false + for _, line := range strings.Split(buf1.String(), "\n") { + line = strings.TrimSpace(line) + if line == "" { + continue + } + var record map[string]any + if err := json.Unmarshal([]byte(line), &record); err != nil { + continue + } + if record["msg"] == "extProc access" { + reqID, exists := record["request_id"] + if exists { + foundReqID = true + if reqID != "req-abc-123" { + t.Errorf("expected request_id req-abc-123, got %v", reqID) + } + } + } + } + if !foundReqID { + t.Errorf("expected request_id to be present in logs, but it wasn't") + } + + // Check buf2 has NO "request_id" key + for _, line := range strings.Split(buf2.String(), "\n") { + line = strings.TrimSpace(line) + if line == "" { + continue + } + var record map[string]any + if err := json.Unmarshal([]byte(line), &record); err != nil { + continue + } + if record["msg"] == "extProc access" { + if _, exists := record["request_id"]; exists { + t.Errorf("expected no request_id field when header is absent, but got %v", record["request_id"]) + } + } + } +} diff --git a/internal/grpcserver/server.go b/internal/grpcserver/server.go index 1105a95..bedb330 100644 --- a/internal/grpcserver/server.go +++ b/internal/grpcserver/server.go @@ -5,6 +5,8 @@ package grpcserver import ( + "crypto/rand" + "encoding/hex" "errors" "fmt" "io" @@ -14,12 +16,14 @@ import ( "strconv" "strings" "sync/atomic" + "time" corev3 "github.com/envoyproxy/go-control-plane/envoy/config/core/v3" epb "github.com/envoyproxy/go-control-plane/envoy/service/ext_proc/v3" typev3 "github.com/envoyproxy/go-control-plane/envoy/type/v3" "google.golang.org/grpc" + "request-validator/internal/accesslog" "request-validator/internal/log" "request-validator/internal/policy" ) @@ -75,6 +79,7 @@ func (s *Server) Stop() { // Process implements the bidirectional processing stream. func (s *Server) Process(stream epb.ExternalProcessor_ProcessServer) error { + streamID := newStreamID() ctx := stream.Context() var req *policy.Request var resp *policy.Response @@ -100,8 +105,20 @@ func (s *Server) Process(stream epb.ExternalProcessor_ProcessServer) error { switch r := reqMsg.Request.(type) { case *epb.ProcessingRequest_RequestHeaders: req = parseRequestHeaders(r.RequestHeaders) + accessArgs := []any{ + "engine", "extProc", + "stream_id", streamID, + "phase", "requestHeaders", + accesslog.RequestAttrs(req, p.Logging), + } + if req != nil { + if reqID := req.Headers.Get("x-request-id"); reqID != "" { + accessArgs = append(accessArgs, "request_id", reqID) + } + } + log.Logger().Info("extProc access", accessArgs...) res := p.EvaluateProc(ctx, "requestHeaders", req, nil) - respMsg := s.handleProcResult("requestHeaders", res, p) + respMsg := s.handleProcResult(streamID, "requestHeaders", res, p) if err := stream.Send(respMsg); err != nil { return err } @@ -117,6 +134,7 @@ func (s *Server) Process(stream epb.ExternalProcessor_ProcessServer) error { if int64(len(req.Body)) > limit { onBodyOverflow := p.Defaults.ExtProc.OnBodyOverflow log.Warnw("ext_proc body overflow", + "stream_id", streamID, "phase", "requestBody", "limit", limit, "body_size", len(req.Body), @@ -169,8 +187,20 @@ func (s *Server) Process(stream epb.ExternalProcessor_ProcessServer) error { continue } + accessArgs := []any{ + "engine", "extProc", + "stream_id", streamID, + "phase", "requestBody", + accesslog.RequestAttrs(req, p.Logging), + } + if req != nil { + if reqID := req.Headers.Get("x-request-id"); reqID != "" { + accessArgs = append(accessArgs, "request_id", reqID) + } + } + log.Logger().Info("extProc access", accessArgs...) res := p.EvaluateProc(ctx, "requestBody", req, nil) - respMsg := s.handleProcResult("requestBody", res, p) + respMsg := s.handleProcResult(streamID, "requestBody", res, p) if err := stream.Send(respMsg); err != nil { return err } @@ -180,8 +210,21 @@ func (s *Server) Process(stream epb.ExternalProcessor_ProcessServer) error { req = &policy.Request{Headers: make(http.Header)} } resp = parseResponseHeaders(r.ResponseHeaders) + accessArgs := []any{ + "engine", "extProc", + "stream_id", streamID, + "phase", "responseHeaders", + accesslog.RequestAttrs(req, p.Logging), + accesslog.ResponseAttrs(resp, p.Logging), + } + if req != nil { + if reqID := req.Headers.Get("x-request-id"); reqID != "" { + accessArgs = append(accessArgs, "request_id", reqID) + } + } + log.Logger().Info("extProc access", accessArgs...) res := p.EvaluateProc(ctx, "responseHeaders", req, resp) - respMsg := s.handleProcResult("responseHeaders", res, p) + respMsg := s.handleProcResult(streamID, "responseHeaders", res, p) if err := stream.Send(respMsg); err != nil { return err } @@ -200,6 +243,7 @@ func (s *Server) Process(stream epb.ExternalProcessor_ProcessServer) error { if int64(len(resp.Body)) > limit { onBodyOverflow := p.Defaults.ExtProc.OnBodyOverflow log.Warnw("ext_proc body overflow", + "stream_id", streamID, "phase", "responseBody", "limit", limit, "body_size", len(resp.Body), @@ -252,8 +296,14 @@ func (s *Server) Process(stream epb.ExternalProcessor_ProcessServer) error { continue } + log.Logger().Info("extProc access", + "engine", "extProc", + "stream_id", streamID, + "phase", "responseBody", + accesslog.ResponseAttrs(resp, p.Logging), + ) res := p.EvaluateProc(ctx, "responseBody", req, resp) - respMsg := s.handleProcResult("responseBody", res, p) + respMsg := s.handleProcResult(streamID, "responseBody", res, p) if err := stream.Send(respMsg); err != nil { return err } @@ -268,7 +318,7 @@ func (s *Server) Process(stream epb.ExternalProcessor_ProcessServer) error { } // handleProcResult filters shadow mutations and logs/builds the processing response. -func (s *Server) handleProcResult(phase string, res policy.ProcResult, p *policy.Config) *epb.ProcessingResponse { +func (s *Server) handleProcResult(streamID string, phase string, res policy.ProcResult, p *policy.Config) *epb.ProcessingResponse { dryGlobal := p.Defaults.DryRun // 1. CORTOCIRCUITO: check for first applied directResponse @@ -277,6 +327,7 @@ func (s *Server) handleProcResult(phase string, res policy.ProcResult, p *policy if m.Op == "directResponse" && !effectiveDry { log.Infow("extProc phase evaluated", "engine", "extProc", + "stream_id", streamID, "phase", phase, "direct_response", fmt.Sprintf("%s:%d", m.Rule, m.RespStatus), "dry_run", false, @@ -341,6 +392,7 @@ func (s *Server) handleProcResult(phase string, res policy.ProcResult, p *policy } logProc("extProc phase evaluated", "engine", "extProc", + "stream_id", streamID, "phase", phase, "applied", appliedLog, "shadow", shadowLog, @@ -629,3 +681,14 @@ func extractClientIP(h http.Header) string { } return "" } + +// newStreamID generates a 16-character hex-encoded correlation ID using 8 random bytes. +// Envoy opens one ext_proc gRPC stream per HTTP request, so a per-stream ID correlates +// all phase logs of one request. If crypto/rand fails, it falls back to a hex-encoded timestamp. +func newStreamID() string { + buf := make([]byte, 8) + if _, err := rand.Read(buf); err != nil { + return fmt.Sprintf("%016x", time.Now().UnixNano()) + } + return hex.EncodeToString(buf) +} diff --git a/internal/httpserver/server.go b/internal/httpserver/server.go index 783b183..9da9f1f 100644 --- a/internal/httpserver/server.go +++ b/internal/httpserver/server.go @@ -37,6 +37,7 @@ import ( "sync/atomic" "time" + "request-validator/internal/accesslog" "request-validator/internal/log" "request-validator/internal/policy" ) @@ -177,7 +178,7 @@ func (s *Server) handle(w http.ResponseWriter, r *http.Request) { "reason", "request body too large", "dry_run", dry, "duration_ms", float64(time.Since(start).Microseconds()) / 1000.0, - accessLogAttrs(req, p.Logging), + accesslog.RequestAttrs(req, p.Logging), } logger.Warn("request decided", rec...) return @@ -235,7 +236,7 @@ func (s *Server) handle(w http.ResponseWriter, r *http.Request) { "reason", d.Reason, "dry_run", effectiveDry, "duration_ms", float64(time.Since(start).Microseconds()) / 1000.0, - accessLogAttrs(req, p.Logging), + accesslog.RequestAttrs(req, p.Logging), } if d.Allowed { logger.Info("request decided", rec...) From c4a9e93f78d0fdc114fc89d3beabe8b857fb6702 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Alby=20Hern=C3=A1ndez?= Date: Tue, 14 Jul 2026 14:51:33 +0100 Subject: [PATCH 2/2] test: cover overflow correlation, nil-policy and edge paths - Overflow WARN records carry the stream_id of their stream (canary: fails if the field is dropped). - Overflow under global dry-run returns CONTINUE instead of an immediate 500, for both request and response bodies, and still logs the overflow with dry_run=true. - A stream processed before any policy is loaded answers CONTINUE and emits no access record. - Unknown phase messages (trailers) fall through to a CONTINUE. - extractClientIP table: XFF single/list, X-Real-Ip fallback, empty. - ResponseAttrs tolerates a nil response; redacted queries leave bare pairs without '=' untouched. --- internal/accesslog/accesslog_test.go | 74 +++++ internal/grpcserver/access_test.go | 417 +++++++++++++++++++++++++++ 2 files changed, 491 insertions(+) diff --git a/internal/accesslog/accesslog_test.go b/internal/accesslog/accesslog_test.go index 967f359..2329e09 100644 --- a/internal/accesslog/accesslog_test.go +++ b/internal/accesslog/accesslog_test.go @@ -295,3 +295,77 @@ func TestResponseAttrsBodyOnlyWhenLogBody(t *testing.T) { } } } + +func TestResponseAttrsNilResponse(t *testing.T) { + var buf bytes.Buffer + if err := log.Configure(log.Options{Level: "info", Format: log.FormatJSON, Writer: &buf}); err != nil { + t.Fatal(err) + } + t.Cleanup(func() { _ = log.Configure(log.Options{}) }) + + attr := ResponseAttrs(nil, policy.Logging{}) + + log.Logger().Info("test_nil_resp", attr) + + out := buf.String() + var rec map[string]any + if err := json.Unmarshal(bytes.TrimSpace(buf.Bytes()), &rec); err != nil { + t.Fatalf("invalid JSON output: %v -- %s", err, out) + } + + respRec, ok := rec["response"].(map[string]any) + if !ok { + t.Fatalf("missing response group in log: %s", out) + } + + if int(respRec["status"].(float64)) != 0 { + t.Fatalf("status mismatch: %v", respRec["status"]) + } + + body, ok := respRec["body"].(map[string]any) + if !ok { + t.Fatalf("missing body in response log: %s", out) + } + + if int(body["size"].(float64)) != 0 { + t.Fatalf("body size mismatch: %v", body["size"]) + } +} + +func TestRedactedQueryPairWithoutEquals(t *testing.T) { + var buf bytes.Buffer + if err := log.Configure(log.Options{Level: "info", Format: log.FormatJSON, Writer: &buf}); err != nil { + t.Fatal(err) + } + t.Cleanup(func() { _ = log.Configure(log.Options{}) }) + + req := &policy.Request{ + RawQuery: "code=secret&standalone&id_token=x", + } + lg := policy.Logging{ + RedactQueryParams: []string{"code", "id_token"}, + } + + log.Logger().Info("test_query", RequestAttrs(req, lg)) + + out := buf.String() + var rec map[string]any + if err := json.Unmarshal(bytes.TrimSpace(buf.Bytes()), &rec); err != nil { + t.Fatalf("invalid JSON output: %v -- %s", err, out) + } + + reqRec, ok := rec["request"].(map[string]any) + if !ok { + t.Fatalf("missing request group in log: %s", out) + } + + queryVal, ok := reqRec["query"].(string) + if !ok { + t.Fatalf("query field missing in request log: %s", out) + } + + expected := "code=***&standalone&id_token=***" + if queryVal != expected { + t.Fatalf("expected query to be %q, got %q", expected, queryVal) + } +} diff --git a/internal/grpcserver/access_test.go b/internal/grpcserver/access_test.go index 802389e..4dfdee4 100644 --- a/internal/grpcserver/access_test.go +++ b/internal/grpcserver/access_test.go @@ -6,6 +6,7 @@ package grpcserver import ( "bytes" "encoding/json" + "net/http" "strings" "testing" @@ -540,3 +541,419 @@ groups: } } } + +func TestOverflowWarnCarriesStreamID(t *testing.T) { + var buf bytes.Buffer + if err := log.Configure(log.Options{Level: "info", Format: log.FormatJSON, Writer: &buf}); err != nil { + t.Fatal(err) + } + t.Cleanup(func() { _ = log.Configure(log.Options{}) }) + + yamlStr := ` +defaults: + extProc: + maxBodyBytes: 10 + onBodyOverflow: fail +groups: + - name: test-group + parameters: + engine: extProc + mode: applyAll + phase: requestHeaders + match: "true" + rules: + - name: dummy + match: "true" +` + cfg := mustLoadConfig(t, yamlStr) + srv := New(cfg) + + stream := &fakeStream{ + incoming: []*epb.ProcessingRequest{ + { + Request: &epb.ProcessingRequest_RequestHeaders{ + RequestHeaders: &epb.HttpHeaders{ + Headers: makeHeaderMap(map[string]string{ + ":method": "POST", + ":path": "/upload", + ":scheme": "http", + ":authority": "localhost", + }), + }, + }, + }, + { + Request: &epb.ProcessingRequest_RequestBody{ + RequestBody: &epb.HttpBody{ + Body: []byte("this is more than ten bytes"), + }, + }, + }, + }, + } + + err := srv.Process(stream) + if err != nil { + t.Fatalf("unexpected error: %v", err) + } + + var records []map[string]any + lines := strings.Split(buf.String(), "\n") + for _, line := range lines { + line = strings.TrimSpace(line) + if line == "" { + continue + } + var record map[string]any + if err := json.Unmarshal([]byte(line), &record); err != nil { + continue + } + records = append(records, record) + } + + var accessStreamID string + var overflowStreamID string + + for _, rec := range records { + msg, _ := rec["msg"].(string) + if msg == "extProc access" { + if ph, ok := rec["phase"].(string); ok && ph == "requestHeaders" { + accessStreamID, _ = rec["stream_id"].(string) + } + } else if msg == "ext_proc body overflow" { + overflowStreamID, _ = rec["stream_id"].(string) + } + } + + if accessStreamID == "" { + t.Fatalf("expected to find 'extProc access' log record for requestHeaders, but didn't") + } + if overflowStreamID == "" { + t.Fatalf("expected to find 'ext_proc body overflow' WARN log record, but didn't") + } + if accessStreamID != overflowStreamID { + t.Errorf("stream_id mismatch: access log had %q, overflow log had %q", accessStreamID, overflowStreamID) + } +} + +func TestOverflowFailDryRunContinues(t *testing.T) { + yamlStr := ` +defaults: + dryRun: true + extProc: + maxBodyBytes: 10 + onBodyOverflow: fail +groups: + - name: test-group + parameters: + engine: extProc + mode: applyAll + phase: requestHeaders + match: "true" + rules: + - name: dummy + match: "true" +` + cfg := mustLoadConfig(t, yamlStr) + + t.Run("RequestBody", func(t *testing.T) { + var buf bytes.Buffer + if err := log.Configure(log.Options{Level: "info", Format: log.FormatJSON, Writer: &buf}); err != nil { + t.Fatal(err) + } + t.Cleanup(func() { _ = log.Configure(log.Options{}) }) + + srv := New(cfg) + stream := &fakeStream{ + incoming: []*epb.ProcessingRequest{ + { + Request: &epb.ProcessingRequest_RequestHeaders{ + RequestHeaders: &epb.HttpHeaders{ + Headers: makeHeaderMap(map[string]string{ + ":method": "POST", + ":path": "/upload", + ":scheme": "http", + ":authority": "localhost", + }), + }, + }, + }, + { + Request: &epb.ProcessingRequest_RequestBody{ + RequestBody: &epb.HttpBody{ + Body: []byte("this is more than ten bytes"), + }, + }, + }, + }, + } + + err := srv.Process(stream) + if err != nil { + t.Fatalf("unexpected error: %v", err) + } + + if len(stream.outgoing) != 2 { + t.Fatalf("expected exactly 2 outgoing responses, got %d", len(stream.outgoing)) + } + + resp := stream.outgoing[1] + rb, ok := resp.Response.(*epb.ProcessingResponse_RequestBody) + if !ok { + t.Fatalf("expected ProcessingResponse_RequestBody, got %T", resp.Response) + } + + if rb.RequestBody.Response.Status != epb.CommonResponse_CONTINUE { + t.Errorf("expected Status to be CONTINUE, got %v", rb.RequestBody.Response.Status) + } + + out := buf.String() + foundWarn := false + for _, line := range strings.Split(out, "\n") { + line = strings.TrimSpace(line) + if line == "" { + continue + } + var record map[string]any + if err := json.Unmarshal([]byte(line), &record); err != nil { + continue + } + if record["msg"] == "ext_proc body overflow" { + foundWarn = true + if record["phase"] != "requestBody" { + t.Errorf("expected phase to be 'requestBody', got %v", record["phase"]) + } + if dryRun, ok := record["dry_run"].(bool); !ok || !dryRun { + t.Errorf("expected dry_run to be true, got %v", record["dry_run"]) + } + } + } + if !foundWarn { + t.Fatalf("expected to find overflow log message, but didn't") + } + }) + + t.Run("ResponseBody", func(t *testing.T) { + var buf bytes.Buffer + if err := log.Configure(log.Options{Level: "info", Format: log.FormatJSON, Writer: &buf}); err != nil { + t.Fatal(err) + } + t.Cleanup(func() { _ = log.Configure(log.Options{}) }) + + srv := New(cfg) + stream := &fakeStream{ + incoming: []*epb.ProcessingRequest{ + { + Request: &epb.ProcessingRequest_ResponseHeaders{ + ResponseHeaders: &epb.HttpHeaders{ + Headers: makeHeaderMap(map[string]string{ + ":status": "200", + "content-type": "application/json", + }), + }, + }, + }, + { + Request: &epb.ProcessingRequest_ResponseBody{ + ResponseBody: &epb.HttpBody{ + Body: []byte("this is more than ten bytes"), + }, + }, + }, + }, + } + + err := srv.Process(stream) + if err != nil { + t.Fatalf("unexpected error: %v", err) + } + + if len(stream.outgoing) != 2 { + t.Fatalf("expected exactly 2 outgoing responses, got %d", len(stream.outgoing)) + } + + resp := stream.outgoing[1] + rb, ok := resp.Response.(*epb.ProcessingResponse_ResponseBody) + if !ok { + t.Fatalf("expected ProcessingResponse_ResponseBody, got %T", resp.Response) + } + + if rb.ResponseBody.Response.Status != epb.CommonResponse_CONTINUE { + t.Errorf("expected Status to be CONTINUE, got %v", rb.ResponseBody.Response.Status) + } + + out := buf.String() + foundWarn := false + for _, line := range strings.Split(out, "\n") { + line = strings.TrimSpace(line) + if line == "" { + continue + } + var record map[string]any + if err := json.Unmarshal([]byte(line), &record); err != nil { + continue + } + if record["msg"] == "ext_proc body overflow" { + foundWarn = true + if record["phase"] != "responseBody" { + t.Errorf("expected phase to be 'responseBody', got %v", record["phase"]) + } + if dryRun, ok := record["dry_run"].(bool); !ok || !dryRun { + t.Errorf("expected dry_run to be true, got %v", record["dry_run"]) + } + } + } + if !foundWarn { + t.Fatalf("expected to find overflow log message, but didn't") + } + }) +} + +func TestNilPolicyContinuesWithoutAccessLog(t *testing.T) { + var buf bytes.Buffer + if err := log.Configure(log.Options{Level: "info", Format: log.FormatJSON, Writer: &buf}); err != nil { + t.Fatal(err) + } + t.Cleanup(func() { _ = log.Configure(log.Options{}) }) + + srv := New(nil) + + stream := &fakeStream{ + incoming: []*epb.ProcessingRequest{ + { + Request: &epb.ProcessingRequest_RequestHeaders{ + RequestHeaders: &epb.HttpHeaders{ + Headers: makeHeaderMap(map[string]string{ + ":method": "GET", + ":path": "/hello", + ":scheme": "http", + ":authority": "localhost", + }), + }, + }, + }, + }, + } + + err := srv.Process(stream) + if err != nil { + t.Fatalf("unexpected error: %v", err) + } + + if len(stream.outgoing) != 1 { + t.Fatalf("expected 1 response, got %d", len(stream.outgoing)) + } + + resp := stream.outgoing[0] + rh, ok := resp.Response.(*epb.ProcessingResponse_RequestHeaders) + if !ok { + t.Fatalf("expected ProcessingResponse_RequestHeaders, got %T", resp.Response) + } + + if rh.RequestHeaders.Response.Status != epb.CommonResponse_CONTINUE { + t.Errorf("expected CONTINUE status, got %v", rh.RequestHeaders.Response.Status) + } + + out := buf.String() + if strings.Contains(out, "extProc access") { + t.Errorf("expected NO 'extProc access' record logged, but got: %s", out) + } +} + +func TestUnknownMessageTypeContinues(t *testing.T) { + yamlStr := ` +defaults: + extProc: + maxBodyBytes: 1024 + onBodyOverflow: fail +groups: + - name: test-group + parameters: + engine: extProc + mode: applyAll + phase: requestHeaders + match: "true" + rules: + - name: dummy + match: "true" +` + cfg := mustLoadConfig(t, yamlStr) + srv := New(cfg) + + stream := &fakeStream{ + incoming: []*epb.ProcessingRequest{ + { + Request: &epb.ProcessingRequest_RequestTrailers{ + RequestTrailers: &epb.HttpTrailers{}, + }, + }, + }, + } + + err := srv.Process(stream) + if err != nil { + t.Fatalf("unexpected error: %v", err) + } + + if len(stream.outgoing) != 1 { + t.Fatalf("expected 1 response, got %d", len(stream.outgoing)) + } + + resp := stream.outgoing[0] + rh, ok := resp.Response.(*epb.ProcessingResponse_RequestHeaders) + if !ok { + t.Fatalf("expected ProcessingResponse_RequestHeaders for unknown message, got %T", resp.Response) + } + + if rh.RequestHeaders.Response.Status != epb.CommonResponse_CONTINUE { + t.Errorf("expected CONTINUE status, got %v", rh.RequestHeaders.Response.Status) + } +} + +func TestExtractClientIP(t *testing.T) { + tests := []struct { + name string + headers map[string]string + expected string + }{ + { + name: "XFF single value", + headers: map[string]string{ + "X-Forwarded-For": " 1.2.3.4 ", + }, + expected: "1.2.3.4", + }, + { + name: "XFF list", + headers: map[string]string{ + "X-Forwarded-For": "1.2.3.4, 5.6.7.8", + }, + expected: "1.2.3.4", + }, + { + name: "no XFF but X-Real-Ip", + headers: map[string]string{ + "X-Real-Ip": " 9.10.11.12 ", + }, + expected: "9.10.11.12", + }, + { + name: "neither", + headers: map[string]string{}, + expected: "", + }, + } + + for _, tc := range tests { + t.Run(tc.name, func(t *testing.T) { + h := make(http.Header) + for k, v := range tc.headers { + h.Set(k, v) + } + got := extractClientIP(h) + if got != tc.expected { + t.Errorf("extractClientIP() = %q, want %q", got, tc.expected) + } + }) + } +}