Coverage for /pythoncovmergedfiles/medio/medio/src/aiohttp/aiohttp/client_middleware_digest_auth.py: 19%

1""" 

2Digest authentication middleware for aiohttp client. 

3 

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""" 

9 

10import hashlib 

11import os 

12import re 

13import time 

14from collections.abc import Callable 

15from typing import Final, Literal, TypedDict 

16 

17from yarl import URL 

18 

19from . import hdrs 

20from .client_exceptions import ClientError 

21from .client_middlewares import ClientHandlerType 

22from .client_reqrep import ClientRequest, ClientResponse 

23from .payload import Payload 

24 

25 

26class DigestAuthChallenge(TypedDict, total=False): 

27 realm: str 

28 nonce: str 

29 qop: str 

30 algorithm: str 

31 opaque: str 

32 domain: str 

33 stale: str 

34 

35 

36DigestFunctions: dict[str, Callable[[bytes], "hashlib._Hash"]] = { 

37 "MD5": hashlib.md5, 

38 "MD5-SESS": hashlib.md5, 

39 "SHA": hashlib.sha1, 

40 "SHA-SESS": hashlib.sha1, 

41 "SHA256": hashlib.sha256, 

42 "SHA256-SESS": hashlib.sha256, 

43 "SHA-256": hashlib.sha256, 

44 "SHA-256-SESS": hashlib.sha256, 

45 "SHA512": hashlib.sha512, 

46 "SHA512-SESS": hashlib.sha512, 

47 "SHA-512": hashlib.sha512, 

48 "SHA-512-SESS": hashlib.sha512, 

49} 

50 

51 

52# Compile the regex pattern once at module level for performance 

53_HEADER_PAIRS_PATTERN = re.compile( 

54 r'(?:^|\s|,\s*)((?>\w+))(?:\s*=\s*(?:"((?:[^"\\]|\\.)*)"|([^\s,]+)))?' 

55 # +------------|--------|--|--||--|--|----|------|---|---||-----|-> Match valid start/sep 

56 # +--------|--|--||--|--|----|------|---|---||-----|-> alphanumeric key (atomic group reduces backtracking) 

57 # +--|--||--|--|----|------|---|---||-----|-> optional value; absent => bare auth-scheme token 

58 # +--||--|--|----|------|---|---||-----|-> maybe whitespace 

59 # +|--|--|----|------|---|---||-----|-> = (delimiter) 

60 # +--|--|----|------|---|---||-----|-> maybe whitespace 

61 # +--|----|------|---|---||-----|-> group quoted or unquoted 

62 # +----|------|---|---||-----|-> if quoted... 

63 # +------|---|---||-----|-> anything but " or \ 

64 # +---|---||-----|-> escaped characters allowed 

65 # +---||-----|-> or can be empty string 

66 # +|-----|-> if unquoted... 

67 # +-----|-> anything but , or <space> 

68 # +-> at least one char req'd 

69) 

70 

71 

72# RFC 7616: Challenge parameters to extract 

73CHALLENGE_FIELDS: Final[ 

74 tuple[ 

75 Literal["realm", "nonce", "qop", "algorithm", "opaque", "domain", "stale"], ... 

76 ] 

77] = ( 

78 "realm", 

79 "nonce", 

80 "qop", 

81 "algorithm", 

82 "opaque", 

83 "domain", 

84 "stale", 

85) 

86 

87# Supported digest authentication algorithms 

88# Use a tuple of sorted keys for predictable documentation and error messages 

89SUPPORTED_ALGORITHMS: Final[tuple[str, ...]] = tuple(sorted(DigestFunctions.keys())) 

90 

91# RFC 7616: Fields that require quoting in the Digest auth header 

92# These fields must be enclosed in double quotes in the Authorization header. 

93# Algorithm, qop, and nc are never quoted per RFC specifications. 

94# This frozen set is used by the template-based header construction to 

95# automatically determine which fields need quotes. 

