fix(python): raise on mid-stream body errors instead of silent EOF

impit-python/src/response.rs

@@ -63,12 +63,11 @@ impl PyResponseBytesIterator {
                     if let Some(parent) = &slf.parent_response {
                         if let Ok(mut parent_ref) = parent.try_borrow_mut(py) {
                             parent_ref.inner_state = InnerResponseState::StreamingClosed;
+                            parent_ref.is_stream_consumed = true;
                             parent_ref.is_closed = true;
                         }
                     }
-                    Err(pyo3::exceptions::PyStopIteration::new_err(format!(
-                        "Stream error: {e}"
-                    )))
+                    Err(ImpitPyError(ImpitError::from(e, None)).into())
                 }
                 None => {
                     slf.content_returned = true;
@@ -155,13 +154,12 @@ impl PyResponseAsyncBytesIterator {
                         Python::attach(|py| {
                             if let Ok(mut parent_ref) = parent.try_borrow_mut(py) {
                                 parent_ref.inner_state = InnerResponseState::StreamingClosed;
+                                parent_ref.is_stream_consumed = true;
                                 parent_ref.is_closed = true;
                             }
                         });
                     }
-                    Err(pyo3::exceptions::PyStopAsyncIteration::new_err(format!(
-                        "Stream error: {e}"
-                    )))
+                    Err(ImpitPyError(ImpitError::from(e, None)).into())
                 }
                 None => {
                     if let Some(parent) = parent_response {

impit-python/test/async_client_test.py

@@ -7,7 +7,7 @@
 
 import pytest
 
-from impit import AsyncClient, Browser, Cookies, StreamClosed, StreamConsumed, TooManyRedirects
+from impit import AsyncClient, Browser, Cookies, RemoteProtocolError, StreamClosed, StreamConsumed, TooManyRedirects
 
 from .httpbin import get_httpbin_url
 from .setup_proxy import start_proxy_server
@@ -30,6 +30,22 @@ def thread_server(port_holder: list[int]) -> None:
     server.close()
 
 
+def truncating_server(port_holder: list[int]) -> None:
+    """Announce a `Content-Length` larger than the body actually sent, then close the socket mid-body."""
+    server = socket.socket(socket.AF_INET6, socket.SOCK_STREAM)
+    server.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
+    server.setsockopt(socket.IPPROTO_IPV6, socket.IPV6_V6ONLY, 0)
+    server.bind(('::', 0))
+    port_holder[0] = server.getsockname()[1]
+    server.listen(1)
+
+    conn, _ = server.accept()
+    conn.recv(1024)
+    conn.send(b'HTTP/1.1 200 OK\r\nContent-Length: 100\r\n\r\n0123456789')
+    conn.close()
+    server.close()
+
+
 @pytest.mark.asyncio
 @pytest.mark.parametrize(
     ('browser', 'ja4'),
@@ -638,3 +654,19 @@ async def test_iter_bytes_without_consumed(self, browser: Browser) -> None:
 
         with pytest.raises(StreamClosed):
             _ = response.content
+
+    async def test_truncated_stream_raises(self, browser: Browser) -> None:
+        port_holder = [0]
+        thread = threading.Thread(target=truncating_server, args=(port_holder,))
+        thread.start()
+        await asyncio.sleep(0.1)
+
+        impit = AsyncClient(browser=browser)
+
+        async with impit.stream('GET', f'http://localhost:{port_holder[0]}/', timeout=5) as response:
+            assert response.status_code == 200
+
+            with pytest.raises(RemoteProtocolError):
+                _ = b''.join([item async for item in response.aiter_bytes()])
+
+        thread.join()

impit-python/test/basic_client_test.py

@@ -14,6 +14,7 @@
     ConnectTimeout,
     Cookies,
     ReadTimeout,
+    RemoteProtocolError,
     StreamClosed,
     StreamConsumed,
     TimeoutException,
@@ -41,6 +42,22 @@ def thread_server(port_holder: list[int]) -> None:
     server.close()
 
 
+def truncating_server(port_holder: list[int]) -> None:
+    """Announce a `Content-Length` larger than the body actually sent, then close the socket mid-body."""
+    server = socket.socket(socket.AF_INET6, socket.SOCK_STREAM)
+    server.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
+    server.setsockopt(socket.IPPROTO_IPV6, socket.IPV6_V6ONLY, 0)
+    server.bind(('::', 0))
+    port_holder[0] = server.getsockname()[1]
+    server.listen(1)
+
+    conn, _ = server.accept()
+    conn.recv(1024)
+    conn.send(b'HTTP/1.1 200 OK\r\nContent-Length: 100\r\n\r\n0123456789')
+    conn.close()
+    server.close()
+
+
 @pytest.mark.parametrize(
     ('browser', 'ja4'),
     [
@@ -611,6 +628,22 @@ def test_iter_bytes_without_consumed(self, browser: Browser) -> None:
         with pytest.raises(StreamClosed):
             _ = response.content
 
+    def test_truncated_stream_raises(self, browser: Browser) -> None:
+        port_holder = [0]
+        thread = threading.Thread(target=truncating_server, args=(port_holder,))
+        thread.start()
+        time.sleep(0.1)
+
+        impit = Client(browser=browser)
+
+        with impit.stream('GET', f'http://localhost:{port_holder[0]}/', timeout=5) as response:
+            assert response.status_code == 200
+
+            with pytest.raises(RemoteProtocolError):
+                _ = b''.join(response.iter_bytes())
+
+        thread.join()
+
 
 def make_slow_server(port_holder: list[int], delay: float = 2.0) -> None:
     """Start a server in a daemon thread that waits `delay` seconds before responding."""

impit/src/errors.rs

@@ -122,7 +122,9 @@ impl ImpitError {
             return ImpitError::TooManyRedirects(context.max_redirects);
         }
 
-        if error.is_body() && (format!("{:?}", error).to_lowercase()).contains("unexpectedeof") {
+        if (error.is_body() || error.is_decode())
+            && (format!("{:?}", error).to_lowercase()).contains("unexpectedeof")
+        {
             return ImpitError::RemoteProtocolError;
         }