mirror of
https://github.com/NousResearch/hermes-agent.git
synced 2026-07-01 12:02:05 +00:00
fix(browser): send Authorization header in Camofox HTTP calls when CAMOFOX_API_KEY is set
The five HTTP call sites in browser_camofox.py (_ensure_tab, _post, _get, _get_raw, _delete) did not include Authorization headers, causing 403 Forbidden when the Camofox server has API key auth enabled. Added _auth_headers() helper and wired it into all five call sites. The health check endpoint (/health) is left without auth since it is a connectivity probe, not a browser operation. Regression test covers: header present when key set, absent when unset, blank key produces empty headers. Fixes #20476
This commit is contained in:
parent
270456308c
commit
babd9168ba
2 changed files with 124 additions and 4 deletions
111
tests/tools/test_browser_camofox_auth.py
Normal file
111
tests/tools/test_browser_camofox_auth.py
Normal file
|
|
@ -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") == {}
|
||||
|
|
@ -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()
|
||||
|
||||
|
|
|
|||
Loading…
Add table
Add a link
Reference in a new issue