96QUOTED_AUTH_FIELDS: Final[frozenset[str]] = frozenset( 

97 {"username", "realm", "nonce", "uri", "response", "opaque", "cnonce"} 

98) 

99 

100 

101def escape_quotes(value: str) -> str: 

102 """Escape backslashes and double quotes for HTTP quoted-strings.""" 

103 return value.replace("\\", "\\\\").replace('"', '\\"') 

104 

105 

106def unescape_quotes(value: str) -> str: 

107 """Unescape backslashes and double quotes in HTTP quoted-strings.""" 

108 return value.replace('\\"', '"').replace("\\\\", "\\") 

109 

110 

111def parse_header_pairs(header: str) -> dict[str, str]: 

112 """ 

113 Parse key-value pairs from the first challenge of a WWW-Authenticate header. 

114 

115 This function handles the complex format of WWW-Authenticate header values, 

116 supporting both quoted and unquoted values, proper handling of commas in 

117 quoted values, and whitespace variations per RFC 7616. 

118 

119 A single header may carry several challenges 

120 (https://www.rfc-editor.org/rfc/rfc7235#section-4.1). Parsing 

121 stops at the next auth-scheme token so a later challenge's parameters cannot 

122 overwrite the first challenge's values; a leading scheme token is skipped. 

123 

124 Examples of supported formats: 

125 - key1="value1", key2=value2 

126 - key1 = "value1" , key2="value, with, commas" 

127 - key1=value1,key2="value2" 

128 - realm="example.com", nonce="12345", qop="auth" 

129 

130 Args: 

131 header: The header value string to parse 

132 

133 Returns: 

134 Dictionary mapping parameter names to their values 

135 """ 

136 pairs: dict[str, str] = {} 

137 for match in _HEADER_PAIRS_PATTERN.finditer(header): 

138 key = match.group(1) 

139 quoted_val, unquoted_val = match.group(2), match.group(3) 

140 if quoted_val is None and unquoted_val is None: 

141 # Bare token with no "=value": an auth-scheme name, not a parameter. 

142 # Skip a leading scheme; once parameters exist, a new scheme marks 

143 # the start of the next challenge, so stop here. 

144 if pairs: 

145 break 

146 continue 

147 pairs[key] = ( 

148 unescape_quotes(quoted_val) if quoted_val is not None else unquoted_val 

149 ) 

150 return pairs 

151 

152 

153class DigestAuthMiddleware: 

154 """ 

155 HTTP digest authentication middleware for aiohttp client. 

156 

157 This middleware intercepts 401 Unauthorized responses containing a Digest 

158 authentication challenge, calculates the appropriate digest credentials, 

159 and automatically retries the request with the proper Authorization header. 

160 

161 Features: 

162 - Handles all aspects of Digest authentication handshake automatically 

163 - Supports all standard hash algorithms: 

164 - MD5, MD5-SESS 

165 - SHA, SHA-SESS 

166 - SHA256, SHA256-SESS, SHA-256, SHA-256-SESS 

167 - SHA512, SHA512-SESS, SHA-512, SHA-512-SESS 

168 - Supports 'auth' and 'auth-int' quality of protection modes 

169 - Properly handles quoted strings and parameter parsing 

170 - Includes replay attack protection with client nonce count tracking 

171 - Supports preemptive authentication per RFC 7616 Section 3.6 

172 

173 Origin scoping: 

174 The credentials are scoped to the origin of the first request the 

175 middleware handles. A request to a different origin is passed through 

176 untouched, so it never receives a digest response computed from those 

177 credentials, unless that origin falls within a protection space the 

178 anchor origin advertised through the RFC 7616 ``domain`` directive. Make 

179 the first request through the middleware against the intended origin, as 

180 the anchor is pinned to it and not reset for the life of the instance. 

181 

182 Standards compliance: 

183 - RFC 7616: HTTP Digest Access Authentication (primary reference) 

184 - RFC 2617: HTTP Authentication (deprecated by RFC 7616) 

185 - RFC 1945: Section 11.1 (username restrictions) 

186 

187 Implementation notes: 

188 The core digest calculation is inspired by the implementation in 

189 https://github.com/requests/requests/blob/v2.18.4/requests/auth.py 

190 with added support for modern digest auth features and error handling. 

191 """ 

