From 6e75ba7fa066c1667cdc09db02c0c1dca8cd4671 Mon Sep 17 00:00:00 2001 From: kshitijk4poor <82637225+kshitijk4poor@users.noreply.github.com> Date: Sat, 11 Jul 2026 11:29:31 +0530 Subject: [PATCH] fix(security): enforce one redirect credential policy --- hermes_cli/models.py | 55 +--- hermes_cli/urllib_security.py | 80 ++++++ providers/base.py | 34 +-- tests/hermes_cli/test_model_validation.py | 28 +- tests/hermes_cli/test_models.py | 60 +--- tests/hermes_cli/test_urllib_security.py | 330 ++++++++++++++++++++++ 6 files changed, 448 insertions(+), 139 deletions(-) create mode 100644 hermes_cli/urllib_security.py create mode 100644 tests/hermes_cli/test_urllib_security.py diff --git a/hermes_cli/models.py b/hermes_cli/models.py index dc7b8cafbd6..3fee59e7fea 100644 --- a/hermes_cli/models.py +++ b/hermes_cli/models.py @@ -18,6 +18,7 @@ from pathlib import Path from typing import Any, NamedTuple, Optional from hermes_cli import __version__ as _HERMES_VERSION +from hermes_cli.urllib_security import open_credentialed_url # Identify ourselves so endpoints fronted by Cloudflare's Browser Integrity # Check (error 1010) don't reject the default ``Python-urllib/*`` signature. @@ -29,43 +30,9 @@ COPILOT_EDITOR_VERSION = "vscode/1.104.1" COPILOT_REASONING_EFFORTS_GPT5 = ["minimal", "low", "medium", "high"] COPILOT_REASONING_EFFORTS_O_SERIES = ["low", "medium", "high"] -_CREDENTIAL_REDIRECT_HEADERS = { - "authorization", - "x-api-key", - "api-key", - "x-goog-api-key", - "cookie", -} - -_ORIGINAL_URLOPEN = urllib.request.urlopen - - -class _StripCredentialRedirectHandler(urllib.request.HTTPRedirectHandler): - """Drop credential headers when urllib follows redirects to another host.""" - - def __init__(self, original_host: str): - self._original_host = original_host - - def redirect_request(self, req, fp, code, msg, headers, newurl): - new_host = (urllib.parse.urlparse(newurl).hostname or "").lower() - if new_host and new_host != self._original_host: - clean_headers = { - name: value - for name, value in req.header_items() - if name.lower() not in _CREDENTIAL_REDIRECT_HEADERS - } - return urllib.request.Request(newurl, headers=clean_headers, method="GET") - return super().redirect_request(req, fp, code, msg, headers, newurl) - - def _urlopen_model_catalog_request(req: urllib.request.Request, *, timeout: float): - if urllib.request.urlopen is not _ORIGINAL_URLOPEN: - return urllib.request.urlopen(req, timeout=timeout) - if not any(name.lower() in _CREDENTIAL_REDIRECT_HEADERS for name, _ in req.header_items()): - return urllib.request.urlopen(req, timeout=timeout) - original_host = (urllib.parse.urlparse(req.full_url).hostname or "").lower() - opener = urllib.request.build_opener(_StripCredentialRedirectHandler(original_host)) - return opener.open(req, timeout=timeout) + """Open catalog requests without forwarding headers across origins.""" + return open_credentialed_url(req, timeout=timeout) # Fallback OpenRouter snapshot used when the live catalog is unavailable. @@ -3154,15 +3121,13 @@ def ensure_lmstudio_model_loaded( load_headers = dict(headers) load_headers["Content-Type"] = "application/json" try: - with urllib.request.urlopen( - urllib.request.Request( - server_root + "/api/v1/models/load", - data=body, - headers=load_headers, - method="POST", - ), - timeout=timeout, - ) as resp: + load_request = urllib.request.Request( + server_root + "/api/v1/models/load", + data=body, + headers=load_headers, + method="POST", + ) + with _urlopen_model_catalog_request(load_request, timeout=timeout) as resp: resp.read() except Exception: return None diff --git a/hermes_cli/urllib_security.py b/hermes_cli/urllib_security.py new file mode 100644 index 00000000000..187ddd6fe43 --- /dev/null +++ b/hermes_cli/urllib_security.py @@ -0,0 +1,80 @@ +"""Security policy for credential-bearing stdlib urllib requests.""" + +from __future__ import annotations + +import urllib.parse +import urllib.request +from collections.abc import Callable, Iterable +from typing import Any + +# Headers safe to forward to a different origin. Everything else is dropped: +# custom provider headers routinely carry credentials under arbitrary names. +_CROSS_ORIGIN_SAFE_HEADERS = frozenset({"accept", "user-agent"}) +_DEFAULT_PORTS = {"http": 80, "https": 443} + + +def url_origin(url: str) -> tuple[str, str, int | None]: + """Return a normalized (scheme, hostname, effective port) origin.""" + parsed = urllib.parse.urlparse(url) + scheme = (parsed.scheme or "").lower() + try: + port = parsed.port + except ValueError: + port = None + return ( + scheme, + (parsed.hostname or "").lower().rstrip("."), + port or _DEFAULT_PORTS.get(scheme), + ) + + +class SafeCredentialRedirectHandler(urllib.request.HTTPRedirectHandler): + """Preserve request headers only while redirects stay on one origin.""" + + def __init__( + self, + original_url: str, + *, + cross_origin_safe_headers: Iterable[str] = _CROSS_ORIGIN_SAFE_HEADERS, + ) -> None: + self._original_origin = url_origin(original_url) + self._cross_origin_safe_headers = frozenset( + str(name).lower() for name in cross_origin_safe_headers + ) + + def redirect_request(self, req, fp, code, msg, headers, newurl): + # Let urllib enforce status/method semantics first (notably 307/308). + redirected = super().redirect_request(req, fp, code, msg, headers, newurl) + if redirected is None: + return None + + resolved_url = urllib.parse.urljoin(req.full_url, newurl) + if url_origin(resolved_url) != self._original_origin: + # Use an allowlist rather than guessing credential header names. + # normalize_extra_headers permits arbitrary secret-bearing names. + for name, _value in list(redirected.header_items()): + if name.lower() not in self._cross_origin_safe_headers: + redirected.remove_header(name) + return redirected + + +def open_credentialed_url( + request: urllib.request.Request, + *, + timeout: float, + opener_factory: Callable[..., Any] = urllib.request.build_opener, +): + """Open a request without forwarding credentials across origins. + + ``opener_factory`` is an explicit instrumentation/test seam. Security is + never disabled based on global ``urlopen`` identity. + """ + opener = opener_factory(SafeCredentialRedirectHandler(request.full_url)) + return opener.open(request, timeout=timeout) + + +__all__ = [ + "SafeCredentialRedirectHandler", + "open_credentialed_url", + "url_origin", +] diff --git a/providers/base.py b/providers/base.py index cdc9abdb745..ea5b397e58c 100644 --- a/providers/base.py +++ b/providers/base.py @@ -194,9 +194,10 @@ class ProviderProfile: url = effective_base.rstrip("/") + "/models" import json - import urllib.parse import urllib.request + from hermes_cli.urllib_security import open_credentialed_url + req = urllib.request.Request(url) if api_key: req.add_header("Authorization", f"Bearer {api_key}") @@ -208,37 +209,8 @@ class ProviderProfile: for k, v in self.default_headers.items(): req.add_header(k, v) - # urllib keeps every header — including Authorization — when it - # follows a redirect, so a catalog endpoint answering 3xx to a - # different origin would receive the provider API key. Drop - # credential headers on cross-origin redirects. Origin = scheme + - # hostname + effective port: a different port on the same host can - # be a different service, so it must not inherit credentials either. - sensitive = {"authorization", "x-api-key", "api-key", "x-goog-api-key", "cookie"} - - def _origin(target_url: str) -> tuple: - parsed = urllib.parse.urlparse(target_url) - scheme = (parsed.scheme or "").lower() - port = parsed.port or {"http": 80, "https": 443}.get(scheme) - return scheme, (parsed.hostname or "").lower(), port - - original_origin = _origin(url) - - class _StripAuthOnCrossOriginRedirect(urllib.request.HTTPRedirectHandler): - def redirect_request(self, redirected, fp, code, msg, hdrs, newurl): - new_req = super().redirect_request( - redirected, fp, code, msg, hdrs, newurl - ) - if new_req is not None and _origin(newurl) != original_origin: - for header in list(new_req.headers): - if header.lower() in sensitive: - new_req.remove_header(header) - return new_req - - opener = urllib.request.build_opener(_StripAuthOnCrossOriginRedirect()) - try: - with opener.open(req, timeout=timeout) as resp: + with open_credentialed_url(req, timeout=timeout) as resp: data = json.loads(resp.read().decode()) items = data if isinstance(data, list) else data.get("data", []) return [m["id"] for m in items if isinstance(m, dict) and "id" in m] diff --git a/tests/hermes_cli/test_model_validation.py b/tests/hermes_cli/test_model_validation.py index 113a4f05a68..93095999d10 100644 --- a/tests/hermes_cli/test_model_validation.py +++ b/tests/hermes_cli/test_model_validation.py @@ -236,7 +236,7 @@ class TestProviderModelIds: } }, ), patch( - "hermes_cli.models.urllib.request.urlopen", + "hermes_cli.models._urlopen_model_catalog_request", return_value=_Resp(), ) as mock_urlopen: assert provider_model_ids("anthropic") == ["enterprise-claude"] @@ -275,7 +275,7 @@ class TestFetchApiModels: assert fetch_api_models("key", None) is None def test_returns_none_on_network_error(self): - with patch("hermes_cli.models.urllib.request.urlopen", side_effect=Exception("timeout")): + with patch("hermes_cli.models._urlopen_model_catalog_request", side_effect=Exception("timeout")): assert fetch_api_models("key", "https://example.com/v1") is None def test_probe_api_models_tries_v1_fallback(self): @@ -297,7 +297,7 @@ class TestFetchApiModels: return _Resp() raise Exception("404") - with patch("hermes_cli.models.urllib.request.urlopen", side_effect=_fake_urlopen): + with patch("hermes_cli.models._urlopen_model_catalog_request", side_effect=_fake_urlopen): probe = probe_api_models("key", "http://localhost:8000") assert calls == ["http://localhost:8000/models", "http://localhost:8000/v1/models"] @@ -316,7 +316,7 @@ class TestFetchApiModels: def read(self): return b'{"data": [{"id": "gpt-5.4", "model_picker_enabled": true, "supported_endpoints": ["/responses"], "capabilities": {"type": "chat", "supports": {"reasoning_effort": ["low", "medium", "high"]}}}, {"id": "claude-sonnet-4.6", "model_picker_enabled": true, "supported_endpoints": ["/chat/completions"], "capabilities": {"type": "chat", "supports": {"reasoning_effort": ["low", "medium", "high"]}}}, {"id": "text-embedding-3-small", "model_picker_enabled": true, "capabilities": {"type": "embedding"}}]}' - with patch("hermes_cli.models.urllib.request.urlopen", return_value=_Resp()) as mock_urlopen: + with patch("hermes_cli.models._urlopen_model_catalog_request", return_value=_Resp()) as mock_urlopen: probe = probe_api_models("gh-token", "https://api.githubcopilot.com") assert mock_urlopen.call_args[0][0].full_url == "https://api.githubcopilot.com/models" @@ -335,7 +335,7 @@ class TestFetchApiModels: def read(self): return b'{"data": [{"id": "gpt-5.4", "model_picker_enabled": true, "supported_endpoints": ["/responses"], "capabilities": {"type": "chat", "supports": {"reasoning_effort": ["low", "medium", "high"]}}}, {"id": "text-embedding-3-small", "model_picker_enabled": true, "capabilities": {"type": "embedding"}}]}' - with patch("hermes_cli.models.urllib.request.urlopen", return_value=_Resp()): + with patch("hermes_cli.models._urlopen_model_catalog_request", return_value=_Resp()): catalog = fetch_github_model_catalog("gh-token") assert catalog is not None @@ -749,7 +749,7 @@ class TestValidateApiFallback: b']}' ) - with patch("hermes_cli.models.urllib.request.urlopen", return_value=mock_resp): + with patch("hermes_cli.models._urlopen_model_catalog_request", return_value=mock_resp): models = fetch_lmstudio_models(base_url="http://localhost:1234/v1") assert models == ["publisher/chat-model"] @@ -760,7 +760,7 @@ class TestValidateApiFallback: mock_resp.__exit__.return_value = False mock_resp.read.return_value = b'{"models":[{"key":"publisher/chat-model","type":"llm"}]}' - with patch("hermes_cli.models.urllib.request.urlopen", return_value=mock_resp) as mock_urlopen: + with patch("hermes_cli.models._urlopen_model_catalog_request", return_value=mock_resp) as mock_urlopen: models = fetch_lmstudio_models(base_url="http://localhost:1234/api/v1") request = mock_urlopen.call_args[0][0] @@ -778,7 +778,7 @@ class TestValidateApiFallback: b']}' ) - with patch("hermes_cli.models.urllib.request.urlopen", return_value=mock_resp): + with patch("hermes_cli.models._urlopen_model_catalog_request", return_value=mock_resp): result = validate_requested_model( "publisher/embed-model", "lmstudio", @@ -802,7 +802,7 @@ class TestValidateApiFallback: fp=None, ) - with patch("hermes_cli.models.urllib.request.urlopen", side_effect=http_error): + with patch("hermes_cli.models._urlopen_model_catalog_request", side_effect=http_error): with pytest.raises(AuthError) as excinfo: fetch_lmstudio_models(base_url="http://localhost:1234/v1") @@ -812,7 +812,7 @@ class TestValidateApiFallback: def test_fetch_lmstudio_models_returns_empty_on_network_error(self): with patch( - "hermes_cli.models.urllib.request.urlopen", + "hermes_cli.models._urlopen_model_catalog_request", side_effect=ConnectionRefusedError(), ): models = fetch_lmstudio_models(base_url="http://localhost:1234/v1") @@ -830,7 +830,7 @@ class TestValidateApiFallback: fp=None, ) - with patch("hermes_cli.models.urllib.request.urlopen", side_effect=http_error): + with patch("hermes_cli.models._urlopen_model_catalog_request", side_effect=http_error): result = validate_requested_model( "publisher/chat-model", "lmstudio", @@ -843,7 +843,7 @@ class TestValidateApiFallback: def test_validate_lmstudio_distinguishes_unreachable(self): with patch( - "hermes_cli.models.urllib.request.urlopen", + "hermes_cli.models._urlopen_model_catalog_request", side_effect=ConnectionRefusedError(), ): result = validate_requested_model( @@ -910,7 +910,7 @@ class TestProbeApiModelsUserAgent: body = b'{"data":[{"id":"claude-opus-4.7"}]}' with patch( - "hermes_cli.models.urllib.request.urlopen", + "hermes_cli.models._urlopen_model_catalog_request", return_value=self._make_mock_response(body), ) as mock_urlopen: result = probe_api_models("sk-test", "https://example.com/v1") @@ -932,7 +932,7 @@ class TestProbeApiModelsUserAgent: body = b'{"data":[]}' with patch( - "hermes_cli.models.urllib.request.urlopen", + "hermes_cli.models._urlopen_model_catalog_request", return_value=self._make_mock_response(body), ) as mock_urlopen: probe_api_models(None, "https://example.com/v1") diff --git a/tests/hermes_cli/test_models.py b/tests/hermes_cli/test_models.py index 3dcb2ee9ab3..bda4807d809 100644 --- a/tests/hermes_cli/test_models.py +++ b/tests/hermes_cli/test_models.py @@ -1,6 +1,5 @@ """Tests for the hermes_cli models module.""" -import urllib.request from unittest.mock import patch, MagicMock from hermes_cli.nous_account import NousPortalAccountInfo @@ -71,7 +70,7 @@ class TestFetchOpenRouterModels: return b'{"data":[{"id":"anthropic/claude-opus-4.8","pricing":{"prompt":"0.000015","completion":"0.000075"}},{"id":"qwen/qwen3.7-max","pricing":{"prompt":"0.000000325","completion":"0.00000195"}},{"id":"nvidia/nemotron-3-super-120b-a12b:free","pricing":{"prompt":"0","completion":"0"}}]}' monkeypatch.setattr(_models_mod, "_openrouter_catalog_cache", None) - with patch("hermes_cli.models.urllib.request.urlopen", return_value=_Resp()): + with patch("hermes_cli.models._urlopen_model_catalog_request", return_value=_Resp()): models = fetch_openrouter_models(force_refresh=True) assert models == [ @@ -83,7 +82,7 @@ class TestFetchOpenRouterModels: def test_falls_back_to_static_snapshot_on_fetch_failure(self, monkeypatch): monkeypatch.setattr(_models_mod, "_openrouter_catalog_cache", None) - with patch("hermes_cli.models.urllib.request.urlopen", side_effect=OSError("boom")): + with patch("hermes_cli.models._urlopen_model_catalog_request", side_effect=OSError("boom")): models = fetch_openrouter_models(force_refresh=True) assert models == OPENROUTER_MODELS @@ -128,7 +127,10 @@ class TestFetchOpenRouterModels: ], ) monkeypatch.setattr(_models_mod, "_openrouter_catalog_cache", None) - with patch("hermes_cli.models.urllib.request.urlopen", return_value=_Resp()): + with ( + patch("hermes_cli.model_catalog.get_curated_openrouter_models", return_value=[]), + patch("hermes_cli.models._urlopen_model_catalog_request", return_value=_Resp()), + ): models = fetch_openrouter_models(force_refresh=True) ids = [mid for mid, _ in models] @@ -162,7 +164,7 @@ class TestFetchOpenRouterModels: ) monkeypatch.setattr(_models_mod, "_openrouter_catalog_cache", None) - with patch("hermes_cli.models.urllib.request.urlopen", return_value=_Resp()): + with patch("hermes_cli.models._urlopen_model_catalog_request", return_value=_Resp()): models = fetch_openrouter_models(force_refresh=True) ids = [mid for mid, _ in models] @@ -781,7 +783,7 @@ class TestNousRecommendedModels: def test_fetch_caches_per_portal_url(self): from hermes_cli.models import fetch_nous_recommended_models mock_cm = self._mock_urlopen(self._SAMPLE_PAYLOAD) - with patch("urllib.request.urlopen", return_value=mock_cm) as mock_urlopen: + with patch("hermes_cli.models._urlopen_model_catalog_request", return_value=mock_cm) as mock_urlopen: a = fetch_nous_recommended_models("https://portal.example.com") b = fetch_nous_recommended_models("https://portal.example.com") assert a == self._SAMPLE_PAYLOAD @@ -791,21 +793,21 @@ class TestNousRecommendedModels: def test_fetch_cache_is_keyed_per_portal(self): from hermes_cli.models import fetch_nous_recommended_models mock_cm = self._mock_urlopen(self._SAMPLE_PAYLOAD) - with patch("urllib.request.urlopen", return_value=mock_cm) as mock_urlopen: + with patch("hermes_cli.models._urlopen_model_catalog_request", return_value=mock_cm) as mock_urlopen: fetch_nous_recommended_models("https://portal.example.com") fetch_nous_recommended_models("https://portal.staging-nousresearch.com") assert mock_urlopen.call_count == 2 # different portals → separate fetches def test_fetch_returns_empty_on_network_failure(self): from hermes_cli.models import fetch_nous_recommended_models - with patch("urllib.request.urlopen", side_effect=OSError("boom")): + with patch("hermes_cli.models._urlopen_model_catalog_request", side_effect=OSError("boom")): result = fetch_nous_recommended_models("https://portal.example.com") assert result == {} def test_fetch_force_refresh_bypasses_cache(self): from hermes_cli.models import fetch_nous_recommended_models mock_cm = self._mock_urlopen(self._SAMPLE_PAYLOAD) - with patch("urllib.request.urlopen", return_value=mock_cm) as mock_urlopen: + with patch("hermes_cli.models._urlopen_model_catalog_request", return_value=mock_cm) as mock_urlopen: fetch_nous_recommended_models("https://portal.example.com") fetch_nous_recommended_models("https://portal.example.com", force_refresh=True) assert mock_urlopen.call_count == 2 @@ -971,43 +973,3 @@ class TestCodexSoftAcceptPlausibilityGate: r = validate_requested_model("gpt-5.5", "openai-codex") assert r["accepted"] is True assert r["recognized"] is True - - -class TestModelCatalogRedirectCredentialStripping: - def test_cross_host_redirect_strips_provider_credentials(self): - req = urllib.request.Request("https://models.example.test/v1/models") - req.add_header("Authorization", "Bearer sk-model-secret") - req.add_header("x-api-key", "anthropic-secret") - req.add_header("Cookie", "session=secret") - req.add_header("User-Agent", "hermes-test") - - handler = _models_mod._StripCredentialRedirectHandler("models.example.test") - redirected = handler.redirect_request( - req, None, 302, "Found", {}, "https://attacker.example.test/leak" - ) - - redirected_headers = { - key.lower(): value - for key, value in redirected.header_items() - } - assert "authorization" not in redirected_headers - assert "x-api-key" not in redirected_headers - assert "cookie" not in redirected_headers - assert redirected_headers["user-agent"] == "hermes-test" - - def test_same_host_redirect_preserves_provider_credentials(self): - req = urllib.request.Request("https://models.example.test/v1/models") - req.add_header("Authorization", "Bearer sk-model-secret") - req.add_header("x-api-key", "anthropic-secret") - - handler = _models_mod._StripCredentialRedirectHandler("models.example.test") - redirected = handler.redirect_request( - req, None, 302, "Found", {}, "https://models.example.test/v1/models/" - ) - - redirected_headers = { - key.lower(): value - for key, value in redirected.header_items() - } - assert redirected_headers["authorization"] == "Bearer sk-model-secret" - assert redirected_headers["x-api-key"] == "anthropic-secret" diff --git a/tests/hermes_cli/test_urllib_security.py b/tests/hermes_cli/test_urllib_security.py new file mode 100644 index 00000000000..da4189a2554 --- /dev/null +++ b/tests/hermes_cli/test_urllib_security.py @@ -0,0 +1,330 @@ +"""Wire-level tests for credential-safe stdlib urllib redirects.""" + +from __future__ import annotations + +import json +from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer +from threading import Thread +import urllib.error +import urllib.request + +import pytest + +from hermes_cli.urllib_security import ( + SafeCredentialRedirectHandler, + open_credentialed_url, + url_origin, +) + + +class _Response: + def __init__(self, payload: bytes = b"{}") -> None: + self._payload = payload + + def __enter__(self): + return self + + def __exit__(self, *_args): + return False + + def read(self) -> bytes: + return self._payload + + +class _RecordingHandler(BaseHTTPRequestHandler): + redirect_to = "" + redirect_status = 302 + requests: list[tuple[str, dict[str, str]]] = [] + + def _record(self) -> None: + type(self).requests.append( + (self.command, {name.lower(): value for name, value in self.headers.items()}) + ) + + def do_GET(self): + if self.path.startswith("/redirect"): + self.send_response(type(self).redirect_status) + self.send_header("Location", type(self).redirect_to) + self.end_headers() + return + self._record() + body = json.dumps({"data": []}).encode() + self.send_response(200) + self.send_header("Content-Length", str(len(body))) + self.end_headers() + self.wfile.write(body) + + def do_POST(self): + self.rfile.read(int(self.headers.get("Content-Length", "0"))) + if self.path == "/redirect": + self.send_response(type(self).redirect_status) + self.send_header("Location", type(self).redirect_to) + self.end_headers() + return + self._record() + self.send_response(200) + self.end_headers() + + def log_message(self, _format, *_args): + pass + + +def _server(): + server = ThreadingHTTPServer(("127.0.0.1", 0), _RecordingHandler) + Thread(target=server.serve_forever, daemon=True).start() + return server + + +def _credential_headers() -> dict[str, str]: + return { + "Authorization": "Bearer secret", + "Cookie": "session=secret", + "CF-Access-Client-Secret": "cloudflare-secret", + "X-Custom-Auth": "tenant-secret", + "Accept": "application/json", + "User-Agent": "hermes-test", + } + + +def test_url_origin_normalizes_default_ports_and_trailing_dot(): + assert url_origin("https://EXAMPLE.test./models") == ( + "https", + "example.test", + 443, + ) + assert url_origin("https://example.test:443/other") == ( + "https", + "example.test", + 443, + ) + assert url_origin("http://example.test") != url_origin("https://example.test") + + +def test_cross_host_redirect_drops_arbitrary_credentials_on_wire(): + source = _server() + sink = _server() + _RecordingHandler.requests = [] + _RecordingHandler.redirect_status = 302 + _RecordingHandler.redirect_to = f"http://localhost:{sink.server_port}/sink" + try: + request = urllib.request.Request( + f"http://127.0.0.1:{source.server_port}/redirect", + headers=_credential_headers(), + ) + with open_credentialed_url(request, timeout=3) as response: + response.read() + finally: + source.shutdown() + sink.shutdown() + + method, headers = _RecordingHandler.requests[-1] + assert method == "GET" + assert headers["accept"] == "application/json" + assert headers["user-agent"] == "hermes-test" + for name in ( + "authorization", + "cookie", + "cf-access-client-secret", + "x-custom-auth", + ): + assert name not in headers + + +def test_same_host_different_port_drops_credentials_on_wire(): + source = _server() + sink = _server() + _RecordingHandler.requests = [] + _RecordingHandler.redirect_status = 302 + _RecordingHandler.redirect_to = f"http://127.0.0.1:{sink.server_port}/sink" + try: + request = urllib.request.Request( + f"http://127.0.0.1:{source.server_port}/redirect", + headers=_credential_headers(), + ) + with open_credentialed_url(request, timeout=3) as response: + response.read() + finally: + source.shutdown() + sink.shutdown() + + _, headers = _RecordingHandler.requests[-1] + assert "authorization" not in headers + assert "cf-access-client-secret" not in headers + + +def test_same_origin_redirect_preserves_headers_on_wire(): + server = _server() + _RecordingHandler.requests = [] + _RecordingHandler.redirect_status = 302 + _RecordingHandler.redirect_to = f"http://127.0.0.1:{server.server_port}/sink" + try: + request = urllib.request.Request( + f"http://127.0.0.1:{server.server_port}/redirect", + headers=_credential_headers(), + ) + with open_credentialed_url(request, timeout=3) as response: + response.read() + finally: + server.shutdown() + + _, headers = _RecordingHandler.requests[-1] + assert headers["authorization"] == "Bearer secret" + assert headers["cf-access-client-secret"] == "cloudflare-secret" + + +def test_scheme_downgrade_is_cross_origin(): + request = urllib.request.Request( + "https://models.example.test/models", headers=_credential_headers() + ) + handler = SafeCredentialRedirectHandler(request.full_url) + redirected = handler.redirect_request( + request, + None, + 302, + "Found", + {}, + "http://models.example.test/models", + ) + assert redirected is not None + headers = {name.lower(): value for name, value in redirected.header_items()} + assert "authorization" not in headers + assert "cf-access-client-secret" not in headers + + +def test_post_302_uses_urllib_semantics_and_drops_credentials(): + request = urllib.request.Request( + "https://models.example.test/load", + data=b"{}", + headers={**_credential_headers(), "Content-Type": "application/json"}, + method="POST", + ) + handler = SafeCredentialRedirectHandler(request.full_url) + redirected = handler.redirect_request( + request, + None, + 302, + "Found", + {}, + "https://other.example.test/load", + ) + assert redirected is not None + assert redirected.get_method() == "GET" + assert redirected.data is None + headers = {name.lower(): value for name, value in redirected.header_items()} + assert "authorization" not in headers + assert "content-type" not in headers + + +def test_post_307_remains_rejected_by_urllib(): + request = urllib.request.Request( + "https://models.example.test/load", + data=b"{}", + headers=_credential_headers(), + method="POST", + ) + handler = SafeCredentialRedirectHandler(request.full_url) + with pytest.raises(urllib.error.HTTPError): + handler.redirect_request( + request, + None, + 307, + "Temporary Redirect", + {}, + "https://other.example.test/load", + ) + + +def test_explicit_opener_factory_is_instrumentable_without_security_bypass(): + calls = [] + + class _Opener: + def open(self, request, *, timeout): + calls.append((request.full_url, timeout)) + return _Response() + + def factory(*handlers): + assert any(isinstance(h, SafeCredentialRedirectHandler) for h in handlers) + return _Opener() + + request = urllib.request.Request( + "https://models.example.test/models", headers={"Authorization": "secret"} + ) + with open_credentialed_url(request, timeout=7, opener_factory=factory): + pass + assert calls == [("https://models.example.test/models", 7)] + + +def test_probe_api_models_drops_custom_credentials_on_wire(): + from hermes_cli.models import probe_api_models + + source = _server() + sink = _server() + _RecordingHandler.requests = [] + _RecordingHandler.redirect_status = 302 + _RecordingHandler.redirect_to = f"http://localhost:{sink.server_port}/sink" + try: + result = probe_api_models( + "provider-key", + f"http://127.0.0.1:{source.server_port}/redirect/..", + timeout=3, + request_headers={ + "CF-Access-Client-Secret": "cloudflare-secret", + "X-Custom-Auth": "tenant-secret", + }, + ) + finally: + source.shutdown() + sink.shutdown() + + assert result["models"] == [] + _, headers = _RecordingHandler.requests[-1] + assert "authorization" not in headers + assert "cf-access-client-secret" not in headers + assert "x-custom-auth" not in headers + + +class _LmStudioSourceHandler(BaseHTTPRequestHandler): + redirect_to = "" + + def do_POST(self): + self.rfile.read(int(self.headers.get("Content-Length", "0"))) + self.send_response(302) + self.send_header("Location", type(self).redirect_to) + self.end_headers() + + def log_message(self, format, *_args): + pass + + +def test_lmstudio_load_post_drops_bearer_on_redirect(monkeypatch): + from hermes_cli import models + + sink = _server() + source = ThreadingHTTPServer(("127.0.0.1", 0), _LmStudioSourceHandler) + Thread(target=source.serve_forever, daemon=True).start() + _RecordingHandler.requests = [] + _LmStudioSourceHandler.redirect_to = f"http://localhost:{sink.server_port}/sink" + monkeypatch.setattr( + models, + "_lmstudio_fetch_raw_models", + lambda **_kwargs: [ + {"id": "model", "max_context_length": 8192, "loaded_instances": []} + ], + ) + try: + loaded = models.ensure_lmstudio_model_loaded( + "model", + f"http://127.0.0.1:{source.server_port}", + api_key="lm-secret", + target_context_length=4096, + timeout=3, + ) + finally: + source.shutdown() + sink.shutdown() + + assert loaded == 4096 + method, headers = _RecordingHandler.requests[-1] + assert method == "GET" + assert "authorization" not in headers + assert "content-type" not in headers