adds a decompressing roundtripper

This roundtripper makes sure to hand over to the application
code a decompressed http.Request.Body. This centralizes
the decompression at the reverse proxy layer.

Author: vroldanbet

pkg/authz/responsefilterer.go

@@ -2,7 +2,6 @@ package authz
 
 import (
 	"bytes"
-	"compress/gzip"
 	"context"
 	"encoding/json"
 	"fmt"
@@ -234,30 +233,7 @@ func (rf *StandardResponseFilterer) FilterResp(resp *http.Response) error {
 			return nil
 		}
 
-		bodyStream := resp.Body
-
-		// NOTE: we need to manually check for gzipped encoding here
-		// because we're proxying - the request/response context would
-		// know about the encoding if we were the ones originating the
-		// request, but since we aren't, we need to manually check this.
-		// This is needed because the k8s API will automatically gzip responses
-		// above 128kb by default.
-		if resp.Header.Get("Content-Encoding") == "gzip" {
-			bodyStream, err = gzip.NewReader(bodyStream)
-			if err != nil {
-				return err
-			}
-			defer func() {
-				_ = bodyStream.Close()
-			}()
-			// We're decompressing the body here, so remove the header.
-			// If we leave it, the downstream HTTP transport will try to
-			// gzip-decompress the already-plain bytes and fail with
-			// "gzip: invalid header".
-			resp.Header.Del("Content-Encoding")
-		}
-
-		body, err := io.ReadAll(bodyStream)
+		body, err := io.ReadAll(resp.Body)
 		if err != nil {
 			return err
 		}

pkg/proxy/embedded_test.go

@@ -558,14 +558,9 @@ func TestGzippedUpstreamResponse(t *testing.T) {
 
 	body, err := io.ReadAll(resp.Body)
 	require.NoError(t, err)
-	require.Equal(t, "gzip", resp.Header.Get("Content-Encoding"), "expecting the response to be gzipped")
+	require.Empty(t, resp.Header.Get("Content-Encoding"),
+		"decompressingTransport must strip Content-Encoding: gzip after decompression")
 	require.Equal(t, responseText+"\n", string(body), "unexpected body value")
-
-	// The proxy should relay the response successfully.
-	// A 502 Bad Gateway here means FilterResp attempted to decode the gzip
-	// bytes as JSON/protobuf without first decompressing them. The fix is to
-	// detect Content-Encoding: gzip in FilterResp and decompress the body
-	// before passing it to the codec decoder.
 	require.Equal(t, http.StatusOK, resp.StatusCode,
 		"proxy must handle gzip-encoded upstream responses; a 502 indicates the body was not decompressed before decoding")
 }

pkg/proxy/server.go

@@ -1,8 +1,11 @@
 package proxy
 
 import (
+	"bytes"
+	"compress/gzip"
 	"context"
 	"fmt"
+	"io"
 	"net/http"
 	"net/http/httputil"
 	"os"
@@ -110,7 +113,7 @@ func NewServer(ctx context.Context, c *CompletedConfig) (*Server, error) {
 			}
 			return responseFilterer.FilterResp(response)
 		},
-		Transport: transport,
+		Transport: &decompressingTransport{base: transport},
 		ErrorHandler: func(writer http.ResponseWriter, h *http.Request, err error) {
 			klog.V(3).InfoSDepth(1, "upstream Kubernetes API error response", "error", err)
 			writer.WriteHeader(http.StatusBadGateway)
@@ -387,3 +390,39 @@ func (t *authHeaderTransport) RoundTrip(req *http.Request) (*http.Response, erro
 
 	return t.base.RoundTrip(newReq)
 }
+
+// decompressingTransport wraps an http.RoundTripper and transparently decompresses
+// Content-Encoding: gzip responses from the upstream.
+type decompressingTransport struct {
+	base http.RoundTripper
+}
+
+func (t *decompressingTransport) RoundTrip(req *http.Request) (*http.Response, error) {
+	base := t.base
+	if base == nil {
+		base = http.DefaultTransport
+	}
+
+	resp, err := base.RoundTrip(req)
+	if err != nil || resp == nil || resp.Header.Get("Content-Encoding") != "gzip" {
+		return resp, err
+	}
+
+	gr, err := gzip.NewReader(resp.Body)
+	if err != nil {
+		_ = resp.Body.Close()
+		return nil, fmt.Errorf("decompressingTransport: gzip.NewReader: %w", err)
+	}
+
+	body, err := io.ReadAll(gr)
+	_ = gr.Close()
+	_ = resp.Body.Close()
+	if err != nil {
+		return nil, fmt.Errorf("decompressingTransport: reading gzip body: %w", err)
+	}
+
+	resp.Body = io.NopCloser(bytes.NewReader(body))
+	resp.Header.Del("Content-Encoding")
+	resp.ContentLength = int64(len(body))
+	return resp, nil
+}

pkg/proxy/transport_test.go

@@ -0,0 +1,80 @@
+package proxy
+
+import (
+	"bytes"
+	"compress/gzip"
+	"io"
+	"net/http"
+	"testing"
+
+	"github.com/stretchr/testify/require"
+)
+
+// TestDecompressingTransport verifies that decompressingTransport transparently
+// decompresses Content-Encoding: gzip responses and strips the header.
+//
+// This prevents the "gzip: invalid header" error that occurs when a real
+// http.Transport (used by kubernetes.Clientset) receives a response with
+// Content-Encoding: gzip but a body that has already been decompressed by
+// FilterResp. The transport would wrap the body in a gzip.Reader and fail.
+func TestDecompressingTransport(t *testing.T) {
+	t.Parallel()
+
+	plain := []byte(`{"apiVersion":"v1","kind":"Secret","metadata":{"name":"s","namespace":"default"}}`)
+
+	var compressed bytes.Buffer
+	gzw := gzip.NewWriter(&compressed)
+	_, err := gzw.Write(plain)
+	require.NoError(t, err)
+	require.NoError(t, gzw.Close())
+
+	t.Run("decompresses gzip and strips header", func(t *testing.T) {
+		t.Parallel()
+
+		base := roundTripFunc(func(_ *http.Request) (*http.Response, error) {
+			return &http.Response{
+				StatusCode:    http.StatusOK,
+				Header:        http.Header{"Content-Encoding": []string{"gzip"}, "Content-Type": []string{"application/json"}},
+				Body:          io.NopCloser(bytes.NewReader(compressed.Bytes())),
+				ContentLength: int64(compressed.Len()),
+			}, nil
+		})
+
+		resp, err := (&decompressingTransport{base: base}).RoundTrip(&http.Request{}) //nolint: bodyclose
+		require.NoError(t, err)
+		require.Empty(t, resp.Header.Get("Content-Encoding"),
+			"Content-Encoding must be stripped after decompression")
+		require.Equal(t, int64(len(plain)), resp.ContentLength)
+
+		got, err := io.ReadAll(resp.Body)
+		require.NoError(t, err)
+		require.Equal(t, plain, got)
+	})
+
+	t.Run("non-gzip response passes through unchanged", func(t *testing.T) {
+		t.Parallel()
+
+		base := roundTripFunc(func(_ *http.Request) (*http.Response, error) {
+			return &http.Response{
+				StatusCode: http.StatusOK,
+				Header:     http.Header{"Content-Type": []string{"application/json"}},
+				Body:       io.NopCloser(bytes.NewReader(plain)),
+			}, nil
+		})
+
+		resp, err := (&decompressingTransport{base: base}).RoundTrip(&http.Request{}) //nolint: bodyclose
+		require.NoError(t, err)
+		require.Empty(t, resp.Header.Get("Content-Encoding"))
+
+		got, err := io.ReadAll(resp.Body)
+		require.NoError(t, err)
+		require.Equal(t, plain, got)
+	})
+}
+
+// roundTripFunc is a helper that implements http.RoundTripper via a function.
+type roundTripFunc func(*http.Request) (*http.Response, error)
+
+func (f roundTripFunc) RoundTrip(req *http.Request) (*http.Response, error) {
+	return f(req)
+}