192 

193 def __init__( 

194 self, 

195 login: str, 

196 password: str, 

197 preemptive: bool = True, 

198 ) -> None: 

199 if login is None: 

200 raise ValueError("None is not allowed as login value") 

201 

202 if password is None: 

203 raise ValueError("None is not allowed as password value") 

204 

205 if ":" in login: 

206 raise ValueError('A ":" is not allowed in username (RFC 1945#section-11.1)') 

207 

208 self._login_str: Final[str] = login 

209 self._login_bytes: Final[bytes] = login.encode("utf-8") 

210 self._password_bytes: Final[bytes] = password.encode("utf-8") 

211 

212 self._last_nonce_bytes = b"" 

213 self._nonce_count = 0 

214 self._challenge: DigestAuthChallenge = {} 

215 self._preemptive: bool = preemptive 

216 # Set of URLs defining the protection space 

217 self._protection_space: list[str] = [] 

218 # Origin the credentials are scoped to; set on the first request. 

219 self._origin: URL | None = None 

220 

221 async def _encode(self, method: str, url: URL, body: Payload | Literal[b""]) -> str: 

222 """ 

223 Build digest authorization header for the current challenge. 

224 

225 Args: 

226 method: The HTTP method (GET, POST, etc.) 

227 url: The request URL 

228 body: The request body (used for qop=auth-int) 

229 

230 Returns: 

231 A fully formatted Digest authorization header string 

232 

233 Raises: 

234 ClientError: If the challenge is missing required parameters or 

235 contains unsupported values 

236 

237 """ 

238 challenge = self._challenge 

239 if "realm" not in challenge: 

240 raise ClientError( 

241 "Malformed Digest auth challenge: Missing 'realm' parameter" 

242 ) 

243 

244 if "nonce" not in challenge: 

245 raise ClientError( 

246 "Malformed Digest auth challenge: Missing 'nonce' parameter" 

247 ) 

248 

249 # Empty realm values are allowed per RFC 7616 (SHOULD, not MUST, contain host name) 

250 realm = challenge["realm"] 

251 nonce = challenge["nonce"] 

252 

253 # Empty nonce values are not allowed as they are security-critical for replay protection 

254 if not nonce: 

255 raise ClientError( 

256 "Security issue: Digest auth challenge contains empty 'nonce' value" 

257 ) 

258 

259 qop_raw = challenge.get("qop", "") 

260 # Preserve original algorithm case for response while using uppercase for processing 

261 algorithm_original = challenge.get("algorithm", "MD5") 

262 algorithm = algorithm_original.upper() 

263 opaque = challenge.get("opaque", "") 

264 

265 # Convert string values to bytes once 

266 nonce_bytes = nonce.encode("utf-8") 

267 realm_bytes = realm.encode("utf-8") 

268 # Use the encoded request-target (raw_path_qs) since that is what is 

269 # transmitted on the wire and what the server signs against. Using the 

270 # decoded form would cause digest verification to fail when the path 

271 # or query string contains percent-encoded reserved characters. 

272 path = URL(url).raw_path_qs 

273 

274 # Process QoP 

275 qop = "" 

276 qop_bytes = b"" 

277 if qop_raw: 

278 valid_qops = {"auth", "auth-int"}.intersection( 

279 {q.strip() for q in qop_raw.split(",") if q.strip()} 

280 ) 

281 if not valid_qops: 

282 raise ClientError( 

283 f"Digest auth error: Unsupported Quality of Protection (qop) value(s): {qop_raw}" 

284 ) 

285 

286 qop = "auth-int" if "auth-int" in valid_qops else "auth" 

287 qop_bytes = qop.encode("utf-8") 

288 

289 if algorithm not in DigestFunctions: 

