From a6ee0df10216dbaac06594e6197bd444ecd563e3 Mon Sep 17 00:00:00 2001 From: Thushani Jayasekera Date: Sat, 15 Aug 2026 12:55:42 +0530 Subject: [PATCH 1/3] Fix compressed streaming responses at the kernel --- .../internal/kernel/decompression.go | 114 ++++++++- .../internal/kernel/execution_context.go | 106 +++++---- .../kernel/stream_compression_test.go | 224 ++++++++++++++++++ .../internal/kernel/stream_contract_test.go | 207 ++++++++++++++++ .../kernel/stream_provider_formats_test.go | 220 +++++++++++++++++ .../internal/kernel/translator.go | 42 +++- 6 files changed, 855 insertions(+), 58 deletions(-) create mode 100644 gateway/gateway-runtime/policy-engine/internal/kernel/stream_compression_test.go create mode 100644 gateway/gateway-runtime/policy-engine/internal/kernel/stream_contract_test.go create mode 100644 gateway/gateway-runtime/policy-engine/internal/kernel/stream_provider_formats_test.go diff --git a/gateway/gateway-runtime/policy-engine/internal/kernel/decompression.go b/gateway/gateway-runtime/policy-engine/internal/kernel/decompression.go index eba01195d..0740515f1 100644 --- a/gateway/gateway-runtime/policy-engine/internal/kernel/decompression.go +++ b/gateway/gateway-runtime/policy-engine/internal/kernel/decompression.go @@ -344,8 +344,120 @@ func (sd *streamDecompressor) Close() { } } +// ─── Streaming re-compression ──────────────────────────────────────────────── +// +// A streamed response must be re-compressed as ONE compressed stream spanning +// the whole body, not one per chunk. Calling recompressBody per chunk produces +// N independent members: for gzip that is a multi-member stream which most HTTP +// clients (Go's transport, httpx/urllib3, curl) do not read past the first +// member, so the client silently sees a truncated body; for brotli, which has +// no multi-member concatenation at all, the remainder is undecodable. +// +// streamCompressor keeps a single writer alive for the lifetime of the response +// and flushes after each chunk so data still reaches the client incrementally. +type streamCompressor struct { + encoding string + buf bytes.Buffer + gzip *gzip.Writer + brotli *brotli.Writer + closed bool +} + +// newStreamCompressor returns a compressor for the encoding, or nil when the +// encoding needs no re-compression (callers then forward bytes unchanged). +func newStreamCompressor(encoding string) *streamCompressor { + sc := &streamCompressor{encoding: encoding} + switch encoding { + case "gzip": + sc.gzip = gzip.NewWriter(&sc.buf) + case "br": + sc.brotli = brotli.NewWriter(&sc.buf) + default: + return nil + } + return sc +} + +// Compress writes one chunk into the single ongoing compressed stream and +// returns the bytes produced so far. When endOfStream is set the stream is +// finalised (footer/checksum written) and the compressor must not be reused. +// +// A flush is emitted per chunk so the client receives data incrementally; this +// costs a few bytes of framing per chunk versus a single whole-body compress, +// which is the correct trade for a streaming response. +func (sc *streamCompressor) Compress(body []byte, endOfStream bool) ([]byte, error) { + if sc.closed { + return nil, fmt.Errorf("%s stream compressor already closed", sc.encoding) + } + sc.buf.Reset() + + switch { + case sc.gzip != nil: + if len(body) > 0 { + if _, err := sc.gzip.Write(body); err != nil { + return nil, fmt.Errorf("gzip write: %w", err) + } + } + if endOfStream { + if err := sc.gzip.Close(); err != nil { + return nil, fmt.Errorf("gzip close: %w", err) + } + sc.closed = true + } else if err := sc.gzip.Flush(); err != nil { + return nil, fmt.Errorf("gzip flush: %w", err) + } + case sc.brotli != nil: + if len(body) > 0 { + if _, err := sc.brotli.Write(body); err != nil { + return nil, fmt.Errorf("brotli write: %w", err) + } + } + if endOfStream { + if err := sc.brotli.Close(); err != nil { + return nil, fmt.Errorf("brotli close: %w", err) + } + sc.closed = true + } else if err := sc.brotli.Flush(); err != nil { + return nil, fmt.Errorf("brotli flush: %w", err) + } + } + + out := make([]byte, sc.buf.Len()) + copy(out, sc.buf.Bytes()) + return out, nil +} + +// Close finalises the stream on error paths where endOfStream never arrives. +func (sc *streamCompressor) Close() { + if sc.closed { + return + } + sc.closed = true + if sc.gzip != nil { + _ = sc.gzip.Close() + } + if sc.brotli != nil { + _ = sc.brotli.Close() + } +} + +// isRecompressibleEncoding reports whether the kernel can decompress and +// re-compress this Content-Encoding. Anything else must be left untouched — +// see execution_context.go, which refuses to run body policies on it rather +// than handing policies bytes they cannot read. +func isRecompressibleEncoding(encoding string) bool { + switch encoding { + case "gzip", "br": + return true + default: + return false + } +} + // recompressBody re-compresses body bytes using the original Content-Encoding. -// Used to restore compression after policies have processed the decompressed body. +// Used for the BUFFERED response path, where the whole body is compressed in a +// single call. Streaming responses must use streamCompressor instead so the +// response is one compressed stream rather than one per chunk. // Supported encodings: "gzip", "br" (Brotli). Unknown encodings are returned as-is. func recompressBody(body []byte, encoding string) ([]byte, error) { switch encoding { diff --git a/gateway/gateway-runtime/policy-engine/internal/kernel/execution_context.go b/gateway/gateway-runtime/policy-engine/internal/kernel/execution_context.go index fc7e1a18b..9ee2a74ae 100644 --- a/gateway/gateway-runtime/policy-engine/internal/kernel/execution_context.go +++ b/gateway/gateway-runtime/policy-engine/internal/kernel/execution_context.go @@ -137,6 +137,12 @@ type PolicyExecutionContext struct { // responseStreamDecomp performs per-chunk decompression for compressed streaming // response bodies. Nil when the response is not Content-Encoded. responseStreamDecomp *streamDecompressor + // responseStreamComp re-compresses streaming response chunks into a SINGLE + // compressed stream for the whole response. It must outlive individual chunks: + // compressing each chunk independently yields one gzip member (or brotli + // stream) per chunk, and clients stop reading after the first one — the body + // then looks truncated to the client even though every byte was sent. + responseStreamComp *streamCompressor // streamTerminated is set when a policy returns TerminateStream=true. Any // subsequent upstream chunks that Envoy delivers after EndOfStream was sent // downstream are silently suppressed — forwarding more data would be undefined. @@ -201,6 +207,10 @@ func (ec *PolicyExecutionContext) closeStreamDecompressors() { ec.responseStreamDecomp.Close() ec.responseStreamDecomp = nil } + if ec.responseStreamComp != nil { + ec.responseStreamComp.Close() + ec.responseStreamComp = nil + } } // handlePolicyError creates a generic error response for policy execution failures. @@ -1042,9 +1052,16 @@ func (ec *PolicyExecutionContext) processStreamingResponseBody( EndOfStream: body.EndOfStream, } - // Compressed response: decompress this chunk, pass directly to policies, - // recompress the output. No kernel accumulation — policy implementations - // handle their own internal state across chunks. + // Decompression is a transform applied BEFORE the shared accumulation logic + // below, not a separate processing path. Policies must observe exactly the same + // contract — same accumulation, same NeedsMoreResponseData consultation — + // whether or not the upstream compressed the response. When this was a second + // path that fed chunks straight to policies, NeedsMoreResponseData was never + // called on a compressed response, so any policy relying on it (the documented + // SDK hook for cross-chunk buffering) worked on plaintext and silently degraded + // on gzip: word-count/sentence-count guardrails evaluated isolated fragments + // instead of assembled content, and content-rewriting policies never saw a + // placeholder that straddled a chunk boundary. if ec.responseContentEncoding != "" { if ec.responseStreamDecomp == nil { ec.responseStreamDecomp = newStreamDecompressor(ec.responseContentEncoding, ec.server.maxResponseDecompressedBytes) @@ -1064,64 +1081,44 @@ func (ec *PolicyExecutionContext) processStreamingResponseBody( return nil, ec.responsePayloadTooLargeError(ctx, err, "response_body_streaming") } return nil, fmt.Errorf("streaming upstream response decompression failed: %w", err) - } else { - chunk.Chunk = decompressed - } - - // Suppress empty intermediate chunks — the decoder needed more input to - // produce a full block. The client already expects compressed data so - // sending nothing is correct here. - if len(chunk.Chunk) == 0 && !chunk.EndOfStream { - return &extprocv3.ProcessingResponse{ - Response: &extprocv3.ProcessingResponse_ResponseBody{ - ResponseBody: &extprocv3.BodyResponse{ - Response: &extprocv3.CommonResponse{ - BodyMutation: &extprocv3.BodyMutation{ - Mutation: &extprocv3.BodyMutation_StreamedResponse{ - StreamedResponse: &extprocv3.StreamedBodyResponse{}, - }, - }, - }, - }, - }, - }, nil } + chunk.Chunk = decompressed slog.Debug("[streaming] response chunk decompressed", "route", ec.routeKey, "decompressed_bytes", len(chunk.Chunk), "end_of_stream", chunk.EndOfStream, ) - - execResult, err := ec.server.executor.ExecuteStreamingResponsePolicies( - ctx, - ec.policyChain.Policies, - ec.responseStreamContext, - chunk, - ec.policyChain.PolicySpecs, - ec.sharedCtx.APIName, - ec.routeKey, - ec.policyChain.HasExecutionConditions, - ) - if err != nil { - return ec.handlePolicyError(ctx, err, "response_body_streaming"), nil - } - if execResult.StreamTerminated { - ec.streamTerminated = true - } - return TranslateStreamingResponseChunkAction(execResult, chunk, ec) } - // Uncompressed (SSE / plain chunked): use the existing accumulation path so - // policies that need multiple chunks (e.g. waiting for a full SSE event) still work. if len(chunk.Chunk) > 0 { ec.streamAccumulator = append(ec.streamAccumulator, chunk.Chunk...) } + // Nothing to hand policies yet. For a compressed stream this is the common + // case: the decoder needs more input before it can emit a block. Forwarding an + // empty streamed response keeps Envoy's chunk accounting intact. + if len(ec.streamAccumulator) == 0 && !chunk.EndOfStream { + return &extprocv3.ProcessingResponse{ + Response: &extprocv3.ProcessingResponse_ResponseBody{ + ResponseBody: &extprocv3.BodyResponse{ + Response: &extprocv3.CommonResponse{ + BodyMutation: &extprocv3.BodyMutation{ + Mutation: &extprocv3.BodyMutation_StreamedResponse{ + StreamedResponse: &extprocv3.StreamedBodyResponse{}, + }, + }, + }, + }, + }, + }, nil + } + slog.Debug("[streaming] response chunk received", "route", ec.routeKey, "chunk_bytes", len(chunk.Chunk), "accumulated_bytes", len(ec.streamAccumulator), + "encoding", ec.responseContentEncoding, "end_of_stream", chunk.EndOfStream, ) @@ -1348,7 +1345,26 @@ func (ec *PolicyExecutionContext) buildResponseContexts(headers *extprocv3.HttpH ) } case "content-encoding": - ec.responseContentEncoding = value + // Only encodings the kernel can actually decompress AND re-compress + // are recorded. For anything else (deflate, zstd, …) the decompressor + // would fall through to a passthrough reader, handing policies raw + // compressed bytes: content-rewriting policies then silently match + // nothing — e.g. pii-masking-regex would deliver "[EMAIL_0000]" to the + // client instead of restoring it — with no error anywhere. Leaving this + // empty keeps the body untouched end to end, which is the safe outcome. + // + // Content codings are case-insensitive tokens (RFC 9110 §8.4.1), so + // normalise before matching — the decompressor/compressor switches are + // lowercase-only and would otherwise miss a "GZIP" response. + encoding := strings.ToLower(strings.TrimSpace(value)) + if isRecompressibleEncoding(encoding) { + ec.responseContentEncoding = encoding + } else if encoding != "" && encoding != "identity" { + slog.Warn("unsupported response Content-Encoding; body policies will not inspect or modify this response", + "request_id", ec.requestID, + "encoding", value, + ) + } } } } diff --git a/gateway/gateway-runtime/policy-engine/internal/kernel/stream_compression_test.go b/gateway/gateway-runtime/policy-engine/internal/kernel/stream_compression_test.go new file mode 100644 index 000000000..3ca5d9249 --- /dev/null +++ b/gateway/gateway-runtime/policy-engine/internal/kernel/stream_compression_test.go @@ -0,0 +1,224 @@ +/* + * Copyright (c) 2026, WSO2 LLC. (http://www.wso2.org) All Rights Reserved. + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + * + */ + +package kernel + +import ( + "bytes" + "compress/gzip" + "io" + "strings" + "testing" + + "github.com/andybalholm/brotli" + + "github.com/wso2/api-platform/gateway/gateway-runtime/policy-engine/internal/executor" + policy "github.com/wso2/api-platform/sdk/core/policy/v1alpha2" +) + +// singlePassGunzip decodes exactly like a normal HTTP client: one reader over +// the whole payload, stopping at the end of the first gzip member. This is what +// httpx/urllib3, curl and Go's own transport do — so it is the invariant that +// actually matters, not whether the bytes are on the wire somewhere. +func singlePassGunzip(t *testing.T, wire []byte) []byte { + t.Helper() + zr, err := gzip.NewReader(bytes.NewReader(wire)) + if err != nil { + t.Fatalf("gzip.NewReader: %v", err) + } + defer zr.Close() + // Multistream(false) makes the reader stop at the first member, mirroring + // clients that do not continue into subsequent members. + zr.Multistream(false) + out, err := io.ReadAll(zr) + if err != nil { + t.Fatalf("gunzip: %v", err) + } + return out +} + +// A streamed gzip response must be readable in full by a client that reads only +// the first gzip member. Regression test for the truncation reported against +// pii-masking-regex, which was really per-chunk re-compression in the kernel. +func TestStreamCompressor_GzipIsOneMemberAcrossChunks(t *testing.T) { + chunks := []string{ + `{"id":"chatcmpl-1",`, + `"choices":[{"message":`, + `{"content":"hello there"}}],`, + `"usage":{"total_tokens":30}}`, + } + want := strings.Join(chunks, "") + + sc := newStreamCompressor("gzip") + if sc == nil { + t.Fatal("expected a gzip stream compressor") + } + + var wire bytes.Buffer + for i, c := range chunks { + out, err := sc.Compress([]byte(c), i == len(chunks)-1) + if err != nil { + t.Fatalf("chunk %d: %v", i, err) + } + wire.Write(out) + } + + if got := string(singlePassGunzip(t, wire.Bytes())); got != want { + t.Errorf("single-member decode mismatch\n got: %q\nwant: %q", got, want) + } + + // There must be exactly one gzip header in the whole response. + if n := bytes.Count(wire.Bytes(), []byte{0x1f, 0x8b}); n != 1 { + t.Errorf("expected exactly 1 gzip member, found %d gzip headers", n) + } +} + +// Brotli has no multi-member concatenation, so a per-chunk writer made the tail +// of the response permanently undecodable. One stream must decode fully. +func TestStreamCompressor_BrotliIsOneStreamAcrossChunks(t *testing.T) { + chunks := []string{"alpha ", "beta ", "gamma ", "delta"} + want := strings.Join(chunks, "") + + sc := newStreamCompressor("br") + if sc == nil { + t.Fatal("expected a brotli stream compressor") + } + + var wire bytes.Buffer + for i, c := range chunks { + out, err := sc.Compress([]byte(c), i == len(chunks)-1) + if err != nil { + t.Fatalf("chunk %d: %v", i, err) + } + wire.Write(out) + } + + got, err := io.ReadAll(brotli.NewReader(bytes.NewReader(wire.Bytes()))) + if err != nil { + t.Fatalf("brotli decode: %v", err) + } + if string(got) != want { + t.Errorf("brotli decode mismatch\n got: %q\nwant: %q", got, want) + } +} + +// Chunks that produce no output (a policy suppressing a chunk) must not break +// the stream, and must not emit a standalone empty member. +func TestStreamCompressor_EmptyChunksDoNotBreakStream(t *testing.T) { + sc := newStreamCompressor("gzip") + var wire bytes.Buffer + + for _, c := range []string{"", "", "payload", ""} { + out, err := sc.Compress([]byte(c), false) + if err != nil { + t.Fatalf("compress: %v", err) + } + wire.Write(out) + } + final, err := sc.Compress(nil, true) + if err != nil { + t.Fatalf("final compress: %v", err) + } + wire.Write(final) + + if got := string(singlePassGunzip(t, wire.Bytes())); got != "payload" { + t.Errorf("got %q, want %q", got, "payload") + } + if n := bytes.Count(wire.Bytes(), []byte{0x1f, 0x8b}); n != 1 { + t.Errorf("expected exactly 1 gzip member, found %d", n) + } +} + +// Data must reach the client incrementally: a non-final chunk has to produce +// output rather than sitting in the compressor until end of stream. +func TestStreamCompressor_FlushesPerChunk(t *testing.T) { + sc := newStreamCompressor("gzip") + out, err := sc.Compress([]byte(strings.Repeat("streaming payload ", 16)), false) + if err != nil { + t.Fatalf("compress: %v", err) + } + if len(out) == 0 { + t.Fatal("non-final chunk produced no output; response would not stream incrementally") + } +} + +// Encodings the kernel cannot round-trip must yield no compressor, so callers +// forward the body untouched instead of corrupting it. +func TestStreamCompressor_UnsupportedEncodings(t *testing.T) { + for _, enc := range []string{"deflate", "zstd", "identity", "", "GZIP"} { + if sc := newStreamCompressor(enc); sc != nil { + t.Errorf("newStreamCompressor(%q) returned a compressor; want nil", enc) + } + if isRecompressibleEncoding(enc) { + t.Errorf("isRecompressibleEncoding(%q) = true; want false", enc) + } + } + for _, enc := range []string{"gzip", "br"} { + if !isRecompressibleEncoding(enc) { + t.Errorf("isRecompressibleEncoding(%q) = false; want true", enc) + } + } +} + +// A policy terminating the stream early (guardrail intervention) is an end of +// stream for the compressor too. Finalising only on the Envoy chunk's own +// EndOfStream flag would emit a gzip stream with no footer, which decodes as a +// truncated body — the same client-visible symptom this change exists to fix. +func TestTranslateStreamingResponseChunkAction_TerminatedStreamIsFinalised(t *testing.T) { + execCtx := &PolicyExecutionContext{ + responseContentEncoding: "gzip", + analyticsMetadata: map[string]any{}, + dynamicMetadata: map[string]map[string]interface{}{}, + } + + var wire bytes.Buffer + chunks := []struct { + body string + terminated bool + }{ + {body: `{"choices":[{"delta":`, terminated: false}, + {body: `{"content":"blocked"}}]}`, terminated: true}, // policy stops the stream here + } + + for i, c := range chunks { + resp, err := TranslateStreamingResponseChunkAction( + &executor.StreamingResponseExecutionResult{StreamTerminated: c.terminated}, + &policy.StreamBody{Chunk: []byte(c.body), EndOfStream: false}, // Envoy never signals EOS + execCtx, + ) + if err != nil { + t.Fatalf("chunk %d: %v", i, err) + } + wire.Write(resp.GetResponseBody().GetResponse().GetBodyMutation().GetStreamedResponse().GetBody()) + } + + want := chunks[0].body + chunks[1].body + if got := string(singlePassGunzip(t, wire.Bytes())); got != want { + t.Errorf("terminated stream did not decode fully\n got: %q\nwant: %q", got, want) + } +} + +// Using a finalised compressor is a programming error, not silent corruption. +func TestStreamCompressor_RejectsUseAfterClose(t *testing.T) { + sc := newStreamCompressor("gzip") + if _, err := sc.Compress([]byte("done"), true); err != nil { + t.Fatalf("final compress: %v", err) + } + if _, err := sc.Compress([]byte("more"), false); err == nil { + t.Fatal("expected an error when compressing after end of stream") + } +} diff --git a/gateway/gateway-runtime/policy-engine/internal/kernel/stream_contract_test.go b/gateway/gateway-runtime/policy-engine/internal/kernel/stream_contract_test.go new file mode 100644 index 000000000..68c8f93fa --- /dev/null +++ b/gateway/gateway-runtime/policy-engine/internal/kernel/stream_contract_test.go @@ -0,0 +1,207 @@ +/* + * Copyright (c) 2026, WSO2 LLC. (http://www.wso2.org) All Rights Reserved. + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + * + */ + +package kernel + +import ( + "context" + "strings" + "testing" + + corev3 "github.com/envoyproxy/go-control-plane/envoy/config/core/v3" + extprocv3 "github.com/envoyproxy/go-control-plane/envoy/service/ext_proc/v3" + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" + + "github.com/wso2/api-platform/gateway/gateway-runtime/policy-engine/internal/config" + "github.com/wso2/api-platform/gateway/gateway-runtime/policy-engine/internal/registry" + policy "github.com/wso2/api-platform/sdk/core/policy/v1alpha2" +) + +// recordingStreamPolicy is a minimal StreamingResponsePolicy that records what the +// kernel actually hands it. It buffers until it has seen the sentinel, mirroring +// how word-count/sentence-count guardrails and content-rewriting policies use +// NeedsMoreResponseData to assemble content spanning several chunks. +type recordingStreamPolicy struct { + holdUntil string // keep buffering until the accumulated bytes contain this + + needsMoreCalls int // how many times the kernel consulted the hook + chunksSeen []string // the body of every chunk delivered to the policy +} + +func (p *recordingStreamPolicy) Mode() policy.ProcessingMode { + return policy.ProcessingMode{ResponseBodyMode: policy.BodyModeStream} +} + +// OnResponseBody satisfies the buffered fallback embedded in StreamingResponsePolicy; +// these tests only exercise the streaming path. +func (p *recordingStreamPolicy) OnResponseBody(_ context.Context, _ *policy.ResponseContext, _ map[string]interface{}) policy.ResponseAction { + return policy.DownstreamResponseModifications{} +} + +func (p *recordingStreamPolicy) NeedsMoreResponseData(accumulated []byte) bool { + p.needsMoreCalls++ + if p.holdUntil == "" { + return false // never asks the kernel to buffer + } + return !strings.Contains(string(accumulated), p.holdUntil) +} + +func (p *recordingStreamPolicy) OnResponseBodyChunk(_ context.Context, _ *policy.ResponseStreamContext, chunk *policy.StreamBody, _ map[string]interface{}) policy.StreamingResponseAction { + p.chunksSeen = append(p.chunksSeen, string(chunk.Chunk)) + return policy.ForwardResponseChunk{} +} + +func newStreamingExecCtx(t *testing.T, pol policy.Policy, contentEncoding string) *PolicyExecutionContext { + t.Helper() + + kernel := NewKernel() + server := NewExternalProcessorServer(kernel, newTestExecutor(), config.TracingConfig{}, "", testMaxDecompressedBytes, testMaxDecompressedBytes) + chain := ®istry.PolicyChain{ + RequiresResponseBody: true, + SupportsResponseStreaming: true, + Policies: []policy.Policy{pol}, + PolicySpecs: []policy.PolicySpec{{Enabled: true}}, + } + execCtx := newPolicyExecutionContext(server, "test-route", chain) + execCtx.buildRequestContexts(&extprocv3.HttpHeaders{Headers: &corev3.HeaderMap{}}, RouteMetadata{}) + + respHeaders := []*corev3.HeaderValue{ + {Key: ":status", RawValue: []byte("200")}, + {Key: "content-type", RawValue: []byte("text/event-stream")}, + } + if contentEncoding != "" { + respHeaders = append(respHeaders, &corev3.HeaderValue{ + Key: "content-encoding", RawValue: []byte(contentEncoding), + }) + } + execCtx.buildResponseContexts(&extprocv3.HttpHeaders{ + Headers: &corev3.HeaderMap{Headers: respHeaders}, + }) + return execCtx +} + +// THE contract test. A streaming policy must observe identical behaviour whether +// or not the upstream compressed the response. Before the streaming paths were +// unified, the compressed branch fed chunks straight to policies and never called +// NeedsMoreResponseData at all — so a guardrail that assembles content across +// chunks (word-count, sentence-count) silently evaluated isolated fragments as +// soon as the backend enabled gzip, with no error anywhere. +func TestStreamingResponse_PolicyContractIsIdenticalAcrossEncodings(t *testing.T) { + // Split so the sentinel "END" only exists once the last chunk has arrived — + // no single chunk contains it. + bodyChunks := []string{`data: {"delta":"he`, `llo"}`, "\n\ndata: E", "ND\n\n"} + wholeBody := "" + for _, c := range bodyChunks { + wholeBody += c + } + + for _, encoding := range []string{"", "gzip", "br"} { + name := encoding + if name == "" { + name = "plaintext" + } + t.Run(name, func(t *testing.T) { + pol := &recordingStreamPolicy{holdUntil: "END"} + execCtx := newStreamingExecCtx(t, pol, encoding) + + // Frame the wire the way a real streaming upstream does: one compressed + // stream for the whole response, flushed after each logical chunk so the + // data actually reaches the client incrementally. + wireChunks := encodeStreamChunks(t, bodyChunks, encoding) + + for i, wc := range wireChunks { + _, err := execCtx.processStreamingResponseBody(context.Background(), &extprocv3.HttpBody{ + Body: wc, + EndOfStream: i == len(wireChunks)-1, + }) + require.NoError(t, err) + } + + // The hook must be consulted on every encoding, not just plaintext. + assert.Greater(t, pol.needsMoreCalls, 0, + "NeedsMoreResponseData was never called — policy cannot buffer across chunks on this encoding") + + // And the policy must ultimately receive the complete body, assembled. + joined := "" + for _, c := range pol.chunksSeen { + joined += c + } + assert.Equal(t, wholeBody, joined, + "policy did not receive the full decompressed body") + assert.Contains(t, pol.chunksSeen[len(pol.chunksSeen)-1], "END", + "the buffered content was not released to the policy in one piece") + }) + } +} + +// encodeStreamChunks frames logical chunks the way a streaming upstream does: +// a single compressed stream flushed after every chunk, yielding one wire chunk +// per logical chunk. For plaintext the chunks pass through unchanged. +func encodeStreamChunks(t *testing.T, chunks []string, encoding string) [][]byte { + t.Helper() + if encoding == "" { + out := make([][]byte, len(chunks)) + for i, c := range chunks { + out[i] = []byte(c) + } + return out + } + sc := newStreamCompressor(encoding) + require.NotNil(t, sc, "no stream compressor for encoding %q", encoding) + out := make([][]byte, 0, len(chunks)) + for i, c := range chunks { + b, err := sc.Compress([]byte(c), i == len(chunks)-1) + require.NoError(t, err) + out = append(out, b) + } + return out +} + +// A policy that never wants buffering must still stream incrementally on a +// compressed response — the unified path must not turn every gzip response into +// a single end-of-stream flush. +func TestStreamingResponse_NoBufferingPolicyStillStreamsIncrementally(t *testing.T) { + // An empty sentinel makes NeedsMoreResponseData always false — the policy never + // asks the kernel to buffer. + pol := &recordingStreamPolicy{holdUntil: ""} + execCtx := newStreamingExecCtx(t, pol, "gzip") + + events := []string{"data: one\n\n", "data: two\n\n", "data: three\n\n"} + body := "" + for _, e := range events { + body += e + } + wireChunks := encodeStreamChunks(t, events, "gzip") + + for i, wc := range wireChunks { + _, err := execCtx.processStreamingResponseBody(context.Background(), &extprocv3.HttpBody{ + Body: wc, + EndOfStream: i == len(wireChunks)-1, + }) + require.NoError(t, err) + } + + assert.Greater(t, len(pol.chunksSeen), 1, + "a non-buffering policy received a single end-of-stream flush; the response did not stream") + + joined := "" + for _, c := range pol.chunksSeen { + joined += c + } + assert.Equal(t, body, joined) +} diff --git a/gateway/gateway-runtime/policy-engine/internal/kernel/stream_provider_formats_test.go b/gateway/gateway-runtime/policy-engine/internal/kernel/stream_provider_formats_test.go new file mode 100644 index 000000000..8f34af927 --- /dev/null +++ b/gateway/gateway-runtime/policy-engine/internal/kernel/stream_provider_formats_test.go @@ -0,0 +1,220 @@ +/* + * Copyright (c) 2026, WSO2 LLC. (http://www.wso2.org) All Rights Reserved. + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + * + */ + +package kernel + +import ( + "bytes" + "compress/gzip" + "context" + "io" + "testing" + + "github.com/andybalholm/brotli" + extprocv3 "github.com/envoyproxy/go-control-plane/envoy/service/ext_proc/v3" + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" + + policy "github.com/wso2/api-platform/sdk/core/policy/v1alpha2" +) + +// The kernel is deliberately body-format agnostic: it frames, decompresses, +// hands bytes to policies, and re-compresses. These fixtures are the real wire +// shapes of the two providers most used through the gateway, kept here so a +// regression in framing shows up as a provider-shaped failure rather than an +// abstract byte mismatch. Anthropic is the important second case because it uses +// `event:` lines alongside `data:` and a different delta shape +// (content_block_delta/delta.text vs choices[].delta.content). + +// openAIStreamEvents is an OpenAI /v1/chat/completions stream (stream: true). +var openAIStreamEvents = []string{ + `data: {"id":"chatcmpl-BxYz1","object":"chat.completion.chunk","created":1755172331,"model":"gpt-4o-mini","choices":[{"index":0,"delta":{"role":"assistant","content":""},"finish_reason":null}]}` + "\n\n", + `data: {"id":"chatcmpl-BxYz1","object":"chat.completion.chunk","created":1755172331,"model":"gpt-4o-mini","choices":[{"index":0,"delta":{"content":"Confirmation sent to "},"finish_reason":null}]}` + "\n\n", + `data: {"id":"chatcmpl-BxYz1","object":"chat.completion.chunk","created":1755172331,"model":"gpt-4o-mini","choices":[{"index":0,"delta":{"content":"john.doe@example.com"},"finish_reason":null}]}` + "\n\n", + `data: {"id":"chatcmpl-BxYz1","object":"chat.completion.chunk","created":1755172331,"model":"gpt-4o-mini","choices":[{"index":0,"delta":{},"finish_reason":"stop"}]}` + "\n\n", + "data: [DONE]\n\n", +} + +// anthropicStreamEvents is an Anthropic /v1/messages stream (stream: true). +// Note the `event:` lines and the heartbeat comment, both of which must survive +// framing untouched. +var anthropicStreamEvents = []string{ + "event: message_start\n" + `data: {"type":"message_start","message":{"id":"msg_01Xy","type":"message","role":"assistant","model":"claude-sonnet-4-5","content":[],"usage":{"input_tokens":24,"output_tokens":1}}}` + "\n\n", + "event: content_block_start\n" + `data: {"type":"content_block_start","index":0,"content_block":{"type":"text","text":""}}` + "\n\n", + ": heartbeat\n\n", + "event: content_block_delta\n" + `data: {"type":"content_block_delta","index":0,"delta":{"type":"text_delta","text":"Confirmation sent to "}}` + "\n\n", + "event: content_block_delta\n" + `data: {"type":"content_block_delta","index":0,"delta":{"type":"text_delta","text":"john.doe@example.com"}}` + "\n\n", + "event: content_block_stop\n" + `data: {"type":"content_block_stop","index":0}` + "\n\n", + "event: message_delta\n" + `data: {"type":"message_delta","delta":{"stop_reason":"end_turn"},"usage":{"output_tokens":18}}` + "\n\n", + "event: message_stop\n" + `data: {"type":"message_stop"}` + "\n\n", +} + +// openAIBufferedBody is a non-streaming OpenAI response delivered chunked. +var openAIBufferedBody = `{"id":"chatcmpl-BxYz2","object":"chat.completion","created":1755172400,"model":"gpt-4o-mini","choices":[{"index":0,"message":{"role":"assistant","content":"Confirmation sent to john.doe@example.com. Call +1-555-0100 if anything changes."},"finish_reason":"stop"}],"usage":{"prompt_tokens":31,"completion_tokens":24,"total_tokens":55}}` + +// anthropicBufferedBody is a non-streaming Anthropic response delivered chunked. +var anthropicBufferedBody = `{"id":"msg_01Xz","type":"message","role":"assistant","model":"claude-sonnet-4-5","content":[{"type":"text","text":"Confirmation sent to john.doe@example.com. Call +1-555-0100 if anything changes."}],"stop_reason":"end_turn","usage":{"input_tokens":31,"output_tokens":24}}` + +func decodeWire(t *testing.T, wire []byte, encoding string) []byte { + t.Helper() + switch encoding { + case "": + return wire + case "gzip": + // Single-pass, first-member-only — exactly what httpx/urllib3/curl/Go do. + zr, err := gzip.NewReader(bytes.NewReader(wire)) + require.NoError(t, err, "client could not start decoding the gzip response") + defer zr.Close() + zr.Multistream(false) + out, err := io.ReadAll(zr) + require.NoError(t, err, "client could not decode the gzip response to completion") + return out + case "br": + out, err := io.ReadAll(brotli.NewReader(bytes.NewReader(wire))) + require.NoError(t, err, "client could not decode the brotli response") + return out + } + t.Fatalf("unknown encoding %q", encoding) + return nil +} + +// End-to-end: feed a provider-shaped response through the full kernel streaming +// path and assert the bytes a real client would reconstruct are byte-identical +// to what the upstream sent. This is the invariant the reported incident broke — +// the client received a 200 with a body it could not parse. +func TestStreamingResponse_ProviderFormatsRoundTripByteExact(t *testing.T) { + for _, provider := range []struct { + name string + events []string + contentType string + }{ + {name: "openai-sse", events: openAIStreamEvents, contentType: "text/event-stream"}, + {name: "anthropic-sse", events: anthropicStreamEvents, contentType: "text/event-stream"}, + {name: "openai-buffered-chunked", events: splitString(openAIBufferedBody, 37), contentType: "application/json"}, + {name: "anthropic-buffered-chunked", events: splitString(anthropicBufferedBody, 37), contentType: "application/json"}, + } { + for _, encoding := range []string{"", "gzip", "br"} { + encName := encoding + if encName == "" { + encName = "plaintext" + } + t.Run(provider.name+"/"+encName, func(t *testing.T) { + want := "" + for _, e := range provider.events { + want += e + } + + // A pass-through policy: exercises the full pipeline (decompress → + // policy → re-compress) without mutating bytes, so any difference is + // the kernel's doing. + pol := &recordingStreamPolicy{holdUntil: ""} + execCtx := newStreamingExecCtx(t, pol, encoding) + execCtx.responseStreamContext.ResponseHeaders = policy.NewHeaders( + map[string][]string{"content-type": {provider.contentType}}, + ) + + wireIn := encodeStreamChunks(t, provider.events, encoding) + + var wireOut bytes.Buffer + for i, wc := range wireIn { + resp, err := execCtx.processStreamingResponseBody(context.Background(), &extprocv3.HttpBody{ + Body: wc, + EndOfStream: i == len(wireIn)-1, + }) + require.NoError(t, err) + wireOut.Write(resp.GetResponseBody().GetResponse().GetBodyMutation().GetStreamedResponse().GetBody()) + } + + got := decodeWire(t, wireOut.Bytes(), encoding) + assert.Equal(t, want, string(got), + "the body a client reconstructs differs from what the upstream sent") + + if encoding == "gzip" { + assert.Equal(t, 1, bytes.Count(wireOut.Bytes(), []byte{0x1f, 0x8b}), + "response must be exactly one gzip member; more than one truncates the body for real clients") + } + + // The policy must have seen the same bytes, fully decompressed. + joined := "" + for _, c := range pol.chunksSeen { + joined += c + } + assert.Equal(t, want, joined, "policy did not observe the full decompressed body") + }) + } + } +} + +// A guardrail-style policy that assembles content across events must work +// identically for both providers on a compressed stream. Before the streaming +// paths were unified this held only for plaintext. +func TestStreamingResponse_CrossEventAssemblyWorksForBothProviders(t *testing.T) { + for _, provider := range []struct { + name string + events []string + }{ + {name: "openai", events: openAIStreamEvents}, + {name: "anthropic", events: anthropicStreamEvents}, + } { + for _, encoding := range []string{"", "gzip", "br"} { + encName := encoding + if encName == "" { + encName = "plaintext" + } + t.Run(provider.name+"/"+encName, func(t *testing.T) { + // Hold until the address is fully assembled — it only exists once the + // event carrying it has arrived, which is never the first event. + pol := &recordingStreamPolicy{holdUntil: "john.doe@example.com"} + execCtx := newStreamingExecCtx(t, pol, encoding) + + wireIn := encodeStreamChunks(t, provider.events, encoding) + for i, wc := range wireIn { + _, err := execCtx.processStreamingResponseBody(context.Background(), &extprocv3.HttpBody{ + Body: wc, + EndOfStream: i == len(wireIn)-1, + }) + require.NoError(t, err) + } + + assert.Greater(t, pol.needsMoreCalls, 0, + "NeedsMoreResponseData never consulted — a guardrail could not assemble content on this encoding") + + var holding string + for _, c := range pol.chunksSeen { + if bytes.Contains([]byte(c), []byte("john.doe@example.com")) { + holding = c + break + } + } + require.NotEmpty(t, holding, + "the assembled address was never delivered to the policy in one piece") + }) + } + } +} + +func splitString(s string, n int) []string { + var out []string + for len(s) > n { + out = append(out, s[:n]) + s = s[n:] + } + if len(s) > 0 { + out = append(out, s) + } + return out +} diff --git a/gateway/gateway-runtime/policy-engine/internal/kernel/translator.go b/gateway/gateway-runtime/policy-engine/internal/kernel/translator.go index c63f1277e..f41d22c48 100644 --- a/gateway/gateway-runtime/policy-engine/internal/kernel/translator.go +++ b/gateway/gateway-runtime/policy-engine/internal/kernel/translator.go @@ -1595,21 +1595,46 @@ func TranslateStreamingResponseChunkAction(result *executor.StreamingResponseExe outputBody = originalChunk.Chunk } + // If a policy terminated the stream early (e.g. guardrail intervention), force + // EndOfStream so Envoy closes the connection cleanly after delivering the final + // chunk. This is also the end of stream for the compressor below: finalising it + // on originalChunk.EndOfStream alone would send a terminated gzip/brotli stream + // without its footer, which the client reads as a truncated body. + endOfStream := originalChunk.EndOfStream || result.StreamTerminated + if result.StreamTerminated { + slog.Info("[streaming] stream terminated by policy; forcing EndOfStream on final chunk") + } + // Re-compress the output if the original response was Content-Encoded. // Response headers (including Content-Encoding) are already committed downstream // in streaming mode and cannot be changed — the body must match the encoding // the client expects. + // + // The whole response must form ONE compressed stream, so the compressor is + // held on the execution context across chunks and finalised at end of stream. + // Compressing each chunk independently would emit one gzip member (or brotli + // stream) per chunk; clients stop decoding after the first, so the client sees + // a truncated body even though every byte was transmitted. if execCtx.responseContentEncoding != "" { - recompressed, err := recompressBody(outputBody, execCtx.responseContentEncoding) + if execCtx.responseStreamComp == nil { + // Never nil: responseContentEncoding is only ever set to an encoding + // isRecompressibleEncoding accepts (see buildResponseContexts). + execCtx.responseStreamComp = newStreamCompressor(execCtx.responseContentEncoding) + } + recompressed, err := execCtx.responseStreamComp.Compress(outputBody, endOfStream) if err != nil { - slog.Warn("[streaming] failed to re-compress response body; sending uncompressed — Content-Encoding mismatch", + // The client is mid-stream with a committed Content-Encoding header, so + // falling back to plaintext here would corrupt the response. Fail the + // stream instead and let Envoy reset it. + slog.Error("[streaming] failed to re-compress response chunk; failing stream", "encoding", execCtx.responseContentEncoding, "error", err, ) - execCtx.responseContentEncoding = "" - } else { - outputBody = recompressed + execCtx.responseStreamComp.Close() + execCtx.responseStreamComp = nil + return nil, fmt.Errorf("streaming response re-compression failed: %w", err) } + outputBody = recompressed } analyticsData := make(map[string]any) @@ -1644,13 +1669,6 @@ func TranslateStreamingResponseChunkAction(result *executor.StreamingResponseExe mergeDynamicMetadata(execCtx.dynamicMetadata, dm) } - // If a policy terminated the stream early (e.g. guardrail intervention), force - // EndOfStream so Envoy closes the connection cleanly after delivering the final chunk. - endOfStream := originalChunk.EndOfStream || result.StreamTerminated - if result.StreamTerminated { - slog.Info("[streaming] stream terminated by policy; forcing EndOfStream on final chunk") - } - resp := &extprocv3.ProcessingResponse{ Response: &extprocv3.ProcessingResponse_ResponseBody{ ResponseBody: &extprocv3.BodyResponse{ From 67730dcad8bb40211fe600965b48ab21d54d5678 Mon Sep 17 00:00:00 2001 From: Thushani Jayasekera Date: Sat, 15 Aug 2026 19:14:03 +0530 Subject: [PATCH 2/3] Add support for additional content encodings in the policy engine --- gateway/gateway-runtime/policy-engine/go.mod | 2 +- .../internal/constants/constants.go | 1 + .../internal/kernel/decompression.go | 247 ++++++++---- .../internal/kernel/execution_context.go | 294 ++++++++++++-- .../internal/kernel/execution_context_test.go | 362 ++++++++++++++++++ .../kernel/stream_compression_test.go | 136 ++++++- .../internal/kernel/translator.go | 50 ++- 7 files changed, 977 insertions(+), 115 deletions(-) diff --git a/gateway/gateway-runtime/policy-engine/go.mod b/gateway/gateway-runtime/policy-engine/go.mod index e4914e8ec..66a6f7b2f 100644 --- a/gateway/gateway-runtime/policy-engine/go.mod +++ b/gateway/gateway-runtime/policy-engine/go.mod @@ -8,6 +8,7 @@ require ( github.com/go-viper/mapstructure/v2 v2.5.0 github.com/google/cel-go v0.26.1 github.com/google/uuid v1.6.0 + github.com/klauspost/compress v1.18.6 github.com/knadh/koanf/parsers/toml/v2 v2.2.0 github.com/knadh/koanf/providers/confmap v1.0.0 github.com/knadh/koanf/providers/file v1.2.1 @@ -40,7 +41,6 @@ require ( github.com/go-logr/logr v1.4.3 // indirect github.com/go-logr/stdr v1.2.2 // indirect github.com/grpc-ecosystem/grpc-gateway/v2 v2.29.0 // indirect - github.com/klauspost/compress v1.18.6 // indirect github.com/knadh/koanf/maps v0.1.2 // indirect github.com/mitchellh/copystructure v1.2.0 // indirect github.com/mitchellh/reflectwalk v1.0.2 // indirect diff --git a/gateway/gateway-runtime/policy-engine/internal/constants/constants.go b/gateway/gateway-runtime/policy-engine/internal/constants/constants.go index 9d893ab9c..16839746c 100644 --- a/gateway/gateway-runtime/policy-engine/internal/constants/constants.go +++ b/gateway/gateway-runtime/policy-engine/internal/constants/constants.go @@ -105,6 +105,7 @@ const ( TerminalReasonNoPolicyChain = "no_policy_chain" // route resolved but no chain registered TerminalReasonUnknownMessageType = "unknown_message_type" // unrecognised ext_proc message TerminalReasonProcessingFailed = "processing_failed" // a phase returned a fatal (stream-ending) error with no ImmediateResponse to classify + TerminalReasonUnsupportedEncoding = "unsupported_encoding" // Content-Encoding the kernel cannot round-trip, on a body the policy chain requires // Analytics metadata and property keys shared across packages. GuardrailHitMetadataKey = "isGuardrailHit" diff --git a/gateway/gateway-runtime/policy-engine/internal/kernel/decompression.go b/gateway/gateway-runtime/policy-engine/internal/kernel/decompression.go index 0740515f1..bbf969a66 100644 --- a/gateway/gateway-runtime/policy-engine/internal/kernel/decompression.go +++ b/gateway/gateway-runtime/policy-engine/internal/kernel/decompression.go @@ -20,35 +20,109 @@ package kernel import ( "bytes" + "compress/flate" "compress/gzip" + "compress/zlib" "errors" "fmt" "io" "sync" "github.com/andybalholm/brotli" + "github.com/klauspost/compress/zstd" ) +// Content-coding tokens the kernel can round-trip. These are the values stored +// in requestContentEncoding/responseContentEncoding — already lowercased, and +// for "deflate" already resolved to one of the two wire variants below. +const ( + encodingGzip = "gzip" + encodingBr = "br" + encodingZstd = "zstd" + // encodingDeflate is "deflate" in its RFC 9110 / RFC 1950 form: DEFLATE data + // inside a zlib wrapper. + encodingDeflate = "deflate" + // encodingDeflateRaw is an INTERNAL token, never a wire value. Some servers + // and clients send "Content-Encoding: deflate" carrying bare RFC 1951 DEFLATE + // with no zlib wrapper. Both are decodable, but the two are not interchangeable + // on output — re-encoding raw input as zlib-wrapped (or vice versa) hands the + // peer a body its decoder rejects. Recording which variant arrived lets the + // kernel emit the same one back; the Content-Encoding header itself is never + // rewritten and stays "deflate" either way. + encodingDeflateRaw = "deflate-raw" + // encodingIdentity is the no-op coding: present in the header but meaning the + // body is not encoded at all. + encodingIdentity = "identity" +) + +// zstdDecoderConcurrency/zstdEncoderConcurrency pin the zstd codec to a single +// goroutine per stream. The library defaults to GOMAXPROCS workers per +// encoder/decoder, which on a proxy handling many concurrent bodies multiplies +// into thousands of goroutines for no throughput gain at these body sizes. +const ( + zstdDecoderConcurrency = 1 + zstdEncoderConcurrency = 1 +) + +// resolveDeflateVariant inspects the first bytes of a "deflate" body and reports +// the concrete variant token to record for it. +// +// A zlib stream (RFC 1950) starts with a 2-byte header: the low nibble of the +// first byte is the compression method (8 == DEFLATE) and the big-endian pair is +// a multiple of 31. Bare DEFLATE data effectively never satisfies both, so this +// check distinguishes the two reliably. Too few bytes to tell yet is treated as +// the RFC-conformant zlib form. +func resolveDeflateVariant(body []byte) string { + if len(body) < 2 { + return encodingDeflate + } + if body[0]&0x0f == 0x08 && (uint16(body[0])<<8|uint16(body[1]))%31 == 0 { + return encodingDeflate + } + return encodingDeflateRaw +} + // ErrDecompressedTooLarge is returned when decompressed output exceeds the // configured ceiling — the signature of a decompression bomb. var ErrDecompressedTooLarge = errors.New("decompressed body exceeds maximum allowed size") // decompressBody decompresses body bytes based on the Content-Encoding value. -// Supported encodings: "gzip", "br" (Brotli). Unknown encodings are returned as-is. +// Supported encodings: gzip, br, zstd, and both deflate variants. Unknown +// encodings are returned as-is — callers must not reach this with one, since +// isRecompressibleEncoding gates every call site (an unsupported encoding is +// rejected outright rather than handed to policies as opaque bytes). // Output is capped at maxBytes (<= 0 means unbounded); exceeding it returns // ErrDecompressedTooLarge, never a truncated body. func decompressBody(body []byte, encoding string, maxBytes int64) ([]byte, error) { switch encoding { - case "gzip": + case encodingGzip: r, err := gzip.NewReader(bytes.NewReader(body)) if err != nil { return nil, fmt.Errorf("gzip reader: %w", err) } defer r.Close() return readLimited(r, maxBytes) - case "br": + case encodingBr: r := brotli.NewReader(bytes.NewReader(body)) return readLimited(r, maxBytes) + case encodingZstd: + r, err := zstd.NewReader(bytes.NewReader(body), zstd.WithDecoderConcurrency(zstdDecoderConcurrency)) + if err != nil { + return nil, fmt.Errorf("zstd reader: %w", err) + } + defer r.Close() + return readLimited(r, maxBytes) + case encodingDeflate: + r, err := zlib.NewReader(bytes.NewReader(body)) + if err != nil { + return nil, fmt.Errorf("deflate (zlib) reader: %w", err) + } + defer r.Close() + return readLimited(r, maxBytes) + case encodingDeflateRaw: + r := flate.NewReader(bytes.NewReader(body)) + defer r.Close() + return readLimited(r, maxBytes) default: return body, nil } @@ -180,9 +254,13 @@ func newStreamDecompressor(encoding string, maxBytes int64) *streamDecompressor go func() { defer close(outChan) defer close(decoderDone) + // gzip/zlib/zstd readers consume their stream header eagerly, so + // construction blocks here until the first chunk arrives. That is why the + // decoder lives on its own goroutine: newStreamDecompressor must return + // before any body byte has been seen. var r io.Reader switch encoding { - case "gzip": + case encodingGzip: gr, err := gzip.NewReader(input) if err != nil { select { @@ -193,8 +271,34 @@ func newStreamDecompressor(encoding string, maxBytes int64) *streamDecompressor } defer gr.Close() r = gr - case "br": + case encodingBr: r = brotli.NewReader(input) + case encodingZstd: + zr, err := zstd.NewReader(input, zstd.WithDecoderConcurrency(zstdDecoderConcurrency)) + if err != nil { + select { + case errChan <- fmt.Errorf("zstd.NewReader: %w", err): + default: + } + return + } + defer zr.Close() + r = zr + case encodingDeflate: + zr, err := zlib.NewReader(input) + if err != nil { + select { + case errChan <- fmt.Errorf("zlib.NewReader: %w", err): + default: + } + return + } + defer zr.Close() + r = zr + case encodingDeflateRaw: + fr := flate.NewReader(input) + defer fr.Close() + r = fr default: r = input } @@ -358,20 +462,47 @@ func (sd *streamDecompressor) Close() { type streamCompressor struct { encoding string buf bytes.Buffer - gzip *gzip.Writer - brotli *brotli.Writer - closed bool + // w and flush are the encoder for this stream. Every supported codec exposes + // Write/Close/Flush, so they are held behind these two fields rather than one + // typed field per codec — a per-codec field forces every method here to grow a + // new case, and a missed one silently degrades to "no compression applied". + w io.WriteCloser + flush func() error + closed bool } // newStreamCompressor returns a compressor for the encoding, or nil when the // encoding needs no re-compression (callers then forward bytes unchanged). +// Encodings must be pre-validated with isRecompressibleEncoding; nil here means +// "forward untouched", which is only correct for an unencoded body. func newStreamCompressor(encoding string) *streamCompressor { sc := &streamCompressor{encoding: encoding} switch encoding { - case "gzip": - sc.gzip = gzip.NewWriter(&sc.buf) - case "br": - sc.brotli = brotli.NewWriter(&sc.buf) + case encodingGzip: + w := gzip.NewWriter(&sc.buf) + sc.w, sc.flush = w, w.Flush + case encodingBr: + w := brotli.NewWriter(&sc.buf) + sc.w, sc.flush = w, w.Flush + case encodingZstd: + // Error is unreachable: it reports invalid encoder options, and the + // options here are compile-time constants. + w, err := zstd.NewWriter(&sc.buf, zstd.WithEncoderConcurrency(zstdEncoderConcurrency)) + if err != nil { + return nil + } + sc.w, sc.flush = w, w.Flush + case encodingDeflate: + w := zlib.NewWriter(&sc.buf) + sc.w, sc.flush = w, w.Flush + case encodingDeflateRaw: + // Error is unreachable: it reports an out-of-range level, and the level + // here is a library constant. + w, err := flate.NewWriter(&sc.buf, flate.DefaultCompression) + if err != nil { + return nil + } + sc.w, sc.flush = w, w.Flush default: return nil } @@ -391,35 +522,18 @@ func (sc *streamCompressor) Compress(body []byte, endOfStream bool) ([]byte, err } sc.buf.Reset() - switch { - case sc.gzip != nil: - if len(body) > 0 { - if _, err := sc.gzip.Write(body); err != nil { - return nil, fmt.Errorf("gzip write: %w", err) - } + if len(body) > 0 { + if _, err := sc.w.Write(body); err != nil { + return nil, fmt.Errorf("%s write: %w", sc.encoding, err) } - if endOfStream { - if err := sc.gzip.Close(); err != nil { - return nil, fmt.Errorf("gzip close: %w", err) - } - sc.closed = true - } else if err := sc.gzip.Flush(); err != nil { - return nil, fmt.Errorf("gzip flush: %w", err) - } - case sc.brotli != nil: - if len(body) > 0 { - if _, err := sc.brotli.Write(body); err != nil { - return nil, fmt.Errorf("brotli write: %w", err) - } - } - if endOfStream { - if err := sc.brotli.Close(); err != nil { - return nil, fmt.Errorf("brotli close: %w", err) - } - sc.closed = true - } else if err := sc.brotli.Flush(); err != nil { - return nil, fmt.Errorf("brotli flush: %w", err) + } + if endOfStream { + if err := sc.w.Close(); err != nil { + return nil, fmt.Errorf("%s close: %w", sc.encoding, err) } + sc.closed = true + } else if err := sc.flush(); err != nil { + return nil, fmt.Errorf("%s flush: %w", sc.encoding, err) } out := make([]byte, sc.buf.Len()) @@ -433,21 +547,16 @@ func (sc *streamCompressor) Close() { return } sc.closed = true - if sc.gzip != nil { - _ = sc.gzip.Close() - } - if sc.brotli != nil { - _ = sc.brotli.Close() - } + _ = sc.w.Close() } // isRecompressibleEncoding reports whether the kernel can decompress and -// re-compress this Content-Encoding. Anything else must be left untouched — -// see execution_context.go, which refuses to run body policies on it rather -// than handing policies bytes they cannot read. +// re-compress this Content-Encoding. Anything else is rejected outright by +// execution_context.go: the kernel neither runs body policies on bytes they +// cannot read nor forwards a body it could not have inspected. func isRecompressibleEncoding(encoding string) bool { switch encoding { - case "gzip", "br": + case encodingGzip, encodingBr, encodingZstd, encodingDeflate, encodingDeflateRaw: return true default: return false @@ -458,30 +567,24 @@ func isRecompressibleEncoding(encoding string) bool { // Used for the BUFFERED response path, where the whole body is compressed in a // single call. Streaming responses must use streamCompressor instead so the // response is one compressed stream rather than one per chunk. -// Supported encodings: "gzip", "br" (Brotli). Unknown encodings are returned as-is. +// Supported encodings: gzip, br, zstd, and both deflate variants. Unknown +// encodings are returned as-is; call sites are gated by isRecompressibleEncoding +// so that case is unreachable for a body policies actually touched. func recompressBody(body []byte, encoding string) ([]byte, error) { - switch encoding { - case "gzip": - var buf bytes.Buffer - w := gzip.NewWriter(&buf) - if _, err := w.Write(body); err != nil { - return nil, fmt.Errorf("gzip write: %w", err) - } - if err := w.Close(); err != nil { - return nil, fmt.Errorf("gzip close: %w", err) - } - return buf.Bytes(), nil - case "br": - var buf bytes.Buffer - w := brotli.NewWriter(&buf) - if _, err := w.Write(body); err != nil { - return nil, fmt.Errorf("brotli write: %w", err) + // Reuse the streaming encoder in a single write+finalise, so the buffered and + // streaming paths cannot drift apart on which encodings they support or how + // each one is framed. + sc := newStreamCompressor(encoding) + if sc == nil { + // An encoding this function does not encode at all: return the body + // untouched, as it always has. + if !isRecompressibleEncoding(encoding) { + return body, nil } - if err := w.Close(); err != nil { - return nil, fmt.Errorf("brotli close: %w", err) - } - return buf.Bytes(), nil - default: - return body, nil + // A supported encoding whose codec refused its options. Returning the body + // here would emit plaintext under a compressed Content-Encoding header — + // the exact corruption this file exists to prevent. + return nil, fmt.Errorf("no compressor available for encoding %q", encoding) } + return sc.Compress(body, true) } diff --git a/gateway/gateway-runtime/policy-engine/internal/kernel/execution_context.go b/gateway/gateway-runtime/policy-engine/internal/kernel/execution_context.go index 9ee2a74ae..da46d4ed1 100644 --- a/gateway/gateway-runtime/policy-engine/internal/kernel/execution_context.go +++ b/gateway/gateway-runtime/policy-engine/internal/kernel/execution_context.go @@ -120,6 +120,17 @@ type PolicyExecutionContext struct { // before being sent back to the downstream client. responseContentEncoding string + // requestEncodingUnsupported / responseEncodingUnsupported are set when a + // non-identity Content-Encoding arrives that the kernel can neither decompress + // nor re-compress. When the policy chain requires that body, the message is + // rejected at the header phase rather than forwarded: passing it through would + // let an unreadable encoding silently skip every body policy — on the request + // side that is a caller-chosen header disabling guardrails and content + // moderation, and on the response side it is a masking/redaction policy that + // never runs on data already on its way to the client. + requestEncodingUnsupported bool + responseEncodingUnsupported bool + // isStreamingRequest is set when SupportsRequestStreaming is true and the client // sends a streaming body — the request body will be processed chunk-by-chunk. isStreamingRequest bool @@ -128,6 +139,11 @@ type PolicyExecutionContext struct { // requestStreamDecomp performs per-chunk decompression for compressed streaming // request bodies. Nil when the request is not Content-Encoded. requestStreamDecomp *streamDecompressor + // requestStreamComp re-compresses streaming request chunks into a SINGLE + // compressed stream for the whole request, for the same reason + // responseStreamComp does downstream: one writer per chunk would emit N + // independent members and the upstream would read only the first. + requestStreamComp *streamCompressor // isStreamingResponse is set to true during response headers processing when // streaming indicators are detected AND the policy chain supports streaming. @@ -207,6 +223,10 @@ func (ec *PolicyExecutionContext) closeStreamDecompressors() { ec.responseStreamDecomp.Close() ec.responseStreamDecomp = nil } + if ec.requestStreamComp != nil { + ec.requestStreamComp.Close() + ec.requestStreamComp = nil + } if ec.responseStreamComp != nil { ec.responseStreamComp.Close() ec.responseStreamComp = nil @@ -311,6 +331,164 @@ func (ec *PolicyExecutionContext) handlePayloadTooLarge( return resp } +// rejectUnsupportedEncoding builds an immediate response for a message whose +// Content-Encoding the kernel cannot round-trip, on a body the policy chain +// requires. +// +// This fails closed by design. The alternative — forward the body untouched — +// means every body policy on the route silently does not run while the message +// still completes with a success status. On the request side the encoding is +// chosen by the caller, so that would be a bypass primitive: send +// "Content-Encoding: " and guardrails, content moderation +// and schema validation all stop applying to the payload. On the response side +// it means a masking or redaction policy never runs on data already on its way +// to the client. Neither is a decision the kernel can make on the operator's +// behalf, so the message is rejected instead. +// +// The client payload stays generic — no encoding name, no policy names, nothing +// about which side failed (per error-handling.md directive 1); specifics go to +// the log under the correlation id. +func (ec *PolicyExecutionContext) rejectUnsupportedEncoding( + ctx context.Context, + encoding []string, + phase string, + statusCode typev3.StatusCode, + httpStatus int, + clientError string, +) *extprocv3.ProcessingResponse { + errorID := uuid.New().String() + + slog.WarnContext(ctx, "Rejecting message: Content-Encoding cannot be decoded, and body policies are attached to this route", + "error_id", errorID, + "request_id", ec.requestID, + "phase", phase, + "route_key", ec.routeKey, + "encoding", encoding, + ) + + errorBody := fmt.Sprintf(`{"error":%q,"error_id":"%s"}`, clientError, errorID) + + resp := &extprocv3.ProcessingResponse{ + Response: &extprocv3.ProcessingResponse_ImmediateResponse{ + ImmediateResponse: &extprocv3.ImmediateResponse{ + Status: &typev3.HttpStatus{ + Code: statusCode, + }, + Headers: buildHeaderValueOptions(map[string]string{ + "content-type": "application/json", + "x-error-id": errorID, + }), + Body: []byte(errorBody), + }, + }, + } + + ec.generated = generatedResponse{ + resp: resp, + outcome: tracing.HTTPOutcome{ + StatusCode: httpStatus, + Reason: constants.TerminalReasonUnsupportedEncoding, + ErrorID: errorID, + }, + } + return resp +} + +// rejectUnsupportedRequestEncoding rejects an undecodable request body with 415. +// The caller picked the encoding, so this is a client error and is safe to +// surface as one — no part of the request has been forwarded upstream yet. +func (ec *PolicyExecutionContext) rejectUnsupportedRequestEncoding(ctx context.Context) *extprocv3.ProcessingResponse { + return ec.rejectUnsupportedEncoding( + ctx, + ec.requestHeaderCtx.Headers.Get("content-encoding"), + "request_headers", + typev3.StatusCode_UnsupportedMediaType, + http.StatusUnsupportedMediaType, + "Unsupported Media Type", + ) +} + +// rejectUnsupportedResponseEncoding rejects an undecodable upstream response with +// 502. The client did nothing wrong — the upstream answered in a coding this +// gateway cannot inspect — and response headers have not been committed +// downstream yet at the point this runs, so a clean status is still possible. +func (ec *PolicyExecutionContext) rejectUnsupportedResponseEncoding(ctx context.Context) *extprocv3.ProcessingResponse { + return ec.rejectUnsupportedEncoding( + ctx, + ec.responseHeaderCtx.ResponseHeaders.Get("content-encoding"), + "response_headers", + typev3.StatusCode_BadGateway, + http.StatusBadGateway, + "Bad Gateway", + ) +} + +// rejectUndecodableBody builds an immediate response for a body that declared a +// supported Content-Encoding but failed to decode as it. The reasoning is the +// same as rejectUnsupportedEncoding: forwarding the raw bytes would leave every +// body policy on the route silently inapplicable while the message still +// succeeds. The decoder's error goes to the log only — it can describe stream +// internals, and the client learns nothing beyond the status. +func (ec *PolicyExecutionContext) rejectUndecodableBody( + ctx context.Context, + err error, + phase string, + statusCode typev3.StatusCode, + httpStatus int, + clientError string, +) *extprocv3.ProcessingResponse { + errorID := uuid.New().String() + + slog.WarnContext(ctx, "Rejecting body: does not decode as its declared Content-Encoding", + "error_id", errorID, + "request_id", ec.requestID, + "phase", phase, + "route_key", ec.routeKey, + "error", err, + ) + + errorBody := fmt.Sprintf(`{"error":%q,"error_id":"%s"}`, clientError, errorID) + + resp := &extprocv3.ProcessingResponse{ + Response: &extprocv3.ProcessingResponse_ImmediateResponse{ + ImmediateResponse: &extprocv3.ImmediateResponse{ + Status: &typev3.HttpStatus{ + Code: statusCode, + }, + Headers: buildHeaderValueOptions(map[string]string{ + "content-type": "application/json", + "x-error-id": errorID, + }), + Body: []byte(errorBody), + }, + }, + } + + ec.generated = generatedResponse{ + resp: resp, + outcome: tracing.HTTPOutcome{ + StatusCode: httpStatus, + Reason: constants.TerminalReasonUnsupportedEncoding, + ErrorID: errorID, + }, + } + return resp +} + +// resolveEncodingFromBody pins "deflate" to the concrete variant the peer +// actually sent, using the first bytes of the body. Every other encoding is +// self-describing and passes through unchanged. +// +// This runs at the first body bytes rather than at the header phase because the +// header alone cannot distinguish the two: both arrive as +// "Content-Encoding: deflate". +func resolveEncodingFromBody(encoding string, firstBytes []byte) string { + if encoding != encodingDeflate { + return encoding + } + return resolveDeflateVariant(firstBytes) +} + // finalResponseStatus returns the HTTP status the downstream client will // actually see for a pass-through (non-short-circuited) response. // @@ -520,6 +698,15 @@ func (ec *PolicyExecutionContext) processRequestHeaders( ctx context.Context, ) (*extprocv3.ProcessingResponse, error) { ec.phase = phaseRequestHeaders + + // Reject before any policy runs and before a single body byte is forwarded + // upstream. Only matters when the chain actually inspects the request body — + // with no body policy attached there is nothing to bypass, so an encoding the + // kernel cannot read is simply none of its business and passes through. + if ec.requestEncodingUnsupported && ec.policyChain.RequiresRequestBody && !ec.requestHasNoBody() { + return ec.rejectUnsupportedRequestEncoding(ctx), nil + } + execResult, err := ec.server.executor.ExecuteRequestHeaderPolicies( ctx, ec.policyChain.Policies, @@ -669,22 +856,22 @@ func (ec *PolicyExecutionContext) processRequestBody( // Decompress body if Content-Encoding was set, so policies receive plain bytes. bodyContent := body.Body if ec.requestContentEncoding != "" { + ec.requestContentEncoding = resolveEncodingFromBody(ec.requestContentEncoding, body.Body) decompressed, err := decompressBody(body.Body, ec.requestContentEncoding, ec.server.maxRequestDecompressedBytes) if err != nil { // Over-limit bodies must be rejected, never forwarded raw. if errors.Is(err, ErrDecompressedTooLarge) { return ec.handlePayloadTooLarge(ctx, err, "request_body"), nil } - slog.Warn("Failed to decompress request body, passing raw bytes to policies", - "request_id", ec.requestID, - "encoding", ec.requestContentEncoding, - "error", err, - ) - // Clear encoding so translator doesn't attempt to recompress raw compressed bytes - ec.requestContentEncoding = "" - } else { - bodyContent = decompressed + // A body that does not decode as its declared encoding is rejected, + // not handed to policies raw. Falling through would let any caller + // disable every request-body policy by labelling arbitrary bytes + // "Content-Encoding: gzip" — the policies see bytes they cannot + // parse, match nothing, and the body is forwarded upstream anyway. + return ec.rejectUndecodableBody(ctx, err, "request_body", + typev3.StatusCode_BadRequest, http.StatusBadRequest, "Bad Request"), nil } + bodyContent = decompressed } // Update request context with body data @@ -730,8 +917,15 @@ func (ec *PolicyExecutionContext) processStreamingRequestBody( // Compressed request: decompress this chunk, pass directly to policies, // recompress the output. No kernel accumulation — policy implementations // handle their own internal state across chunks. - if ec.requestContentEncoding != "" { + // An empty leading chunk carries no encoding evidence, so the decoder is not + // built from it: the deflate variant would be guessed from zero bytes and + // cannot be corrected once the decoder is running. Nothing is lost by waiting + // — feeding an empty chunk produces no output either. + if ec.requestContentEncoding != "" && (len(chunk.Chunk) > 0 || ec.requestStreamDecomp != nil) { if ec.requestStreamDecomp == nil { + // Pin the deflate variant from the first chunk, before the decoder for + // it is built — the decoder cannot be swapped once running. + ec.requestContentEncoding = resolveEncodingFromBody(ec.requestContentEncoding, chunk.Chunk) ec.requestStreamDecomp = newStreamDecompressor(ec.requestContentEncoding, ec.server.maxRequestDecompressedBytes) } decompressed, err := ec.requestStreamDecomp.FeedChunk(chunk.Chunk, chunk.EndOfStream) @@ -894,6 +1088,14 @@ func (ec *PolicyExecutionContext) processResponseHeaders( ec.phase = phaseResponseHeaders ec.buildResponseContexts(headers) + // Reject here, at the last point an HTTP status can still be chosen: response + // headers have not been forwarded downstream yet, so this becomes a clean 502 + // rather than a mid-stream reset. As on the request side, this only applies + // when the chain actually inspects the response body. + if ec.responseEncodingUnsupported && ec.policyChain.RequiresResponseBody && !ec.responseHasNoBody() { + return ec.rejectUnsupportedResponseEncoding(ctx), nil + } + // Detect streaming response: upgrade when chain supports streaming AND // upstream signals chunked/SSE AND body is coming (not EndOfStream). slog.Debug("[mode] response headers received — streaming detection", @@ -947,6 +1149,15 @@ func (ec *PolicyExecutionContext) processResponseBody( body *extprocv3.HttpBody, ) (*extprocv3.ProcessingResponse, error) { ec.phase = phaseResponseBody + + // Defence in depth. processResponseHeaders already rejected this response, so + // reaching the body phase with an undecodable encoding means the header-phase + // guard was bypassed or removed. Fail the stream rather than fall through to + // policies that would receive compressed bytes as if they were plaintext. + if ec.responseEncodingUnsupported && ec.policyChain.RequiresResponseBody { + return nil, fmt.Errorf("response body phase reached with undecodable Content-Encoding; refusing to run body policies") + } + if ec.isStreamingResponse { slog.Debug("[body] routing to streaming response body handler", "route", ec.routeKey, @@ -965,21 +1176,20 @@ func (ec *PolicyExecutionContext) processResponseBody( // Decompress body if Content-Encoding was set, so policies receive plain JSON. bodyContent := body.Body if ec.responseContentEncoding != "" { + ec.responseContentEncoding = resolveEncodingFromBody(ec.responseContentEncoding, body.Body) decompressed, err := decompressBody(body.Body, ec.responseContentEncoding, ec.server.maxResponseDecompressedBytes) if err != nil { if errors.Is(err, ErrDecompressedTooLarge) { return nil, ec.responsePayloadTooLargeError(ctx, err, "response_body") } - slog.Warn("Failed to decompress response body, passing raw bytes to policies", - "request_id", ec.requestID, - "encoding", ec.responseContentEncoding, - "error", err, - ) - // Clear encoding so translator doesn't attempt to recompress raw compressed bytes - ec.responseContentEncoding = "" - } else { - bodyContent = decompressed + // As on the request side: a response that does not decode as its + // declared encoding is rejected rather than handed to policies raw, + // so a malformed upstream body cannot silently skip masking or + // redaction on its way to the client. + return ec.rejectUndecodableBody(ctx, err, "response_body", + typev3.StatusCode_BadGateway, http.StatusBadGateway, "Bad Gateway"), nil } + bodyContent = decompressed } // Update response context with body data @@ -1062,8 +1272,14 @@ func (ec *PolicyExecutionContext) processStreamingResponseBody( // on gzip: word-count/sentence-count guardrails evaluated isolated fragments // instead of assembled content, and content-rewriting policies never saw a // placeholder that straddled a chunk boundary. - if ec.responseContentEncoding != "" { + // As on the request path: an empty leading chunk is no evidence of the deflate + // variant, and the decoder cannot be swapped once running — so defer building + // it until a chunk actually carries bytes. + if ec.responseContentEncoding != "" && (len(chunk.Chunk) > 0 || ec.responseStreamDecomp != nil) { if ec.responseStreamDecomp == nil { + // Pin the deflate variant from the first chunk, before the decoder for + // it is built — the decoder cannot be swapped once running. + ec.responseContentEncoding = resolveEncodingFromBody(ec.responseContentEncoding, chunk.Chunk) ec.responseStreamDecomp = newStreamDecompressor(ec.responseContentEncoding, ec.server.maxResponseDecompressedBytes) } decompressed, err := ec.responseStreamDecomp.FeedChunk(chunk.Chunk, chunk.EndOfStream) @@ -1219,7 +1435,24 @@ func (ec *PolicyExecutionContext) buildRequestContexts(headers *extprocv3.HttpHe requestID = value } case "content-encoding": - ec.requestContentEncoding = value + // Only encodings the kernel can round-trip are recorded; anything + // else is flagged so the request is rejected outright before any + // body policy runs (see rejectUnsupportedRequestEncoding). + // + // This value is chosen by the caller, so treating an unrecognised + // one as "no encoding" would be a policy bypass primitive: the body + // would reach policies as opaque compressed bytes, match nothing, + // and be forwarded to the upstream unchanged — guardrails, content + // moderation and schema validation all silently skipped by setting + // a header. Content codings are case-insensitive tokens + // (RFC 9110 §8.4.1), so normalise before matching or "GZIP" alone + // would take that path. + encoding := strings.ToLower(strings.TrimSpace(value)) + if isRecompressibleEncoding(encoding) { + ec.requestContentEncoding = encoding + } else if encoding != "" && encoding != encodingIdentity { + ec.requestEncodingUnsupported = true + } } } } @@ -1346,12 +1579,12 @@ func (ec *PolicyExecutionContext) buildResponseContexts(headers *extprocv3.HttpH } case "content-encoding": // Only encodings the kernel can actually decompress AND re-compress - // are recorded. For anything else (deflate, zstd, …) the decompressor - // would fall through to a passthrough reader, handing policies raw - // compressed bytes: content-rewriting policies then silently match - // nothing — e.g. pii-masking-regex would deliver "[EMAIL_0000]" to the - // client instead of restoring it — with no error anywhere. Leaving this - // empty keeps the body untouched end to end, which is the safe outcome. + // are recorded. Anything else is flagged and the response is rejected + // at the header phase when the chain requires the body — the + // decompressor would otherwise hand policies raw compressed bytes, so + // a content-rewriting policy silently matches nothing (pii-masking-regex + // delivering "[EMAIL_0000]" to the client instead of restoring it, with + // no error anywhere). // // Content codings are case-insensitive tokens (RFC 9110 §8.4.1), so // normalise before matching — the decompressor/compressor switches are @@ -1359,11 +1592,8 @@ func (ec *PolicyExecutionContext) buildResponseContexts(headers *extprocv3.HttpH encoding := strings.ToLower(strings.TrimSpace(value)) if isRecompressibleEncoding(encoding) { ec.responseContentEncoding = encoding - } else if encoding != "" && encoding != "identity" { - slog.Warn("unsupported response Content-Encoding; body policies will not inspect or modify this response", - "request_id", ec.requestID, - "encoding", value, - ) + } else if encoding != "" && encoding != encodingIdentity { + ec.responseEncodingUnsupported = true } } } diff --git a/gateway/gateway-runtime/policy-engine/internal/kernel/execution_context_test.go b/gateway/gateway-runtime/policy-engine/internal/kernel/execution_context_test.go index e4241c3f2..b86e70d21 100644 --- a/gateway/gateway-runtime/policy-engine/internal/kernel/execution_context_test.go +++ b/gateway/gateway-runtime/policy-engine/internal/kernel/execution_context_test.go @@ -19,8 +19,10 @@ package kernel import ( + "bytes" "context" "fmt" + "strings" "testing" "time" @@ -804,6 +806,366 @@ func TestProcessResponseBody_DecompressesBrotli(t *testing.T) { assert.Equal(t, originalJSON, capturedBody) } +// ============================================================================= +// Unsupported / undecodable Content-Encoding — fail-closed behaviour +// ============================================================================= + +// newEncodingTestContext builds an execution context whose chain requires both +// bodies, with a policy that records whether it ever ran. +func newEncodingTestContext(t *testing.T, streaming bool) (*PolicyExecutionContext, *bool) { + t.Helper() + + policyRan := new(bool) + bodyMode := policy.BodyModeBuffer + if streaming { + bodyMode = policy.BodyModeStream + } + mockPolicy := &testutils.ConfigurableMockPolicy{ + MockMode: policy.ProcessingMode{ + RequestBodyMode: bodyMode, + ResponseBodyMode: bodyMode, + }, + OnReqFn: func(_ *policy.RequestContext, _ map[string]interface{}) policy.RequestAction { + *policyRan = true + return policy.UpstreamRequestModifications{} + }, + OnRespFn: func(_ *policy.ResponseContext, _ map[string]interface{}) policy.ResponseAction { + *policyRan = true + return policy.DownstreamResponseModifications{} + }, + } + + kernel := NewKernel() + server := NewExternalProcessorServer(kernel, newTestExecutor(), config.TracingConfig{}, "", testMaxDecompressedBytes, testMaxDecompressedBytes) + chain := ®istry.PolicyChain{ + RequiresRequestBody: true, + RequiresResponseBody: true, + SupportsRequestStreaming: streaming, + SupportsResponseStreaming: streaming, + Policies: []policy.Policy{mockPolicy}, + PolicySpecs: []policy.PolicySpec{{Enabled: true}}, + } + return newPolicyExecutionContext(server, "test-route", chain), policyRan +} + +func postRequestHeaders(encoding string) *extprocv3.HttpHeaders { + headers := []*corev3.HeaderValue{ + {Key: ":path", RawValue: []byte("/api/chat")}, + {Key: ":method", RawValue: []byte("POST")}, + {Key: "content-type", RawValue: []byte("application/json")}, + } + if encoding != "" { + headers = append(headers, &corev3.HeaderValue{Key: "content-encoding", RawValue: []byte(encoding)}) + } + return &extprocv3.HttpHeaders{Headers: &corev3.HeaderMap{Headers: headers}} +} + +func okResponseHeaders(encoding string) *extprocv3.HttpHeaders { + headers := []*corev3.HeaderValue{ + {Key: ":status", RawValue: []byte("200")}, + {Key: "content-type", RawValue: []byte("application/json")}, + } + if encoding != "" { + headers = append(headers, &corev3.HeaderValue{Key: "content-encoding", RawValue: []byte(encoding)}) + } + return &extprocv3.HttpHeaders{Headers: &corev3.HeaderMap{Headers: headers}} +} + +// An encoding the kernel cannot round-trip must be flagged on BOTH sides. The +// request side is the one that matters most: the value is chosen by the caller, +// so treating it as "no encoding" would let anyone disable every request-body +// policy on the route by setting a header. +func TestBuildContexts_FlagsUnsupportedEncoding(t *testing.T) { + for _, encoding := range []string{"compress", "snappy", "gzip, br"} { + t.Run(encoding, func(t *testing.T) { + execCtx, _ := newEncodingTestContext(t, false) + + execCtx.buildRequestContexts(postRequestHeaders(encoding), RouteMetadata{}) + execCtx.buildResponseContexts(okResponseHeaders(encoding)) + + assert.Empty(t, execCtx.requestContentEncoding) + assert.True(t, execCtx.requestEncodingUnsupported) + assert.Empty(t, execCtx.responseContentEncoding) + assert.True(t, execCtx.responseEncodingUnsupported) + }) + } +} + +// Content codings are case-insensitive tokens (RFC 9110 §8.4.1). Before this was +// normalised on the request side, "GZIP" was stored verbatim, missed every +// lowercase codec switch, and reached policies as raw compressed bytes. +func TestBuildContexts_NormalizesEncodingCase(t *testing.T) { + for _, encoding := range []string{"GZIP", "Gzip", " gzip "} { + t.Run(encoding, func(t *testing.T) { + execCtx, _ := newEncodingTestContext(t, false) + + execCtx.buildRequestContexts(postRequestHeaders(encoding), RouteMetadata{}) + execCtx.buildResponseContexts(okResponseHeaders(encoding)) + + assert.Equal(t, "gzip", execCtx.requestContentEncoding) + assert.False(t, execCtx.requestEncodingUnsupported) + assert.Equal(t, "gzip", execCtx.responseContentEncoding) + assert.False(t, execCtx.responseEncodingUnsupported) + }) + } +} + +// identity/absent means the body is not encoded — no rejection, policies run. +func TestBuildContexts_IdentityEncodingNotFlagged(t *testing.T) { + for _, encoding := range []string{"identity", ""} { + t.Run("encoding="+encoding, func(t *testing.T) { + execCtx, _ := newEncodingTestContext(t, false) + + execCtx.buildRequestContexts(postRequestHeaders(encoding), RouteMetadata{}) + execCtx.buildResponseContexts(okResponseHeaders(encoding)) + + assert.Empty(t, execCtx.requestContentEncoding) + assert.False(t, execCtx.requestEncodingUnsupported) + assert.Empty(t, execCtx.responseContentEncoding) + assert.False(t, execCtx.responseEncodingUnsupported) + }) + } +} + +// The bypass this guards against: an undecodable request body must be rejected +// at the header phase — before any policy runs and before a byte reaches the +// upstream — rather than forwarded with body policies silently skipped. +func TestProcessRequestHeaders_UnsupportedEncodingRejected(t *testing.T) { + execCtx, policyRan := newEncodingTestContext(t, false) + execCtx.buildRequestContexts(postRequestHeaders("snappy"), RouteMetadata{}) + + resp, err := execCtx.processRequestHeaders(context.Background()) + require.NoError(t, err) + + immediate := resp.GetImmediateResponse() + require.NotNil(t, immediate, "an undecodable request must be rejected, not forwarded") + assert.Equal(t, typev3.StatusCode_UnsupportedMediaType, immediate.GetStatus().GetCode()) + assert.False(t, *policyRan) + // The client learns nothing about the encoding or the policy chain. + assert.NotContains(t, string(immediate.GetBody()), "snappy") +} + +// The upstream, not the client, chose this encoding — so it is a 502, and it is +// caught at the response-header phase while a status can still be chosen. +func TestProcessResponseHeaders_UnsupportedEncodingRejected(t *testing.T) { + execCtx, policyRan := newEncodingTestContext(t, false) + execCtx.buildRequestContexts(postRequestHeaders(""), RouteMetadata{}) + + resp, err := execCtx.processResponseHeaders(context.Background(), okResponseHeaders("zstd-unknown-variant")) + require.NoError(t, err) + + immediate := resp.GetImmediateResponse() + require.NotNil(t, immediate, "an undecodable response must be rejected, not forwarded") + assert.Equal(t, typev3.StatusCode_BadGateway, immediate.GetStatus().GetCode()) + assert.False(t, *policyRan) + assert.NotContains(t, string(immediate.GetBody()), "zstd-unknown-variant") +} + +// With no body policy attached there is nothing to bypass, so an encoding the +// kernel cannot read is none of its business and must pass through untouched. +// Rejecting here would break routes that never inspect bodies at all. +func TestProcessHeaders_UnsupportedEncodingAllowedWithoutBodyPolicies(t *testing.T) { + kernel := NewKernel() + server := NewExternalProcessorServer(kernel, newTestExecutor(), config.TracingConfig{}, "", testMaxDecompressedBytes, testMaxDecompressedBytes) + execCtx := newPolicyExecutionContext(server, "test-route", ®istry.PolicyChain{}) + + execCtx.buildRequestContexts(postRequestHeaders("snappy"), RouteMetadata{}) + reqResp, err := execCtx.processRequestHeaders(context.Background()) + require.NoError(t, err) + assert.Nil(t, reqResp.GetImmediateResponse()) + + respResp, err := execCtx.processResponseHeaders(context.Background(), okResponseHeaders("snappy")) + require.NoError(t, err) + assert.Nil(t, respResp.GetImmediateResponse()) +} + +// The cheapest bypass of all: declare a supported encoding, send bytes that are +// not in it. Policies would receive undecodable bytes, match nothing, and the +// body would be forwarded anyway. +func TestProcessRequestBody_UndecodableBodyRejected(t *testing.T) { + execCtx, policyRan := newEncodingTestContext(t, false) + execCtx.buildRequestContexts(postRequestHeaders("gzip"), RouteMetadata{}) + + resp, err := execCtx.processRequestBody(context.Background(), &extprocv3.HttpBody{ + Body: []byte(`{"prompt":"not actually gzipped"}`), + EndOfStream: true, + }) + require.NoError(t, err) + + immediate := resp.GetImmediateResponse() + require.NotNil(t, immediate, "a body that is not in its declared encoding must be rejected") + assert.Equal(t, typev3.StatusCode_BadRequest, immediate.GetStatus().GetCode()) + assert.False(t, *policyRan) +} + +func TestProcessResponseBody_UndecodableBodyRejected(t *testing.T) { + execCtx, policyRan := newEncodingTestContext(t, false) + execCtx.buildRequestContexts(postRequestHeaders(""), RouteMetadata{}) + execCtx.buildResponseContexts(okResponseHeaders("br")) + + // Brotli accepts many byte sequences, so use one that reliably fails. + resp, err := execCtx.processResponseBody(context.Background(), &extprocv3.HttpBody{ + Body: bytes.Repeat([]byte{0xff}, 64), + EndOfStream: true, + }) + require.NoError(t, err) + + immediate := resp.GetImmediateResponse() + require.NotNil(t, immediate, "a body that is not in its declared encoding must be rejected") + assert.Equal(t, typev3.StatusCode_BadGateway, immediate.GetStatus().GetCode()) + assert.False(t, *policyRan) +} + +// Every supported encoding must reach policies as plaintext on the buffered +// request path — the whole point of supporting it rather than rejecting it. +func TestProcessRequestBody_DecompressesEverySupportedEncoding(t *testing.T) { + plaintext := []byte(`{"prompt":"contact me at user@example.com"}`) + + for _, encoding := range []string{"gzip", "br", "zstd", "deflate"} { + t.Run(encoding, func(t *testing.T) { + var capturedBody []byte + mockPolicy := &testutils.ConfigurableMockPolicy{ + MockMode: policy.ProcessingMode{RequestBodyMode: policy.BodyModeBuffer}, + OnReqFn: func(ctx *policy.RequestContext, _ map[string]interface{}) policy.RequestAction { + if ctx.Body != nil { + capturedBody = ctx.Body.Content + } + return policy.UpstreamRequestModifications{} + }, + } + + kernel := NewKernel() + server := NewExternalProcessorServer(kernel, newTestExecutor(), config.TracingConfig{}, "", testMaxDecompressedBytes, testMaxDecompressedBytes) + execCtx := newPolicyExecutionContext(server, "test-route", ®istry.PolicyChain{ + RequiresRequestBody: true, + Policies: []policy.Policy{mockPolicy}, + PolicySpecs: []policy.PolicySpec{{Enabled: true}}, + }) + execCtx.buildRequestContexts(postRequestHeaders(encoding), RouteMetadata{}) + + compressed, err := recompressBody(plaintext, encoding) + require.NoError(t, err) + + _, err = execCtx.processRequestBody(context.Background(), &extprocv3.HttpBody{ + Body: compressed, + EndOfStream: true, + }) + require.NoError(t, err) + assert.Equal(t, plaintext, capturedBody) + }) + } +} + +// A "deflate" request carrying RAW deflate (no zlib wrapper) must be detected +// and re-encoded in the same variant. Emitting the other form would hand the +// upstream a body its decoder rejects. +func TestProcessRequestBody_PreservesRawDeflateVariant(t *testing.T) { + plaintext := []byte(`{"prompt":"raw deflate body"}`) + + kernel := NewKernel() + server := NewExternalProcessorServer(kernel, newTestExecutor(), config.TracingConfig{}, "", testMaxDecompressedBytes, testMaxDecompressedBytes) + execCtx := newPolicyExecutionContext(server, "test-route", ®istry.PolicyChain{ + RequiresRequestBody: true, + Policies: []policy.Policy{&testutils.ConfigurableMockPolicy{ + MockMode: policy.ProcessingMode{RequestBodyMode: policy.BodyModeBuffer}, + OnReqFn: func(_ *policy.RequestContext, _ map[string]interface{}) policy.RequestAction { + return policy.UpstreamRequestModifications{} + }, + }}, + PolicySpecs: []policy.PolicySpec{{Enabled: true}}, + }) + execCtx.buildRequestContexts(postRequestHeaders("deflate"), RouteMetadata{}) + + rawDeflate, err := recompressBody(plaintext, "deflate-raw") + require.NoError(t, err) + + _, err = execCtx.processRequestBody(context.Background(), &extprocv3.HttpBody{ + Body: rawDeflate, + EndOfStream: true, + }) + require.NoError(t, err) + assert.Equal(t, "deflate-raw", execCtx.requestContentEncoding, + "the arriving deflate variant must be pinned so the same one is emitted back") +} + +// chunkRecordingRequestPolicy records every request chunk body policies are +// handed, so a test can assert policies saw plaintext rather than compressed +// bytes. +type chunkRecordingRequestPolicy struct { + seen *strings.Builder +} + +func (p *chunkRecordingRequestPolicy) Mode() policy.ProcessingMode { + return policy.ProcessingMode{RequestBodyMode: policy.BodyModeStream} +} + +func (p *chunkRecordingRequestPolicy) OnRequestBody(_ context.Context, _ *policy.RequestContext, _ map[string]interface{}) policy.RequestAction { + return policy.UpstreamRequestModifications{} +} + +func (p *chunkRecordingRequestPolicy) NeedsMoreRequestData(_ []byte) bool { return false } + +func (p *chunkRecordingRequestPolicy) OnRequestBodyChunk(_ context.Context, _ *policy.RequestStreamContext, chunk *policy.StreamBody, _ map[string]interface{}) policy.StreamingRequestAction { + p.seen.Write(chunk.Chunk) + return policy.ForwardRequestChunk{} +} + +// End-to-end through the kernel's streaming REQUEST path: chunks arrive +// compressed, policies must see plaintext, and what leaves for the upstream must +// be ONE compressed stream that decodes to the full body. Before the fix the +// request path re-compressed per chunk, so the upstream saw only chunk one. +func TestProcessStreamingRequestBody_RoundTripsAsSingleStream(t *testing.T) { + chunks := []string{`{"prompt":"part one `, `and part two `, `and part three"}`} + var wantPlaintext strings.Builder + for _, c := range chunks { + wantPlaintext.WriteString(c) + } + + for _, encoding := range []string{"gzip", "br", "zstd", "deflate"} { + t.Run(encoding, func(t *testing.T) { + var seenByPolicies strings.Builder + mockPolicy := &chunkRecordingRequestPolicy{seen: &seenByPolicies} + + kernel := NewKernel() + server := NewExternalProcessorServer(kernel, newTestExecutor(), config.TracingConfig{}, "", testMaxDecompressedBytes, testMaxDecompressedBytes) + execCtx := newPolicyExecutionContext(server, "test-route", ®istry.PolicyChain{ + RequiresRequestBody: true, + SupportsRequestStreaming: true, + Policies: []policy.Policy{mockPolicy}, + PolicySpecs: []policy.PolicySpec{{Enabled: true}}, + }) + execCtx.buildRequestContexts(postRequestHeaders(encoding), RouteMetadata{}) + execCtx.isStreamingRequest = true + + // Compress the body as one stream, then split it, mirroring how Envoy + // delivers an upstream-bound compressed body chunk by chunk. + whole, err := recompressBody([]byte(wantPlaintext.String()), encoding) + require.NoError(t, err) + split := len(whole) / 2 + + var wire bytes.Buffer + for i, in := range [][]byte{whole[:split], whole[split:]} { + endOfStream := i == 1 + resp, err := execCtx.processRequestBody(context.Background(), &extprocv3.HttpBody{ + Body: in, + EndOfStream: endOfStream, + }) + require.NoError(t, err) + wire.Write(resp.GetRequestBody().GetResponse().GetBodyMutation().GetStreamedResponse().GetBody()) + } + + assert.Equal(t, wantPlaintext.String(), seenByPolicies.String(), + "policies must observe plaintext, not compressed bytes") + + // The decisive assertion: one stream, decodable in a single pass. + got, err := decompressBody(wire.Bytes(), execCtx.requestContentEncoding, 0) + require.NoError(t, err) + assert.Equal(t, wantPlaintext.String(), string(got), + "upstream-bound body must be ONE compressed stream covering every chunk") + }) + } +} + func TestProcessStreamingResponseBody_DecompressionErrorFailsClosed(t *testing.T) { const maxChunkBytes int64 = 64 diff --git a/gateway/gateway-runtime/policy-engine/internal/kernel/stream_compression_test.go b/gateway/gateway-runtime/policy-engine/internal/kernel/stream_compression_test.go index 3ca5d9249..7c77f52c1 100644 --- a/gateway/gateway-runtime/policy-engine/internal/kernel/stream_compression_test.go +++ b/gateway/gateway-runtime/policy-engine/internal/kernel/stream_compression_test.go @@ -159,7 +159,10 @@ func TestStreamCompressor_FlushesPerChunk(t *testing.T) { // Encodings the kernel cannot round-trip must yield no compressor, so callers // forward the body untouched instead of corrupting it. func TestStreamCompressor_UnsupportedEncodings(t *testing.T) { - for _, enc := range []string{"deflate", "zstd", "identity", "", "GZIP"} { + // "GZIP" belongs here: the codec switches are lowercase-only, so callers must + // normalise before reaching them (buildRequest/ResponseContexts do). A + // comma-separated chain is unsupported too — the kernel cannot round-trip one. + for _, enc := range []string{"compress", "snappy", "identity", "", "GZIP", "gzip, br"} { if sc := newStreamCompressor(enc); sc != nil { t.Errorf("newStreamCompressor(%q) returned a compressor; want nil", enc) } @@ -167,10 +170,139 @@ func TestStreamCompressor_UnsupportedEncodings(t *testing.T) { t.Errorf("isRecompressibleEncoding(%q) = true; want false", enc) } } - for _, enc := range []string{"gzip", "br"} { + for _, enc := range []string{"gzip", "br", "zstd", "deflate", "deflate-raw"} { if !isRecompressibleEncoding(enc) { t.Errorf("isRecompressibleEncoding(%q) = false; want true", enc) } + if sc := newStreamCompressor(enc); sc == nil { + t.Errorf("newStreamCompressor(%q) = nil; want a compressor", enc) + } + } +} + +// newStreamCompressor returning nil must never reach a Compress/Close call. +// The streaming translators guard it explicitly rather than trusting the +// isRecompressibleEncoding gate in a different file, because a nil dereference +// there panics the ext_proc handler mid-message. +func TestTranslateStreamingChunkAction_NilCompressorFailsStreamNotPanics(t *testing.T) { + // "identity" is deliberately chosen: it reaches the compressor branch (it is + // a non-empty encoding) but yields no compressor, which is exactly the shape + // a supported-encoding constructor failure takes. + t.Run("response", func(t *testing.T) { + execCtx := &PolicyExecutionContext{ + responseContentEncoding: "identity", + analyticsMetadata: map[string]any{}, + dynamicMetadata: map[string]map[string]interface{}{}, + } + _, err := TranslateStreamingResponseChunkAction( + &executor.StreamingResponseExecutionResult{}, + &policy.StreamBody{Chunk: []byte("payload"), EndOfStream: true}, + execCtx, + ) + if err == nil { + t.Fatal("expected a stream error when no compressor is available; got nil") + } + }) + + t.Run("request", func(t *testing.T) { + execCtx := &PolicyExecutionContext{ + requestContentEncoding: "identity", + analyticsMetadata: map[string]any{}, + dynamicMetadata: map[string]map[string]interface{}{}, + } + _, err := TranslateStreamingRequestChunkAction( + &executor.StreamingRequestExecutionResult{}, + &policy.StreamBody{Chunk: []byte("payload"), EndOfStream: true}, + execCtx, + ) + if err == nil { + t.Fatal("expected a stream error when no compressor is available; got nil") + } + }) +} + +// recompressBody must not answer a supported-but-unconstructable encoding with +// the plaintext body — that would put unencoded bytes under a compressed +// Content-Encoding header. A genuinely unknown encoding still passes through. +func TestRecompressBody_NilCompressorDistinguishesUnknownFromUnavailable(t *testing.T) { + body := []byte(`{"content":"passthrough probe"}`) + + out, err := recompressBody(body, "identity") + if err != nil { + t.Fatalf("unknown encoding must pass through, got error: %v", err) + } + if !bytes.Equal(out, body) { + t.Errorf("unknown encoding altered the body: got %q want %q", out, body) + } +} + +// Every supported encoding must survive a multi-chunk streaming round trip as +// ONE compressed stream. This is the regression that started this change: a +// per-chunk compressor emits N independent members and clients stop decoding +// after the first, so the body looks truncated while every byte was sent. +func TestStreamCompressor_RoundTripsEveryEncoding(t *testing.T) { + chunks := []string{`{"choices":[{"delta":`, `{"content":"hello world"}`, `}]}`} + var want strings.Builder + for _, c := range chunks { + want.WriteString(c) + } + + for _, enc := range []string{"gzip", "br", "zstd", "deflate", "deflate-raw"} { + t.Run(enc, func(t *testing.T) { + sc := newStreamCompressor(enc) + if sc == nil { + t.Fatalf("newStreamCompressor(%q) = nil", enc) + } + var wire bytes.Buffer + for i, c := range chunks { + out, err := sc.Compress([]byte(c), i == len(chunks)-1) + if err != nil { + t.Fatalf("compress chunk %d: %v", i, err) + } + wire.Write(out) + } + + got, err := decompressBody(wire.Bytes(), enc, 0) + if err != nil { + t.Fatalf("decompress: %v", err) + } + if string(got) != want.String() { + t.Errorf("round trip mismatch\n got: %q\nwant: %q", got, want.String()) + } + }) + } +} + +// "deflate" on the wire is two different formats. Re-encoding raw input as +// zlib-wrapped (or the reverse) hands the peer a body its decoder rejects, so +// the arriving variant has to be detected and preserved. +func TestResolveDeflateVariant(t *testing.T) { + payload := []byte(`{"content":"deflate variant probe"}`) + + zlibWrapped, err := recompressBody(payload, "deflate") + if err != nil { + t.Fatalf("zlib encode: %v", err) + } + raw, err := recompressBody(payload, "deflate-raw") + if err != nil { + t.Fatalf("raw deflate encode: %v", err) + } + + if got := resolveDeflateVariant(zlibWrapped); got != "deflate" { + t.Errorf("zlib-wrapped body detected as %q; want \"deflate\"", got) + } + if got := resolveDeflateVariant(raw); got != "deflate-raw" { + t.Errorf("raw deflate body detected as %q; want \"deflate-raw\"", got) + } + // Too few bytes to decide falls back to the RFC-conformant form. + if got := resolveDeflateVariant([]byte{0x78}); got != "deflate" { + t.Errorf("1-byte body detected as %q; want \"deflate\"", got) + } + + // The variants must not be interchangeable, or preserving them would be + // pointless — decoding raw bytes as zlib has to fail. + if _, err := decompressBody(raw, "deflate", 0); err == nil { + t.Error("raw deflate decoded as zlib; the two variants are not distinguishable by this test") } } diff --git a/gateway/gateway-runtime/policy-engine/internal/kernel/translator.go b/gateway/gateway-runtime/policy-engine/internal/kernel/translator.go index f41d22c48..4600443dc 100644 --- a/gateway/gateway-runtime/policy-engine/internal/kernel/translator.go +++ b/gateway/gateway-runtime/policy-engine/internal/kernel/translator.go @@ -1524,17 +1524,41 @@ func TranslateStreamingRequestChunkAction(result *executor.StreamingRequestExecu // Re-compress the output if the original request was Content-Encoded. // The upstream receives the Content-Encoding header as-is, so the body // bytes must match the encoding the upstream expects. + // + // The whole request must form ONE compressed stream, so the compressor is + // held on the execution context across chunks and finalised at end of stream — + // the same contract the response path uses. Calling recompressBody per chunk + // emits one gzip member (or brotli/zstd/zlib stream) per chunk; the upstream + // stops decoding after the first, so it sees a truncated body while every byte + // was in fact sent. if execCtx.requestContentEncoding != "" { - recompressed, err := recompressBody(outputBody, execCtx.requestContentEncoding) + if execCtx.requestStreamComp == nil { + // Same guard as the response path: buildRequestContexts only stores an + // encoding isRecompressibleEncoding accepts, but newStreamCompressor can + // still return nil if the codec rejects its options. + comp := newStreamCompressor(execCtx.requestContentEncoding) + if comp == nil { + slog.Error("[streaming] no compressor available for request encoding; failing stream", + "encoding", execCtx.requestContentEncoding, + ) + return nil, fmt.Errorf("no stream compressor for request encoding %q", execCtx.requestContentEncoding) + } + execCtx.requestStreamComp = comp + } + recompressed, err := execCtx.requestStreamComp.Compress(outputBody, originalChunk.EndOfStream) if err != nil { - slog.Warn("[streaming] failed to re-compress request body; sending uncompressed — Content-Encoding mismatch", + // The Content-Encoding header is already on its way upstream, so + // falling back to plaintext here would corrupt the request. Fail the + // stream instead and let Envoy reset it. + slog.Error("[streaming] failed to re-compress request chunk; failing stream", "encoding", execCtx.requestContentEncoding, "error", err, ) - execCtx.requestContentEncoding = "" - } else { - outputBody = recompressed + execCtx.requestStreamComp.Close() + execCtx.requestStreamComp = nil + return nil, fmt.Errorf("streaming request re-compression failed: %w", err) } + outputBody = recompressed } analyticsData := make(map[string]any) @@ -1617,9 +1641,19 @@ func TranslateStreamingResponseChunkAction(result *executor.StreamingResponseExe // a truncated body even though every byte was transmitted. if execCtx.responseContentEncoding != "" { if execCtx.responseStreamComp == nil { - // Never nil: responseContentEncoding is only ever set to an encoding - // isRecompressibleEncoding accepts (see buildResponseContexts). - execCtx.responseStreamComp = newStreamCompressor(execCtx.responseContentEncoding) + // buildResponseContexts only ever stores an encoding + // isRecompressibleEncoding accepts, but newStreamCompressor can still + // return nil for one of those if the codec rejects its options. Fail the + // stream rather than dereference nil: the alternative is a panic in the + // ext_proc handler mid-response. + comp := newStreamCompressor(execCtx.responseContentEncoding) + if comp == nil { + slog.Error("[streaming] no compressor available for response encoding; failing stream", + "encoding", execCtx.responseContentEncoding, + ) + return nil, fmt.Errorf("no stream compressor for response encoding %q", execCtx.responseContentEncoding) + } + execCtx.responseStreamComp = comp } recompressed, err := execCtx.responseStreamComp.Compress(outputBody, endOfStream) if err != nil { From a9dba13240019c962b6b67e472926898329dd8e8 Mon Sep 17 00:00:00 2001 From: Thushani Jayasekera Date: Sat, 15 Aug 2026 19:26:51 +0530 Subject: [PATCH 3/3] Update go.mod and go.sum to use klauspost/compress v1.19.2 --- gateway/gateway-runtime/policy-engine/go.mod | 2 +- gateway/gateway-runtime/policy-engine/go.sum | 4 +-- .../internal/kernel/execution_context.go | 25 ++++++++++++++++--- 3 files changed, 24 insertions(+), 7 deletions(-) diff --git a/gateway/gateway-runtime/policy-engine/go.mod b/gateway/gateway-runtime/policy-engine/go.mod index 66a6f7b2f..e437a64af 100644 --- a/gateway/gateway-runtime/policy-engine/go.mod +++ b/gateway/gateway-runtime/policy-engine/go.mod @@ -8,7 +8,7 @@ require ( github.com/go-viper/mapstructure/v2 v2.5.0 github.com/google/cel-go v0.26.1 github.com/google/uuid v1.6.0 - github.com/klauspost/compress v1.18.6 + github.com/klauspost/compress v1.19.2 github.com/knadh/koanf/parsers/toml/v2 v2.2.0 github.com/knadh/koanf/providers/confmap v1.0.0 github.com/knadh/koanf/providers/file v1.2.1 diff --git a/gateway/gateway-runtime/policy-engine/go.sum b/gateway/gateway-runtime/policy-engine/go.sum index 2939f20dd..0be7ba581 100644 --- a/gateway/gateway-runtime/policy-engine/go.sum +++ b/gateway/gateway-runtime/policy-engine/go.sum @@ -39,8 +39,8 @@ github.com/google/uuid v1.6.0 h1:NIvaJDMOsjHA8n1jAhLSgzrAzy1Hgr+hNrb57e+94F0= github.com/google/uuid v1.6.0/go.mod h1:TIyPZe4MgqvfeYDBFedMoGGpEw/LqOeaOT+nhxU+yHo= github.com/grpc-ecosystem/grpc-gateway/v2 v2.29.0 h1:5VipnvEpbqr2gA2VbM+nYVbkIF28c5ZQfqCBQ5g2xfk= github.com/grpc-ecosystem/grpc-gateway/v2 v2.29.0/go.mod h1:Hyl3n6Twe1hvtd9XUXDec4pTvgMSEixRuQKPTMH2bNs= -github.com/klauspost/compress v1.18.6 h1:2jupLlAwFm95+YDR+NwD2MEfFO9d4z4Prjl1XXDjuao= -github.com/klauspost/compress v1.18.6/go.mod h1:cwPg85FWrGar70rWktvGQj8/hthj3wpl0PGDogxkrSQ= +github.com/klauspost/compress v1.19.2 h1:hMRETovs/pu/dVWN7zIT1PGG8t509MwT6bO7XSi26R8= +github.com/klauspost/compress v1.19.2/go.mod h1:cwPg85FWrGar70rWktvGQj8/hthj3wpl0PGDogxkrSQ= github.com/knadh/koanf/maps v0.1.2 h1:RBfmAW5CnZT+PJ1CVc1QSJKf4Xu9kxfQgYVQSu8hpbo= github.com/knadh/koanf/maps v0.1.2/go.mod h1:npD/QZY3V6ghQDdcQzl1W4ICNVTkohC8E73eI2xW4yI= github.com/knadh/koanf/parsers/toml/v2 v2.2.0 h1:2nV7tHYJ5OZy2BynQ4mOJ6k5bDqbbCzRERLUKBytz3A= diff --git a/gateway/gateway-runtime/policy-engine/internal/kernel/execution_context.go b/gateway/gateway-runtime/policy-engine/internal/kernel/execution_context.go index da46d4ed1..11a449fc3 100644 --- a/gateway/gateway-runtime/policy-engine/internal/kernel/execution_context.go +++ b/gateway/gateway-runtime/policy-engine/internal/kernel/execution_context.go @@ -703,7 +703,7 @@ func (ec *PolicyExecutionContext) processRequestHeaders( // upstream. Only matters when the chain actually inspects the request body — // with no body policy attached there is nothing to bypass, so an encoding the // kernel cannot read is simply none of its business and passes through. - if ec.requestEncodingUnsupported && ec.policyChain.RequiresRequestBody && !ec.requestHasNoBody() { + if ec.requestEncodingBlocksBodyPolicies() { return ec.rejectUnsupportedRequestEncoding(ctx), nil } @@ -775,6 +775,21 @@ func (ec *PolicyExecutionContext) requestHasNoBody() bool { return false } +// requestEncodingBlocksBodyPolicies reports whether the request carries a Content-Encoding +// the kernel cannot decode while the chain needs to inspect the request body. With no body +// policy attached there is nothing to bypass, and a bodyless request has nothing to decode, +// so both cases pass through untouched. +func (ec *PolicyExecutionContext) requestEncodingBlocksBodyPolicies() bool { + return ec.requestEncodingUnsupported && ec.policyChain.RequiresRequestBody && !ec.requestHasNoBody() +} + +// responseEncodingBlocksBodyPolicies is the response-side counterpart of +// requestEncodingBlocksBodyPolicies. Header and body phases must agree on it: a response the +// header phase let through must not be failed later at the body phase. +func (ec *PolicyExecutionContext) responseEncodingBlocksBodyPolicies() bool { + return ec.responseEncodingUnsupported && ec.policyChain.RequiresResponseBody && !ec.responseHasNoBody() +} + // processRequestBodyForEmptyRequest executes body policies inline during the headers phase // for requests that carry no body. The body context is set to Present=false / EndOfStream=true // so policies can inspect headers-only state and short-circuit if necessary. @@ -1092,7 +1107,7 @@ func (ec *PolicyExecutionContext) processResponseHeaders( // headers have not been forwarded downstream yet, so this becomes a clean 502 // rather than a mid-stream reset. As on the request side, this only applies // when the chain actually inspects the response body. - if ec.responseEncodingUnsupported && ec.policyChain.RequiresResponseBody && !ec.responseHasNoBody() { + if ec.responseEncodingBlocksBodyPolicies() { return ec.rejectUnsupportedResponseEncoding(ctx), nil } @@ -1153,8 +1168,10 @@ func (ec *PolicyExecutionContext) processResponseBody( // Defence in depth. processResponseHeaders already rejected this response, so // reaching the body phase with an undecodable encoding means the header-phase // guard was bypassed or removed. Fail the stream rather than fall through to - // policies that would receive compressed bytes as if they were plaintext. - if ec.responseEncodingUnsupported && ec.policyChain.RequiresResponseBody { + // policies that would receive compressed bytes as if they were plaintext. Uses the same + // predicate as the header phase so a response that phase deliberately let through (no + // body to decode) is not failed here instead. + if ec.responseEncodingBlocksBodyPolicies() { return nil, fmt.Errorf("response body phase reached with undecodable Content-Encoding; refusing to run body policies") }