Skip to content
Open
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
31 changes: 25 additions & 6 deletions internal/endpointspec/endpointspec.go
Original file line number Diff line number Diff line change
Expand Up @@ -17,6 +17,7 @@ import (
"strconv"
"strings"

"github.com/tidwall/gjson"
"github.com/tidwall/sjson"
"k8s.io/utils/ptr"

Expand Down Expand Up @@ -145,19 +146,37 @@ 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)
}
}
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
Expand Down
23 changes: 23 additions & 0 deletions internal/endpointspec/endpointspec_test.go
Original file line number Diff line number Diff line change
Expand Up @@ -9,6 +9,7 @@ import (
"bytes"
"errors"
"mime/multipart"
"strings"
"testing"

"github.com/stretchr/testify/require"
Expand Down Expand Up @@ -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)
Expand Down
4 changes: 2 additions & 2 deletions tests/data-plane/testupstream_test.go
Original file line number Diff line number Diff line change
Expand Up @@ -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}}}
Expand Down Expand Up @@ -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}}}
Expand Down
Loading