290 raise ClientError( 

291 f"Digest auth error: Unsupported hash algorithm: {algorithm}. " 

292 f"Supported algorithms: {', '.join(SUPPORTED_ALGORITHMS)}" 

293 ) 

294 hash_fn: Final = DigestFunctions[algorithm] 

295 

296 def H(x: bytes) -> bytes: 

297 """RFC 7616 Section 3: Hash function H(data) = hex(hash(data)).""" 

298 return hash_fn(x).hexdigest().encode() 

299 

300 def KD(s: bytes, d: bytes) -> bytes: 

301 """RFC 7616 Section 3: KD(secret, data) = H(concat(secret, ":", data)).""" 

302 return H(b":".join((s, d))) 

303 

304 # Calculate A1 and A2 

305 A1 = b":".join((self._login_bytes, realm_bytes, self._password_bytes)) 

306 A2 = f"{method.upper()}:{path}".encode() 

307 if qop == "auth-int": 

308 if isinstance(body, Payload): # will always be empty bytes unless Payload 

309 entity_bytes = await body.as_bytes() # Get bytes from Payload 

310 else: 

311 entity_bytes = body 

312 entity_hash = H(entity_bytes) 

313 A2 = b":".join((A2, entity_hash)) 

314 

315 HA1 = H(A1) 

316 HA2 = H(A2) 

317 

318 # Nonce count handling 

319 if nonce_bytes == self._last_nonce_bytes: 

320 self._nonce_count += 1 

321 else: 

322 self._nonce_count = 1 

323 

324 self._last_nonce_bytes = nonce_bytes 

325 ncvalue = f"{self._nonce_count:08x}" 

326 ncvalue_bytes = ncvalue.encode("utf-8") 

327 

328 # Generate client nonce 

329 cnonce = hashlib.sha1( 

330 b"".join( 

331 [ 

332 str(self._nonce_count).encode("utf-8"), 

333 nonce_bytes, 

334 time.ctime().encode("utf-8"), 

335 os.urandom(8), 

336 ] 

337 ) 

338 ).hexdigest()[:16] 

339 cnonce_bytes = cnonce.encode("utf-8") 

340 

341 # Special handling for session-based algorithms 

342 if algorithm.upper().endswith("-SESS"): 

343 HA1 = H(b":".join((HA1, nonce_bytes, cnonce_bytes))) 

344 

345 # Calculate the response digest 

346 if qop: 

347 noncebit = b":".join( 

348 (nonce_bytes, ncvalue_bytes, cnonce_bytes, qop_bytes, HA2) 

349 ) 

350 response_digest = KD(HA1, noncebit) 

351 else: 

352 response_digest = KD(HA1, b":".join((nonce_bytes, HA2))) 

353 

354 # Define a dict mapping of header fields to their values 

355 # Group fields into always-present, optional, and qop-dependent 

356 header_fields = { 

357 # Always present fields 

358 "username": escape_quotes(self._login_str), 

359 "realm": escape_quotes(realm), 

360 "nonce": escape_quotes(nonce), 

361 "uri": path, 

362 "response": response_digest.decode(), 

363 "algorithm": algorithm_original, 

364 } 

365 

366 # Optional fields 

367 if opaque: 

368 header_fields["opaque"] = escape_quotes(opaque) 

369 

370 # QoP-dependent fields 

371 if qop: 

372 header_fields["qop"] = qop 

373 header_fields["nc"] = ncvalue 

374 header_fields["cnonce"] = cnonce 

375 

376 # Build header using templates for each field type 

377 pairs: list[str] = [] 

378 for field, value in header_fields.items(): 

379 if field in QUOTED_AUTH_FIELDS: 

380 pairs.append(f'{field}="{value}"') 

381 else: 

382 pairs.append(f"{field}={value}") 

383 

384 return f"Digest {', '.join(pairs)}" 

385 

386 def _in_protection_space(self, url: URL) -> bool: 

387 """ 

388 Check if the given URL is within the current protection space. 

389 

390 According to RFC 7616, a URI is in the protection space if any URI 

391 in the protection space is a prefix of it (after both have been made absolute). 

392 """ 

