extproc: reject invalid server flow control window increments

Author: nvxbug

internal/xds/httpfilter/extproc/ext_proc.go

@@ -24,6 +24,7 @@ import (
 	"errors"
 	"fmt"
 	"io"
+	"math"
 	"slices"
 	"strings"
 	"sync"
@@ -1513,33 +1514,13 @@ func (cs *clientStream) recvFromProcServerLoop(newStream func(context.Context, .
 		}
 
 		if windowUpdate := resp.GetServerWindowUpdate(); windowUpdate != nil {
-			deltaDownstreamToSidestream := windowUpdate.GetWindowIncrementDownstreamToSidestream()
-			if deltaDownstreamToSidestream != 0 {
-				// Add the window update to the existing window.
-				newQuota := cs.downstreamToSidestreamWindow.Add(deltaDownstreamToSidestream)
-				previousQuota := newQuota - deltaDownstreamToSidestream
-				// If the window turned from negative to positive, send a signal to the
-				// `acquireDownstreamToSidestreamWindow` function to wake it up.
-				if previousQuota <= 0 && newQuota > 0 {
-					select {
-					case cs.downstreamToSidestreamPositiveUpdateCh <- struct{}{}:
-					default:
-					}
-				}
+			if err := applyServerWindowUpdate(&cs.downstreamToSidestreamWindow, cs.downstreamToSidestreamPositiveUpdateCh, windowUpdate.GetWindowIncrementDownstreamToSidestream()); err != nil {
+				cs.failProcStream(fmt.Errorf("external processor sent an invalid downstream-to-sidestream window update: %v", err))
+				return
 			}
-			deltaUpstreamToSidestream := windowUpdate.GetWindowIncrementUpstreamToSidestream()
-			if deltaUpstreamToSidestream != 0 {
-				// Add the window update to the existing window.
-				newQuota := cs.upstreamToSidestreamWindow.Add(deltaUpstreamToSidestream)
-				previousQuota := newQuota - deltaUpstreamToSidestream
-				// If the window turned from negative to positive, send a signal to the
-				// `acquireUpstreamToSidestreamWindow` function to wake it up.
-				if previousQuota <= 0 && newQuota > 0 {
-					select {
-					case cs.upstreamToSidestreamPositiveUpdateCh <- struct{}{}:
-					default:
-					}
-				}
+			if err := applyServerWindowUpdate(&cs.upstreamToSidestreamWindow, cs.upstreamToSidestreamPositiveUpdateCh, windowUpdate.GetWindowIncrementUpstreamToSidestream()); err != nil {
+				cs.failProcStream(fmt.Errorf("external processor sent an invalid upstream-to-sidestream window update: %v", err))
+				return
 			}
 		}
 
@@ -2057,6 +2038,43 @@ func (cs *clientStream) waitForTrailerProcessing(recvErr error) error {
 	}
 }
 
+// applyServerWindowUpdate adds a flow control window increment received from
+// the external processor to window, waking a blocked acquirer if the window
+// becomes positive. The increment is server-controlled, so it must be
+// non-negative and must not overflow the int64 window; an increment that
+// violates either is a protocol error and returns one instead of wrapping the
+// accounting, matching how HTTP/2 flow control rejects a window update that
+// exceeds the maximum. This recv loop is the only writer that grows the
+// window, so the compare-and-swap only contends with a concurrent deduction in
+// acquire{Downstream,Upstream}ToSidestreamWindow.
+func applyServerWindowUpdate(window *atomic.Int64, positiveUpdateCh chan struct{}, delta int64) error {
+	if delta == 0 {
+		return nil
+	}
+	if delta < 0 {
+		return fmt.Errorf("negative window increment %d", delta)
+	}
+	for {
+		previousQuota := window.Load()
+		if previousQuota > math.MaxInt64-delta {
+			return fmt.Errorf("window increment %d overflows the current window %d", delta, previousQuota)
+		}
+		newQuota := previousQuota + delta
+		if !window.CompareAndSwap(previousQuota, newQuota) {
+			continue
+		}
+		// If the window turned from non-positive to positive, wake the acquirer
+		// blocked in acquire{Downstream,Upstream}ToSidestreamWindow.
+		if previousQuota <= 0 && newQuota > 0 {
+			select {
+			case positiveUpdateCh <- struct{}{}:
+			default:
+			}
+		}
+		return nil
+	}
+}
+
 // acquireDownstreamToSidestreamWindow checks available flow control window for
 // downstream to sidestream. If window is positive, it deducts bodySize and
 // returns true immediately. If window is <= 0, it blocks until a positive

internal/xds/httpfilter/extproc/ext_proc_test.go

@@ -0,0 +1,101 @@
+/*
+ *
+ * Copyright 2026 gRPC authors.
+ *
+ * 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 extproc
+
+import (
+	"math"
+	"sync/atomic"
+	"testing"
+
+	iextproc "google.golang.org/grpc/internal/xds/httpfilter/extproc/internal"
+)
+
+// TestApplyServerWindowUpdate verifies that a flow control window increment
+// from the external processor is applied only when it is a non-negative value
+// that does not overflow the window. An out-of-range increment must be
+// rejected without mutating the window, so the accounting cannot wrap to a
+// value that permanently blocks acquireDownstreamToSidestreamWindow.
+func TestApplyServerWindowUpdate(t *testing.T) {
+	tests := []struct {
+		name       string
+		start      int64
+		delta      int64
+		wantErr    bool
+		wantWindow int64
+		wantSignal bool
+	}{
+		{
+			name:       "zero increment is a no-op",
+			start:      iextproc.DefaultFlowControlWindowSize,
+			delta:      0,
+			wantWindow: iextproc.DefaultFlowControlWindowSize,
+		},
+		{
+			name:       "valid increment grows the window",
+			start:      iextproc.DefaultFlowControlWindowSize,
+			delta:      1024,
+			wantWindow: iextproc.DefaultFlowControlWindowSize + 1024,
+		},
+		{
+			name:       "increment that crosses zero signals the acquirer",
+			start:      -1024,
+			delta:      2048,
+			wantWindow: 1024,
+			wantSignal: true,
+		},
+		{
+			name:       "overflowing increment is rejected and leaves the window unchanged",
+			start:      iextproc.DefaultFlowControlWindowSize,
+			delta:      math.MaxInt64,
+			wantErr:    true,
+			wantWindow: iextproc.DefaultFlowControlWindowSize,
+		},
+		{
+			name:       "negative increment is rejected and leaves the window unchanged",
+			start:      iextproc.DefaultFlowControlWindowSize,
+			delta:      -1,
+			wantErr:    true,
+			wantWindow: iextproc.DefaultFlowControlWindowSize,
+		},
+	}
+	for _, test := range tests {
+		t.Run(test.name, func(t *testing.T) {
+			var window atomic.Int64
+			window.Store(test.start)
+			ch := make(chan struct{}, 1)
+
+			err := applyServerWindowUpdate(&window, ch, test.delta)
+			if (err != nil) != test.wantErr {
+				t.Fatalf("applyServerWindowUpdate(%d, %d) error = %v, wantErr %v", test.start, test.delta, err, test.wantErr)
+			}
+			if got := window.Load(); got != test.wantWindow {
+				t.Errorf("window = %d, want %d", got, test.wantWindow)
+			}
+			gotSignal := false
+			select {
+			case <-ch:
+				gotSignal = true
+			default:
+			}
+			if gotSignal != test.wantSignal {
+				t.Errorf("positive-update signal = %v, want %v", gotSignal, test.wantSignal)
+			}
+		})
+	}
+}