From d93e4764af14268338e51590a5dcb84d35c6c6b7 Mon Sep 17 00:00:00 2001 From: Ignasi Barrera Date: Sat, 26 Sep 2026 13:38:19 -0400 Subject: [PATCH 1/4] fix: make sure there are no duplicates when setting stream_options.include_usage Signed-off-by: Ignasi Barrera --- internal/endpointspec/endpointspec.go | 31 +++++++++++++++++----- internal/endpointspec/endpointspec_test.go | 23 ++++++++++++++++ 2 files changed, 48 insertions(+), 6 deletions(-) diff --git a/internal/endpointspec/endpointspec.go b/internal/endpointspec/endpointspec.go index 545f7c323d..83714b4364 100644 --- a/internal/endpointspec/endpointspec.go +++ b/internal/endpointspec/endpointspec.go @@ -17,6 +17,7 @@ import ( "strconv" "strings" + "github.com/tidwall/gjson" "github.com/tidwall/sjson" "k8s.io/utils/ptr" @@ -145,12 +146,7 @@ func (ChatCompletionsEndpointSpec) ParseBody( // Rewrite the original bytes to include the stream_options.include_usage=true so that forcing the request body // mutation, which uses this raw body, will also result in the stream_options.include_usage=true. var err error - mutatedBody, err = sjson.SetBytesOptions(body, "stream_options.include_usage", true, &sjson.Options{ - Optimistic: true, - // Note: it is safe to do in-place replacement since this route level processor is executed once per request, - // and the result can be safely shared among possible multiple retries. - ReplaceInPlace: true, - }) + mutatedBody, err = forceStreamOptionsIncludeUsage(body) if err != nil { return "", nil, false, nil, fmt.Errorf("%w: failed to set stream_options.include_usage", internalapi.ErrMalformedRequest) } @@ -158,6 +154,29 @@ func (ChatCompletionsEndpointSpec) ParseBody( return req.Model, &req, req.Stream, mutatedBody, nil } +// forceStreamOptionsIncludeUsage rewrites body so that it has exactly one top-level +// "stream_options" key set to {"include_usage": true}. +func forceStreamOptionsIncludeUsage(body []byte) ([]byte, error) { + mutatedBody := body + for gjson.GetBytes(mutatedBody, "stream_options").Exists() { + var err error + mutatedBody, err = sjson.DeleteBytes(mutatedBody, "stream_options") + if err != nil { + return nil, fmt.Errorf("failed to remove existing stream_options: %w", err) + } + } + mutatedBody, err := sjson.SetBytesOptions(mutatedBody, "stream_options.include_usage", true, &sjson.Options{ + Optimistic: true, + // Note: it is safe to do in-place replacement since this route level processor is executed once per request, + // and the result can be safely shared among possible multiple retries. + ReplaceInPlace: true, + }) + if err != nil { + return nil, fmt.Errorf("failed to set stream_options.include_usage: %w", err) + } + return mutatedBody, nil +} + // ParseMultipartBody implements [Spec.ParseMultipartBody]. func (ChatCompletionsEndpointSpec) ParseMultipartBody([]byte, string, bool) (internalapi.OriginalModel, *openai.ChatCompletionRequest, bool, []byte, error) { return "", nil, false, nil, errMultipartNotSupported diff --git a/internal/endpointspec/endpointspec_test.go b/internal/endpointspec/endpointspec_test.go index 46a9158273..52d73b6b95 100644 --- a/internal/endpointspec/endpointspec_test.go +++ b/internal/endpointspec/endpointspec_test.go @@ -9,6 +9,7 @@ import ( "bytes" "errors" "mime/multipart" + "strings" "testing" "github.com/stretchr/testify/require" @@ -67,6 +68,28 @@ func TestChatCompletionsEndpointSpec_ParseBody(t *testing.T) { require.Nil(t, mutated) }) + t.Run("streaming_with_duplicate_stream_options", func(t *testing.T) { + body := []byte(`{"model":"gpt-4o","stream":true,"stream_options":{"include_usage":true},"stream_options":{"include_usage":false}}`) + + model, parsed, stream, mutated, err := spec.ParseBody(body, true) + require.NoError(t, err) + require.Equal(t, "gpt-4o", model) + require.True(t, stream) + require.NotNil(t, parsed) + require.NotNil(t, parsed.StreamOptions) + require.True(t, parsed.StreamOptions.IncludeUsage) + require.NotNil(t, mutated) + + // The mutated body -- which is what actually gets forwarded to the upstream provider -- + // must contain a single, unambiguous stream_options.include_usage=true and must not retain + // any attacker-controlled duplicate "stream_options" key. + require.Equal(t, 1, strings.Count(string(mutated), "stream_options")) + var mutatedReq openai.ChatCompletionRequest + require.NoError(t, json.Unmarshal(mutated, &mutatedReq)) + require.NotNil(t, mutatedReq.StreamOptions) + require.True(t, mutatedReq.StreamOptions.IncludeUsage) + }) + t.Run("non_streaming", func(t *testing.T) { req := openai.ChatCompletionRequest{Model: "gpt-4-mini", Stream: false} body, err := json.Marshal(req) From 7f59e6cbb97dd4c2493e2287b2f73daa8c99044d Mon Sep 17 00:00:00 2001 From: Ignasi Barrera Date: Sat, 26 Sep 2026 14:53:22 -0400 Subject: [PATCH 2/4] fix tests Signed-off-by: Ignasi Barrera --- tests/data-plane/testupstream_test.go | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/tests/data-plane/testupstream_test.go b/tests/data-plane/testupstream_test.go index f46da64ee0..2b6b06ffd7 100644 --- a/tests/data-plane/testupstream_test.go +++ b/tests/data-plane/testupstream_test.go @@ -1744,7 +1744,7 @@ func TestStreamingUsageInclusionWithCosts(t *testing.T) { name: "streaming - forced to include usage", backend: "openai", requestBody: `{"model":"something","messages":[{"role":"system","content":"You are a chatbot."}], "stream": true, "stream_options": {"include_usage": false}}`, - expRequestBody: `{"model":"something","messages":[{"role":"system","content":"You are a chatbot."}], "stream": true, "stream_options": {"include_usage": true}}`, + expRequestBody: `{"model":"something","messages":[{"role":"system","content":"You are a chatbot."}], "stream": true,"stream_options":{"include_usage":true}}`, responseBody: ` {"id":"chatcmpl-foo","object":"chat.completion.chunk","created":1731618222,"model":"gpt-4o-mini-2024-07-18","system_fingerprint":"fp_0ba0d124f1","choices":[{"index":0,"delta":{"role":"assistant","content":"","refusal":null},"logprobs":null,"finish_reason":null}],"usage":null} {"id":"chatcmpl-foo","object":"chat.completion.chunk","created":1731618222,"model":"gpt-4o-mini-2024-07-18","system_fingerprint":"fp_0ba0d124f1","choices":[],"usage":{"prompt_tokens":13,"completion_tokens":12,"total_tokens":25,"prompt_tokens_details":{"cached_tokens":0,"audio_tokens":0},"completion_tokens_details":{"reasoning_tokens":0,"audio_tokens":0,"accepted_prediction_tokens":0,"rejected_prediction_tokens":0}}} @@ -1780,7 +1780,7 @@ data: [DONE] name: "streaming - model override forced to include usage", backend: "modelname-override", requestBody: `{"model":"requested-model","messages":[{"role":"system","content":"You are a chatbot."}], "stream": true, "stream_options": {"include_usage": false}}`, - expRequestBody: `{"model":"override-model","messages":[{"role":"system","content":"You are a chatbot."}], "stream": true, "stream_options": {"include_usage": true}}`, + expRequestBody: `{"model":"override-model","messages":[{"role":"system","content":"You are a chatbot."}], "stream": true,"stream_options":{"include_usage":true}}`, responseBody: ` {"id":"chatcmpl-foo","object":"chat.completion.chunk","created":1731618222,"model":"gpt-4o-mini-2024-07-18","system_fingerprint":"fp_0ba0d124f1","choices":[{"index":0,"delta":{"role":"assistant","content":"","refusal":null},"logprobs":null,"finish_reason":null}],"usage":null} {"id":"chatcmpl-foo","object":"chat.completion.chunk","created":1731618222,"model":"gpt-4o-mini-2024-07-18","system_fingerprint":"fp_0ba0d124f1","choices":[],"usage":{"prompt_tokens":13,"completion_tokens":12,"total_tokens":25,"prompt_tokens_details":{"cached_tokens":0,"audio_tokens":0},"completion_tokens_details":{"reasoning_tokens":0,"audio_tokens":0,"accepted_prediction_tokens":0,"rejected_prediction_tokens":0}}} From b56443311006bb5363582b7bf137951b1447223e Mon Sep 17 00:00:00 2001 From: Ignasi Barrera Date: Tue, 29 Sep 2026 11:52:53 -0400 Subject: [PATCH 3/4] fixes Signed-off-by: Ignasi Barrera --- internal/endpointspec/endpointspec.go | 22 ++++++++++++++--- internal/endpointspec/endpointspec_test.go | 28 ++++++++++++++++++++++ 2 files changed, 47 insertions(+), 3 deletions(-) diff --git a/internal/endpointspec/endpointspec.go b/internal/endpointspec/endpointspec.go index 83714b4364..92840ece6d 100644 --- a/internal/endpointspec/endpointspec.go +++ b/internal/endpointspec/endpointspec.go @@ -155,17 +155,33 @@ func (ChatCompletionsEndpointSpec) ParseBody( } // forceStreamOptionsIncludeUsage rewrites body so that it has exactly one top-level -// "stream_options" key set to {"include_usage": true}. +// "stream_options" key with include_usage forced to true. Other fields on it (e.g. vLLM's +// continuous_usage_stats) are preserved. If the body contains duplicate top-level +// "stream_options" keys, only the last one is kept -- matching the semantics of +// json.Unmarshal, which is what populates the already-parsed request -- so that the mutated +// body cannot disagree with the parsed request about which stream_options applies. func forceStreamOptionsIncludeUsage(body []byte) ([]byte, error) { mutatedBody := body - for gjson.GetBytes(mutatedBody, "stream_options").Exists() { + streamOptions := "{}" + for { + res := gjson.GetBytes(mutatedBody, "stream_options") + if !res.Exists() { + break + } + streamOptions = res.Raw var err error mutatedBody, err = sjson.DeleteBytes(mutatedBody, "stream_options") if err != nil { return nil, fmt.Errorf("failed to remove existing stream_options: %w", err) } } - mutatedBody, err := sjson.SetBytesOptions(mutatedBody, "stream_options.include_usage", true, &sjson.Options{ + + streamOptions, err := sjson.SetRawOptions(streamOptions, "include_usage", "true", &sjson.Options{Optimistic: true}) + if err != nil { + return nil, fmt.Errorf("failed to set include_usage on stream_options: %w", err) + } + + mutatedBody, err = sjson.SetRawBytesOptions(mutatedBody, "stream_options", []byte(streamOptions), &sjson.Options{ Optimistic: true, // Note: it is safe to do in-place replacement since this route level processor is executed once per request, // and the result can be safely shared among possible multiple retries. diff --git a/internal/endpointspec/endpointspec_test.go b/internal/endpointspec/endpointspec_test.go index 52d73b6b95..b5c1ecee60 100644 --- a/internal/endpointspec/endpointspec_test.go +++ b/internal/endpointspec/endpointspec_test.go @@ -90,6 +90,34 @@ func TestChatCompletionsEndpointSpec_ParseBody(t *testing.T) { require.True(t, mutatedReq.StreamOptions.IncludeUsage) }) + t.Run("streaming_preserves_extra_stream_options_fields", func(t *testing.T) { + // vLLM supports additional stream_options fields beyond include_usage, e.g. + // continuous_usage_stats. Forcing include_usage must not drop them. + body := []byte(`{"model":"gpt-4o","stream":true,"stream_options":{"include_usage":false,"continuous_usage_stats":true}}`) + + _, parsed, _, mutated, err := spec.ParseBody(body, true) + require.NoError(t, err) + require.NotNil(t, parsed) + require.True(t, parsed.StreamOptions.IncludeUsage) + require.NotNil(t, mutated) + require.Equal(t, 1, strings.Count(string(mutated), "stream_options")) + require.JSONEq(t, `{"model":"gpt-4o","stream":true,"stream_options":{"include_usage":true,"continuous_usage_stats":true}}`, string(mutated)) + }) + + t.Run("streaming_with_duplicate_stream_options_preserves_last_fields", func(t *testing.T) { + // With duplicate top-level keys, json.Unmarshal (and therefore `parsed`) takes the + // last occurrence. The mutated body must match that behavior and keep its other fields. + body := []byte(`{"model":"gpt-4o","stream":true,"stream_options":{"continuous_usage_stats":true},"stream_options":{"include_usage":false,"continuous_usage_stats":false}}`) + + _, parsed, _, mutated, err := spec.ParseBody(body, true) + require.NoError(t, err) + require.NotNil(t, parsed) + require.True(t, parsed.StreamOptions.IncludeUsage) + require.NotNil(t, mutated) + require.Equal(t, 1, strings.Count(string(mutated), "stream_options")) + require.JSONEq(t, `{"model":"gpt-4o","stream":true,"stream_options":{"include_usage":true,"continuous_usage_stats":false}}`, string(mutated)) + }) + t.Run("non_streaming", func(t *testing.T) { req := openai.ChatCompletionRequest{Model: "gpt-4-mini", Stream: false} body, err := json.Marshal(req) From fea56db68d43e7f8e603a9d8420a685a3b75b517 Mon Sep 17 00:00:00 2001 From: Ignasi Barrera Date: Tue, 29 Sep 2026 12:53:34 -0400 Subject: [PATCH 4/4] fix tests Signed-off-by: Ignasi Barrera --- tests/data-plane/testupstream_test.go | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/tests/data-plane/testupstream_test.go b/tests/data-plane/testupstream_test.go index 2b6b06ffd7..32596a2fe0 100644 --- a/tests/data-plane/testupstream_test.go +++ b/tests/data-plane/testupstream_test.go @@ -1744,7 +1744,7 @@ func TestStreamingUsageInclusionWithCosts(t *testing.T) { name: "streaming - forced to include usage", backend: "openai", requestBody: `{"model":"something","messages":[{"role":"system","content":"You are a chatbot."}], "stream": true, "stream_options": {"include_usage": false}}`, - expRequestBody: `{"model":"something","messages":[{"role":"system","content":"You are a chatbot."}], "stream": true,"stream_options":{"include_usage":true}}`, + expRequestBody: `{"model":"something","messages":[{"role":"system","content":"You are a chatbot."}], "stream": true,"stream_options":{"include_usage": true}}`, responseBody: ` {"id":"chatcmpl-foo","object":"chat.completion.chunk","created":1731618222,"model":"gpt-4o-mini-2024-07-18","system_fingerprint":"fp_0ba0d124f1","choices":[{"index":0,"delta":{"role":"assistant","content":"","refusal":null},"logprobs":null,"finish_reason":null}],"usage":null} {"id":"chatcmpl-foo","object":"chat.completion.chunk","created":1731618222,"model":"gpt-4o-mini-2024-07-18","system_fingerprint":"fp_0ba0d124f1","choices":[],"usage":{"prompt_tokens":13,"completion_tokens":12,"total_tokens":25,"prompt_tokens_details":{"cached_tokens":0,"audio_tokens":0},"completion_tokens_details":{"reasoning_tokens":0,"audio_tokens":0,"accepted_prediction_tokens":0,"rejected_prediction_tokens":0}}} @@ -1780,7 +1780,7 @@ data: [DONE] name: "streaming - model override forced to include usage", backend: "modelname-override", requestBody: `{"model":"requested-model","messages":[{"role":"system","content":"You are a chatbot."}], "stream": true, "stream_options": {"include_usage": false}}`, - expRequestBody: `{"model":"override-model","messages":[{"role":"system","content":"You are a chatbot."}], "stream": true,"stream_options":{"include_usage":true}}`, + expRequestBody: `{"model":"override-model","messages":[{"role":"system","content":"You are a chatbot."}], "stream": true,"stream_options":{"include_usage": true}}`, responseBody: ` {"id":"chatcmpl-foo","object":"chat.completion.chunk","created":1731618222,"model":"gpt-4o-mini-2024-07-18","system_fingerprint":"fp_0ba0d124f1","choices":[{"index":0,"delta":{"role":"assistant","content":"","refusal":null},"logprobs":null,"finish_reason":null}],"usage":null} {"id":"chatcmpl-foo","object":"chat.completion.chunk","created":1731618222,"model":"gpt-4o-mini-2024-07-18","system_fingerprint":"fp_0ba0d124f1","choices":[],"usage":{"prompt_tokens":13,"completion_tokens":12,"total_tokens":25,"prompt_tokens_details":{"cached_tokens":0,"audio_tokens":0},"completion_tokens_details":{"reasoning_tokens":0,"audio_tokens":0,"accepted_prediction_tokens":0,"rejected_prediction_tokens":0}}}