x

internal/experiment/webconnectivitylte/cleartextflow.go

@@ -267,7 +267,7 @@ func (t *CleartextFlow) httpTransaction(ctx context.Context, network, address, a
 			t.CookieJar.SetCookies(req.URL, cookies)
 		}
 		reader := io.LimitReader(resp.Body, maxbody)
-		body, err = StreamAllContext(ctx, reader)
+		body, err = netxlite.StreamAllContext(ctx, reader)
 	}
 	if err == nil && httpRedirectIsRedirect(resp) {
 		err = httpValidateRedirect(resp)

internal/experiment/webconnectivitylte/secureflow.go

@@ -322,7 +322,7 @@ func (t *SecureFlow) httpTransaction(ctx context.Context, network, address, alpn
 			t.CookieJar.SetCookies(req.URL, cookies)
 		}
 		reader := io.LimitReader(resp.Body, maxbody)
-		body, err = StreamAllContext(ctx, reader)
+		body, err = netxlite.StreamAllContext(ctx, reader)
 	}
 	if err == nil && httpRedirectIsRedirect(resp) {
 		err = httpValidateRedirect(resp)

internal/experiment/webconnectivityqa/testcase.go

@@ -92,6 +92,8 @@ func AllTestCases() []*TestCase {
 		tcpBlockingConnectTimeout(),
 		tcpBlockingConnectionRefusedWithInconsistentDNS(),
 
+		throttlingWithHTTPS(),
+
 		tlsBlockingConnectionResetWithConsistentDNS(),
 		tlsBlockingConnectionResetWithInconsistentDNS(),
 

internal/experiment/webconnectivityqa/throttling.go

@@ -0,0 +1,36 @@
+package webconnectivityqa
+
+import (
+	"time"
+
+	"github.com/apex/log"
+	"github.com/ooni/netem"
+	"github.com/ooni/probe-cli/v3/internal/netemx"
+)
+
+// throttlingWithHTTPS is the case where the website has throttling configured for it.
+func throttlingWithHTTPS() *TestCase {
+	return &TestCase{
+		Name:  "throttlingWithHTTPS",
+		Flags: TestCaseFlagNoV04,
+		Input: "https://largefile.com/",
+		Configure: func(env *netemx.QAEnv) {
+
+			env.DPIEngine().AddRule(&netem.DPIThrottleTrafficForTLSSNI{
+				Delay:  300 * time.Millisecond,
+				Logger: log.Log,
+				PLR:    0.2,
+				SNI:    "largefile.com",
+			})
+
+		},
+		ExpectErr: false,
+		ExpectTestKeys: &testKeys{
+			DNSConsistency:        "consistent",
+			HTTPExperimentFailure: "generic_timeout_error",
+			XBlockingFlags:        8, // AnalysisBlockingFlagHTTPBlocking
+			Accessible:            false,
+			Blocking:              "http-failure",
+		},
+	}
+}

internal/netxlite/iox_test.go

@@ -270,3 +270,133 @@ func TestCopyContext(t *testing.T) {
 		wg.Wait()
 	})
 }
+
+func TestStreamAllContext(t *testing.T) {
+	t.Run("with success and background context", func(t *testing.T) {
+		r := strings.NewReader("deadbeef")
+		ctx := context.Background()
+		out, err := StreamAllContext(ctx, r)
+		if err != nil {
+			t.Fatal(err)
+		}
+		if len(out) != 8 {
+			t.Fatal("not the expected number of bytes")
+		}
+	})
+
+	t.Run("with success and wrapped io.EOF", func(t *testing.T) {
+		// See https://github.com/ooni/probe/issues/1965
+		wg := &sync.WaitGroup{}
+		wg.Add(1)
+		r := &mocks.Reader{
+			MockRead: func(b []byte) (int, error) {
+				defer wg.Done()
+				// "When Read encounters an error or end-of-file condition
+				// after successfully reading n > 0 bytes, it returns
+				// the number of bytes read. It may return the (non-nil)
+				// error from the same call or return the error (and n == 0)
+				// from a subsequent call.""
+				//
+				// See https://pkg.go.dev/io#Reader
+				//
+				// Note: Returning a wrapped error to ensure we address
+				// https://github.com/ooni/probe/issues/1965
+				return len(b), NewErrWrapper(ClassifyGenericError,
+					ReadOperation, io.EOF)
+			},
+		}
+		out, err := StreamAllContext(context.Background(), r)
+		if err != nil {
+			t.Fatal(err)
+		}
+		if len(out) <= 0 {
+			t.Fatal("we expected to see a positive number of bytes here")
+		}
+		wg.Wait()
+	})
+
+	t.Run("with failure and background context", func(t *testing.T) {
+		expected := errors.New("mocked error")
+		r := &mocks.Reader{
+			MockRead: func(b []byte) (int, error) {
+				return 0, expected
+			},
+		}
+		ctx := context.Background()
+		out, err := StreamAllContext(ctx, r)
+		if !errors.Is(err, expected) {
+			t.Fatal("not the error we expected", err)
+		}
+		var errWrapper *ErrWrapper
+		if !errors.As(err, &errWrapper) {
+			t.Fatal("the returned error is not wrapped")
+		}
+		if len(out) != 0 {
+			t.Fatal("not the expected number of bytes")
+		}
+	})
+
+	t.Run("with success and cancelled context", func(t *testing.T) {
+		wg := &sync.WaitGroup{}
+		wg.Add(1)
+		sigch := make(chan interface{})
+		r := &mocks.Reader{
+			MockRead: func(b []byte) (int, error) {
+				defer wg.Done()
+				<-sigch
+				// "When Read encounters an error or end-of-file condition
+				// after successfully reading n > 0 bytes, it returns
+				// the number of bytes read. It may return the (non-nil)
+				// error from the same call or return the error (and n == 0)
+				// from a subsequent call.""
+				//
+				// See https://pkg.go.dev/io#Reader
+				return len(b), io.EOF
+			},
+		}
+		ctx, cancel := context.WithCancel(context.Background())
+		cancel() // fail immediately
+		out, err := StreamAllContext(ctx, r)
+		if !errors.Is(err, context.Canceled) {
+			t.Fatal("not the error we expected", err)
+		}
+		var errWrapper *ErrWrapper
+		if !errors.As(err, &errWrapper) {
+			t.Fatal("the returned error is not wrapped")
+		}
+		if len(out) != 0 {
+			t.Fatal("not the expected number of bytes")
+		}
+		close(sigch)
+		wg.Wait()
+	})
+
+	t.Run("with failure and cancelled context", func(t *testing.T) {
+		wg := &sync.WaitGroup{}
+		wg.Add(1)
+		sigch := make(chan interface{})
+		expected := errors.New("mocked error")
+		r := &mocks.Reader{
+			MockRead: func(b []byte) (int, error) {
+				defer wg.Done()
+				<-sigch
+				return 0, expected
+			},
+		}
+		ctx, cancel := context.WithCancel(context.Background())
+		cancel() // fail immediately
+		out, err := StreamAllContext(ctx, r)
+		if !errors.Is(err, context.Canceled) {
+			t.Fatal("not the error we expected", err)
+		}
+		var errWrapper *ErrWrapper
+		if !errors.As(err, &errWrapper) {
+			t.Fatal("the returned error is not wrapped")
+		}
+		if len(out) != 0 {
+			t.Fatal("not the expected number of bytes")
+		}
+		close(sigch)
+		wg.Wait()
+	})
+}