mirrors/hermes-agent
1
0
Fork 0
mirror of https://github.com/NousResearch/hermes-agent.git synced 2026-07-15 14:22:43 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions 5

fix(security): enforce one redirect credential policy

Browse source
This commit is contained in:
kshitijk4poor 2026-07-11 11:29:31 +05:30 • committed by kshitij
parent b8dd1bf3a5
commit 6e75ba7fa0
6 changed files with 448 additions and 139 deletions
Download patch file Download diff file Expand all files Collapse all files

55
hermes_cli/models.py
View file

@ -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

80
hermes_cli/urllib_security.py Normal file
View file

@ -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",
]

34
providers/base.py
View file

@ -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]

28
tests/hermes_cli/test_model_validation.py
View file

@ -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")

60
tests/hermes_cli/test_models.py
View file

@ -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"

330
tests/hermes_cli/test_urllib_security.py Normal file
View file

@ -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