Harden transport api_key normalization error boundaries

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

Author: cursoragent

hyperbrowser/transport/async_transport.py

@@ -4,7 +4,7 @@
 from hyperbrowser.exceptions import HyperbrowserError
 from hyperbrowser.header_utils import merge_headers
 from hyperbrowser.version import __version__
-from .base import APIResponse, AsyncTransportStrategy
+from .base import APIResponse, AsyncTransportStrategy, _normalize_transport_api_key
 from .error_utils import (
     extract_error_message,
     format_generic_request_failure_message,
@@ -19,16 +19,7 @@ class AsyncTransport(AsyncTransportStrategy):
     _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")
-        normalized_api_key = api_key.strip()
-        if not normalized_api_key:
-            raise HyperbrowserError("api_key must not be empty")
-        if any(
-            ord(character) < 32 or ord(character) == 127
-            for character in normalized_api_key
-        ):
-            raise HyperbrowserError("api_key must not contain control characters")
+        normalized_api_key = _normalize_transport_api_key(api_key)
         merged_headers = merge_headers(
             {
                 "x-api-key": normalized_api_key,

hyperbrowser/transport/base.py

@@ -51,6 +51,44 @@ def _format_mapping_key_for_error(key: str) -> str:
     return "<blank key>"
 
 
+def _normalize_transport_api_key(api_key: str) -> str:
+    if not isinstance(api_key, str):
+        raise HyperbrowserError("api_key must be a string")
+
+    try:
+        normalized_api_key = api_key.strip()
+        if not isinstance(normalized_api_key, str):
+            raise TypeError("normalized api_key must be a string")
+    except HyperbrowserError:
+        raise
+    except Exception as exc:
+        raise HyperbrowserError(
+            "Failed to normalize api_key",
+            original_error=exc,
+        ) from exc
+
+    if not normalized_api_key:
+        raise HyperbrowserError("api_key must not be empty")
+
+    try:
+        contains_control_character = any(
+            ord(character) < 32 or ord(character) == 127
+            for character in normalized_api_key
+        )
+    except HyperbrowserError:
+        raise
+    except Exception as exc:
+        raise HyperbrowserError(
+            "Failed to validate api_key characters",
+            original_error=exc,
+        ) from exc
+
+    if contains_control_character:
+        raise HyperbrowserError("api_key must not contain control characters")
+
+    return normalized_api_key
+
+
 class APIResponse(Generic[T]):
     """
     Wrapper for API responses to standardize sync/async handling.

hyperbrowser/transport/sync.py

@@ -4,7 +4,7 @@
 from hyperbrowser.exceptions import HyperbrowserError
 from hyperbrowser.header_utils import merge_headers
 from hyperbrowser.version import __version__
-from .base import APIResponse, SyncTransportStrategy
+from .base import APIResponse, SyncTransportStrategy, _normalize_transport_api_key
 from .error_utils import (
     extract_error_message,
     format_generic_request_failure_message,
@@ -19,16 +19,7 @@ class SyncTransport(SyncTransportStrategy):
     _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")
-        normalized_api_key = api_key.strip()
-        if not normalized_api_key:
-            raise HyperbrowserError("api_key must not be empty")
-        if any(
-            ord(character) < 32 or ord(character) == 127
-            for character in normalized_api_key
-        ):
-            raise HyperbrowserError("api_key must not contain control characters")
+        normalized_api_key = _normalize_transport_api_key(api_key)
         merged_headers = merge_headers(
             {
                 "x-api-key": normalized_api_key,

tests/test_transport_api_key.py

@@ -0,0 +1,80 @@
+import pytest
+
+from hyperbrowser.exceptions import HyperbrowserError
+from hyperbrowser.transport.async_transport import AsyncTransport
+from hyperbrowser.transport.sync import SyncTransport
+
+
+@pytest.mark.parametrize("transport_class", [SyncTransport, AsyncTransport])
+def test_transport_wraps_api_key_strip_runtime_errors(transport_class):
+    class _BrokenStripApiKey(str):
+        def strip(self, chars=None):  # type: ignore[override]
+            _ = chars
+            raise RuntimeError("api key strip exploded")
+
+    with pytest.raises(HyperbrowserError, match="Failed to normalize api_key") as exc_info:
+        transport_class(api_key=_BrokenStripApiKey("test-key"))
+
+    assert isinstance(exc_info.value.original_error, RuntimeError)
+
+
+@pytest.mark.parametrize("transport_class", [SyncTransport, AsyncTransport])
+def test_transport_preserves_hyperbrowser_api_key_strip_errors(transport_class):
+    class _BrokenStripApiKey(str):
+        def strip(self, chars=None):  # type: ignore[override]
+            _ = chars
+            raise HyperbrowserError("custom strip failure")
+
+    with pytest.raises(HyperbrowserError, match="custom strip failure") as exc_info:
+        transport_class(api_key=_BrokenStripApiKey("test-key"))
+
+    assert exc_info.value.original_error is None
+
+
+@pytest.mark.parametrize("transport_class", [SyncTransport, AsyncTransport])
+def test_transport_wraps_non_string_api_key_strip_results(transport_class):
+    class _NonStringStripResultApiKey(str):
+        def strip(self, chars=None):  # type: ignore[override]
+            _ = chars
+            return object()
+
+    with pytest.raises(HyperbrowserError, match="Failed to normalize api_key") as exc_info:
+        transport_class(api_key=_NonStringStripResultApiKey("test-key"))
+
+    assert isinstance(exc_info.value.original_error, TypeError)
+
+
+@pytest.mark.parametrize("transport_class", [SyncTransport, AsyncTransport])
+def test_transport_wraps_api_key_character_iteration_failures(transport_class):
+    class _BrokenIterApiKey(str):
+        def strip(self, chars=None):  # type: ignore[override]
+            _ = chars
+            return self
+
+        def __iter__(self):
+            raise RuntimeError("api key iteration exploded")
+
+    with pytest.raises(
+        HyperbrowserError, match="Failed to validate api_key characters"
+    ) as exc_info:
+        transport_class(api_key=_BrokenIterApiKey("test-key"))
+
+    assert isinstance(exc_info.value.original_error, RuntimeError)
+
+
+@pytest.mark.parametrize("transport_class", [SyncTransport, AsyncTransport])
+def test_transport_preserves_hyperbrowser_api_key_character_iteration_failures(
+    transport_class,
+):
+    class _BrokenIterApiKey(str):
+        def strip(self, chars=None):  # type: ignore[override]
+            _ = chars
+            return self
+
+        def __iter__(self):
+            raise HyperbrowserError("custom iteration failure")
+
+    with pytest.raises(HyperbrowserError, match="custom iteration failure") as exc_info:
+        transport_class(api_key=_BrokenIterApiKey("test-key"))
+
+    assert exc_info.value.original_error is None