Add limit http body reader

go.sum

@@ -406,8 +406,6 @@ github.com/ooni/minivpn v0.0.7 h1:fRL6lOivKM+Q23HcN/FFiBftbKTAtz7U8r6cOypBAeM=
 github.com/ooni/minivpn v0.0.7/go.mod h1:0KNwmK2Wg9lDbk936XjtxvCq4tPNbK4C3IJvyLwIMrE=
 github.com/ooni/netem v0.0.0-20250905215919-3882eda4fb66 h1:i2sw2lTJKT4dbeTNW8GD+bKT8CFLdjZT6sLxWlKZbrw=
 github.com/ooni/netem v0.0.0-20250905215919-3882eda4fb66/go.mod h1:CKKtaYpxFRzD2I0Cxlkz9msGyc6p/ZkOLu6TfXneDyU=
-github.com/ooni/probe-assets v0.28.0 h1:r1hlPmC9PwfKqX0of9T0tx9xTIczPvqx5N6Bt0LK2LA=
-github.com/ooni/probe-assets v0.28.0/go.mod h1:m0k2FFzcLfFm7dhgyYkLCUR3R0CoRPr0jcjctDS2+gU=
 github.com/ooni/probe-assets v0.29.0 h1:+jUBn5xZ5bHOP4++0tAd6J7w6kLS1q2fFeKyAXsfMFE=
 github.com/ooni/probe-assets v0.29.0/go.mod h1:m0k2FFzcLfFm7dhgyYkLCUR3R0CoRPr0jcjctDS2+gU=
 github.com/oschwald/geoip2-golang v1.9.0 h1:uvD3O6fXAXs+usU+UGExshpdP13GAqp4GBrzN7IgKZc=

internal/bytecounter/http_test.go

