fix(security): preserve installed urllib policies

This commit is contained in:
kshitijk4poor 2026-07-11 11:37:39 +05:30 committed by kshitij
parent 6fe2c8b78c
commit 27a1042b13
2 changed files with 65 additions and 3 deletions

View file

@ -2,6 +2,7 @@
from __future__ import annotations
import copy
import urllib.parse
import urllib.request
from collections.abc import Callable, Iterable
@ -58,18 +59,38 @@ class SafeCredentialRedirectHandler(urllib.request.HTTPRedirectHandler):
return redirected
def _secure_opener_from_installed_policy(original_url: str):
"""Clone the installed opener's handlers, replacing redirect policy only."""
installed = getattr(urllib.request, "_opener", None)
if installed is None:
installed = urllib.request.build_opener()
handlers = [
copy.copy(handler)
for handler in getattr(installed, "handlers", ())
if not isinstance(handler, urllib.request.HTTPRedirectHandler)
]
handlers.append(SafeCredentialRedirectHandler(original_url))
return urllib.request.build_opener(*handlers)
def open_credentialed_url(
request: urllib.request.Request,
*,
timeout: float,
opener_factory: Callable[..., Any] = urllib.request.build_opener,
opener_factory: Callable[..., Any] | None = None,
):
"""Open a request without forwarding credentials across origins.
``opener_factory`` is an explicit instrumentation/test seam. Security is
The default preserves an application-installed opener's proxy, TLS,
cookies, custom protocol handlers, and instrumentation while replacing its
redirect handler. ``opener_factory`` is an explicit test seam; security is
never disabled based on global ``urlopen`` identity.
"""
opener = opener_factory(SafeCredentialRedirectHandler(request.full_url))
if opener_factory is None:
opener = _secure_opener_from_installed_policy(request.full_url)
else:
opener = opener_factory(SafeCredentialRedirectHandler(request.full_url))
return opener.open(request, timeout=timeout)

View file

@ -254,6 +254,47 @@ def test_explicit_opener_factory_is_instrumentable_without_security_bypass():
assert calls == [("https://models.example.test/models", 7)]
def test_installed_custom_opener_policy_is_preserved(monkeypatch):
opened = []
class FooHandler(urllib.request.BaseHandler):
def foo_open(self, request):
opened.append(request.full_url)
return _Response(b"custom")
installed = urllib.request.build_opener(FooHandler())
monkeypatch.setattr(urllib.request, "_opener", installed)
request = urllib.request.Request(
"foo://models.example.test/catalog", headers={"Authorization": "secret"}
)
with open_credentialed_url(request, timeout=3) as response:
assert response.read() == b"custom"
assert opened == ["foo://models.example.test/catalog"]
def test_installed_proxy_handler_is_preserved(monkeypatch):
installed = urllib.request.build_opener(
urllib.request.ProxyHandler({"https": "http://proxy.example.test:8443"})
)
monkeypatch.setattr(urllib.request, "_opener", installed)
from hermes_cli.urllib_security import _secure_opener_from_installed_policy
secured = _secure_opener_from_installed_policy(
"https://models.example.test/catalog"
)
proxy_handlers = [
handler
for handler in getattr(secured, "handlers", ())
if isinstance(handler, urllib.request.ProxyHandler)
]
assert proxy_handlers
assert getattr(proxy_handlers[0], "proxies", {}) == {
"https": "http://proxy.example.test:8443"
}
def test_probe_api_models_drops_custom_credentials_on_wire():
from hermes_cli.models import probe_api_models