Enforce HTTP status range in transport responses

Co-authored-by: Shri Sukhani <shrisukhani@users.noreply.github.com>

Author: cursoragent

hyperbrowser/transport/async_transport.py

@@ -15,6 +15,9 @@
 class AsyncTransport(AsyncTransportStrategy):
     """Asynchronous transport implementation using httpx"""
 
+    _MIN_HTTP_STATUS_CODE = 100
+    _MAX_HTTP_STATUS_CODE = 599
+
     def __init__(self, api_key: str, headers: Optional[Mapping[str, str]] = None):
         if not isinstance(api_key, str):
             raise HyperbrowserError("api_key must be a string")
@@ -42,7 +45,14 @@ def _normalize_response_status_code(self, response: httpx.Response) -> int:
             status_code = response.status_code
             if isinstance(status_code, bool):
                 raise TypeError("boolean status code is invalid")
-            return int(status_code)
+            normalized_status_code = int(status_code)
+            if not (
+                self._MIN_HTTP_STATUS_CODE
+                <= normalized_status_code
+                <= self._MAX_HTTP_STATUS_CODE
+            ):
+                raise ValueError("status code is outside HTTP range")
+            return normalized_status_code
         except HyperbrowserError:
             raise
         except Exception as exc:

hyperbrowser/transport/sync.py

@@ -15,6 +15,9 @@
 class SyncTransport(SyncTransportStrategy):
     """Synchronous transport implementation using httpx"""
 
+    _MIN_HTTP_STATUS_CODE = 100
+    _MAX_HTTP_STATUS_CODE = 599
+
     def __init__(self, api_key: str, headers: Optional[Mapping[str, str]] = None):
         if not isinstance(api_key, str):
             raise HyperbrowserError("api_key must be a string")
@@ -41,7 +44,14 @@ def _normalize_response_status_code(self, response: httpx.Response) -> int:
             status_code = response.status_code
             if isinstance(status_code, bool):
                 raise TypeError("boolean status code is invalid")
-            return int(status_code)
+            normalized_status_code = int(status_code)
+            if not (
+                self._MIN_HTTP_STATUS_CODE
+                <= normalized_status_code
+                <= self._MAX_HTTP_STATUS_CODE
+            ):
+                raise ValueError("status code is outside HTTP range")
+            return normalized_status_code
         except HyperbrowserError:
             raise
         except Exception as exc:

tests/test_transport_response_handling.py

@@ -78,6 +78,20 @@ def json(self):
         return {}
 
 
+class _OutOfRangeStatusNoContentResponse:
+    content = b""
+    text = ""
+
+    def __init__(self, status_code: int) -> None:
+        self.status_code = status_code
+
+    def raise_for_status(self) -> None:
+        return None
+
+    def json(self):
+        return {}
+
+
 class _BrokenStatusCodeHttpErrorResponse:
     content = b""
     text = "status error"
@@ -171,6 +185,22 @@ def test_sync_handle_response_with_http_status_error_and_broken_status_code():
         transport.close()
 
 
+@pytest.mark.parametrize("status_code", [99, 600])
+def test_sync_handle_response_with_out_of_range_status_raises_hyperbrowser_error(
+    status_code: int,
+):
+    transport = SyncTransport(api_key="test-key")
+    try:
+        with pytest.raises(
+            HyperbrowserError, match="Failed to process response status code"
+        ):
+            transport._handle_response(
+                _OutOfRangeStatusNoContentResponse(status_code)  # type: ignore[arg-type]
+            )
+    finally:
+        transport.close()
+
+
 def test_sync_handle_response_with_request_error_includes_method_and_url():
     transport = SyncTransport(api_key="test-key")
     try:
@@ -322,6 +352,25 @@ async def run() -> None:
     asyncio.run(run())
 
 
+@pytest.mark.parametrize("status_code", [99, 600])
+def test_async_handle_response_with_out_of_range_status_raises_hyperbrowser_error(
+    status_code: int,
+):
+    async def run() -> None:
+        transport = AsyncTransport(api_key="test-key")
+        try:
+            with pytest.raises(
+                HyperbrowserError, match="Failed to process response status code"
+            ):
+                await transport._handle_response(
+                    _OutOfRangeStatusNoContentResponse(status_code)  # type: ignore[arg-type]
+                )
+        finally:
+            await transport.close()
+
+    asyncio.run(run())
+
+
 def test_async_handle_response_with_request_error_includes_method_and_url():
     async def run() -> None:
         transport = AsyncTransport(api_key="test-key")