Harden retry status-code normalization for string and bytes inputs

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

Author: cursoragent

hyperbrowser/client/polling.py

@@ -18,6 +18,7 @@
 _CLIENT_ERROR_STATUS_MIN = 400
 _CLIENT_ERROR_STATUS_MAX = 500
 _RETRYABLE_CLIENT_ERROR_STATUS_CODES = {408, 429}
+_MAX_STATUS_CODE_TEXT_LENGTH = 6
 
 
 class _NonRetryablePollingError(HyperbrowserError):
@@ -177,14 +178,28 @@ def _normalize_status_code_for_retry(status_code: object) -> Optional[int]:
         return None
     if isinstance(status_code, int):
         return status_code
-    if isinstance(status_code, str):
-        normalized_status = status_code.strip()
+    status_text: Optional[str] = None
+    if isinstance(status_code, (bytes, bytearray)):
+        try:
+            status_text = bytes(status_code).decode("ascii")
+        except UnicodeDecodeError:
+            return None
+    elif isinstance(status_code, str):
+        status_text = status_code
+
+    if status_text is not None:
+        normalized_status = status_text.strip()
         if not normalized_status:
             return None
-        try:
-            return int(normalized_status, 10)
-        except ValueError:
+        if len(normalized_status) > _MAX_STATUS_CODE_TEXT_LENGTH:
+            return None
+        if normalized_status[0] in {"+", "-"}:
+            digits = normalized_status[1:]
+        else:
+            digits = normalized_status
+        if not digits or not digits.isdigit():
             return None
+        return int(normalized_status, 10)
     return None
 
 

tests/test_polling.py

@@ -144,6 +144,31 @@ def get_status() -> str:
     assert attempts["count"] == 1
 
 
+def test_poll_until_terminal_status_retries_overlong_numeric_status_codes():
+    attempts = {"count": 0}
+
+    def get_status() -> str:
+        attempts["count"] += 1
+        if attempts["count"] < 3:
+            raise HyperbrowserError(
+                "oversized status metadata",
+                status_code="4000000000000",  # type: ignore[arg-type]
+            )
+        return "completed"
+
+    status = poll_until_terminal_status(
+        operation_name="sync poll oversized numeric status retries",
+        get_status=get_status,
+        is_terminal_status=lambda value: value == "completed",
+        poll_interval_seconds=0.0001,
+        max_wait_seconds=1.0,
+        max_status_failures=5,
+    )
+
+    assert status == "completed"
+    assert attempts["count"] == 3
+
+
 def test_poll_until_terminal_status_does_not_retry_stop_iteration_errors():
     attempts = {"count": 0}
 
@@ -602,6 +627,27 @@ def operation() -> str:
     assert attempts["count"] == 3
 
 
+def test_retry_operation_does_not_retry_numeric_bytes_client_errors():
+    attempts = {"count": 0}
+
+    def operation() -> str:
+        attempts["count"] += 1
+        raise HyperbrowserError(
+            "client failure",
+            status_code=b"400",  # type: ignore[arg-type]
+        )
+
+    with pytest.raises(HyperbrowserError, match="client failure"):
+        retry_operation(
+            operation_name="sync retry numeric-bytes client error",
+            operation=operation,
+            max_attempts=5,
+            retry_delay_seconds=0.0001,
+        )
+
+    assert attempts["count"] == 1
+
+
 def test_retry_operation_does_not_retry_stop_iteration_errors():
     attempts = {"count": 0}
 
@@ -951,6 +997,34 @@ async def get_status() -> str:
     asyncio.run(run())
 
 
+def test_poll_until_terminal_status_async_retries_overlong_numeric_status_codes():
+    async def run() -> None:
+        attempts = {"count": 0}
+
+        async def get_status() -> str:
+            attempts["count"] += 1
+            if attempts["count"] < 3:
+                raise HyperbrowserError(
+                    "oversized status metadata",
+                    status_code="4000000000000",  # type: ignore[arg-type]
+                )
+            return "completed"
+
+        status = await poll_until_terminal_status_async(
+            operation_name="async poll oversized numeric status retries",
+            get_status=get_status,
+            is_terminal_status=lambda value: value == "completed",
+            poll_interval_seconds=0.0001,
+            max_wait_seconds=1.0,
+            max_status_failures=5,
+        )
+
+        assert status == "completed"
+        assert attempts["count"] == 3
+
+    asyncio.run(run())
+
+
 def test_poll_until_terminal_status_async_does_not_retry_stop_async_iteration_errors():
     async def run() -> None:
         attempts = {"count": 0}
@@ -1181,6 +1255,32 @@ async def operation() -> str:
     asyncio.run(run())
 
 
+def test_retry_operation_async_retries_numeric_bytes_rate_limit_errors():
+    async def run() -> None:
+        attempts = {"count": 0}
+
+        async def operation() -> str:
+            attempts["count"] += 1
+            if attempts["count"] < 3:
+                raise HyperbrowserError(
+                    "rate limited",
+                    status_code=b"429",  # type: ignore[arg-type]
+                )
+            return "ok"
+
+        result = await retry_operation_async(
+            operation_name="async retry numeric-bytes rate limit",
+            operation=operation,
+            max_attempts=5,
+            retry_delay_seconds=0.0001,
+        )
+
+        assert result == "ok"
+        assert attempts["count"] == 3
+
+    asyncio.run(run())
+
+
 def test_retry_operation_async_does_not_retry_stop_async_iteration_errors():
     async def run() -> None:
         attempts = {"count": 0}