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/accesslog/accesslog_test.go b/internal/accesslog/accesslog_test.go new file mode 100644 index 0000000..2329e09 --- /dev/null +++ b/internal/accesslog/accesslog_test.go @@ -0,0 +1,371 @@ +// SPDX-FileCopyrightText: 2026 Alby Hernández +// SPDX-License-Identifier: Apache-2.0 + +package accesslog + +import ( + "bytes" + "encoding/json" + "log/slog" + "net/http" + "strings" + "testing" + + "request-validator/internal/log" + "request-validator/internal/policy" +) + +func TestMask(t *testing.T) { + cases := []struct { + in string + reveal int + want string + comment string + }{ + {"", 6, "", "empty"}, + {"abc", 6, "***", "shorter than reveal"}, + {"abcdef", 6, "******", "exactly reveal -> fully masked (len<2*n)"}, + {"abcdefghi", 6, "*********", "len 9 < 12, fully masked"}, + {"abcdefghijkl", 6, "abcdef******", "len 12 == 2*n, prefix shown"}, + {"abcdefghijklmno", 6, "abcdef*********", "long enough, prefix shown"}, + {"abcdefghijklmno", 0, "***************", "reveal=0 -> all masked"}, + {"abcdefghijklmno", -3, "***************", "reveal<0 -> all masked"}, + } + for _, c := range cases { + if got := mask(c.in, c.reveal); got != c.want { + t.Errorf("mask(%q,%d) = %q want %q (%s)", c.in, c.reveal, got, c.want, c.comment) + } + } +} + +func TestAccessLogAttrsExcludeAndRedact(t *testing.T) { + // Capture log output into a buffer. + 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("Authorization", "Bearer eyJxxxxxxxxxxxxxxxx") + hdrs.Set("X-Api-Key", "abc") + hdrs.Set("Cookie", "session=verysecret") + hdrs.Set("User-Agent", "Antigravity/1.15") + + req := &policy.Request{ + Method: "POST", + Host: "auth.example-1.com", + Path: "/realms/mcp/clients-registrations", + RawQuery: "code=abc&debug=1", + RemoteIP: "203.0.113.5", + Headers: hdrs, + Body: []byte(`{"redirect_uris":["https://localhost:51234/cb"]}`), + } + lg := policy.Logging{ + ExcludeHeaders: []string{"cookie"}, + RedactHeaders: []string{"authorization", "x-api-key"}, + RedactReveal: 6, + RedactQueryParams: []string{"code"}, + } + + log.Logger().Info("test", "decision", "allow", RequestAttrs(req, lg)) + + out := buf.String() + if !strings.Contains(out, `"decision":"allow"`) { + t.Fatalf("missing decision: %s", out) + } + // Cookie must be gone entirely. + if strings.Contains(strings.ToLower(out), "cookie") { + t.Fatalf("cookie should have been excluded: %s", out) + } + // Authorization is long enough so we reveal the leading 6 chars then mask. + if !strings.Contains(out, `"authorization":"Bearer*`) { + t.Fatalf("authorization not redacted as expected: %s", out) + } + // X-Api-Key is shorter than 2*reveal -> fully masked. + if !strings.Contains(out, `"x-api-key":"***"`) { + t.Fatalf("x-api-key short value should be fully masked: %s", out) + } + // Query: code redacted, debug untouched. + if !strings.Contains(out, `"query":"code=***&debug=1"`) { + t.Fatalf("query not redacted as expected: %s", out) + } + // Body size is logged, body content is not (logBody=false). + if !strings.Contains(out, `"size":48`) { + t.Fatalf("body size missing: %s", out) + } + if strings.Contains(out, "redirect_uris") { + t.Fatalf("body content leaked despite LogBody=false: %s", out) + } +} + +func TestAccessLogAttrsLogBody(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") + req := &policy.Request{ + Method: "POST", + Headers: hdrs, + Body: []byte(`{"a":1}`), + } + lg := policy.Logging{LogBody: true} + log.Logger().Info("test", RequestAttrs(req, lg)) + + out := buf.String() + // Parse the JSON to assert structurally (less fragile than substring matching). + 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, _ := rec["request"].(map[string]any) + body, _ := reqRec["body"].(map[string]any) + if body["raw"] != `{"a":1}` { + t.Fatalf("body.raw missing or wrong: %v", body) + } + if int(body["size"].(float64)) != 7 { + t.Fatalf("body.size wrong: %v", body) + } +} + +// Ensure header keys are always lowercased in the output regardless of how +// they were set on the http.Header (which canonicalises to Title-Case). +func TestAccessLogHeaderKeysLowercased(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("X-Custom-Header", "value") + req := &policy.Request{Headers: hdrs} + 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()) + } +} + +// Sanity check: console format produces parseable output (no panics). +func TestConsoleFormatProducesLine(t *testing.T) { + var buf bytes.Buffer + if err := log.Configure(log.Options{Level: "info", Format: log.FormatConsole, Writer: &buf}); err != nil { + t.Fatal(err) + } + t.Cleanup(func() { _ = log.Configure(log.Options{}) }) + + log.Logger().Info("hello", slog.String("k", "v")) + + out := buf.String() + if !strings.Contains(out, "INFO ") || !strings.Contains(out, "hello") || !strings.Contains(out, "k=v") { + 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) + } + } +} + +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 new file mode 100644 index 0000000..4dfdee4 --- /dev/null +++ b/internal/grpcserver/access_test.go @@ -0,0 +1,959 @@ +// SPDX-FileCopyrightText: 2026 Alby Hernández +// SPDX-License-Identifier: Apache-2.0 + +package grpcserver + +import ( + "bytes" + "encoding/json" + "net/http" + "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"]) + } + } + } +} + +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) + } + }) + } +} 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/access_test.go b/internal/httpserver/access_test.go deleted file mode 100644 index a8760cc..0000000 --- a/internal/httpserver/access_test.go +++ /dev/null @@ -1,169 +0,0 @@ -// SPDX-FileCopyrightText: 2026 Alby Hernández -// SPDX-License-Identifier: Apache-2.0 - -package httpserver - -import ( - "bytes" - "encoding/json" - "log/slog" - "net/http" - "strings" - "testing" - - "request-validator/internal/log" - "request-validator/internal/policy" -) - -func TestMask(t *testing.T) { - cases := []struct { - in string - reveal int - want string - comment string - }{ - {"", 6, "", "empty"}, - {"abc", 6, "***", "shorter than reveal"}, - {"abcdef", 6, "******", "exactly reveal -> fully masked (len<2*n)"}, - {"abcdefghi", 6, "*********", "len 9 < 12, fully masked"}, - {"abcdefghijkl", 6, "abcdef******", "len 12 == 2*n, prefix shown"}, - {"abcdefghijklmno", 6, "abcdef*********", "long enough, prefix shown"}, - {"abcdefghijklmno", 0, "***************", "reveal=0 -> all masked"}, - {"abcdefghijklmno", -3, "***************", "reveal<0 -> all masked"}, - } - for _, c := range cases { - if got := mask(c.in, c.reveal); got != c.want { - t.Errorf("mask(%q,%d) = %q want %q (%s)", c.in, c.reveal, got, c.want, c.comment) - } - } -} - -func TestAccessLogAttrsExcludeAndRedact(t *testing.T) { - // Capture log output into a buffer. - 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("Authorization", "Bearer eyJxxxxxxxxxxxxxxxx") - hdrs.Set("X-Api-Key", "abc") - hdrs.Set("Cookie", "session=verysecret") - hdrs.Set("User-Agent", "Antigravity/1.15") - - req := &policy.Request{ - Method: "POST", - Host: "auth.example-1.com", - Path: "/realms/mcp/clients-registrations", - RawQuery: "code=abc&debug=1", - RemoteIP: "203.0.113.5", - Headers: hdrs, - Body: []byte(`{"redirect_uris":["https://localhost:51234/cb"]}`), - } - lg := policy.Logging{ - ExcludeHeaders: []string{"cookie"}, - RedactHeaders: []string{"authorization", "x-api-key"}, - RedactReveal: 6, - RedactQueryParams: []string{"code"}, - } - - log.Logger().Info("test", "decision", "allow", accessLogAttrs(req, lg)) - - out := buf.String() - if !strings.Contains(out, `"decision":"allow"`) { - t.Fatalf("missing decision: %s", out) - } - // Cookie must be gone entirely. - if strings.Contains(strings.ToLower(out), "cookie") { - t.Fatalf("cookie should have been excluded: %s", out) - } - // Authorization is long enough so we reveal the leading 6 chars then mask. - if !strings.Contains(out, `"authorization":"Bearer*`) { - t.Fatalf("authorization not redacted as expected: %s", out) - } - // X-Api-Key is shorter than 2*reveal -> fully masked. - if !strings.Contains(out, `"x-api-key":"***"`) { - t.Fatalf("x-api-key short value should be fully masked: %s", out) - } - // Query: code redacted, debug untouched. - if !strings.Contains(out, `"query":"code=***&debug=1"`) { - t.Fatalf("query not redacted as expected: %s", out) - } - // Body size is logged, body content is not (logBody=false). - if !strings.Contains(out, `"size":48`) { - t.Fatalf("body size missing: %s", out) - } - if strings.Contains(out, "redirect_uris") { - t.Fatalf("body content leaked despite LogBody=false: %s", out) - } -} - -func TestAccessLogAttrsLogBody(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") - req := &policy.Request{ - Method: "POST", - Headers: hdrs, - Body: []byte(`{"a":1}`), - } - lg := policy.Logging{LogBody: true} - log.Logger().Info("test", accessLogAttrs(req, lg)) - - out := buf.String() - // Parse the JSON to assert structurally (less fragile than substring matching). - 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, _ := rec["request"].(map[string]any) - body, _ := reqRec["body"].(map[string]any) - if body["raw"] != `{"a":1}` { - t.Fatalf("body.raw missing or wrong: %v", body) - } - if int(body["size"].(float64)) != 7 { - t.Fatalf("body.size wrong: %v", body) - } -} - -// Ensure header keys are always lowercased in the output regardless of how -// they were set on the http.Header (which canonicalises to Title-Case). -func TestAccessLogHeaderKeysLowercased(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("X-Custom-Header", "value") - req := &policy.Request{Headers: hdrs} - log.Logger().Info("test", accessLogAttrs(req, policy.Logging{})) - - if !strings.Contains(buf.String(), `"x-custom-header":"value"`) { - t.Fatalf("expected lowercase key in: %s", buf.String()) - } -} - -// Sanity check: console format produces parseable output (no panics). -func TestConsoleFormatProducesLine(t *testing.T) { - var buf bytes.Buffer - if err := log.Configure(log.Options{Level: "info", Format: log.FormatConsole, Writer: &buf}); err != nil { - t.Fatal(err) - } - t.Cleanup(func() { _ = log.Configure(log.Options{}) }) - - log.Logger().Info("hello", slog.String("k", "v")) - - out := buf.String() - if !strings.Contains(out, "INFO ") || !strings.Contains(out, "hello") || !strings.Contains(out, "k=v") { - t.Fatalf("console output unexpected: %q", out) - } -} 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...)