compare full origin (scheme, host, port), not hostname, before keeping credentials

Review feedback: a same-host redirect to a different port can land on a
different service, which must not inherit the provider API key. Compare
(scheme, hostname, effective port) — with 80/443 defaults — instead of
hostname alone, and add a two-server regression test for the
same-host/different-port case.

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
This commit is contained in:
solyanviktor-star 2026-07-07 22:17:10 +03:00 committed by kshitij
parent a061788d43
commit 61e3bd67d6
2 changed files with 46 additions and 24 deletions

View file

@ -210,25 +210,32 @@ class ProviderProfile:
# urllib keeps every header — including Authorization — when it
# follows a redirect, so a catalog endpoint answering 3xx to a
# different host would receive the provider API key. Drop
# credential headers on cross-host redirects.
# 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"}
original_host = (urllib.parse.urlparse(url).hostname or "").lower()
class _StripAuthOnCrossHostRedirect(urllib.request.HTTPRedirectHandler):
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:
new_host = (urllib.parse.urlparse(newurl).hostname or "").lower()
if new_host != original_host:
for header in list(new_req.headers):
if header.lower() in sensitive:
new_req.remove_header(header)
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(_StripAuthOnCrossHostRedirect())
opener = urllib.request.build_opener(_StripAuthOnCrossOriginRedirect())
try:
with opener.open(req, timeout=timeout) as resp:

View file

@ -118,21 +118,18 @@ class TestCustomProviderBaseUrlPassthrough:
class _RedirectingHandler(BaseHTTPRequestHandler):
"""Redirects /models to another hostname and records received headers."""
"""Redirects /models to a configurable target and records received headers."""
redirect_host = "localhost" # resolves to the same server, different hostname
redirect_to = "" # full URL to redirect /models to (set per test)
received_headers: dict = {}
def do_GET(self):
if self.path.rstrip("/") == "/models":
self.send_response(302)
self.send_header(
"Location",
f"http://{self.redirect_host}:{self.server.server_address[1]}/redirected",
)
self.send_header("Location", type(self).redirect_to)
self.end_headers()
else:
type(self).received_headers = dict(self.headers)
_RedirectingHandler.received_headers = dict(self.headers)
body = json.dumps({"data": [{"id": "redirected-model"}]}).encode()
self.send_response(200)
self.send_header("Content-Type", "application/json")
@ -144,14 +141,18 @@ class _RedirectingHandler(BaseHTTPRequestHandler):
class TestFetchModelsRedirectCredentialStripping:
"""Credential headers must not follow a redirect to a different host."""
"""Credential headers must not follow a redirect outside the original origin."""
def _run(self, redirect_host):
_RedirectingHandler.redirect_host = redirect_host
def _run(self, redirect_to):
"""redirect_to is a callable (first_port, second_port) -> Location URL."""
_RedirectingHandler.received_headers = {}
server = HTTPServer(("127.0.0.1", 0), _RedirectingHandler)
second_server = HTTPServer(("127.0.0.1", 0), _RedirectingHandler)
port = server.server_address[1]
second_port = second_server.server_address[1]
_RedirectingHandler.redirect_to = redirect_to(port, second_port)
Thread(target=server.serve_forever, daemon=True).start()
Thread(target=second_server.serve_forever, daemon=True).start()
try:
profile = ProviderProfile(
name="test",
@ -161,17 +162,31 @@ class TestFetchModelsRedirectCredentialStripping:
result = profile.fetch_models(api_key="bearer-secret")
finally:
server.shutdown()
second_server.shutdown()
headers = {k.lower(): v for k, v in _RedirectingHandler.received_headers.items()}
return result, headers
def test_cross_host_redirect_strips_credentials(self):
result, headers = self._run(redirect_host="localhost")
result, headers = self._run(
lambda port, _: f"http://localhost:{port}/redirected"
)
assert result == ["redirected-model"] # fetch itself still works
assert "authorization" not in headers
assert "x-api-key" not in headers
def test_same_host_redirect_keeps_credentials(self):
result, headers = self._run(redirect_host="127.0.0.1")
def test_same_host_different_port_redirect_strips_credentials(self):
"""A different port is a different origin — it can be a different service."""
result, headers = self._run(
lambda _, second_port: f"http://127.0.0.1:{second_port}/redirected"
)
assert result == ["redirected-model"]
assert "authorization" not in headers
assert "x-api-key" not in headers
def test_same_origin_redirect_keeps_credentials(self):
result, headers = self._run(
lambda port, _: f"http://127.0.0.1:{port}/redirected"
)
assert result == ["redirected-model"]
assert headers.get("authorization") == "Bearer bearer-secret"
assert headers.get("x-api-key") == "default-header-secret"