meta-openembedded/meta-python/recipes-devtools/python/python3-aiohttp/CVE-2026-54276.patch

From cc464f0ffe7ac8054f3992f8f53bb78b66632aca Mon Sep 17 00:00:00 2001
From: "patchback[bot]" <45432694+patchback[bot]@users.noreply.github.com>
Date: Sun, 7 Jun 2026 00:30:39 -0500
Subject: [PATCH] [PR #12825/cb1d6a53 backport][3.14] Scope
 DigestAuthMiddleware credentials to the request origin (#12839)

CVE: CVE-2026-54276
Upstream-Status: Backport [https://github.com/aio-libs/aiohttp/commit/38d16060037e1bfcd6d677abababa3c2a4bb58fa]

Co-authored-by: J. Nick Koston <nick@koston.org>
Co-authored-by: J. Nick Koston <nick@home-assistant.io>
(cherry picked from commit 38d16060037e1bfcd6d677abababa3c2a4bb58fa)
Signed-off-by: Darsh Kelaiya <dkelaiya@cisco.com>
---
 CHANGES/12825.bugfix.rst                    |   1 +
 aiohttp/client_middleware_digest_auth.py    |  21 +++
 docs/client_reference.rst                   |  14 ++
 tests/test_client_middleware_digest_auth.py | 170 ++++++++++++++++++++
 4 files changed, 206 insertions(+)
 create mode 100644 CHANGES/12825.bugfix.rst

diff --git a/CHANGES/12825.bugfix.rst b/CHANGES/12825.bugfix.rst
new file mode 100644
index 000000000..88d1bfe8c
--- /dev/null
+++ b/CHANGES/12825.bugfix.rst
@@ -0,0 +1 @@
+Scoped :class:`~aiohttp.DigestAuthMiddleware` credentials to the origin of the first request it handles, so a redirect to a different origin no longer triggers a digest response computed from the configured credentials; a challenge from another origin is only answered when that origin falls within a protection space advertised by the anchor origin through the RFC 7616 ``domain`` directive -- by :user:`bdraco`.
diff --git a/aiohttp/client_middleware_digest_auth.py b/aiohttp/client_middleware_digest_auth.py
index d7f2f1eb9..a818e57cd 100644
--- a/aiohttp/client_middleware_digest_auth.py
+++ b/aiohttp/client_middleware_digest_auth.py
@@ -171,6 +171,15 @@ class DigestAuthMiddleware:
     - Includes replay attack protection with client nonce count tracking
     - Supports preemptive authentication per RFC 7616 Section 3.6
 
+    Origin scoping:
+    The credentials are scoped to the origin of the first request the
+    middleware handles. A request to a different origin is passed through
+    untouched, so it never receives a digest response computed from those
+    credentials, unless that origin falls within a protection space the
+    anchor origin advertised through the RFC 7616 ``domain`` directive. Make
+    the first request through the middleware against the intended origin, as
+    the anchor is pinned to it and not reset for the life of the instance.
+
     Standards compliance:
     - RFC 7616: HTTP Digest Access Authentication (primary reference)
     - RFC 2617: HTTP Authentication (deprecated by RFC 7616)
@@ -207,6 +216,8 @@ class DigestAuthMiddleware:
         self._preemptive: bool = preemptive
         # Set of URLs defining the protection space
         self._protection_space: List[str] = []
+        # Origin the credentials are scoped to; set on the first request.
+        self._origin: URL | None = None
 
     async def _encode(
         self, method: str, url: URL, body: Union[Payload, Literal[b""]]
@@ -454,6 +465,16 @@ class DigestAuthMiddleware:
         self, request: ClientRequest, handler: ClientHandlerType
     ) -> ClientResponse:
         """Run the digest auth middleware."""
+        # Credentials are scoped to the first request's origin. Other origins
+        # pass through untouched unless a challenge from the anchor origin
+        # advertised them via RFC 7616 domain; mirrors aiohttp stripping
+        # Authorization on cross-origin redirects.
+        origin = request.url.origin()
+        if self._origin is None:
+            self._origin = origin
+        elif origin != self._origin and not self._in_protection_space(request.url):
+            return await handler(request)
+
         response = None
         for retry_count in range(2):
             # Apply authorization header if:
diff --git a/docs/client_reference.rst b/docs/client_reference.rst
index 374796f40..63ee375ca 100644
--- a/docs/client_reference.rst
+++ b/docs/client_reference.rst
@@ -2367,6 +2367,16 @@ Utilities
    The server may still respond with a 401 status and ``stale=true`` if the nonce
    has expired, in which case the middleware will automatically retry with the new nonce.
 
+   **Origin scoping**
+
+   The credentials are scoped to the origin of the first request the middleware
+   handles. A request to a different origin is passed through untouched, so it
+   never receives a digest response computed from those credentials, unless that
+   origin falls within a protection space the anchor origin advertised through
+   the RFC 7616 ``domain`` directive. Make the first request through the
+   middleware against the intended origin, as the anchor is pinned to it and not
+   reset for the life of the instance.
+
    To disable preemptive authentication and require a 401 challenge for every request,
    set ``preemptive=False``::
 
@@ -2392,6 +2402,10 @@ Utilities
    .. versionadded:: 3.12
    .. versionchanged:: 3.12.8
       Added ``preemptive`` parameter to enable/disable preemptive authentication.
+   .. versionchanged:: 3.14.1
+      Credentials are scoped to the origin of the first request the middleware
+      handles; other origins are passed through untouched unless covered by an
+      RFC 7616 ``domain`` directive from the anchor origin.
 
 
 .. class:: CookieJar(*, unsafe=False, quote_cookie=True, treat_as_secure_origin = [])
diff --git a/tests/test_client_middleware_digest_auth.py b/tests/test_client_middleware_digest_auth.py
index 65e7d667e..03fab2691 100644
--- a/tests/test_client_middleware_digest_auth.py
+++ b/tests/test_client_middleware_digest_auth.py
@@ -1156,6 +1156,176 @@ async def test_preemptive_auth_without_domain_uses_origin(
     )  # Second request - preemptive auth (entire origin)
 
 
+async def test_does_not_answer_cross_origin_redirect_challenge(
+    aiohttp_server: AiohttpServer,
+) -> None:
+    """A cross-origin redirect target must not receive a digest response.
+
+    aiohttp strips the Authorization header on cross-origin redirects; the
+    digest middleware must not re-add one for the redirect target, otherwise
+    the configured credentials leak to an origin the caller never targeted.
+    """
+    target_auth_headers: list[str | None] = []
+
+    async def target_handler(request: Request) -> Response:
+        auth_header = request.headers.get(hdrs.AUTHORIZATION)
+        target_auth_headers.append(auth_header)
+        assert auth_header is None
+        return Response(
+            status=401,
+            headers={
+                hdrs.WWW_AUTHENTICATE: 'Digest realm="evil", nonce="cross-origin"'
+            },
+        )
+
+    target_app = Application()
+    target_app.router.add_get("/", target_handler)
+    target_server = await aiohttp_server(target_app)
+
+    async def source_handler(request: Request) -> Response:
+        return Response(
+            status=302, headers={hdrs.LOCATION: str(target_server.make_url("/"))}
+        )
+
+    source_app = Application()
+    source_app.router.add_get("/", source_handler)
+    source_server = await aiohttp_server(source_app)
+
+    digest_auth = DigestAuthMiddleware("victim", "secret")
+    async with (
+        ClientSession(middlewares=(digest_auth,)) as session,
+        session.get(source_server.make_url("/")) as response,
+    ):
+        await response.text()
+
+    assert target_auth_headers == [None]
+
+
+async def test_answers_same_origin_redirect_challenge(
+    aiohttp_server: AiohttpServer,
+) -> None:
+    """A same-origin redirect that issues a challenge must still authenticate."""
+    auth_headers: list[str | None] = []
+
+    async def handler(request: Request) -> Response:
+        if request.path == "/start":
+            return Response(status=302, headers={hdrs.LOCATION: "/protected"})
+        auth_header = request.headers.get(hdrs.AUTHORIZATION)
+        auth_headers.append(auth_header)
+        if auth_header is None:
+            return Response(
+                status=401,
+                headers={hdrs.WWW_AUTHENTICATE: 'Digest realm="good", nonce="abc"'},
+            )
+        return Response(text="OK")
+
+    app = Application()
+    app.router.add_get("/start", handler)
+    app.router.add_get("/protected", handler)
+    server = await aiohttp_server(app)
+
+    digest_auth = DigestAuthMiddleware("user", "pass")
+    async with (
+        ClientSession(middlewares=(digest_auth,)) as session,
+        session.get(server.make_url("/start")) as response,
+    ):
+        assert response.status == 200
+        assert await response.text() == "OK"
+
+    assert auth_headers[0] is None
+    assert auth_headers[1] is not None
+    assert auth_headers[1].startswith("Digest")
+
+
+async def test_answers_cross_origin_within_domain_protection_space(
+    aiohttp_server: AiohttpServer,
+) -> None:
+    """A different origin advertised via the ``domain`` directive is honored.
+
+    RFC 7616 allows a challenge to define a protection space spanning other
+    servers through the ``domain`` directive. The anchor origin vouches for
+    those URIs, so preemptive auth to them is expected.
+    """
+    other_auth_headers: list[str | None] = []
+
+    async def other_handler(request: Request) -> Response:
+        other_auth_headers.append(request.headers.get(hdrs.AUTHORIZATION))
+        return Response(text="other")
+
+    other_app = Application()
+    other_app.router.add_get("/", other_handler)
+    other_server = await aiohttp_server(other_app)
+    other_origin = str(other_server.make_url("/").origin())
+
+    async def anchor_handler(request: Request) -> Response:
+        if request.headers.get(hdrs.AUTHORIZATION) is None:
+            challenge = f'Digest realm="anchor", nonce="n1", domain="{other_origin}/"'
+            return Response(status=401, headers={hdrs.WWW_AUTHENTICATE: challenge})
+        return Response(text="anchor")
+
+    anchor_app = Application()
+    anchor_app.router.add_get("/", anchor_handler)
+    anchor_server = await aiohttp_server(anchor_app)
+
+    digest_auth = DigestAuthMiddleware("user", "pass")
+    async with ClientSession(middlewares=(digest_auth,)) as session:
+        async with session.get(anchor_server.make_url("/")) as response:
+            assert response.status == 200
+        async with session.get(other_server.make_url("/")) as response:
+            assert response.status == 200
+
+    assert other_auth_headers[0] is not None
+    assert other_auth_headers[0].startswith("Digest")
+
+
+async def test_does_not_answer_cross_origin_challenge_without_redirect(
+    aiohttp_server: AiohttpServer,
+) -> None:
+    """Origin scoping applies to any cross-origin request, not just redirects.
+
+    After authenticating against the anchor origin, a direct request to a
+    different origin that issues its own challenge must not be answered with a
+    digest response computed from the configured credentials.
+    """
+    other_auth_headers: list[str | None] = []
+
+    async def other_handler(request: Request) -> Response:
+        auth_header = request.headers.get(hdrs.AUTHORIZATION)
+        other_auth_headers.append(auth_header)
+        assert auth_header is None
+        return Response(
+            status=401,
+            headers={hdrs.WWW_AUTHENTICATE: 'Digest realm="evil", nonce="x"'},
+        )
+
+    other_app = Application()
+    other_app.router.add_get("/", other_handler)
+    other_server = await aiohttp_server(other_app)
+
+    async def anchor_handler(request: Request) -> Response:
+        if request.headers.get(hdrs.AUTHORIZATION) is None:
+            return Response(
+                status=401,
+                headers={hdrs.WWW_AUTHENTICATE: 'Digest realm="anchor", nonce="n1"'},
+            )
+        return Response(text="anchor")
+
+    anchor_app = Application()
+    anchor_app.router.add_get("/", anchor_handler)
+    anchor_server = await aiohttp_server(anchor_app)
+
+    digest_auth = DigestAuthMiddleware("user", "pass")
+    async with ClientSession(middlewares=(digest_auth,)) as session:
+        async with session.get(anchor_server.make_url("/")) as response:
+            assert response.status == 200
+        async with session.get(other_server.make_url("/")) as response:
+            assert response.status == 401
+
+    # The other origin only ever saw the unauthenticated request; the
+    # middleware never answered its challenge.
+    assert other_auth_headers == [None]
+
+
 @pytest.mark.parametrize(
     ("status", "headers", "expected"),
     [