@@ -94,7 +94,7 @@ func TestHTTPTransport(t *testing.T) {
 			if err != nil {
 				t.Fatal(err)
 			}
-			data, err := netxlite.ReadAllContext(context.Background(), resp.Body)
+			data, err := netxlite.ReadAllContext(context.Background(), netxlite.LimitBodyReader(resp))
 			if err != nil {
 				t.Fatal(err)
 			}
@@ -141,7 +141,7 @@ func TestHTTPTransport(t *testing.T) {
 			if err != nil {
 				t.Fatal(err)
 			}
-			data, err := netxlite.ReadAllContext(context.Background(), resp.Body)
+			data, err := netxlite.ReadAllContext(context.Background(), netxlite.LimitBodyReader(resp))
 			if err != nil {
 				t.Fatal(err)
 			}

internal/cmd/gardener/internal/aggregationapi/aggregationapi.go

@@ -57,7 +57,7 @@ func Query(
 	runtimex.Assert(resp.StatusCode == 200, "aggregationapi: http request failed")
 
 	// read the response body
-	data := runtimex.Try1(netxlite.ReadAllContext(ctx, resp.Body))
+	data := runtimex.Try1(netxlite.ReadAllContext(ctx, netxlite.LimitBodyReader(resp)))
 
 	// parse the response body
 	var apiResp Response

internal/cmd/oohelper/internal/client.go

@@ -130,7 +130,7 @@ func (oo OOClient) Do(ctx context.Context, config OOConfig) (*CtrlResponse, erro
 	if resp.StatusCode != 200 {
 		return nil, ErrHTTPStatusCode
 	}
-	data, err = netxlite.ReadAllContext(ctx, resp.Body)
+	data, err = netxlite.ReadAllContext(ctx, netxlite.LimitBodyReader(resp))
 	if err != nil {
 		return nil, err
 	}

internal/experiment/dash/collect.go

@@ -53,7 +53,7 @@ func collect(ctx context.Context, baseURL, authorization string,
 	// read, parse, and ignore the response body. Historically the
 	// most userful data has always been on the server side, therefore,
 	// it doesn't matter much that we're discarding server results.
-	data, err = netxlite.ReadAllContext(ctx, resp.Body)
+	data, err = netxlite.ReadAllContext(ctx, netxlite.LimitBodyReader(resp))
 	if err != nil {
 		return err
 	}

internal/experiment/dash/download.go

@@ -89,7 +89,7 @@ func download(ctx context.Context, config downloadConfig) (downloadResult, error
 	}
 
 	// read the response body
-	data, err := netxlite.ReadAllContext(ctx, resp.Body)
+	data, err := netxlite.ReadAllContext(ctx, netxlite.LimitBodyReader(resp))
 	if err != nil {
 		return result, err
 	}

internal/experiment/dash/model.go

@@ -25,7 +25,7 @@ const (
 	testName = "dash"
 
 	// testVersion is the version of the experiment.
-	testVersion = "0.14.0"
+	testVersion = "0.14.1"
 
 	// totalStep is the total number of steps we should run
 	// during the download experiment.

internal/experiment/dash/negotiate.go

@@ -49,7 +49,7 @@ func negotiate(
 	}
 
 	// read the response body
-	data, err = netxlite.ReadAllContext(ctx, resp.Body)
+	data, err = netxlite.ReadAllContext(ctx, netxlite.LimitBodyReader(resp))
 	if err != nil {
 		return negotiateResp, err
 	}

internal/experiment/hhfm/hhfm.go

@@ -22,7 +22,7 @@ import (
 
 const (
 	testName    = "http_header_field_manipulation"
-	testVersion = "0.2.0"
+	testVersion = "0.2.1"
 )
 
 // Config contains the experiment config.
@@ -192,7 +192,7 @@ func transact(txp Transport, req *http.Request,
 		return nil, nil, urlgetter.ErrHTTPRequestFailed
 	}
 	callbacks.OnProgress(0.75, "reading response body...")
-	data, err := netxlite.ReadAllContext(req.Context(), resp.Body)
+	data, err := netxlite.ReadAllContext(req.Context(), netxlite.LimitBodyReader(resp))
 	callbacks.OnProgress(1.00, fmt.Sprintf("got reseponse body... %+v", err))
 	if err != nil {
 		return nil, nil, err

internal/experiment/ndt7/download.go

@@ -72,7 +72,8 @@ func (mgr downloadManager) doRun(ctx context.Context) error {
 			return err
 		}
 		if kind == websocket.TextMessage {
-			data, err := netxlite.ReadAllContext(ctx, reader)
+			limitReader := io.LimitReader(reader, netxlite.MaxPayloadSize)
+			data, err := netxlite.ReadAllContext(ctx, limitReader)
 			if err != nil {
 				return err
 			}

internal/experiment/ndt7/ndt7.go

@@ -15,7 +15,7 @@ import (
 
 const (
 	testName    = "ndt"
-	testVersion = "0.10.1"
+	testVersion = "0.10.2"
 )
 
 // Config contains the experiment settings

internal/netxlite/httplimit.go

@@ -0,0 +1,18 @@
+package netxlite
+
+import (
+	"io"
+	"net/http"
+)
+
+// MaxPayloadSize is 128MB
+const MaxPayloadSize = int64(1 << 27)
+
+// LimitBodyReader returns a LimitedReader capped at min(MaxPayloadSize, http.Response.ContentLength)
+func LimitBodyReader(resp *http.Response) io.Reader {
+	size := MaxPayloadSize
+	if (0 < resp.ContentLength) && (resp.ContentLength < MaxPayloadSize) {
+		size = resp.ContentLength
+	}
+	return io.LimitReader(resp.Body, size)
+}

internal/netxlite/httplogger_test.go

@@ -80,7 +80,7 @@ func TestHTTPTransportLogger(t *testing.T) {
 			if err != nil {
 				t.Fatal(err)
 			}
-			ReadAllContext(context.Background(), resp.Body)
+			ReadAllContext(context.Background(), LimitBodyReader(resp))
 			resp.Body.Close()
 			if count < 1 {
 				t.Fatal("no logs?!")

internal/netxlite/integration_test.go

@@ -621,3 +621,63 @@ func TestHTTP3Transport(t *testing.T) {
 		txp.CloseIdleConnections()
 	})
 }
+
+func TestHTTPTransportWithLimitBodyReader(t *testing.T) {
+	if testing.Short() {
+		t.Skip("skip test in short mode")
+	}
+
+	t.Run("works as intended", func(t *testing.T) {
+		netx := &netxlite.Netx{}
+		d := netx.NewDialerWithResolver(log.Log, netx.NewStdlibResolver(log.Log))
+		td := netxlite.NewTLSDialer(d, netx.NewTLSHandshakerStdlib(log.Log))
+		txp := netxlite.NewHTTPTransport(log.Log, d, td)
+		client := &http.Client{Transport: txp}
+		resp, err := client.Get("https://www.google.com/robots.txt")
+		if err != nil {
+			t.Fatal(err)
+		}
+		resp.Body.Close()
+		client.CloseIdleConnections()
+	})
+	t.Run("we can send larger body than netxlite.MaxPayloadSize", func(t *testing.T) {
+		srvr := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+			header := w.Header()
+			header["Content-Length"] = []string{fmt.Sprintf("%d", 1<<28)}
+			header["Content-Type"] = []string{"application/octet-stream"}
+			w.WriteHeader(200)
+
+			for i := 0; i < 256; i++ {
+				mbBuf := make([]byte, 1<<20)
+				n, err := w.Write(mbBuf)
+				if n < len(mbBuf) || err != nil {
+					t.Log("test server received err", err)
+				}
+			}
+		}))
+		defer srvr.Close()
+		// TODO(https://github.com/ooni/probe/issues/2534): NewHTTPTransportStdlib has QUIRKS but we
+		// don't actually care about those QUIRKS in this context
+		netx := &netxlite.Netx{}
+		txp := netx.NewHTTPTransportStdlib(model.DiscardLogger)
+		req, err := http.NewRequest("GET", srvr.URL, nil)
+		if err != nil {
+			t.Fatal(err)
+		}
+		resp, err := txp.RoundTrip(req)
+		if err != nil {
+			t.Fatal(err)
+		}
+		defer resp.Body.Close()
+		data, err := netxlite.ReadAllContext(req.Context(), netxlite.LimitBodyReader(resp))
+		if err != nil {
+			t.Fatal(err)
+		}
+		if int64(len(data)) > netxlite.MaxPayloadSize {
+			t.Fatal("Body payload exceeded", netxlite.MaxPayloadSize)
+		}
+		if int64(len(data)) != netxlite.MaxPayloadSize {
+			t.Fatal("Body payload was prematurely truncated to ", len(data))
+		}
+	})
+}

internal/netxlite/internal/gencertifi/main.go

@@ -50,7 +50,7 @@ func main() {
 	}
 	defer resp.Body.Close()
 
-	bundle, err := netxlite.ReadAllContext(context.Background(), resp.Body)
+	bundle, err := netxlite.ReadAllContext(context.Background(), netxlite.LimitBodyReader(resp))
 	if err != nil {
 		log.Fatal(err)
 	}

pkg/oonimkall/httpx.go

@@ -71,7 +71,7 @@ func (sess *Session) httpDoLocked(ctx *Context, jreq *HTTPRequest) (*HTTPRespons
 		return nil, errors.New("httpx: HTTP request failed")
 	}
 
-	rawResp, err := netxlite.ReadAllContext(ctx.ctx, resp.Body)
+	rawResp, err := netxlite.ReadAllContext(ctx.ctx, netxlite.LimitBodyReader(resp))
 	if err != nil {
 		return nil, err
 	}