diff --git a/tests/tools/test_browser_camofox_auth.py b/tests/tools/test_browser_camofox_auth.py new file mode 100644 index 00000000000..590bea47028 --- /dev/null +++ b/tests/tools/test_browser_camofox_auth.py @@ -0,0 +1,111 @@ +"""Tests that Camofox browser sends Authorization header when CAMOFOX_API_KEY is set. + +Regression test for https://github.com/NousResearch/hermes-agent/issues/20476 +""" + +import json +from unittest.mock import MagicMock, patch + +import pytest + +from tools.browser_camofox import ( + _auth_headers, + camofox_back, + camofox_click, + camofox_close, + camofox_navigate, + camofox_press, + camofox_scroll, + camofox_snapshot, + camofox_type, +) + + +def _mock_response(status=200, json_data=None): + resp = MagicMock() + resp.status_code = status + resp.json.return_value = json_data or {} + resp.content = b"\x89PNG\r\n\x1a\nfake" + resp.raise_for_status = MagicMock() + return resp + + +class TestAuthHeaders: + """Unit tests for _auth_headers() helper.""" + + def test_empty_when_no_key(self, monkeypatch): + monkeypatch.delenv("CAMOFOX_API_KEY", raising=False) + assert _auth_headers() == {} + + def test_bearer_when_key_set(self, monkeypatch): + monkeypatch.setenv("CAMOFOX_API_KEY", "test-secret-123") + assert _auth_headers() == {"Authorization": "Bearer test-secret-123"} + + def test_empty_when_key_blank(self, monkeypatch): + monkeypatch.setenv("CAMOFOX_API_KEY", " ") + assert _auth_headers() == {} + + +class TestAuthHeadersSent: + """Verify all HTTP call sites include auth headers when CAMOFOX_API_KEY is set.""" + + @pytest.fixture(autouse=True) + def _set_key(self, monkeypatch): + monkeypatch.setenv("CAMOFOX_URL", "http://localhost:9377") + monkeypatch.setenv("CAMOFOX_API_KEY", "my-api-key") + + @patch("tools.browser_camofox.requests.post") + def test_ensure_tab_sends_auth(self, mock_post): + mock_post.return_value = _mock_response(json_data={"tabId": "t1"}) + camofox_navigate("https://example.com", task_id="auth_test_1") + _, kwargs = mock_post.call_args + assert kwargs["headers"] == {"Authorization": "Bearer my-api-key"} + + @patch("tools.browser_camofox.requests.post") + def test_post_sends_auth(self, mock_post): + mock_post.return_value = _mock_response(json_data={"tabId": "t2"}) + camofox_navigate("https://example.com", task_id="auth_test_2") + mock_post.return_value = _mock_response(json_data={"ok": True, "url": "https://x.com"}) + camofox_navigate("https://x.com", task_id="auth_test_2") + # The second call is a POST to /tabs/{tabId}/navigate + last_call = mock_post.call_args_list[-1] + assert last_call.kwargs.get("headers") == {"Authorization": "Bearer my-api-key"} + + @patch("tools.browser_camofox.requests.post") + @patch("tools.browser_camofox.requests.get") + def test_get_sends_auth(self, mock_get, mock_post): + mock_post.return_value = _mock_response(json_data={"tabId": "t3"}) + camofox_navigate("https://example.com", task_id="auth_test_3") + mock_get.return_value = _mock_response(json_data={ + "snapshot": '- heading "Hello"', + "refsCount": 1, + }) + camofox_snapshot(task_id="auth_test_3") + _, kwargs = mock_get.call_args + assert kwargs["headers"] == {"Authorization": "Bearer my-api-key"} + + @patch("tools.browser_camofox.requests.post") + @patch("tools.browser_camofox.requests.delete") + def test_delete_sends_auth(self, mock_delete, mock_post): + mock_post.return_value = _mock_response(json_data={"tabId": "t4"}) + camofox_navigate("https://example.com", task_id="auth_test_4") + mock_delete.return_value = _mock_response(json_data={"ok": True}) + camofox_close(task_id="auth_test_4") + _, kwargs = mock_delete.call_args + assert kwargs["headers"] == {"Authorization": "Bearer my-api-key"} + + +class TestNoAuthHeadersWhenKeyUnset: + """Verify HTTP calls send empty headers when CAMOFOX_API_KEY is not set.""" + + @pytest.fixture(autouse=True) + def _unset_key(self, monkeypatch): + monkeypatch.setenv("CAMOFOX_URL", "http://localhost:9377") + monkeypatch.delenv("CAMOFOX_API_KEY", raising=False) + + @patch("tools.browser_camofox.requests.post") + def test_no_auth_on_tab_creation(self, mock_post): + mock_post.return_value = _mock_response(json_data={"tabId": "t5"}) + camofox_navigate("https://example.com", task_id="noauth_test_1") + _, kwargs = mock_post.call_args + assert kwargs.get("headers") == {} diff --git a/tools/browser_camofox.py b/tools/browser_camofox.py index 1ac1a41fe30..717e0420d06 100644 --- a/tools/browser_camofox.py +++ b/tools/browser_camofox.py @@ -52,6 +52,14 @@ _vnc_url: Optional[str] = None # cached from /health response _vnc_url_checked = False # only probe once per process +def _auth_headers() -> Dict[str, str]: + """Return Authorization header when CAMOFOX_API_KEY is set.""" + key = os.getenv("CAMOFOX_API_KEY", "").strip() + if key: + return {"Authorization": f"Bearer {key}"} + return {} + + def get_camofox_url() -> str: """Return the configured Camofox server URL, or empty string.""" return os.getenv("CAMOFOX_URL", "").rstrip("/") @@ -349,6 +357,7 @@ def _ensure_tab(task_id: Optional[str], url: str = "about:blank") -> Dict[str, A "url": url, }, timeout=_DEFAULT_TIMEOUT, + headers=_auth_headers(), ) resp.raise_for_status() data = resp.json() @@ -387,7 +396,7 @@ def camofox_soft_cleanup(task_id: Optional[str] = None) -> bool: def _post(path: str, body: dict, timeout: int = _DEFAULT_TIMEOUT) -> dict: """POST JSON to camofox and return parsed response.""" url = f"{get_camofox_url()}{path}" - resp = requests.post(url, json=body, timeout=timeout) + resp = requests.post(url, json=body, timeout=timeout, headers=_auth_headers()) resp.raise_for_status() return resp.json() @@ -395,7 +404,7 @@ def _post(path: str, body: dict, timeout: int = _DEFAULT_TIMEOUT) -> dict: def _get(path: str, params: dict = None, timeout: int = _DEFAULT_TIMEOUT) -> dict: """GET from camofox and return parsed response.""" url = f"{get_camofox_url()}{path}" - resp = requests.get(url, params=params, timeout=timeout) + resp = requests.get(url, params=params, timeout=timeout, headers=_auth_headers()) resp.raise_for_status() return resp.json() @@ -403,7 +412,7 @@ def _get(path: str, params: dict = None, timeout: int = _DEFAULT_TIMEOUT) -> dic def _get_raw(path: str, params: dict = None, timeout: int = _DEFAULT_TIMEOUT) -> requests.Response: """GET from camofox and return raw response (for binary data).""" url = f"{get_camofox_url()}{path}" - resp = requests.get(url, params=params, timeout=timeout) + resp = requests.get(url, params=params, timeout=timeout, headers=_auth_headers()) resp.raise_for_status() return resp @@ -411,7 +420,7 @@ def _get_raw(path: str, params: dict = None, timeout: int = _DEFAULT_TIMEOUT) -> def _delete(path: str, body: dict = None, timeout: int = _DEFAULT_TIMEOUT) -> dict: """DELETE to camofox and return parsed response.""" url = f"{get_camofox_url()}{path}" - resp = requests.delete(url, json=body, timeout=timeout) + resp = requests.delete(url, json=body, timeout=timeout, headers=_auth_headers()) resp.raise_for_status() return resp.json()