fix(security): sanitize after installed request hooks

hermes_cli/urllib_security.py

@@ -58,6 +58,28 @@ class SafeCredentialRedirectHandler(urllib.request.HTTPRedirectHandler):
         return redirected
 
 
+class _CrossOriginRequestSanitizer(urllib.request.BaseHandler):
+    """Strip headers after installed request processors have run."""
+
+    # Request processors run in ascending order. Keep this last so an installed
+    # cookie/auth/instrumentation processor cannot re-add a secret after the
+    # redirect handler sanitizes the new Request.
+    handler_order = 1_000_000_000
+
+    def __init__(self, original_url: str) -> None:
+        self._original_origin = url_origin(original_url)
+
+    def _sanitize(self, request: urllib.request.Request):
+        if url_origin(request.full_url) != self._original_origin:
+            for name, _value in list(request.header_items()):
+                if name.lower() not in _CROSS_ORIGIN_SAFE_HEADERS:
+                    request.remove_header(name)
+        return request
+
+    http_request = _sanitize
+    https_request = _sanitize
+
+
 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)
@@ -70,8 +92,17 @@ def _secure_opener_from_installed_policy(original_url: str):
         if not isinstance(handler, urllib.request.HTTPRedirectHandler)
     ]
     handlers.append(SafeCredentialRedirectHandler(original_url))
+    handlers.append(_CrossOriginRequestSanitizer(original_url))
     secured = urllib.request.build_opener(*handlers)
-    secured.addheaders = list(getattr(installed, "addheaders", ()))
+    # OpenerDirector injects addheaders after request processors, which would
+    # bypass the sanitizer on redirects. Carry them on the initial request
+    # instead, then leave the rebuilt opener's late-injection list empty.
+    setattr(
+        secured,
+        "_hermes_initial_addheaders",
+        list(getattr(installed, "addheaders", ())),
+    )
+    secured.addheaders = []
     return secured
 
 
@@ -90,6 +121,9 @@ def open_credentialed_url(
     """
     if opener_factory is None:
         opener = _secure_opener_from_installed_policy(request.full_url)
+        for name, value in getattr(opener, "_hermes_initial_addheaders", ()):
+            if not request.has_header(name):
+                request.add_header(name, value)
     else:
         opener = opener_factory(SafeCredentialRedirectHandler(request.full_url))
     return opener.open(request, timeout=timeout)

tests/hermes_cli/test_urllib_security.py

@@ -281,13 +281,19 @@ def test_installed_custom_opener_policy_is_preserved(monkeypatch):
     secured = _secure_opener_from_installed_policy(
         "foo://models.example.test/catalog"
     )
-    assert secured.addheaders == installed.addheaders
+    assert secured.addheaders == []
+    assert getattr(secured, "_hermes_initial_addheaders") == installed.addheaders
 
     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"
+    request_headers = {
+        name.lower(): value for name, value in request.header_items()
+    }
+    assert request_headers["x-trace-policy"] == "installed"
+    assert request_headers["user-agent"] == "enterprise-client"
     assert opened == ["foo://models.example.test/catalog"]
 
 
@@ -313,6 +319,40 @@ def test_installed_proxy_handler_is_preserved(monkeypatch):
     }
 
 
+def test_installed_request_processor_cannot_resurrect_cross_origin_secret(
+    monkeypatch,
+):
+    source = _server()
+    sink = _server()
+    _RecordingHandler.requests = []
+    _RecordingHandler.redirect_status = 302
+    _RecordingHandler.redirect_to = f"http://localhost:{sink.server_port}/sink"
+
+    class SecretProcessor(urllib.request.BaseHandler):
+        def http_request(self, request):
+            request.add_header("X-Installed-Secret", "must-not-cross")
+            return request
+
+    installed = urllib.request.build_opener(SecretProcessor())
+    installed.addheaders = [("X-Opener-Secret", "also-must-not-cross")]
+    monkeypatch.setattr(urllib.request, "_opener", installed)
+    try:
+        request = urllib.request.Request(
+            f"http://127.0.0.1:{source.server_port}/redirect",
+            headers={"Authorization": "Bearer secret"},
+        )
+        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 "x-installed-secret" not in headers
+    assert "x-opener-secret" not in headers
+
+
 def test_multihop_redirects_never_resurrect_credentials():
     request = urllib.request.Request(
         "https://a.example.test/models", headers=_credential_headers()