diff --git a/.github/dependabot.yml b/.github/dependabot.yml new file mode 100644 index 0000000..e590722 --- /dev/null +++ b/.github/dependabot.yml @@ -0,0 +1,28 @@ +version: 2 + +updates: + # The workflow pins actions to commit SHAs, which is the point — a mutable + # tag is a stranger's push away from running in CI. Pinning without this + # file just trades a supply-chain hole for a stale one, so let Dependabot + # move the pins; it rewrites the SHA and the trailing version comment + # together. + - package-ecosystem: github-actions + directory: / + schedule: + interval: weekly + day: monday + # Two first-party actions. Separate PRs for them is noise, not signal. + groups: + actions: + patterns: ["*"] + + - package-ecosystem: gomod + directory: / + schedule: + interval: weekly + day: monday + groups: + # Patch and minor bumps land as one PR that CI either passes or does + # not. Majors stay separate, because those are the ones worth reading. + go-minor: + update-types: [minor, patch] diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index 43f5e19..1300f8d 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -11,32 +11,70 @@ on: - cron: "0 6 * * 1" workflow_dispatch: +# go test runs arbitrary code from the PR, so its token gets read and nothing +# else. Also set in repo settings; this copy is the one that shows up in a diff. +permissions: + contents: read + +# Cancel a superseded PR run on force-push. Not main or the weekly sweep: +# each of those runs is the record for one commit. +concurrency: + group: ${{ github.workflow }}-${{ github.ref }} + cancel-in-progress: ${{ github.event_name == 'pull_request' }} + jobs: check: runs-on: ubuntu-latest + # The default is 6 hours. The race suite takes a couple of minutes. + timeout-minutes: 15 + env: + # The runner image ships no golangci-lint, and `make lint` used to fall + # back to `go vet` when it was missing, so this step passed for months + # without linting. Pinned like gitleaks, and to the version developers + # run locally, so CI and a laptop disagree only when the code differs. + GOLANGCI_VERSION: 2.13.2 + GOLANGCI_SHA256: 2277d43b98ec0054280f2ac26b53268bae97682444678a59a657dd565da021d6 steps: - - uses: actions/checkout@v7 + - uses: actions/checkout@3d3c42e5aac5ba805825da76410c181273ba90b1 # v7.0.1 + with: + # Without this the job's token is left behind in .git/config, where + # the pull request's own test code can read it. + persist-credentials: false - - uses: actions/setup-go@v7 + - uses: actions/setup-go@b7ad1dad31e06c5925ef5d2fc7ad053ef454303e # v7.0.0 with: go-version-file: go.mod - # fmt + vet + lint + the full suite under -race. + - name: Install golangci-lint + run: | + set -euo pipefail + tarball="golangci-lint-${GOLANGCI_VERSION}-linux-amd64.tar.gz" + curl -sSfL -o "$tarball" \ + "https://github.com/golangci/golangci-lint/releases/download/v${GOLANGCI_VERSION}/${tarball}" + echo "${GOLANGCI_SHA256} ${tarball}" | sha256sum -c - + tar -xzf "$tarball" --strip-components=1 \ + "golangci-lint-${GOLANGCI_VERSION}-linux-amd64/golangci-lint" + sudo install golangci-lint /usr/local/bin/golangci-lint + + # fmt + vet + lint + the full suite under -race. CI is set by Actions, so + # a missing linter fails here instead of quietly downgrading to go vet. - run: make check secrets: runs-on: ubuntu-latest + timeout-minutes: 10 env: # Pinned and checksummed. An unpinned scanner is a supply-chain hole in # the one job whose purpose is supply-chain hygiene. GITLEAKS_VERSION: 8.30.1 GITLEAKS_SHA256: 551f6fc83ea457d62a0d98237cbad105af8d557003051f41f3e7ca7b3f2470eb steps: - - uses: actions/checkout@v7 + - uses: actions/checkout@3d3c42e5aac5ba805825da76410c181273ba90b1 # v7.0.1 with: # gitleaks walks real history; the default shallow clone gives it one # commit and nothing to walk. fetch-depth: 0 + persist-credentials: false - name: Install gitleaks run: | diff --git a/.golangci.yml b/.golangci.yml new file mode 100644 index 0000000..f8553c8 --- /dev/null +++ b/.golangci.yml @@ -0,0 +1,10 @@ +version: "2" + +issues: + # The defaults stop at 3 of a kind and 50 per linter, which quietly turns a + # full run into a sample. A gate has to report everything it found. + max-issues-per-linter: 0 + max-same-issues: 0 + # One issue per line by default, so a second linter's finding on an already + # flagged line is dropped. That hid an SA1019 deprecation behind a govet hint. + uniq-by-line: false diff --git a/CLAUDE.md b/CLAUDE.md index b6e548e..65ee72d 100644 --- a/CLAUDE.md +++ b/CLAUDE.md @@ -31,6 +31,13 @@ A confidently wrong verdict is worse than no verdict. When evidence is unavailable — an undecodable body, a bodyless response — report that, never guess. +`routes[].strip_params` (`internal/proxy/rewrite.go`) is the single exception to +byte-for-byte passthrough, and the only code that edits a client's body. It +exists because some vendors reject a parameter the client's SDK sends +unconditionally. Keep it narrow — top-level keys, JSON objects, opt-in per +route — and keep every strip in the report. An edit the report does not +disclose turns every verdict on that route into a guess. + ## Layout | Package | Responsibility | diff --git a/Makefile b/Makefile index 2d57761..815079b 100644 --- a/Makefile +++ b/Makefile @@ -64,11 +64,15 @@ bench: vet: go vet ./... -## lint: golangci-lint when installed, otherwise go vet +## lint: golangci-lint when installed, otherwise go vet (an error under CI) .PHONY: lint lint: ifdef GOLANGCI golangci-lint run ./... +else ifdef CI + @echo "golangci-lint is not installed; the go vet fallback would report a" >&2 + @echo "green lint step that linted almost nothing. Install it in the workflow." >&2 + @exit 1 else @echo "golangci-lint not installed; falling back to go vet (make tools to install)" @go vet ./... diff --git a/README.md b/README.md index ec3026b..d5b53cc 100644 --- a/README.md +++ b/README.md @@ -116,6 +116,7 @@ Everything else has a default. The settings worth knowing: | `log.full_trace` | off | Every header, body and SSE frame with arrival times. Keys stay redacted. | | `routes[].retry` | off | Retry 429s and 5xx, honouring `Retry-After`. Off means transparent. | | `routes[].timeouts.*` | no limit | See [Waiting](#waiting). | +| `routes[].strip_params` | none | Delete top-level JSON keys from request bodies. See [When a vendor rejects a parameter](#when-a-vendor-rejects-a-parameter). | | `auth.tokens` | none | See [Exposing it](#exposing-it). `auth.enabled: false` ignores tokens and the environment entirely. | ## Waiting @@ -144,6 +145,43 @@ A deadline the proxy enforces still blames the vendor for going silent, and the report names the setting that fired, so you can tell it apart from a real vendor failure. +## When a vendor rejects a parameter + +Some OpenAI-compatible backends refuse a parameter your client sends +unconditionally, and there is no way to turn it off client-side: + +```json +{"message":"Validation: Unsupported parameter(s): `prompt_cache_key`","type":"Bad Request","code":400} +``` + +`strip_params` deletes top-level keys from the request body before forwarding: + +```yaml +routes: + - name: nebul + upstream: https://api.nebul.example + api_key_env: NEBUL_API_KEY + strip_params: + - prompt_cache_key +``` + +This is the one setting that makes the proxy edit what a client sent, so it +comes with strings attached: + +- Only top-level keys, and only when the body is a JSON object. Anything else + goes through untouched. +- Values that survive are re-encoded from their original bytes, so numbers keep + their precision and prompts keep their ``. Top-level key order and + whitespace do change. +- A body over `max_request_body` is streamed rather than buffered and cannot be + rewritten. The request goes through intact and the report says the strip did + not run. +- `model`, `messages` and `stream` are refused at startup: removing those breaks + the request instead of fixing it. +- Startup names the route as rewriting requests, and every strip that removed + something appears in that request's report as `stripped=…`. A verdict about + which side broke a request has to admit the proxy edited it first. + ## Exposing it With no tokens configured the proxy is open, which is fine on `127.0.0.1`. If diff --git a/cmd/demo/main.go b/cmd/demo/main.go index ec9a2e9..ec999cb 100644 --- a/cmd/demo/main.go +++ b/cmd/demo/main.go @@ -252,14 +252,14 @@ func streamHandler(n int, end ending, finish string) http.HandlerFunc { if err != nil { return } - defer conn.Close() + defer func() { _ = conn.Close() }() writeHead(rw) for i := 0; i < n; i++ { frame := fmt.Sprintf( `data: {"id":"chatcmpl-demo","model":"demo-model","choices":[{"index":0,"delta":{"content":"tok%d"}}]}`+"\n\n", i) if end == endTruncate && i == n-1 { - fmt.Fprintf(rw, "%x\r\n%s", len(frame), frame[:len(frame)/2]) + _, _ = fmt.Fprintf(rw, "%x\r\n%s", len(frame), frame[:len(frame)/2]) _ = rw.Flush() return } @@ -291,7 +291,7 @@ func slowFirstToken(think time.Duration) http.HandlerFunc { if err != nil { return } - defer conn.Close() + defer func() { _ = conn.Close() }() writeHead(rw) time.Sleep(think) @@ -311,7 +311,7 @@ func errorMidStream() http.HandlerFunc { if err != nil { return } - defer conn.Close() + defer func() { _ = conn.Close() }() writeHead(rw) writeChunk(rw, `data: {"choices":[{"index":0,"delta":{"content":"partial answer"}}]}`+"\n\n") writeChunk(rw, `data: {"error":{"message":"the model backend crashed","type":"server_error"}}`+"\n\n") @@ -358,7 +358,7 @@ func writeHead(rw *bufio.ReadWriter) { } func writeChunk(rw *bufio.ReadWriter, data string) { - fmt.Fprintf(rw, "%x\r\n%s\r\n", len(data), data) + _, _ = fmt.Fprintf(rw, "%x\r\n%s\r\n", len(data), data) _ = rw.Flush() } @@ -370,7 +370,7 @@ func hangUpMidStream(addr string) { return } body := `{"model":"demo-model","stream":true,"messages":[{"role":"user","content":"hello"}]}` - fmt.Fprintf(conn, "POST /demo/v1/chat/completions HTTP/1.1\r\nHost: demo\r\n"+ + _, _ = fmt.Fprintf(conn, "POST /demo/v1/chat/completions HTTP/1.1\r\nHost: demo\r\n"+ "Content-Type: application/json\r\nContent-Length: %d\r\n\r\n%s", len(body), body) buf := make([]byte, 4096) diff --git a/cmd/llm-proxy/main.go b/cmd/llm-proxy/main.go index fe928b8..4cf59ca 100644 --- a/cmd/llm-proxy/main.go +++ b/cmd/llm-proxy/main.go @@ -57,7 +57,7 @@ func run() error { fs.BoolVar(&f.check, "check", false, "validate the config, print the route table and exit") fs.BoolVar(&f.showVersion, "version", false, "print the version and exit") fs.Usage = func() { - fmt.Fprintf(fs.Output(), "llm-proxy — an OpenAI-compatible wire-level debugging proxy\n\nusage: llm-proxy [flags]\n\n") + _, _ = fmt.Fprintf(fs.Output(), "llm-proxy — an OpenAI-compatible wire-level debugging proxy\n\nusage: llm-proxy [flags]\n\n") fs.PrintDefaults() } if err := fs.Parse(os.Args[1:]); err != nil { @@ -75,7 +75,7 @@ func run() error { " see llm-proxy.example.yaml for a starting point") } - cfg, warnings, err := config.Load(path) + cfg, _, err := config.Load(path) if err != nil { return fmt.Errorf("config %s:\n%w", path, err) } @@ -83,7 +83,7 @@ func run() error { // Re-validate after the flags, and keep *these* warnings: -listen can move // the proxy onto a public address, and the warning about an unguarded // listener has to reflect where it will actually listen. - warnings, err = cfg.Validate() + warnings, err := cfg.Validate() if err != nil { return fmt.Errorf("config %s:\n%w", path, err) } @@ -187,7 +187,7 @@ func printRouteTable(c *config.Config, path string, warnings []config.Warning) { } for _, r := range c.Routes { - auth := "none" + var auth string switch { case r.APIKeyEnv == "": auth = "client-supplied only" @@ -211,6 +211,11 @@ func printRouteTable(c *config.Config, path string, warnings []config.Warning) { fmt.Printf(" auth=%s %s retry=%s\n", auth, proto, retry) fmt.Printf(" waits: first byte %s, between chunks %s, progress every %s\n", limit(r.Timeouts.ResponseHeader), limit(r.Timeouts.StreamIdle), limit(r.Timeouts.GapWarn)) + // Only printed when it is on: this is the one setting under which the + // route no longer forwards what the client actually sent. + if len(r.StripParams) > 0 { + fmt.Printf(" rewriting requests: stripping %s\n", strings.Join(r.StripParams, ", ")) + } fmt.Printf(" client base_url: http://%s/%s/v1\n\n", c.Listen, r.Name) } diff --git a/internal/config/config.go b/internal/config/config.go index db3050c..6ef2734 100644 --- a/internal/config/config.go +++ b/internal/config/config.go @@ -214,6 +214,21 @@ type RouteOpts struct { // which many OpenAI-compatible servers do. ExpectDone string `yaml:"expect_done"` + // StripParams are top-level JSON keys deleted from a request body before it + // is forwarded. This is the only place llm-proxy edits what a client sent, + // and it is off unless a route asks for it by name. + // + // It exists for one specific interop failure: an OpenAI-compatible vendor + // that rejects a parameter the client's SDK sends unconditionally — + // `Unsupported parameter(s): prompt_cache_key` — where the client offers no + // way to stop sending it. Stripping it at the proxy is the only place the + // fix can go. + // + // Every strip that actually removes something is recorded and shown in the + // report, because a verdict about which side broke a request is worthless + // if the report does not admit the request was rewritten on the way past. + StripParams []string `yaml:"strip_params"` + // AbortOnTruncation aborts the downstream connection when a stream is // found truncated, so the client cannot mistake it for a complete answer. AbortOnTruncation *bool `yaml:"abort_on_truncation"` diff --git a/internal/config/config_test.go b/internal/config/config_test.go index 9843798..7efb299 100644 --- a/internal/config/config_test.go +++ b/internal/config/config_test.go @@ -223,6 +223,31 @@ func TestValidate(t *testing.T) { "max_attempts", }, {"bad pattern", minimalRoute + "context_length_patterns:\n - \"a(\"\n", "does not compile"}, + { + "strip_params empty entry", + "routes:\n - name: a\n upstream: https://a.example.com\n strip_params:\n - \"\"", + "strip_params[0] is empty", + }, + { + "strip_params padded key", + "routes:\n - name: a\n upstream: https://a.example.com\n strip_params:\n - \" store \"", + "whitespace", + }, + { + "strip_params duplicate key", + "routes:\n - name: a\n upstream: https://a.example.com\n strip_params:\n - store\n - store", + "twice", + }, + { + "strip_params messages", + "routes:\n - name: a\n upstream: https://a.example.com\n strip_params:\n - messages", + "must not contain \"messages\"", + }, + { + "strip_params stream", + "routes:\n - name: a\n upstream: https://a.example.com\n strip_params:\n - stream", + "must not contain \"stream\"", + }, } for _, tc := range tests { t.Run(tc.name, func(t *testing.T) { @@ -238,6 +263,56 @@ func TestValidate(t *testing.T) { } } +// strip_params is the one setting that makes the proxy edit what a client sent, +// so configuring it has to be visible at startup rather than only in a report +// somebody may never read. +func TestStripParamsWarnsThatRequestsAreRewritten(t *testing.T) { + t.Parallel() + c, warns, err := Load(writeConfig(t, ` +routes: + - name: nebul + upstream: https://api.nebul.example + strip_params: + - prompt_cache_key +`)) + if err != nil { + t.Fatalf("Load: %v", err) + } + if got := c.Routes[0].StripParams; len(got) != 1 || got[0] != "prompt_cache_key" { + t.Errorf("strip_params = %v, want [prompt_cache_key]", got) + } + if !hasWarning(warns, "prompt_cache_key") { + t.Errorf("configuring strip_params produced no warning: %+v", warns) + } +} + +// Like every other route setting, it can be set once under defaults. +func TestStripParamsInheritsFromDefaults(t *testing.T) { + t.Parallel() + c, _, err := Load(writeConfig(t, ` +defaults: + strip_params: + - prompt_cache_key +routes: + - name: inherits + upstream: https://a.example.com + - name: overrides + upstream: https://b.example.com + strip_params: [] +`)) + if err != nil { + t.Fatalf("Load: %v", err) + } + if got := c.Routes[0].StripParams; len(got) != 1 || got[0] != "prompt_cache_key" { + t.Errorf("route[0].strip_params = %v, want it inherited", got) + } + // An explicit empty list is a decision, not an absence: a route that says + // it strips nothing must not have the default handed back to it. + if got := c.Routes[1].StripParams; len(got) != 0 { + t.Errorf("route[1].strip_params = %v, want an explicit empty list to win", got) + } +} + // A missing key must not stop the proxy from starting: watching the 401 happen // is more useful than a refusal to boot. func TestMissingAPIKeyWarnsButDoesNotFail(t *testing.T) { diff --git a/internal/config/default_config.yaml b/internal/config/default_config.yaml index 6cc8c55..2493f5b 100644 --- a/internal/config/default_config.yaml +++ b/internal/config/default_config.yaml @@ -67,6 +67,13 @@ defaults: max_request_body: 8MiB expect_done: auto abort_on_truncation: true + # Top-level JSON keys deleted from request bodies before forwarding. Empty, + # and it belongs empty: this is the only setting that makes the proxy edit + # what a client sent, so it stops being a passive observer on any route that + # uses it. Set it when a vendor rejects a parameter your client sends + # unconditionally — `Unsupported parameter(s): prompt_cache_key` — and there + # is no way to turn it off client-side. Every strip is named in the report. + strip_params: [] timeouts: # Establishing the connection is bounded: if a TCP or TLS handshake has not # completed in ten seconds, nothing is going to happen. diff --git a/internal/config/load.go b/internal/config/load.go index 232d47c..0819bee 100644 --- a/internal/config/load.go +++ b/internal/config/load.go @@ -125,6 +125,9 @@ func (c *Config) applyRouteDefaults() { if r.Headers == nil { r.Headers = d.Headers } + if r.StripParams == nil { + r.StripParams = d.StripParams + } inheritBool(&r.ForwardClientAuth, d.ForwardClientAuth) inheritBool(&r.HTTP2, d.HTTP2) @@ -331,6 +334,15 @@ func (c *Config) Validate() ([]Warning, error) { errs = append(errs, fmt.Errorf("%s: expect_done %q: want auto, true or false", where, r.ExpectDone)) } + if err := validateStripParams(where, r.StripParams); err != nil { + errs = append(errs, err) + } else if len(r.StripParams) > 0 { + warns = append(warns, Warning{Route: r.Name, Text: fmt.Sprintf( + "strip_params is set (%s): request bodies on this route are rewritten before "+ + "being forwarded, so what the vendor sees is not byte-for-byte what the "+ + "client sent", strings.Join(r.StripParams, ", "))}) + } + t := r.Timeouts idle, header, gap := Dur(t.StreamIdle), Dur(t.ResponseHeader), Dur(t.GapWarn) if idle > 0 && header > 0 && idle > header { @@ -497,6 +509,43 @@ func validateUpstream(where, upstream string) error { return nil } +// loadBearingParams may not be stripped. The first two make the request invalid +// on arrival, and the third changes the response protocol out from under a +// client that asked for SSE — three ways to turn an interop shim into a fault +// the proxy itself caused, and then reports as the vendor's. +var loadBearingParams = map[string]string{ + "model": "the vendor has nothing to route the request to", + "messages": "there is no prompt left to answer", + "stream": "the vendor would answer a client that asked for SSE with a single JSON body", +} + +func validateStripParams(where string, keys []string) error { + var errs []error + seen := make(map[string]bool, len(keys)) + for i, k := range keys { + switch { + case strings.TrimSpace(k) == "": + errs = append(errs, fmt.Errorf("%s: strip_params[%d] is empty", where, i)) + case k != strings.TrimSpace(k): + // A stray space is a key that silently never matches, which looks + // exactly like the feature not working. + errs = append(errs, fmt.Errorf( + "%s: strip_params[%d] %q has leading or trailing whitespace; JSON keys are exact", + where, i, k)) + case seen[k]: + errs = append(errs, fmt.Errorf("%s: strip_params lists %q twice", where, k)) + } + seen[k] = true + + if why, bad := loadBearingParams[k]; bad { + errs = append(errs, fmt.Errorf( + "%s: strip_params must not contain %q — %s. strip_params is for parameters a "+ + "vendor rejects, not for reshaping the request", where, k, why)) + } + } + return errors.Join(errs...) +} + func validateRetry(where string, r *Retry) error { var errs []error if r.MaxAttempts < 1 { diff --git a/internal/config/main_test.go b/internal/config/main_test.go index 37686ea..0fd68ef 100644 --- a/internal/config/main_test.go +++ b/internal/config/main_test.go @@ -10,6 +10,6 @@ import ( // llm-proxy is likely to have it exported — and without this every test that // assumes an unguarded proxy would fail on their machine and pass in CI. func TestMain(m *testing.M) { - os.Unsetenv("LLM_PROXY_TOKENS") + _ = os.Unsetenv("LLM_PROXY_TOKENS") os.Exit(m.Run()) } diff --git a/internal/fault/fault.go b/internal/fault/fault.go index 06ec279..158f3ed 100644 --- a/internal/fault/fault.go +++ b/internal/fault/fault.go @@ -90,6 +90,7 @@ const ( KindProxyConfig Kind = "proxy_misconfigured" KindProxyInternal Kind = "proxy_internal" KindBodyTooLarge Kind = "request_body_too_large" + KindStripSkipped Kind = "strip_params_skipped" ) func (k Kind) String() string { return string(k) } @@ -325,6 +326,8 @@ func defaultVerdict(k Kind) string { return "the proxy is misconfigured — this is llm-proxy's own fault, not the vendor's." case KindBodyTooLarge: return "the request body exceeded max_request_body." + case KindStripSkipped: + return "a configured strip_params rewrite could not be applied, so the body went upstream unmodified." case KindProxyInternal: return "an internal proxy error. This is llm-proxy's own fault, not the vendor's." default: diff --git a/internal/obs/console.go b/internal/obs/console.go index 18baa55..0693481 100644 --- a/internal/obs/console.go +++ b/internal/obs/console.go @@ -538,6 +538,13 @@ func (r *Renderer) requestSummary(snap record.Snapshot) string { if !snap.BodyReplayable { parts = append(parts, r.paint(ansiDim, "retry=unavailable (body too large to replay)")) } + if len(snap.StrippedParams) > 0 { + // A fault report that blames a side has to disclose that the proxy + // edited the request first. Yellow, not dim: this is the one thing on + // the line that was not the client's doing. + parts = append(parts, r.paint(ansiYellow, + "stripped="+strings.Join(snap.StrippedParams, ","))) + } return strings.Join(parts, " ") } diff --git a/internal/obs/logging.go b/internal/obs/logging.go index 3b58d2e..93d5060 100644 --- a/internal/obs/logging.go +++ b/internal/obs/logging.go @@ -60,7 +60,7 @@ func NewLogger(o Options) *Logger { var consoleHandler slog.Handler if pretty { - consoleHandler = tint.NewHandler(out, &tint.Options{ + consoleHandler = tint.NewTextHandler(out, &tint.Options{ Level: level, TimeFormat: "15:04:05.000", NoColor: !color, diff --git a/internal/obs/obs_test.go b/internal/obs/obs_test.go index b708a06..23cbea2 100644 --- a/internal/obs/obs_test.go +++ b/internal/obs/obs_test.go @@ -183,6 +183,22 @@ func TestVerdictIsAlwaysPresent(t *testing.T) { } } +// A verdict that blames the vendor for rejecting a request must disclose that +// the proxy rewrote that request on the way past. Without this the report reads +// as an observation when it is partly a consequence of our own edit. +func TestFaultReportDisclosesAStrippedRequest(t *testing.T) { + t.Parallel() + r := NewRenderer(RendererOptions{Color: false, Symbols: "ascii"}) + snap := sampleSnapshot() + snap.StrippedParams = []string{"prompt_cache_key"} + snap.Fault = fault.New(fault.SideUpstream, fault.KindHTTPStatus, "status", nil) + + out := r.Render(snap) + if !strings.Contains(out, "stripped=prompt_cache_key") { + t.Errorf("the report does not say the request was rewritten:\n%s", out) + } +} + // A mangled glyph in a CI log or a terminal with the wrong locale is worse than // a plain arrow. func TestAsciiSymbolsAvoidUnicode(t *testing.T) { diff --git a/internal/proxy/auth_test.go b/internal/proxy/auth_test.go index ab0b261..e6bfa66 100644 --- a/internal/proxy/auth_test.go +++ b/internal/proxy/auth_test.go @@ -32,7 +32,7 @@ func TestUnauthenticatedRequestIsRejected(t *testing.T) { if err != nil { t.Fatal(err) } - defer resp.Body.Close() + defer func() { _ = resp.Body.Close() }() if resp.StatusCode != http.StatusUnauthorized { t.Fatalf("Status = %d, want 401", resp.StatusCode) @@ -154,7 +154,7 @@ func TestWrongTokenIsRejected(t *testing.T) { if err != nil { t.Fatal(err) } - defer resp.Body.Close() + defer func() { _ = resp.Body.Close() }() if resp.StatusCode != http.StatusUnauthorized { t.Errorf("Status = %d, want 401", resp.StatusCode) diff --git a/internal/proxy/handler.go b/internal/proxy/handler.go index 45eabd0..a92be86 100644 --- a/internal/proxy/handler.go +++ b/internal/proxy/handler.go @@ -163,6 +163,12 @@ func (h *routeHandler) proxy( h.writeProxyError(cw, rec, http.StatusBadRequest, f) return } + // Before the peek, so everything reported below describes the bytes that + // actually go upstream. Not limited to chat requests: a vendor that rejects + // a parameter rejects it on /v1/embeddings too, and a body that is not a + // JSON object comes back untouched anyway. + body = h.applyStripParams(rec, body, rest) + if rec.Chat { peekPayload(rec, body) // Announced only now: before the peek there is no model, stream flag or @@ -360,7 +366,7 @@ func (h *routeHandler) deliver( cancelUp context.CancelCauseFunc, prefix []byte, ) { - defer resp.Body.Close() + defer func() { _ = resp.Body.Close() }() streaming := isEventStream(resp.Header) analyzer := analyze.New(h.srv.now) @@ -518,7 +524,16 @@ func (h *routeHandler) postmortem(rec *record.Request, status int, a *record.Att // consequence of that departure, not an independent vendor failure. Left // alone it would blame the vendor for whatever its socket did in response // to a cancelled read. - if f != nil && f.Side == fault.SideUpstream && rec.ClientGone() { + // + // "Had already gone" has to mean gone *before the response was finished*, + // which is why this asks whether bytes were left undelivered rather than + // whether the client is gone now. The watcher stays armed until the handler + // returns — well past this point — so a client that took every byte the + // upstream sent and then closed, which is what any client without + // keep-alives does on success, would otherwise reach in and rewrite a + // vendor truncation as its own fault, depending purely on whether its FIN + // landed before this line ran. + if f != nil && f.Side == fault.SideUpstream && rec.ClientLeftMidResponse() { f = fault.AsInduced(f) } @@ -540,8 +555,8 @@ func (h *routeHandler) postmortem(rec *record.Request, status int, a *record.Att if status >= 400 && rec.Fault() == nil { env := oaierr.Parse(errBody) - switch { - case status == http.StatusTooManyRequests: + switch status { + case http.StatusTooManyRequests: rec.SetFault(statusFault(fault.KindRateLimited, status, env)) default: if which, ok := h.srv.ctxMatchers.Match(env, errBody); ok { @@ -776,7 +791,7 @@ func decodeForDisplay(body []byte, encoding string) []byte { if err != nil { return body } - defer zr.Close() + defer func() { _ = zr.Close() }() // A capped peek is usually a partial gzip stream, so a read error still // leaves whatever decoded successfully. out, _ := io.ReadAll(io.LimitReader(zr, maxErrorBody)) diff --git a/internal/proxy/main_test.go b/internal/proxy/main_test.go index 3676125..a16e187 100644 --- a/internal/proxy/main_test.go +++ b/internal/proxy/main_test.go @@ -8,6 +8,6 @@ import ( // See the note in internal/config: the suite must not inherit the operator's // LLM_PROXY_TOKENS, or every test would get a 401 from its own proxy. func TestMain(m *testing.M) { - os.Unsetenv("LLM_PROXY_TOKENS") + _ = os.Unsetenv("LLM_PROXY_TOKENS") os.Exit(m.Run()) } diff --git a/internal/proxy/rewrite.go b/internal/proxy/rewrite.go new file mode 100644 index 0000000..ad1ae5e --- /dev/null +++ b/internal/proxy/rewrite.go @@ -0,0 +1,108 @@ +package proxy + +import ( + "bytes" + "encoding/json" + "fmt" + "io" + "strings" + + "github.com/peterhoneder/llm-proxy/internal/fault" + "github.com/peterhoneder/llm-proxy/internal/record" +) + +// applyStripParams removes the route's strip_params from the request body. +// +// This is the only code path in llm-proxy that edits what a client sent, so it +// is deliberately narrow: it is off unless a route names keys, it only ever +// deletes top-level keys, and anything it cannot do cleanly it does not do at +// all. Whatever happens is recorded — a rewritten request that is not reported +// as rewritten would undermine every verdict this tool exists to give. +func (h *routeHandler) applyStripParams(rec *record.Request, body []byte, rest io.Reader) []byte { + keys := h.target.route.StripParams + if len(keys) == 0 { + return body + } + + if rest != nil { + // Past max_request_body the remainder is streamed straight through and + // was never buffered, so there is no whole document to rewrite. Half a + // rewrite would corrupt the body outright, so the request goes as the + // client wrote it and the report says the shim did not run. + rec.Warn(fault.KindStripSkipped, fmt.Sprintf( + "strip_params (%s) not applied: the body exceeds max_request_body, so it is streamed "+ + "rather than buffered and cannot be rewritten", strings.Join(keys, ", "))) + return body + } + + out, removed, err := stripParams(body, keys) + if err != nil { + // Re-encoding a document that decoded cleanly should not fail. If it + // somehow does, forward the client's original bytes rather than + // anything half-built. + rec.Warn(fault.KindStripSkipped, fmt.Sprintf( + "strip_params (%s) not applied: the body could not be re-encoded: %v", + strings.Join(keys, ", "), err)) + return body + } + if len(removed) == 0 { + return body + } + + rec.SetStrippedParams(removed) + // Both of these describe what goes upstream, not what arrived: the report + // names the stripped keys separately, so the body it shows should be the + // one the vendor is answering. ReqBody is non-nil only under full_trace. + rec.ReqBodyBytes = int64(len(out)) + if rec.ReqBody != nil { + rec.ReqBody = out + } + return out +} + +// stripParams deletes top-level keys from a JSON object body and returns the +// re-encoded document along with the keys that were actually present. +// +// Values are held as json.RawMessage so everything that is kept survives +// byte-for-byte: number precision, string escaping, the exact shape of nested +// objects. Only the top level is rebuilt, which reorders keys alphabetically +// and drops the client's whitespace. That is the smallest edit that can remove +// a key, and it is why the caller reports what it did. +// +// A body that is not a JSON object is not an error: there is nothing to strip +// from, and inventing a rewrite for it is not this function's call. It goes +// upstream untouched, exactly as peekPayload leaves an unparseable body alone. +func stripParams(body []byte, keys []string) ([]byte, []string, error) { + if len(keys) == 0 || len(body) == 0 { + return body, nil, nil + } + + var obj map[string]json.RawMessage + if err := json.Unmarshal(body, &obj); err != nil || obj == nil { + return body, nil, nil + } + + var removed []string + for _, k := range keys { + if _, ok := obj[k]; ok { + delete(obj, k) + removed = append(removed, k) + } + } + if len(removed) == 0 { + return body, nil, nil + } + + // encoding/json escapes <, > and & by default, which would rewrite every + // HTML or XML tag inside a prompt into < escapes — semantically equal, + // but a gratuitous change to bytes the operator may well be diffing, and a + // sizeable one for the tag-heavy prompts agents send. + var buf bytes.Buffer + enc := json.NewEncoder(&buf) + enc.SetEscapeHTML(false) + if err := enc.Encode(obj); err != nil { + return body, nil, err + } + // Encode appends a newline that the client did not send. + return bytes.TrimRight(buf.Bytes(), "\n"), removed, nil +} diff --git a/internal/proxy/rewrite_test.go b/internal/proxy/rewrite_test.go new file mode 100644 index 0000000..2062a87 --- /dev/null +++ b/internal/proxy/rewrite_test.go @@ -0,0 +1,316 @@ +package proxy + +import ( + "bytes" + "encoding/json" + "net/http" + "strconv" + "strings" + "testing" + "time" + + "github.com/peterhoneder/llm-proxy/internal/config" + "github.com/peterhoneder/llm-proxy/internal/fault" + "github.com/peterhoneder/llm-proxy/internal/testutil" +) + +func withStrip(keys ...string) func(*config.Route) { + return func(r *config.Route) { r.StripParams = keys } +} + +// The case the feature exists for: a vendor answers +// +// {"message":"Validation: Unsupported parameter(s): `prompt_cache_key`", ...} +// +// and the client sends the parameter unconditionally. +func TestStripParamsRemovesTheParameterTheVendorRejects(t *testing.T) { + t.Parallel() + h := newHarnessTuned(t, withStrip("prompt_cache_key"), + testutil.Script{Status: http.StatusOK, Body: `{"id":"ok"}`}) + + snap := h.chat(t, `{"model":"m","prompt_cache_key":"abc","messages":[{"role":"user","content":"hi"}]}`) + + seen := h.upstream.Seen() + if len(seen) != 1 { + t.Fatalf("upstream saw %d requests, want 1", len(seen)) + } + if bytes.Contains(seen[0].Body, []byte("prompt_cache_key")) { + t.Errorf("the vendor still received the parameter: %s", seen[0].Body) + } + if snap.Fault != nil { + t.Errorf("unexpected fault: %s — %s", snap.Fault.Kind, snap.Fault.Detail) + } + + // Everything else has to survive, or the shim broke the request it was + // supposed to rescue. + var got map[string]any + if err := json.Unmarshal(seen[0].Body, &got); err != nil { + t.Fatalf("the forwarded body is not valid JSON: %v\n%s", err, seen[0].Body) + } + if got["model"] != "m" { + t.Errorf("model = %v, want it untouched", got["model"]) + } + if msgs, ok := got["messages"].([]any); !ok || len(msgs) != 1 { + t.Errorf("messages = %v, want the one message the client sent", got["messages"]) + } + + // A rewritten body must go out under its own length, not the client's, and + // not as a chunked stream — the wire form is evidence in this tool. + if te := seen[0].Header.Get("Transfer-Encoding"); te != "" { + t.Errorf("the rewritten request went out chunked (%q), changing the wire form", te) + } + if cl := seen[0].Header.Get("Content-Length"); cl != "" && + cl != strconv.Itoa(len(seen[0].Body)) { + t.Errorf("Content-Length = %q, want %d — the length of what was actually sent", + cl, len(seen[0].Body)) + } + + if want := []string{"prompt_cache_key"}; !equalStrings(snap.StrippedParams, want) { + t.Errorf("StrippedParams = %v, want %v — a rewritten request must be reported as rewritten", + snap.StrippedParams, want) + } + if snap.ReqBodyBytes != int64(len(seen[0].Body)) { + t.Errorf("ReqBodyBytes = %d, want %d — the report should describe what went upstream", + snap.ReqBodyBytes, len(seen[0].Body)) + } +} + +// The whole product rests on passing bytes through unaltered. Nothing may be +// rewritten unless a route asked for it by name. +func TestStripParamsIsOffByDefault(t *testing.T) { + t.Parallel() + h := newHarness(t, testutil.Script{Status: http.StatusOK, Body: `{"id":"ok"}`}) + + body := `{"model":"m","prompt_cache_key":"abc","messages":[]}` + snap := h.chat(t, body) + + seen := h.upstream.Seen() + if len(seen) != 1 { + t.Fatalf("upstream saw %d requests, want 1", len(seen)) + } + if string(seen[0].Body) != body { + t.Errorf("body = %s, want the client's bytes unaltered", seen[0].Body) + } + if len(snap.StrippedParams) != 0 { + t.Errorf("StrippedParams = %v, want none", snap.StrippedParams) + } +} + +// A configured key that is not in this particular body must not cost the +// request a re-encode: an unrelated request going through a stripping route +// still deserves byte-for-byte passthrough. +func TestStripParamsLeavesTheBodyAloneWhenTheKeyIsAbsent(t *testing.T) { + t.Parallel() + h := newHarnessTuned(t, withStrip("prompt_cache_key"), + testutil.Script{Status: http.StatusOK, Body: `{"id":"ok"}`}) + + body := `{ "model":"m", "messages":[] }` + snap := h.chat(t, body) + + seen := h.upstream.Seen() + if string(seen[0].Body) != body { + t.Errorf("body = %s, want it untouched down to the whitespace", seen[0].Body) + } + if len(snap.StrippedParams) != 0 { + t.Errorf("StrippedParams = %v, want none: the key was not there", snap.StrippedParams) + } +} + +// Everything the rewrite keeps must survive byte-for-byte. Number precision and +// prompts full of XML tags are the two that a naive re-encode quietly mangles. +func TestStripParamsPreservesTheValuesItKeeps(t *testing.T) { + t.Parallel() + h := newHarnessTuned(t, withStrip("prompt_cache_key"), + testutil.Script{Status: http.StatusOK, Body: `{"id":"ok"}`}) + + const prompt = `use a & b, not a>b` + h.chat(t, `{"model":"m","prompt_cache_key":"abc","temperature":0.10000000000000000555,`+ + `"messages":[{"role":"user","content":"`+prompt+`"}],"metadata":{"nested":{"deep":[1,2,3]}}}`) + + got := h.upstream.Seen()[0].Body + if !bytes.Contains(got, []byte(prompt)) { + t.Errorf("the prompt was re-escaped on the way through:\n%s", got) + } + if !bytes.Contains(got, []byte("0.10000000000000000555")) { + t.Errorf("a number lost precision in the rewrite:\n%s", got) + } + if !bytes.Contains(got, []byte(`{"nested":{"deep":[1,2,3]}}`)) { + t.Errorf("a nested value was reshaped:\n%s", got) + } +} + +// Retry replays whatever went out the first time. If the two attempts differed, +// the second would not be the same request and nothing could be concluded from +// comparing them. +func TestStripParamsSurvivesARetryIdentically(t *testing.T) { + t.Parallel() + h := newHarnessTuned(t, func(r *config.Route) { + withRetry(3, time.Minute)(r) + withStrip("prompt_cache_key")(r) + }, + testutil.Script{Status: http.StatusInternalServerError, Body: `{"error":"boom"}`}, + testutil.Script{Status: http.StatusOK, Body: `{"id":"ok"}`}, + ) + + h.chat(t, `{"model":"m","prompt_cache_key":"abc","messages":[]}`) + + seen := h.upstream.Seen() + if len(seen) != 2 { + t.Fatalf("upstream saw %d requests, want 2", len(seen)) + } + if !bytes.Equal(seen[0].Body, seen[1].Body) { + t.Errorf("the replayed body differs:\n first: %s\nsecond: %s", seen[0].Body, seen[1].Body) + } + if bytes.Contains(seen[1].Body, []byte("prompt_cache_key")) { + t.Errorf("the retry re-sent the stripped parameter: %s", seen[1].Body) + } +} + +// A vendor that rejects a parameter rejects it everywhere, so the shim is not +// limited to chat completions. +func TestStripParamsAppliesOutsideChatCompletions(t *testing.T) { + t.Parallel() + h := newHarnessTuned(t, withStrip("prompt_cache_key"), + testutil.Script{Status: http.StatusOK, Body: `{"data":[]}`}) + + h.do(t, "POST", "/vendor/v1/embeddings", `{"model":"e","prompt_cache_key":"abc","input":"hi"}`) + + if got := h.upstream.Seen()[0].Body; bytes.Contains(got, []byte("prompt_cache_key")) { + t.Errorf("the parameter survived on a non-chat path: %s", got) + } +} + +// There is nothing to strip from a body that is not a JSON object, and +// inventing a rewrite for one is not the proxy's call. +func TestStripParamsLeavesANonJSONBodyAlone(t *testing.T) { + t.Parallel() + h := newHarnessTuned(t, withStrip("prompt_cache_key"), + testutil.Script{Status: http.StatusOK, Body: `{"id":"ok"}`}) + + body := "prompt_cache_key=abc&model=m" + h.do(t, "POST", "/vendor/v1/anything", body) + + if got := string(h.upstream.Seen()[0].Body); got != body { + t.Errorf("body = %s, want it forwarded untouched", got) + } +} + +// Past max_request_body the remainder is streamed and never buffered, so there +// is no document to rewrite. Half a rewrite would corrupt the body; the request +// goes through intact and the report says the shim did not run. +func TestStripParamsIsReportedWhenTheBodyIsTooLargeToRewrite(t *testing.T) { + t.Parallel() + h := newHarnessTuned(t, func(r *config.Route) { + withStrip("prompt_cache_key")(r) + r.MaxRequestBody = 256 + }, testutil.Script{Status: http.StatusOK, Body: `{"id":"ok"}`}) + + body := `{"model":"m","prompt_cache_key":"abc","messages":[{"role":"user","content":"` + + strings.Repeat("x", 512) + `"}]}` + snap := h.chat(t, body) + + if got := string(h.upstream.Seen()[0].Body); got != body { + t.Errorf("an unrewritable body was altered:\n%s", got) + } + if len(snap.StrippedParams) != 0 { + t.Errorf("StrippedParams = %v, want none: nothing was stripped", snap.StrippedParams) + } + if !hasWarningKind(snap.Warnings, fault.KindStripSkipped) { + t.Errorf("no warning that strip_params did not run; warnings = %+v", snap.Warnings) + } +} + +func TestStripParamsRemovesEveryConfiguredKey(t *testing.T) { + t.Parallel() + h := newHarnessTuned(t, withStrip("prompt_cache_key", "safety_identifier", "store"), + testutil.Script{Status: http.StatusOK, Body: `{"id":"ok"}`}) + + snap := h.chat(t, `{"model":"m","prompt_cache_key":"a","store":false,"messages":[]}`) + + got := h.upstream.Seen()[0].Body + for _, k := range []string{"prompt_cache_key", "store"} { + if bytes.Contains(got, []byte(k)) { + t.Errorf("%q survived: %s", k, got) + } + } + // Only what was actually present gets reported, so the report never claims + // to have removed something the client never sent. + if want := []string{"prompt_cache_key", "store"}; !equalStrings(snap.StrippedParams, want) { + t.Errorf("StrippedParams = %v, want %v", snap.StrippedParams, want) + } +} + +func TestStripParamsUnit(t *testing.T) { + t.Parallel() + tests := []struct { + name string + body string + keys []string + want string + wantRemoved []string + }{ + { + name: "no keys configured", + body: `{"a":1}`, keys: nil, want: `{"a":1}`, + }, + { + name: "key absent leaves the bytes alone", + body: `{ "a" : 1 }`, keys: []string{"b"}, want: `{ "a" : 1 }`, + }, + { + name: "empty body", + body: ``, keys: []string{"a"}, want: ``, + }, + { + name: "JSON null is not an object", + body: `null`, keys: []string{"a"}, want: `null`, + }, + { + name: "a JSON array is not an object", + body: `[{"a":1}]`, keys: []string{"a"}, want: `[{"a":1}]`, + }, + { + name: "only the top level is searched", + body: `{"a":1,"nested":{"a":2}}`, keys: []string{"a"}, + want: `{"nested":{"a":2}}`, wantRemoved: []string{"a"}, + }, + { + name: "a null value still counts as present", + body: `{"a":null,"b":1}`, keys: []string{"a"}, + want: `{"b":1}`, wantRemoved: []string{"a"}, + }, + { + name: "stripping the only key leaves an empty object", + body: `{"a":1}`, keys: []string{"a"}, + want: `{}`, wantRemoved: []string{"a"}, + }, + } + + for _, tc := range tests { + t.Run(tc.name, func(t *testing.T) { + t.Parallel() + got, removed, err := stripParams([]byte(tc.body), tc.keys) + if err != nil { + t.Fatalf("stripParams: %v", err) + } + if string(got) != tc.want { + t.Errorf("body = %s, want %s", got, tc.want) + } + if !equalStrings(removed, tc.wantRemoved) { + t.Errorf("removed = %v, want %v", removed, tc.wantRemoved) + } + }) + } +} + +func equalStrings(a, b []string) bool { + if len(a) != len(b) { + return false + } + for i := range a { + if a[i] != b[i] { + return false + } + } + return true +} diff --git a/internal/proxy/routing_test.go b/internal/proxy/routing_test.go index 9ce9fe0..794ade7 100644 --- a/internal/proxy/routing_test.go +++ b/internal/proxy/routing_test.go @@ -147,7 +147,7 @@ func TestUnknownRouteReturnsParseableError(t *testing.T) { if err != nil { t.Fatal(err) } - defer resp.Body.Close() + defer func() { _ = resp.Body.Close() }() if resp.StatusCode != http.StatusNotFound { t.Errorf("Status = %d, want 404", resp.StatusCode) @@ -189,7 +189,7 @@ func TestHealthAndRoutesEndpoints(t *testing.T) { if err != nil { t.Fatal(err) } - defer resp.Body.Close() + defer func() { _ = resp.Body.Close() }() var routes struct { Routes []struct { diff --git a/internal/proxy/server.go b/internal/proxy/server.go index abe426d..c75626c 100644 --- a/internal/proxy/server.go +++ b/internal/proxy/server.go @@ -360,6 +360,12 @@ func (s *Server) StartupBanner() string { width, "/"+r.Name+"/*", r.Upstream, authDescription(t), proto, retry) fmt.Fprintf(&b, " %-*s client base_url: http://%s/%s/v1\n", width, "", s.cfg.Listen, r.Name) + // Printed only when it is on, because it is the one setting that stops + // this route being a passive observer of what the client sent. + if len(r.StripParams) > 0 { + fmt.Fprintf(&b, " %-*s rewriting requests: stripping %s\n", + width, "", strings.Join(r.StripParams, ", ")) + } } return b.String() } diff --git a/internal/proxy/statuses_test.go b/internal/proxy/statuses_test.go index 80fc485..908c714 100644 --- a/internal/proxy/statuses_test.go +++ b/internal/proxy/statuses_test.go @@ -5,6 +5,7 @@ import ( "net/http" "strings" "testing" + "time" "github.com/peterhoneder/llm-proxy/internal/fault" "github.com/peterhoneder/llm-proxy/internal/record" @@ -149,6 +150,37 @@ func TestNonStreamingTruncatedBody(t *testing.T) { requireKind(t, snap, fault.KindTruncatedBody) } +// The same truncation, but with the client hanging up the instant it has the +// body — which is what every client without keep-alives does on every request. +// +// The client watcher stays armed until the handler returns, so its stamp can +// land while the verdict is still being drawn. It must not change the answer: +// the client took every byte the upstream sent, so it interrupted nothing, and +// the vendor's body still ended mid-JSON. Attribution here is decided by +// undelivered bytes, not by which goroutine got there first. +func TestClientHangUpAfterTruncatedBodyStillBlamesUpstream(t *testing.T) { + t.Parallel() + upstream := newRawUpstream(t, func(w http.ResponseWriter, r *http.Request) { + w.Header().Set("Content-Type", "application/json") + _, _ = io.WriteString(w, `{"id":"cmpl-9","choices":[{"index":0,"message":{"content":"cut o`) + }) + h := newHarnessWithUpstream(t, upstream) + + c := testutil.Dial(t, h.addr) + c.Send("POST", "/vendor/v1/chat/completions", "localhost", `{"model":"m","messages":[]}`) + c.ReadStatusLine() + c.ReadHeaders() + c.ReadSome(time.Second) + c.HangUp() + + snap := h.waitForRecord(t, 3*time.Second) + requireSide(t, snap, fault.SideUpstream) + requireKind(t, snap, fault.KindTruncatedBody) + if snap.Fault.Induced { + t.Error("a client that received the whole body cannot have induced the truncation") + } +} + func TestNonStreamingCompleteResponse(t *testing.T) { t.Parallel() h := newHarness(t, testutil.Script{ diff --git a/internal/record/record.go b/internal/record/record.go index 20f06bc..a9bdd1f 100644 --- a/internal/record/record.go +++ b/internal/record/record.go @@ -205,6 +205,11 @@ type Request struct { ReqBodyBytes int64 BodyReplayable bool + // StrippedParams are the keys a route's strip_params actually removed from + // this body. Kept so the report can say the request was rewritten: a + // verdict on who broke a request must not omit that the proxy edited it. + StrippedParams []string + // ClientRequestID is an inbound X-Request-Id, kept so a harness that // stamps its own correlation id can be matched to this report. ClientRequestID string @@ -342,6 +347,14 @@ func (r *Request) SetKeySource(src string) { r.KeySource = src } +// SetStrippedParams records which request parameters a route's strip_params +// removed on the way upstream. +func (r *Request) SetStrippedParams(keys []string) { + r.mu.Lock() + defer r.mu.Unlock() + r.StrippedParams = append([]string(nil), keys...) +} + // Warn appends a warning. func (r *Request) Warn(kind fault.Kind, text string) { r.mu.Lock() @@ -389,6 +402,21 @@ func (r *Request) ClientGone() bool { return !r.ClientGoneAt.IsZero() } +// ClientLeftMidResponse reports whether the client went away while the proxy +// still had bytes to hand it. +// +// The bare fact that a client is gone says nothing about causation. A client +// with keep-alives disabled closes the connection on every successful request, +// once it has everything it asked for, and the watcher is still armed while the +// verdict is being drawn. Pairing the departure with undelivered bytes is what +// separates a client that interrupted the response from one that was simply +// finished with it. +func (r *Request) ClientLeftMidResponse() bool { + r.mu.Lock() + defer r.mu.Unlock() + return !r.ClientGoneAt.IsZero() && r.BytesToClient < r.BytesFromUpstream +} + // AddRead records bytes read from the upstream. func (r *Request) AddRead(n int, at time.Time) { r.lastUpstreamByte.Store(at.UnixNano()) @@ -481,6 +509,7 @@ type Snapshot struct { ReqBodyBytes int64 BodyReplayable bool + StrippedParams []string Status int HeadersSentAt time.Time @@ -519,6 +548,8 @@ func (r *Request) Snapshot() Snapshot { warnings := make([]Warning, len(r.Warnings)) copy(warnings, r.Warnings) + stripped := append([]string(nil), r.StrippedParams...) + return Snapshot{ ID: r.ID, ConnID: r.ConnID, Start: r.Start, End: r.End, Duration: r.Duration, @@ -533,7 +564,7 @@ func (r *Request) Snapshot() Snapshot { IncludeUsage: r.IncludeUsage, MaxTokens: r.MaxTokens, Temperature: r.Temperature, TopP: r.TopP, NMessages: r.NMessages, ReqBody: r.ReqBody, ReqBodyBytes: r.ReqBodyBytes, - BodyReplayable: r.BodyReplayable, + BodyReplayable: r.BodyReplayable, StrippedParams: stripped, Status: r.Status, HeadersSentAt: r.HeadersSentAt, TTFB: r.TTFB, diff --git a/internal/record/record_test.go b/internal/record/record_test.go new file mode 100644 index 0000000..f61c9ad --- /dev/null +++ b/internal/record/record_test.go @@ -0,0 +1,68 @@ +package record + +import ( + "net/http/httptest" + "testing" + "time" +) + +func newTestRequest(t *testing.T) *Request { + t.Helper() + r := httptest.NewRequest("POST", "/v1/chat/completions", nil) + return New("req-1", "conn-1", "vendor", "openai", r, time.Now()) +} + +// ClientLeftMidResponse is what stops a client's departure from rewriting a +// verdict it had no part in. The distinction it draws is causal, not temporal: +// a client is only implicated if the proxy still had bytes for it. +func TestClientLeftMidResponse(t *testing.T) { + t.Parallel() + + now := time.Now() + + t.Run("gone with bytes still undelivered", func(t *testing.T) { + t.Parallel() + rec := newTestRequest(t) + rec.AddRead(64, now) + rec.AddDelivered(20, now) + rec.SetClientGone(now) + if !rec.ClientLeftMidResponse() { + t.Error("a client that left with 44 bytes still owed to it interrupted the response") + } + }) + + // The case that made the truncated-body verdict a coin toss: a client + // without keep-alives closes the connection on every successful request, + // and the watcher is still armed while the verdict is drawn. + t.Run("gone having received everything", func(t *testing.T) { + t.Parallel() + rec := newTestRequest(t) + rec.AddRead(64, now) + rec.AddDelivered(64, now) + rec.SetClientGone(now) + if rec.ClientLeftMidResponse() { + t.Error("a client that received every byte read from the upstream interrupted nothing") + } + }) + + t.Run("still connected", func(t *testing.T) { + t.Parallel() + rec := newTestRequest(t) + rec.AddRead(64, now) + rec.AddDelivered(20, now) + if rec.ClientLeftMidResponse() { + t.Error("the client never went away") + } + }) + + // A bodyless response reads and delivers nothing. Equal counts, so no + // departure can be blamed on the response being unfinished. + t.Run("gone with nothing to deliver", func(t *testing.T) { + t.Parallel() + rec := newTestRequest(t) + rec.SetClientGone(now) + if rec.ClientLeftMidResponse() { + t.Error("with no bytes on either side there is nothing the client cut short") + } + }) +} diff --git a/internal/testutil/fakeupstream.go b/internal/testutil/fakeupstream.go index 8de9d87..4ebd09e 100644 --- a/internal/testutil/fakeupstream.go +++ b/internal/testutil/fakeupstream.go @@ -91,7 +91,7 @@ func NewUpstream(t *testing.T, scripts ...Script) *Upstream { t.Helper() u := &Upstream{scripts: scripts} u.Server = httptest.NewServer(http.HandlerFunc(u.serve)) - t.Cleanup(u.Server.Close) + t.Cleanup(u.Close) return u } @@ -164,7 +164,7 @@ func (u *Upstream) serveStream(w http.ResponseWriter, s Script) { if err != nil { panic(err) } - defer conn.Close() + defer func() { _ = conn.Close() }() var head bytes.Buffer fmt.Fprintf(&head, "HTTP/1.1 %d %s\r\n", statusOr(s.Status), http.StatusText(statusOr(s.Status))) @@ -204,7 +204,7 @@ func (u *Upstream) serveStream(w http.ResponseWriter, s Script) { // The final frame is where a truncation has to happen, so it is // written as a partial chunk rather than a complete one. if s.Ending == EndTruncateMidFrame && i == len(frames)-1 { - fmt.Fprintf(rw, "%x\r\n%s", len(frame), frame[:len(frame)/2]) + _, _ = fmt.Fprintf(rw, "%x\r\n%s", len(frame), frame[:len(frame)/2]) _ = rw.Flush() resetOrClose(conn, false) return @@ -266,7 +266,7 @@ func splitFrames(payload string) []string { } func writeChunk(rw *bufio.ReadWriter, data string) { - fmt.Fprintf(rw, "%x\r\n%s\r\n", len(data), data) + _, _ = fmt.Fprintf(rw, "%x\r\n%s\r\n", len(data), data) _ = rw.Flush() } diff --git a/llm-proxy.example.yaml b/llm-proxy.example.yaml index b8376c2..b0f1377 100644 --- a/llm-proxy.example.yaml +++ b/llm-proxy.example.yaml @@ -61,6 +61,23 @@ routes: # detection gets stricter. expect_done: auto + # Some OpenAI-compatible vendors reject a parameter the client's SDK sends + # unconditionally: + # + # {"message":"Validation: Unsupported parameter(s): `prompt_cache_key`", ...} + # + # strip_params deletes top-level keys from the request body before forwarding, + # for exactly that case. It is the one setting that makes the proxy edit what + # a client sent, so bodies on this route are no longer byte-for-byte what + # arrived — startup says so, and every strip is named in the request report. + # model, messages and stream are refused: removing those breaks the request + # rather than fixing it. + - name: nebul + upstream: https://api.nebul.example + api_key_env: NEBUL_API_KEY + strip_params: + - prompt_cache_key + # A local model server: no key, and keep-alives off while you are hunting # stale-connection resets. - name: local