Coverage for /pythoncovmergedfiles/medio/medio/src/aiohttp/aiohttp/client_middleware_digest_auth.py: 19%
Shortcuts on this page
r m x toggle line displays
j k next/prev highlighted chunk
0 (zero) top of page
1 (one) first highlighted chunk
1"""
2Digest authentication middleware for aiohttp client.
4This middleware implements HTTP Digest Authentication according to RFC 7616,
5providing a more secure alternative to Basic Authentication. It supports all
6standard hash algorithms including MD5, SHA, SHA-256, SHA-512 and their session
7variants, as well as both 'auth' and 'auth-int' quality of protection (qop) options.
8"""
10import hashlib
11import os
12import re
13import sys
14import time
15from collections.abc import Callable
16from typing import Final, Literal, TypedDict
18from yarl import URL
20from . import hdrs
21from .client_exceptions import ClientError
22from .client_middlewares import ClientHandlerType
23from .client_reqrep import ClientRequest, ClientResponse
24from .payload import Payload
27class DigestAuthChallenge(TypedDict, total=False):
28 realm: str
29 nonce: str
30 qop: str
31 algorithm: str
32 opaque: str
33 domain: str
34 stale: str
37DigestFunctions: dict[str, Callable[[bytes], "hashlib._Hash"]] = {
38 "MD5": hashlib.md5,
39 "MD5-SESS": hashlib.md5,
40 "SHA": hashlib.sha1,
41 "SHA-SESS": hashlib.sha1,
42 "SHA256": hashlib.sha256,
43 "SHA256-SESS": hashlib.sha256,
44 "SHA-256": hashlib.sha256,
45 "SHA-256-SESS": hashlib.sha256,
46 "SHA512": hashlib.sha512,
47 "SHA512-SESS": hashlib.sha512,
48 "SHA-512": hashlib.sha512,
49 "SHA-512-SESS": hashlib.sha512,
50}
53# Compile the regex pattern once at module level for performance
54_HEADER_PAIRS_PATTERN = re.compile(
55 r'(?:^|\s|,\s*)(\w+)(?:\s*=\s*(?:"((?:[^"\\]|\\.)*)"|([^\s,]+)))?'
56 if sys.version_info < (3, 11)
57 else r'(?:^|\s|,\s*)((?>\w+))(?:\s*=\s*(?:"((?:[^"\\]|\\.)*)"|([^\s,]+)))?'
58 # +------------|--------|--|--||--|--|----|------|---|---||-----|-> Match valid start/sep
59 # +--------|--|--||--|--|----|------|---|---||-----|-> alphanumeric key (atomic group reduces backtracking)
60 # +--|--||--|--|----|------|---|---||-----|-> optional value; absent => bare auth-scheme token
61 # +--||--|--|----|------|---|---||-----|-> maybe whitespace
62 # +|--|--|----|------|---|---||-----|-> = (delimiter)
63 # +--|--|----|------|---|---||-----|-> maybe whitespace
64 # +--|----|------|---|---||-----|-> group quoted or unquoted
65 # +----|------|---|---||-----|-> if quoted...
66 # +------|---|---||-----|-> anything but " or \
67 # +---|---||-----|-> escaped characters allowed
68 # +---||-----|-> or can be empty string
69 # +|-----|-> if unquoted...
70 # +-----|-> anything but , or <space>
71 # +-> at least one char req'd
72)
75# RFC 7616: Challenge parameters to extract
76CHALLENGE_FIELDS: Final[
77 tuple[
78 Literal["realm", "nonce", "qop", "algorithm", "opaque", "domain", "stale"], ...
79 ]
80] = (
81 "realm",
82 "nonce",
83 "qop",
84 "algorithm",
85 "opaque",
86 "domain",
87 "stale",
88)
90# Supported digest authentication algorithms
91# Use a tuple of sorted keys for predictable documentation and error messages
92SUPPORTED_ALGORITHMS: Final[tuple[str, ...]] = tuple(sorted(DigestFunctions.keys()))
94# RFC 7616: Fields that require quoting in the Digest auth header
95# These fields must be enclosed in double quotes in the Authorization header.
96# Algorithm, qop, and nc are never quoted per RFC specifications.
97# This frozen set is used by the template-based header construction to
98# automatically determine which fields need quotes.
99QUOTED_AUTH_FIELDS: Final[frozenset[str]] = frozenset(
100 {"username", "realm", "nonce", "uri", "response", "opaque", "cnonce"}
101)
104def escape_quotes(value: str) -> str:
105 """Escape backslashes and double quotes for HTTP quoted-strings."""
106 return value.replace("\\", "\\\\").replace('"', '\\"')
109def unescape_quotes(value: str) -> str:
110 """Unescape backslashes and double quotes in HTTP quoted-strings."""
111 return value.replace('\\"', '"').replace("\\\\", "\\")
114def parse_header_pairs(header: str) -> dict[str, str]:
115 """
116 Parse key-value pairs from the first challenge of a WWW-Authenticate header.
118 This function handles the complex format of WWW-Authenticate header values,
119 supporting both quoted and unquoted values, proper handling of commas in
120 quoted values, and whitespace variations per RFC 7616.
122 A single header may carry several challenges
123 (https://www.rfc-editor.org/rfc/rfc7235#section-4.1). Parsing
124 stops at the next auth-scheme token so a later challenge's parameters cannot
125 overwrite the first challenge's values; a leading scheme token is skipped.
127 Examples of supported formats:
128 - key1="value1", key2=value2
129 - key1 = "value1" , key2="value, with, commas"
130 - key1=value1,key2="value2"
131 - realm="example.com", nonce="12345", qop="auth"
133 Args:
134 header: The header value string to parse
136 Returns:
137 Dictionary mapping parameter names to their values
138 """
139 pairs: dict[str, str] = {}
140 for match in _HEADER_PAIRS_PATTERN.finditer(header):
141 key = match.group(1)
142 quoted_val, unquoted_val = match.group(2), match.group(3)
143 if quoted_val is None and unquoted_val is None:
144 # Bare token with no "=value": an auth-scheme name, not a parameter.
145 # Skip a leading scheme; once parameters exist, a new scheme marks
146 # the start of the next challenge, so stop here.
147 if pairs:
148 break
149 continue
150 pairs[key] = (
151 unescape_quotes(quoted_val) if quoted_val is not None else unquoted_val
152 )
153 return pairs
156class DigestAuthMiddleware:
157 """
158 HTTP digest authentication middleware for aiohttp client.
160 This middleware intercepts 401 Unauthorized responses containing a Digest
161 authentication challenge, calculates the appropriate digest credentials,
162 and automatically retries the request with the proper Authorization header.
164 Features:
165 - Handles all aspects of Digest authentication handshake automatically
166 - Supports all standard hash algorithms:
167 - MD5, MD5-SESS
168 - SHA, SHA-SESS
169 - SHA256, SHA256-SESS, SHA-256, SHA-256-SESS
170 - SHA512, SHA512-SESS, SHA-512, SHA-512-SESS
171 - Supports 'auth' and 'auth-int' quality of protection modes
172 - Properly handles quoted strings and parameter parsing
173 - Includes replay attack protection with client nonce count tracking
174 - Supports preemptive authentication per RFC 7616 Section 3.6
176 Origin scoping:
177 The credentials are scoped to the origin of the first request the
178 middleware handles. A request to a different origin is passed through
179 untouched, so it never receives a digest response computed from those
180 credentials, unless that origin falls within a protection space the
181 anchor origin advertised through the RFC 7616 ``domain`` directive. Make
182 the first request through the middleware against the intended origin, as
183 the anchor is pinned to it and not reset for the life of the instance.
185 Standards compliance:
186 - RFC 7616: HTTP Digest Access Authentication (primary reference)
187 - RFC 2617: HTTP Authentication (deprecated by RFC 7616)
188 - RFC 1945: Section 11.1 (username restrictions)
190 Implementation notes:
191 The core digest calculation is inspired by the implementation in
192 https://github.com/requests/requests/blob/v2.18.4/requests/auth.py
193 with added support for modern digest auth features and error handling.
194 """
196 def __init__(
197 self,
198 login: str,
199 password: str,
200 preemptive: bool = True,
201 ) -> None:
202 if login is None:
203 raise ValueError("None is not allowed as login value")
205 if password is None:
206 raise ValueError("None is not allowed as password value")
208 if ":" in login:
209 raise ValueError('A ":" is not allowed in username (RFC 1945#section-11.1)')
211 self._login_str: Final[str] = login
212 self._login_bytes: Final[bytes] = login.encode("utf-8")
213 self._password_bytes: Final[bytes] = password.encode("utf-8")
215 self._last_nonce_bytes = b""
216 self._nonce_count = 0
217 self._challenge: DigestAuthChallenge = {}
218 self._preemptive: bool = preemptive
219 # Set of URLs defining the protection space
220 self._protection_space: list[str] = []
221 # Origin the credentials are scoped to; set on the first request.
222 self._origin: URL | None = None
224 async def _encode(self, method: str, url: URL, body: Payload | Literal[b""]) -> str:
225 """
226 Build digest authorization header for the current challenge.
228 Args:
229 method: The HTTP method (GET, POST, etc.)
230 url: The request URL
231 body: The request body (used for qop=auth-int)
233 Returns:
234 A fully formatted Digest authorization header string
236 Raises:
237 ClientError: If the challenge is missing required parameters or
238 contains unsupported values
240 """
241 challenge = self._challenge
242 if "realm" not in challenge:
243 raise ClientError(
244 "Malformed Digest auth challenge: Missing 'realm' parameter"
245 )
247 if "nonce" not in challenge:
248 raise ClientError(
249 "Malformed Digest auth challenge: Missing 'nonce' parameter"
250 )
252 # Empty realm values are allowed per RFC 7616 (SHOULD, not MUST, contain host name)
253 realm = challenge["realm"]
254 nonce = challenge["nonce"]
256 # Empty nonce values are not allowed as they are security-critical for replay protection
257 if not nonce:
258 raise ClientError(
259 "Security issue: Digest auth challenge contains empty 'nonce' value"
260 )
262 qop_raw = challenge.get("qop", "")
263 # Preserve original algorithm case for response while using uppercase for processing
264 algorithm_original = challenge.get("algorithm", "MD5")
265 algorithm = algorithm_original.upper()
266 opaque = challenge.get("opaque", "")
268 # Convert string values to bytes once
269 nonce_bytes = nonce.encode("utf-8")
270 realm_bytes = realm.encode("utf-8")
271 # Use the encoded request-target (raw_path_qs) since that is what is
272 # transmitted on the wire and what the server signs against. Using the
273 # decoded form would cause digest verification to fail when the path
274 # or query string contains percent-encoded reserved characters.
275 path = URL(url).raw_path_qs
277 # Process QoP
278 qop = ""
279 qop_bytes = b""
280 if qop_raw:
281 valid_qops = {"auth", "auth-int"}.intersection(
282 {q.strip() for q in qop_raw.split(",") if q.strip()}
283 )
284 if not valid_qops:
285 raise ClientError(
286 f"Digest auth error: Unsupported Quality of Protection (qop) value(s): {qop_raw}"
287 )
289 qop = "auth-int" if "auth-int" in valid_qops else "auth"
290 qop_bytes = qop.encode("utf-8")
292 if algorithm not in DigestFunctions:
293 raise ClientError(
294 f"Digest auth error: Unsupported hash algorithm: {algorithm}. "
295 f"Supported algorithms: {', '.join(SUPPORTED_ALGORITHMS)}"
296 )
297 hash_fn: Final = DigestFunctions[algorithm]
299 def H(x: bytes) -> bytes:
300 """RFC 7616 Section 3: Hash function H(data) = hex(hash(data))."""
301 return hash_fn(x).hexdigest().encode()
303 def KD(s: bytes, d: bytes) -> bytes:
304 """RFC 7616 Section 3: KD(secret, data) = H(concat(secret, ":", data))."""
305 return H(b":".join((s, d)))
307 # Calculate A1 and A2
308 A1 = b":".join((self._login_bytes, realm_bytes, self._password_bytes))
309 A2 = f"{method.upper()}:{path}".encode()
310 if qop == "auth-int":
311 if isinstance(body, Payload): # will always be empty bytes unless Payload
312 entity_bytes = await body.as_bytes() # Get bytes from Payload
313 else:
314 entity_bytes = body
315 entity_hash = H(entity_bytes)
316 A2 = b":".join((A2, entity_hash))
318 HA1 = H(A1)
319 HA2 = H(A2)
321 # Nonce count handling
322 if nonce_bytes == self._last_nonce_bytes:
323 self._nonce_count += 1
324 else:
325 self._nonce_count = 1
327 self._last_nonce_bytes = nonce_bytes
328 ncvalue = f"{self._nonce_count:08x}"
329 ncvalue_bytes = ncvalue.encode("utf-8")
331 # Generate client nonce
332 cnonce = hashlib.sha1(
333 b"".join(
334 [
335 str(self._nonce_count).encode("utf-8"),
336 nonce_bytes,
337 time.ctime().encode("utf-8"),
338 os.urandom(8),
339 ]
340 )
341 ).hexdigest()[:16]
342 cnonce_bytes = cnonce.encode("utf-8")
344 # Special handling for session-based algorithms
345 if algorithm.upper().endswith("-SESS"):
346 HA1 = H(b":".join((HA1, nonce_bytes, cnonce_bytes)))
348 # Calculate the response digest
349 if qop:
350 noncebit = b":".join(
351 (nonce_bytes, ncvalue_bytes, cnonce_bytes, qop_bytes, HA2)
352 )
353 response_digest = KD(HA1, noncebit)
354 else:
355 response_digest = KD(HA1, b":".join((nonce_bytes, HA2)))
357 # Define a dict mapping of header fields to their values
358 # Group fields into always-present, optional, and qop-dependent
359 header_fields = {
360 # Always present fields
361 "username": escape_quotes(self._login_str),
362 "realm": escape_quotes(realm),
363 "nonce": escape_quotes(nonce),
364 "uri": path,
365 "response": response_digest.decode(),
366 "algorithm": algorithm_original,
367 }
369 # Optional fields
370 if opaque:
371 header_fields["opaque"] = escape_quotes(opaque)
373 # QoP-dependent fields
374 if qop:
375 header_fields["qop"] = qop
376 header_fields["nc"] = ncvalue
377 header_fields["cnonce"] = cnonce
379 # Build header using templates for each field type
380 pairs: list[str] = []
381 for field, value in header_fields.items():
382 if field in QUOTED_AUTH_FIELDS:
383 pairs.append(f'{field}="{value}"')
384 else:
385 pairs.append(f"{field}={value}")
387 return f"Digest {', '.join(pairs)}"
389 def _in_protection_space(self, url: URL) -> bool:
390 """
391 Check if the given URL is within the current protection space.
393 According to RFC 7616, a URI is in the protection space if any URI
394 in the protection space is a prefix of it (after both have been made absolute).
395 """
396 request_str = str(url)
397 for space_str in self._protection_space:
398 # Check if request starts with space URL
399 if not request_str.startswith(space_str):
400 continue
401 # Exact match or space ends with / (proper directory prefix)
402 if len(request_str) == len(space_str) or space_str[-1] == "/":
403 return True
404 # Check next char is / to ensure proper path boundary
405 if request_str[len(space_str)] == "/":
406 return True
407 return False
409 def _authenticate(self, response: ClientResponse) -> bool:
410 """
411 Takes the given response and tries digest-auth, if needed.
413 Returns true if the original request must be resent.
414 """
415 if response.status != 401:
416 return False
418 auth_header = response.headers.get("www-authenticate", "")
419 if not auth_header:
420 return False # No authentication header present
422 method, sep, headers = auth_header.partition(" ")
423 if not sep:
424 # No space found in www-authenticate header
425 return False # Malformed auth header, missing scheme separator
427 if method.lower() != "digest":
428 # Not a digest auth challenge (could be Basic, Bearer, etc.)
429 return False
431 if not headers:
432 # We have a digest scheme but no parameters
433 return False # Malformed digest header, missing parameters
435 # We have a digest auth header with content
436 if not (header_pairs := parse_header_pairs(headers)):
437 # Failed to parse any key-value pairs
438 return False # Malformed digest header, no valid parameters
440 # Extract challenge parameters
441 self._challenge = {}
442 for field in CHALLENGE_FIELDS:
443 if (value := header_pairs.get(field)) is not None:
444 self._challenge[field] = value
446 # Update protection space based on domain parameter or default to origin
447 origin = response.url.origin()
448 self._protection_space = []
450 if domain := self._challenge.get("domain"):
451 # Parse space-separated list of URIs
452 for uri in domain.split():
453 # Remove quotes if present
454 uri = uri.strip('"')
455 if not uri:
456 continue
457 if uri.startswith("/"):
458 # Path-absolute, relative to origin
459 self._protection_space.append(str(origin.join(URL(uri))))
460 else:
461 # Absolute URI
462 self._protection_space.append(str(URL(uri)))
464 if not self._protection_space:
465 self._protection_space = [str(origin)]
467 # Return True only if we found at least one challenge parameter
468 return bool(self._challenge)
470 async def __call__(
471 self, request: ClientRequest, handler: ClientHandlerType
472 ) -> ClientResponse:
473 """Run the digest auth middleware."""
474 # Credentials are scoped to the first request's origin. Other origins
475 # pass through untouched unless a challenge from the anchor origin
476 # advertised them via RFC 7616 domain; mirrors aiohttp stripping
477 # Authorization on cross-origin redirects.
478 origin = request.url.origin()
479 if self._origin is None:
480 self._origin = origin
481 elif origin != self._origin and not self._in_protection_space(request.url):
482 return await handler(request)
484 response = None
485 for retry_count in range(2):
486 # Apply authorization header if:
487 # 1. This is a retry after 401 (retry_count > 0), OR
488 # 2. Preemptive auth is enabled AND we have a challenge AND the URL is in protection space
489 if retry_count > 0 or (
490 self._preemptive
491 and self._challenge
492 and self._in_protection_space(request.url)
493 ):
494 request.headers[hdrs.AUTHORIZATION] = await self._encode(
495 request.method, request.url, request.body
496 )
498 # Send the request
499 response = await handler(request)
501 # Check if we need to authenticate
502 if not self._authenticate(response):
503 break
505 # At this point, response is guaranteed to be defined
506 assert response is not None
507 return response