393 request_str = str(url) 

394 for space_str in self._protection_space: 

395 # Check if request starts with space URL 

396 if not request_str.startswith(space_str): 

397 continue 

398 # Exact match or space ends with / (proper directory prefix) 

399 if len(request_str) == len(space_str) or space_str[-1] == "/": 

400 return True 

401 # Check next char is / to ensure proper path boundary 

402 if request_str[len(space_str)] == "/": 

403 return True 

404 return False 

405 

406 def _authenticate(self, response: ClientResponse) -> bool: 

407 """ 

408 Takes the given response and tries digest-auth, if needed. 

409 

410 Returns true if the original request must be resent. 

411 """ 

412 if response.status != 401: 

413 return False 

414 

415 auth_header = response.headers.get("www-authenticate", "") 

416 if not auth_header: 

417 return False # No authentication header present 

418 

419 method, sep, headers = auth_header.partition(" ") 

420 if not sep: 

421 # No space found in www-authenticate header 

422 return False # Malformed auth header, missing scheme separator 

423 

424 if method.lower() != "digest": 

425 # Not a digest auth challenge (could be Basic, Bearer, etc.) 

426 return False 

427 

428 if not headers: 

429 # We have a digest scheme but no parameters 

430 return False # Malformed digest header, missing parameters 

431 

432 # We have a digest auth header with content 

433 if not (header_pairs := parse_header_pairs(headers)): 

434 # Failed to parse any key-value pairs 

435 return False # Malformed digest header, no valid parameters 

436 

437 # Extract challenge parameters 

438 self._challenge = {} 

439 for field in CHALLENGE_FIELDS: 

440 if (value := header_pairs.get(field)) is not None: 

441 self._challenge[field] = value 

442 

443 # Update protection space based on domain parameter or default to origin 

444 origin = response.url.origin() 

445 self._protection_space = [] 

446 

447 if domain := self._challenge.get("domain"): 

448 # Parse space-separated list of URIs 

449 for uri in domain.split(): 

450 # Remove quotes if present 

451 uri = uri.strip('"') 

452 if not uri: 

453 continue 

454 if uri.startswith("/"): 

455 # Path-absolute, relative to origin 

456 self._protection_space.append(str(origin.join(URL(uri)))) 

457 else: 

458 # Absolute URI 

459 self._protection_space.append(str(URL(uri))) 

460 

461 if not self._protection_space: 

462 self._protection_space = [str(origin)] 

463 

464 # Return True only if we found at least one challenge parameter 

465 return bool(self._challenge) 

466 

467 async def __call__( 

468 self, request: ClientRequest, handler: ClientHandlerType 

469 ) -> ClientResponse: 

470 """Run the digest auth middleware.""" 

471 # Credentials are scoped to the first request's origin. Other origins 

472 # pass through untouched unless a challenge from the anchor origin 

473 # advertised them via RFC 7616 domain; mirrors aiohttp stripping 

474 # Authorization on cross-origin redirects. 

475 origin = request.url.origin() 

476 if self._origin is None: 

477 self._origin = origin 

478 elif origin != self._origin and not self._in_protection_space(request.url): 

479 return await handler(request) 

480 

481 response = None 

482 for retry_count in range(2): 

483 # Apply authorization header if: 

484 # 1. This is a retry after 401 (retry_count > 0), OR 

485 # 2. Preemptive auth is enabled AND we have a challenge AND the URL is in protection space 

486 if retry_count > 0 or ( 

487 self._preemptive 

488 and self._challenge 

489 and self._in_protection_space(request.url) 

490 ): 

491 request.headers[hdrs.AUTHORIZATION] = await self._encode( 

492 request.method, request.url, request.body 

493 ) 

494 

495 # Send the request 

496 response = await handler(request) 

497 

498 # Check if we need to authenticate 

499 if not self._authenticate(response): 

500 break 

501 

502 # At this point, response is guaranteed to be defined 

503 assert response is not None 

504 return response