Information disclosure / cross-request gzip stream state leakage via pooled gzip.Writer

diff --git a/staging/src/k8s.io/apiserver/pkg/endpoints/handlers/watch.go b/staging/src/k8s.io/apiserver/pkg/endpoints/handlers/watch.go index 25f5a1235579f..9b8d18ce4fb20 100644 --- a/staging/src/k8s.io/apiserver/pkg/endpoints/handlers/watch.go +++ b/staging/src/k8s.io/apiserver/pkg/endpoints/handlers/watch.go @@ -269,36 +269,45 @@ func (p *plainResponseWriter) Close() error { return nil } -var _ watchStreamWriter = &gzipResponseWriter{} +var _ watchStreamWriter = &perFlushGzipWriter{} -type gzipResponseWriter struct { - gw *gzip.Writer - flusher http.Flusher +type perFlushGzipWriter struct { + delegateRW http.ResponseWriter + flusher http.Flusher + gw *gzip.Writer } -func (g *gzipResponseWriter) Write(b []byte) (int, error) { - return g.gw.Write(b) +func (p *perFlushGzipWriter) Write(b []byte) (int, error) { + if p.gw == nil { + p.gw = watchGzipPool.Get().(*gzip.Writer) + p.gw.Reset(p.delegateRW) + } + return p.gw.Write(b) } -// Flush pushes data through two buffers: gzip's internal buffer into the -// http.ResponseWriter, then the http.ResponseWriter's buffer to the network. -func (g *gzipResponseWriter) Flush() error { - if err := g.gw.Flush(); err != nil { +// Flush writes compressed data to the client and releases the gzip.Writer back to the pool. +func (p *perFlushGzipWriter) Flush() error { + // no writer means Write was not called, nothing to flush + if p.gw == nil { + return nil + } + if err := p.gw.Flush(); err != nil { return err } - g.flusher.Flush() - return nil + err := p.Close() + p.flusher.Flush() + return err } -func (g *gzipResponseWriter) Close() error { - if g.gw == nil { +func (p *perFlushGzipWriter) Close() error { + if p.gw == nil { return nil } - err := g.gw.Close() - g.gw.Reset(nil) - watchGzipPool.Put(g.gw) + err := p.gw.Close() + p.gw.Reset(nil) + watchGzipPool.Put(p.gw) // prevent double-close returning the writer to the pool twice - g.gw = nil + p.gw = nil return err } @@ -328,9 +337,7 @@ func (w *watchResponseWriter) BeginStream(mediaType string) { if w.contentEncoding == "gzip" && w.isWatchListRequest { w.delegateRW.Header().Set("Content-Encoding", "gzip") w.delegateRW.Header().Add("Vary", "Accept-Encoding") - gw := watchGzipPool.Get().(*gzip.Writer) - gw.Reset(w.delegateRW) - w.writer = &gzipResponseWriter{gw: gw, flusher: w.flusher} + w.writer = &perFlushGzipWriter{delegateRW: w.delegateRW, flusher: w.flusher} } w.delegateRW.WriteHeader(http.StatusOK) // Flush HTTP headers only @@ -463,6 +470,11 @@ func (s *WatchServer) HandleHTTP(w http.ResponseWriter, req *http.Request) { span.End(5 * time.Second) s.watchListCompleteHook() } + // release the gzip writer back to the pool so idle watches don't hold gzip state. + if err := rw.Flush(); err != nil { + utilruntime.HandleErrorWithContext(req.Context(), err, "Failed to flush watch response after initial events") + return + } } } } diff --git a/staging/src/k8s.io/apiserver/pkg/endpoints/handlers/watchlist_compression_test.go b/staging/src/k8s.io/apiserver/pkg/endpoints/handlers/watchlist_compression_test.go index 65dbe01bcf6e8..0cb51b1a9e708 100644 --- a/staging/src/k8s.io/apiserver/pkg/endpoints/handlers/watchlist_compression_test.go +++ b/staging/src/k8s.io/apiserver/pkg/endpoints/handlers/watchlist_compression_test.go @@ -116,18 +116,34 @@ func TestWatchResponseWriter(t *testing.T) { } func TestWatchResponseWriterDoubleClose(t *testing.T) { - recorder := httptest.NewRecorder() - rw := newWatchResponseWriter(recorder, recorder, "gzip", true) - rw.BeginStream("application/json") - - if _, err := rw.Write([]byte("data")); err != nil { - t.Fatalf("Write failed: %v", err) - } - if err := rw.Close(); err != nil { - t.Fatalf("first Close failed: %v", err) + scenarios := []struct { + name string + newWriter func(*testing.T, *httptest.ResponseRecorder) watchStreamWriter + }{ + { + name: "perFlushGzipWriter", + newWriter: func(t *testing.T, r *httptest.ResponseRecorder) watchStreamWriter { + return &perFlushGzipWriter{delegateRW: r, flusher: r} + }, + }, + { + name: "watchResponseWriter", + newWriter: func(t *testing.T, r *httptest.ResponseRecorder) watchStreamWriter { + rw := newWatchResponseWriter(r, r, "gzip", true) + rw.BeginStream("application/json") + return rw + }, + }, } - if err := rw.Close(); err != nil { - t.Fatalf("second Close failed: %v", err) + + for _, scenario := range scenarios { + t.Run(scenario.name, func(t *testing.T) { + w := scenario.newWriter(t, httptest.NewRecorder()) + _, err := w.Write([]byte("data")) + require.NoError(t, err) + require.NoError(t, w.Close()) + require.NoError(t, w.Close()) + }) } } @@ -156,13 +172,87 @@ func TestGzipNewReaderFailsOnUncompressedContent(t *testing.T) { require.Error(t, err) } +func TestPerFlushGzipWriter(t *testing.T) { + scenarios := []struct { + name string + writes []string + flushPerWrite bool + expected string + }{ + { + name: "single event with flush", + writes: []string{"hello"}, + flushPerWrite: true, + expected: "hello", + }, + { + name: "flush after every write produces concatenated gzip members", + writes: []string{"first", "second", "third"}, + flushPerWrite: true, + expected: "firstsecondthird", + }, + { + name: "flush only at the end produces a single gzip member", + writes: []string{"first", "second", "third"}, + flushPerWrite: false, + expected: "firstsecondthird", + }, + } + + for _, scenario := range scenarios { + t.Run(scenario.name, func(t *testing.T) { + recorder := httptest.NewRecorder() + w := &perFlushGzipWriter{delegateRW: recorder, flusher: recorder} + for _, msg := range scenario.writes { + _, err := w.Write([]byte(msg)) + require.NoError(t, err) + if scenario.flushPerWrite { + require.NoError(t, w.Flush()) + } + } + if !scenario.flushPerWrite { + require.NoError(t, w.Flush()) + } + require.NoError(t, w.Close()) + gr, err := gzip.NewReader(recorder.Body) + require.NoError(t, err) + got, err := io.ReadAll(gr) + require.NoError(t, err) + require.Equal(t, scenario.expected, string(got)) + }) + } +} + func TestWatchServerCompression(t *testing.T) { + simpleEventFn := func(name string) watch.Event { + return watch.Event{ + Type: watch.Added, + Object: &endpointstesting.Simple{ + ObjectMeta: metav1.ObjectMeta{Name: name}, + TypeMeta: metav1.TypeMeta{APIVersion: testGroupV2.String()}, + }, + } + } + bookmarkEventFn := func() watch.Event { + return watch.Event{ + Type: watch.Bookmark, + Object: &endpointstesting.Simple{ + ObjectMeta: metav1.ObjectMeta{ + ResourceVersion: "100", + Annotations: map[string]string{metav1.InitialEventsAnnotationKey: "true"}, + }, + TypeMeta: metav1.TypeMeta{APIVersion: testGroupV2.String()}, + }, + } + } + scenarios := []struct { name string featureGateEnabled bool isWatchListRequest bool acceptEncoding string expectGzip bool + events []watch.Event }{ { name: "watchlist request is compressed", @@ -170,6 +260,19 @@ func TestWatchServerCompression(t *testing.T) { isWatchListRequest: true, acceptEncoding: "gzip", expectGzip: true, + events: []watch.Event{simpleEventFn("obj-1")}, + }, + { + name: "events after initial-events-end bookmark are readable", + featureGateEnabled: true, + isWatchListRequest: true, + acceptEncoding: "gzip", + expectGzip: true, + events: []watch.Event{ + simpleEventFn("pre-bookmark"), + bookmarkEventFn(), + simpleEventFn("post-bookmark"), + }, }, { name: "regular watch request is not compressed", @@ -177,6 +280,7 @@ func TestWatchServerCompression(t *testing.T) { isWatchListRequest: false, acceptEncoding: "gzip", expectGzip: false, + events: []watch.Event{simpleEventFn("obj-1")}, }, { name: "watchlist request without accept-encoding is not compressed", @@ -184,6 +288,7 @@ func TestWatchServerCompression(t *testing.T) { isWatchListRequest: true, acceptEncoding: "", expectGzip: false, + events: []watch.Event{simpleEventFn("obj-1")}, }, { name: "watchlist request with feature gate disabled is not compressed", @@ -191,6 +296,7 @@ func TestWatchServerCompression(t *testing.T) { isWatchListRequest: true, acceptEncoding: "gzip", expectGzip: false, + events: []watch.Event{simpleEventFn("obj-1")}, }, { name: "watchlist request with multi value accept-encoding is compressed", @@ -198,6 +304,7 @@ func TestWatchServerCompression(t *testing.T) { isWatchListRequest: true, acceptEncoding: "deflate, gzip", expectGzip: true, + events: []watch.Event{simpleEventFn("obj-1")}, }, } @@ -205,7 +312,7 @@ func TestWatchServerCompression(t *testing.T) { t.Run(scenario.name, func(t *testing.T) { featuregatetesting.SetFeatureGateDuringTest(t, utilfeature.DefaultFeatureGate, features.WatchListCompression, scenario.featureGateEnabled) - watcher := watch.NewFakeWithOptions(watch.FakeOptions{ChannelSize: 1}) + watcher := watch.NewFakeWithOptions(watch.FakeOptions{ChannelSize: len(scenario.events)}) info, ok := runtime.SerializerInfoForMediaType(codecs.SupportedMediaTypes(), runtime.ContentTypeJSON) require.True(t, ok) @@ -222,7 +329,9 @@ func TestWatchServerCompression(t *testing.T) { isWatchListRequest: scenario.isWatchListRequest, } - watcher.Add(&endpointstesting.Simple{TypeMeta: metav1.TypeMeta{APIVersion: testGroupV2.String()}}) + for _, event := range scenario.events { + watcher.Action(event.Type, event.Object) + } s := httptest.NewServer(serveWatch(watcher, watchServer, nil)) defer s.Close() @@ -251,12 +360,16 @@ func TestWatchServerCompression(t *testing.T) { require.Empty(t, resp.Header.Get("Content-Encoding")) } - var got watchJSON - require.NoError(t, json.NewDecoder(body).Decode(&got)) - require.Equal(t, watch.Added, got.Type) - var obj endpointstesting.Simple - require.NoError(t, json.Unmarshal(got.Object, &obj)) - require.Equal(t, testGroupV2.String(), obj.APIVersion) + decoder := json.NewDecoder(body) + for i, event := range scenario.events { + var got watchJSON + require.NoError(t, decoder.Decode(&got), "decoding event %d", i) + require.Equal(t, event.Type, got.Type, "event %d type", i) + var obj endpointstesting.Simple + require.NoError(t, json.Unmarshal(got.Object, &obj), "unmarshalling event %d", i) + require.Equal(t, testGroupV2.String(), obj.APIVersion, "event %d", i) + require.Equal(t, event.Object.(*endpointstesting.Simple).Name, obj.Name, "event %d", i) + } }) } }