Keep following a relative redirect when the endpoint URL carries userinfo

A relative Location joined onto a URL with user:pass@ keeps that userinfo,
which is the caller's own credential for the same origin; only refuse
userinfo the redirect itself introduces.

Author: maxisbey

src/mcp/shared/_httpx_utils.py

@@ -90,16 +90,18 @@ def next_request_within_origin(response: httpx2.Response) -> httpx2.Request | No
     GET: httpx2 turns a POST into a body-less GET for 301/302/303, which would
     drop the message), its URL stays within the origin of the request just sent
     (same scheme, host and port, or http to https on the same host with default
-    ports), and the Location carries no userinfo (which httpx2 would otherwise
-    send as Basic auth). None for anything else, including a non-redirect.
+    ports), and the Location does not bring userinfo of its own (which httpx2
+    would otherwise send as Basic auth; userinfo the configured URL already had
+    is kept by a relative Location and is fine). None for anything else,
+    including a non-redirect.
     """
     next_request = response.next_request
     if next_request is None:
         return None
     sent = response.request
     if (
         next_request.method != sent.method
-        or next_request.url.userinfo
+        or (next_request.url.userinfo and next_request.url.userinfo != sent.url.userinfo)
         or not _within_origin(sent.url, next_request.url)
     ):
         return None

tests/shared/test_httpx_utils.py

@@ -208,6 +208,20 @@ async def test_redirect_location_with_userinfo_is_not_followed():
     assert received == [f"POST {url}"]
 
 
+async def test_userinfo_of_the_configured_url_kept_by_a_relative_location_is_followed():
+    """Userinfo the caller put in the endpoint URL is carried over by a relative Location (URL join
+    keeps the authority); that is the caller's own credential for the same origin, so the redirect
+    is followed as httpx2 itself would (SDK-defined)."""
+    url = "http://user:secret@mcp.example/mcp"
+    client, received, _ = _recording_client({url: (307, "/mcp/")})
+
+    async with client, stream_within_origin(client, "POST", url, content=b"payload") as response:
+        await response.aread()
+
+    assert response.status_code == 200
+    assert received == [f"POST {url}", "POST http://user:secret@mcp.example/mcp/"]
+
+
 async def test_request_within_origin_returns_a_read_response():
     """The non-streaming form hands back a response whose body is already read."""
     url = "http://mcp.example/mcp"