Harden transport model/key display fallback sanitization

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

Author: cursoragent

hyperbrowser/transport/base.py

@@ -13,18 +13,23 @@
 
 
 def _sanitize_display_text(value: str, *, max_length: int) -> str:
-    sanitized_value = "".join(
-        "?" if ord(character) < 32 or ord(character) == 127 else character
-        for character in value
-    ).strip()
-    if not sanitized_value:
+    try:
+        sanitized_value = "".join(
+            "?" if ord(character) < 32 or ord(character) == 127 else character
+            for character in value
+        ).strip()
+        if not isinstance(sanitized_value, str):
+            return ""
+        if not sanitized_value:
+            return ""
+        if len(sanitized_value) <= max_length:
+            return sanitized_value
+        available_length = max_length - len(_TRUNCATED_DISPLAY_SUFFIX)
+        if available_length <= 0:
+            return _TRUNCATED_DISPLAY_SUFFIX
+        return f"{sanitized_value[:available_length]}{_TRUNCATED_DISPLAY_SUFFIX}"
+    except Exception:
         return ""
-    if len(sanitized_value) <= max_length:
-        return sanitized_value
-    available_length = max_length - len(_TRUNCATED_DISPLAY_SUFFIX)
-    if available_length <= 0:
-        return _TRUNCATED_DISPLAY_SUFFIX
-    return f"{sanitized_value[:available_length]}{_TRUNCATED_DISPLAY_SUFFIX}"
 
 
 def _safe_model_name(model: object) -> str:
@@ -34,9 +39,12 @@ def _safe_model_name(model: object) -> str:
         return "response model"
     if not isinstance(model_name, str):
         return "response model"
-    normalized_model_name = _sanitize_display_text(
-        model_name, max_length=_MAX_MODEL_NAME_DISPLAY_LENGTH
-    )
+    try:
+        normalized_model_name = _sanitize_display_text(
+            model_name, max_length=_MAX_MODEL_NAME_DISPLAY_LENGTH
+        )
+    except Exception:
+        return "response model"
     if not normalized_model_name:
         return "response model"
     return normalized_model_name
@@ -48,6 +56,10 @@ def _format_mapping_key_for_error(key: str) -> str:
     )
     if normalized_key:
         return normalized_key
+    try:
+        _ = "".join(character for character in key)
+    except Exception:
+        return "<unreadable key>"
     return "<blank key>"
 
 

tests/test_transport_base.py

@@ -101,6 +101,39 @@ def __getitem__(self, key: str) -> object:
         raise KeyError(key)
 
 
+class _BrokenRenderedModelNameString(str):
+    def __iter__(self):
+        raise RuntimeError("cannot iterate rendered model name")
+
+
+class _UnreadableNameCallableModel:
+    __name__ = _BrokenRenderedModelNameString("UnreadableModelName")
+
+    def __call__(self, **kwargs):
+        _ = kwargs
+        raise RuntimeError("call failed")
+
+
+class _BrokenRenderedMappingKey(str):
+    def __iter__(self):
+        raise RuntimeError("cannot iterate rendered mapping key")
+
+
+class _BrokenRenderedKeyValueMapping(Mapping[str, object]):
+    _KEY = _BrokenRenderedMappingKey("name")
+
+    def __iter__(self):
+        return iter([self._KEY])
+
+    def __len__(self) -> int:
+        return 1
+
+    def __getitem__(self, key: str) -> object:
+        if key == self._KEY:
+            raise RuntimeError("cannot read rendered key value")
+        raise KeyError(key)
+
+
 def test_api_response_from_json_parses_model_data() -> None:
     response = APIResponse.from_json(
         {"name": "job-1", "retries": 2}, _SampleResponseModel
@@ -206,6 +239,17 @@ def test_api_response_from_json_sanitizes_and_truncates_model_name_in_errors() -
         )
 
 
+def test_api_response_from_json_falls_back_for_unreadable_model_name_text() -> None:
+    with pytest.raises(
+        HyperbrowserError,
+        match="Failed to parse response data for response model",
+    ):
+        APIResponse.from_json(
+            {"name": "job-1"},
+            cast("type[_SampleResponseModel]", _UnreadableNameCallableModel()),
+        )
+
+
 def test_api_response_from_json_uses_placeholder_for_blank_mapping_key_in_errors() -> (
     None
 ):
@@ -232,6 +276,19 @@ def test_api_response_from_json_sanitizes_and_truncates_mapping_keys_in_errors()
         APIResponse.from_json(_BrokenLongKeyValueMapping(), _SampleResponseModel)
 
 
+def test_api_response_from_json_falls_back_for_unreadable_mapping_keys_in_errors() -> (
+    None
+):
+    with pytest.raises(
+        HyperbrowserError,
+        match=(
+            "Failed to parse response data for _SampleResponseModel: "
+            "unable to read value for key '<unreadable key>'"
+        ),
+    ):
+        APIResponse.from_json(_BrokenRenderedKeyValueMapping(), _SampleResponseModel)
+
+
 def test_api_response_from_json_preserves_hyperbrowser_errors() -> None:
     with pytest.raises(HyperbrowserError, match="model validation failed") as exc_info:
         APIResponse.from_json({}, _RaisesHyperbrowserModel)