Validate computer action endpoint string hygiene

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

Author: cursoragent

hyperbrowser/client/managers/async_manager/computer_action.py

@@ -47,18 +47,56 @@ async def _execute_request(
                 original_error=exc,
             ) from exc
 
-        if not computer_action_endpoint:
+        if computer_action_endpoint is None:
             raise HyperbrowserError(
                 "Computer action endpoint not available for this session"
             )
+        if type(computer_action_endpoint) is not str:
+            raise HyperbrowserError("session computer_action_endpoint must be a string")
+        try:
+            normalized_computer_action_endpoint = computer_action_endpoint.strip()
+            if type(normalized_computer_action_endpoint) is not str:
+                raise TypeError("normalized computer_action_endpoint must be a string")
+        except HyperbrowserError:
+            raise
+        except Exception as exc:
+            raise HyperbrowserError(
+                "Failed to normalize session computer_action_endpoint",
+                original_error=exc,
+            ) from exc
+
+        if not normalized_computer_action_endpoint:
+            raise HyperbrowserError(
+                "Computer action endpoint not available for this session"
+            )
+        if normalized_computer_action_endpoint != computer_action_endpoint:
+            raise HyperbrowserError(
+                "session computer_action_endpoint must not contain leading or trailing whitespace"
+            )
+        try:
+            contains_control_character = any(
+                ord(character) < 32 or ord(character) == 127
+                for character in normalized_computer_action_endpoint
+            )
+        except HyperbrowserError:
+            raise
+        except Exception as exc:
+            raise HyperbrowserError(
+                "Failed to validate session computer_action_endpoint characters",
+                original_error=exc,
+            ) from exc
+        if contains_control_character:
+            raise HyperbrowserError(
+                "session computer_action_endpoint must not contain control characters"
+            )
 
         if isinstance(params, BaseModel):
             payload = params.model_dump(by_alias=True, exclude_none=True)
         else:
             payload = params
 
         response = await self._client.transport.post(
-            computer_action_endpoint,
+            normalized_computer_action_endpoint,
             data=payload,
         )
         return parse_response_model(

hyperbrowser/client/managers/sync_manager/computer_action.py

@@ -47,18 +47,56 @@ def _execute_request(
                 original_error=exc,
             ) from exc
 
-        if not computer_action_endpoint:
+        if computer_action_endpoint is None:
             raise HyperbrowserError(
                 "Computer action endpoint not available for this session"
             )
+        if type(computer_action_endpoint) is not str:
+            raise HyperbrowserError("session computer_action_endpoint must be a string")
+        try:
+            normalized_computer_action_endpoint = computer_action_endpoint.strip()
+            if type(normalized_computer_action_endpoint) is not str:
+                raise TypeError("normalized computer_action_endpoint must be a string")
+        except HyperbrowserError:
+            raise
+        except Exception as exc:
+            raise HyperbrowserError(
+                "Failed to normalize session computer_action_endpoint",
+                original_error=exc,
+            ) from exc
+
+        if not normalized_computer_action_endpoint:
+            raise HyperbrowserError(
+                "Computer action endpoint not available for this session"
+            )
+        if normalized_computer_action_endpoint != computer_action_endpoint:
+            raise HyperbrowserError(
+                "session computer_action_endpoint must not contain leading or trailing whitespace"
+            )
+        try:
+            contains_control_character = any(
+                ord(character) < 32 or ord(character) == 127
+                for character in normalized_computer_action_endpoint
+            )
+        except HyperbrowserError:
+            raise
+        except Exception as exc:
+            raise HyperbrowserError(
+                "Failed to validate session computer_action_endpoint characters",
+                original_error=exc,
+            ) from exc
+        if contains_control_character:
+            raise HyperbrowserError(
+                "session computer_action_endpoint must not contain control characters"
+            )
 
         if isinstance(params, BaseModel):
             payload = params.model_dump(by_alias=True, exclude_none=True)
         else:
             payload = params
 
         response = self._client.transport.post(
-            computer_action_endpoint,
+            normalized_computer_action_endpoint,
             data=payload,
         )
         return parse_response_model(

tests/test_computer_action_manager.py

@@ -64,6 +64,76 @@ async def run() -> None:
     asyncio.run(run())
 
 
+def test_sync_computer_action_manager_rejects_non_string_endpoints():
+    manager = SyncComputerActionManager(_DummyClient())
+
+    with pytest.raises(
+        HyperbrowserError, match="session computer_action_endpoint must be a string"
+    ) as exc_info:
+        manager.screenshot(SimpleNamespace(computer_action_endpoint=123))
+
+    assert exc_info.value.original_error is None
+
+
+def test_async_computer_action_manager_rejects_non_string_endpoints():
+    async def run() -> None:
+        manager = AsyncComputerActionManager(_DummyClient())
+        with pytest.raises(
+            HyperbrowserError, match="session computer_action_endpoint must be a string"
+        ) as exc_info:
+            await manager.screenshot(SimpleNamespace(computer_action_endpoint=123))
+        assert exc_info.value.original_error is None
+
+    asyncio.run(run())
+
+
+def test_sync_computer_action_manager_rejects_string_subclass_endpoints():
+    class _Endpoint(str):
+        pass
+
+    manager = SyncComputerActionManager(_DummyClient())
+
+    with pytest.raises(
+        HyperbrowserError, match="session computer_action_endpoint must be a string"
+    ) as exc_info:
+        manager.screenshot(
+            SimpleNamespace(
+                computer_action_endpoint=_Endpoint("https://example.com/cua")
+            )
+        )
+
+    assert exc_info.value.original_error is None
+
+
+def test_sync_computer_action_manager_rejects_whitespace_wrapped_endpoints():
+    manager = SyncComputerActionManager(_DummyClient())
+
+    with pytest.raises(
+        HyperbrowserError,
+        match="session computer_action_endpoint must not contain leading or trailing whitespace",
+    ) as exc_info:
+        manager.screenshot(
+            SimpleNamespace(computer_action_endpoint="  https://example.com/cua  ")
+        )
+
+    assert exc_info.value.original_error is None
+
+
+def test_async_computer_action_manager_rejects_control_character_endpoints():
+    async def run() -> None:
+        manager = AsyncComputerActionManager(_DummyClient())
+        with pytest.raises(
+            HyperbrowserError,
+            match="session computer_action_endpoint must not contain control characters",
+        ) as exc_info:
+            await manager.screenshot(
+                SimpleNamespace(computer_action_endpoint="https://exa\tmple.com/cua")
+            )
+        assert exc_info.value.original_error is None
+
+    asyncio.run(run())
+
+
 def test_sync_computer_action_manager_rejects_string_subclass_session_ids():
     class _SessionId(str):
         pass