@@ -12,6 +12,8 @@
 from ...pipeline.stage import Stage
 from ...util.clock import ASYNC_SYSTEM_CLOCK, SYSTEM_CLOCK, AsyncClock, Clock
 from .access_token import AccessTokenInfo, TokenRequestOptions
+from .challenge import parse_challenges
+from .challenge_handler import ChallengeHandler
 from .credentials import (
     AsyncTokenCredential,
     BasicAuthCredential,
@@ -91,11 +93,23 @@ class BearerTokenPolicy(Policy):
     at most once per refresh window even under heavy concurrent send pressure.
 
     The ``on_challenge`` hook is a no-op by default; subclasses override it
-    to handle CAE/claims challenges.
+    to handle CAE/claims challenges. Alternatively, callers can pass a
+    ``challenge_handler`` to plug in a scheme-aware handler (e.g.
+    ``DigestChallengeHandler``); the handler is consulted before
+    ``on_challenge`` on 401/407 and, if it satisfies the challenge, the
+    returned ``(name, value)`` pair is stamped on the retried request.
     """
 
     STAGE = Stage.AUTH
-    __slots__ = ("_audience", "_cache", "_clock", "_credential", "_lock", "_scopes")
+    __slots__ = (
+        "_audience",
+        "_cache",
+        "_challenge_handler",
+        "_clock",
+        "_credential",
+        "_lock",
+        "_scopes",
+    )
 
     def __init__(
         self,
@@ -104,6 +118,7 @@ def __init__(
         cache: TokenCache | None = None,
         audience: str | None = None,
         clock: Clock | None = None,
+        challenge_handler: ChallengeHandler | None = None,
     ) -> None:
         if not scopes:
             raise ValueError("at least one scope is required")
@@ -113,21 +128,65 @@ def __init__(
         self._audience = audience
         self._clock: Clock = clock if clock is not None else SYSTEM_CLOCK
         self._lock = threading.Lock()
+        self._challenge_handler = challenge_handler
 
     def send(self, request: Request, ctx: PipelineContext) -> Response:
         request = self._authorize(request, ctx)
         response = self.next.send(request, ctx)
-        if int(response.status) != 401:
+        status = int(response.status)
+        if status not in (401, 407):
+            return response
+        # On 401 the bearer token was rejected: drop it from the cache so the
+        # ``on_challenge`` fallback path forces a refresh. A 407 means the
+        # proxy rejected us, not the origin — leave the cached token alone.
+        if status == 401:
+            self._cache.set(self._scopes, _expired_token(), self._audience)
+        handler_header = self._apply_challenge_handler(request, response, status)
+        if handler_header is not None:
+            request = request.with_header(*handler_header)
+            response = self.next.send(request, ctx)
+            if int(response.status) not in (401, 407):
+                return response
+            raise ClientAuthenticationError(response=response)
+        if status != 401:
+            # No handler match on a 407; surface the response unchanged so
+            # callers can inspect proxy-auth failures themselves.
             return response
-        # Drop cached token and ask subclasses to handle the challenge.
-        self._cache.set(self._scopes, _expired_token(), self._audience)
         if "WWW-Authenticate" in response.headers and self.on_challenge(request, response):
             request = self._authorize(request, ctx, force_refresh=True)
             response = self.next.send(request, ctx)
             if int(response.status) != 401:
                 return response
         raise ClientAuthenticationError(response=response)
 
+    def _apply_challenge_handler(
+        self,
+        request: Request,
+        response: Response,
+        status: int,
+    ) -> tuple[str, str] | None:
+        """Delegate to the configured ``ChallengeHandler``, if any.
+
+        Returns the ``(name, value)`` pair to stamp on the retry, or ``None``
+        when no handler is configured or the handler declines the challenge.
+        """
+        if self._challenge_handler is None:
+            return None
+        is_proxy = status == 407
+        header_name = "Proxy-Authenticate" if is_proxy else "WWW-Authenticate"
+        raw = response.headers.get(header_name)
+        if raw is None:
+            return None
+        challenges = parse_challenges(raw)
+        if not challenges or not self._challenge_handler.can_handle(challenges):
+            return None
+        return self._challenge_handler.handle(
+            request.method,
+            request.url,
+            challenges,
+            is_proxy=is_proxy,
+        )
+
     def on_challenge(self, request: Request, response: Response) -> bool:
         """Override to handle a ``WWW-Authenticate`` challenge.
 
 
@@ -14,9 +14,11 @@
 from dexpace.sdk.core.http.auth import (
     AccessTokenInfo,
     AsyncBearerTokenPolicy,
+    AuthenticateChallenge,
     BasicAuthCredential,
     BasicAuthPolicy,
     BearerTokenPolicy,
+    DigestChallengeHandler,
     KeyCredential,
     KeyCredentialPolicy,
 )
@@ -303,3 +305,116 @@ async def test_async_bearer_token_policy_serializes_concurrent_refresh() -> None
         await asyncio.gather(*(p.run(_request(), DispatchContext(_instr(t))) for t in trace_ids))
 
     assert cred.calls == 1
+
+
+class _ScriptedClient(HttpClient):
+    """Captures requests and replies with a scripted sequence of responses.
+
+    Each call consumes the next ``(status, www_authenticate)`` entry from the
+    queue. The final entry is reused if the queue runs dry.
+    """
+
+    def __init__(self, script: list[tuple[Status, str | None]]) -> None:
+        if not script:
+            raise ValueError("script must not be empty")
+        self._script = script
+        self.calls: list[Request] = []
+        self._index = 0
+
+    def execute(self, request: Request) -> Response:
+        self.calls.append(request)
+        idx = min(self._index, len(self._script) - 1)
+        self._index += 1
+        status, www_auth = self._script[idx]
+        from dexpace.sdk.core.http.common import Headers
+
+        header_pairs: list[tuple[str, str]] = []
+        if www_auth is not None:
+            header_pairs.append(("WWW-Authenticate", www_auth))
+        return Response(
+            request=request,
+            protocol=Protocol.HTTP_1_1,
+            status=status,
+            headers=Headers(header_pairs),
+        )
+
+
+def test_challenge_handler_wires_digest_authentication() -> None:
+    """Digest handler plugged into BearerTokenPolicy negotiates a 401 Digest."""
+
+    digest_challenge = (
+        'Digest realm="testrealm@host.com", '
+        'qop="auth", '
+        'nonce="dcd98b7102dd2f0e8b11d0f600bfb0c093", '
+        'opaque="5ccc069c403ebaf9f0171e9517f40e41", '
+        "algorithm=MD5"
+    )
+    client = _ScriptedClient(
+        [
+            (Status.UNAUTHORIZED, digest_challenge),
+            (Status.OK, None),
+        ]
+    )
+    cred = _StaticCredential()
+    handler = DigestChallengeHandler(
+        "Mufasa",
+        "Circle Of Life",
+        cnonce_factory=lambda: "0a4f113b",
+    )
+    policy = BearerTokenPolicy(cred, "scope-a", challenge_handler=handler)
+    with Pipeline(client, policies=[policy]) as p:
+        p.run(_request(), DispatchContext(_instr("0" * 16 + "c")))
+
+    assert len(client.calls) == 2
+    auth = client.calls[1].headers.get("authorization")
+    assert auth is not None
+    assert auth.startswith("Digest ")
+    assert 'username="Mufasa"' in auth
+    assert 'realm="testrealm@host.com"' in auth
+    assert "algorithm=MD5" in auth
+
+
+def test_challenge_handler_can_handle_false_falls_through() -> None:
+    """Handler that does not recognise the challenge falls back to on_challenge."""
+
+    class _NoopHandler:
+        def __init__(self) -> None:
+            self.can_handle_calls = 0
+            self.handle_calls = 0
+
+        def can_handle(self, challenges: list[AuthenticateChallenge]) -> bool:
+            del challenges
+            self.can_handle_calls += 1
+            return False
+
+        def handle(
+            self,
+            method: Method,
+            url: Url,
+            challenges: list[AuthenticateChallenge],
+            *,
+            is_proxy: bool,
+        ) -> tuple[str, str] | None:
+            del method, url, challenges, is_proxy
+            self.handle_calls += 1
+            return None
+
+    client = _CapturingClient(status=Status.UNAUTHORIZED, www_auth=True)
+    cred = _StaticCredential()
+    handler = _NoopHandler()
+
+    on_challenge_calls = 0
+
+    class _Spy(BearerTokenPolicy):
+        def on_challenge(self, request: Request, response: Response) -> bool:
+            nonlocal on_challenge_calls
+            on_challenge_calls += 1
+            return False
+
+    policy = _Spy(cred, "scope-a", challenge_handler=handler)
+    with Pipeline(client, policies=[policy]) as p, pytest.raises(ClientAuthenticationError):
+        p.run(_request(), DispatchContext(_instr("0" * 16 + "d")))
+
+    assert handler.can_handle_calls == 1
+    assert handler.handle_calls == 0
+    assert on_challenge_calls == 1