Coverage for /pythoncovmergedfiles/medio/medio/src/aiohttp/aiohttp/cookiejar.py: 21%

1import calendar 

2import contextlib 

3import datetime 

4import heapq 

5import itertools 

6import json 

7import math 

8import os 

9import pathlib 

10import re 

11import time 

12import warnings 

13from collections import defaultdict 

14from collections.abc import Iterable, Iterator, Mapping, Sequence 

15from http.cookies import BaseCookie, Morsel, SimpleCookie 

16from types import MappingProxyType 

17from typing import cast 

18 

19from yarl import URL 

20 

21from ._cookie_helpers import ( 

22 _COOKIE_CTL_RE, 

23 _encoded_length, 

24 parse_set_cookie_headers, 

25 preserve_morsel_with_coded_value, 

26) 

27from .abc import AbstractCookieJar, ClearCookiePredicate 

28from .helpers import is_ip_address 

29from .log import internal_logger 

30from .typedefs import LooseCookies, PathLike, StrOrURL 

31 

32__all__ = ("CookieJar", "DummyCookieJar") 

33 

34 

35CookieItem = str | Morsel[str] 

36_JarKey = tuple[str, str] # (domain, path) 

37_CookieKey = tuple[str, str, str] # (domain, path, name) 

38# A built morsel, its encoded name=value length and the cookie's key. 

39_MorselEntry = tuple["Morsel[str]", int, _CookieKey] 

40_UNLIMITED_COOKIE_KEY: _CookieKey = ("", "", "") 

41 

42# We cache these string methods here as their use is in performance critical code. 

43_FORMAT_PATH = "{}/{}".format 

44_FORMAT_DOMAIN_REVERSED = "{1}.{0}".format 

45 

46# The minimum number of scheduled cookie expirations before we start cleaning up 

47# the expiration heap. This is a performance optimization to avoid cleaning up the 

48# heap too often when there are only a few scheduled expirations. 

49_MIN_SCHEDULED_COOKIE_EXPIRATION = 100 

50_SIMPLE_COOKIE = SimpleCookie() 

51 

52# Private, non-configurable limits for the built-in jar. 

53# Eviction follows Firefox's CookieStorage. 

54_MAX_COOKIES_PER_DOMAIN = 180 

55_COOKIE_QUOTA_PER_DOMAIN = 150 

56_MAX_COOKIES_TOTAL = 3300 

57_COOKIE_QUOTA_TOTAL = 3000 

58_MAX_COOKIES_PER_REQUEST = 150 

59_MAX_COOKIE_HEADER_LENGTH = 8190 

60_COOKIE_HEADER_PREFIX_LENGTH = len(b"Cookie: ") 

61 

62# Not persisted; the absolute deadline is saved instead. 

63_RELATIVE_EXPIRY_ATTRS = frozenset(("max-age", "expires")) 

64 

65 

66class CookieJar(AbstractCookieJar): 

67 """Implements cookie storage adhering to RFC 6265.""" 

68 

69 _limits_enabled = True 

70 

71 # https://datatracker.ietf.org/doc/html/rfc6265#section-5.1.1 

72 DATE_TOKENS_RE = re.compile( 

73 r"[\x09\x20-\x2F\x3B-\x40\x5B-\x60\x7B-\x7E]*" 

74 r"(?P<token>[\x00-\x08\x0A-\x1F\d:a-zA-Z\x7F-\xFF]+)", 

75 re.ASCII, 

76 ) 

77 

78 DATE_HMS_TIME_RE = re.compile(r"(\d{1,2}):(\d{1,2}):(\d{1,2})", re.ASCII) 

79 

80 DATE_DAY_OF_MONTH_RE = re.compile(r"(\d{1,2})", re.ASCII) 

81 

82 DATE_MONTH_RE = re.compile( 

83 "(jan)|(feb)|(mar)|(apr)|(may)|(jun)|(jul)|(aug)|(sep)|(oct)|(nov)|(dec)", 

84 re.I | re.ASCII, 

85 ) 

86 

87 DATE_YEAR_RE = re.compile(r"(\d{2,4})", re.ASCII) 

88 

89 # calendar.timegm() fails for timestamps after datetime.datetime.max 

90 # Minus one as a loss of precision occurs when timestamp() is called. 

91 MAX_TIME = int(datetime.datetime.max.replace(tzinfo=datetime.UTC).timestamp()) - 1 

92 try: 

93 calendar.timegm(time.gmtime(MAX_TIME)) 

94 except OSError: 

95 # Hit the maximum representable time on Windows 

96 # https://learn.microsoft.com/en-us/cpp/c-runtime-library/reference/localtime-localtime32-localtime64 

97 MAX_TIME = calendar.timegm((3000, 12, 31, 23, 59, 59, -1, -1, -1)) 

98 except OverflowError: 

99 # #4515: datetime.max may not be representable on 32-bit platforms 

100 MAX_TIME = 2**31 - 1 

101 # Avoid minuses in the future, 3x faster 

102 SUB_MAX_TIME = MAX_TIME - 1 

103 

104 def __init__( 

105 self, 

106 *, 

107 unsafe: bool = False, 

108 quote_cookie: bool = True, 

109 treat_as_secure_origin: StrOrURL | Iterable[StrOrURL] | None = None, 

110 ) -> None: 

111 self._cookies: defaultdict[_JarKey, SimpleCookie] = defaultdict(SimpleCookie) 

112 self._morsel_cache: defaultdict[_JarKey, dict[str, _MorselEntry]] = defaultdict( 

113 dict 

114 ) 

115 # Cookie identity is (domain, path, name). 

116 self._host_only_cookies: set[_CookieKey] = set() 

117 self._unsafe = unsafe 

118 self._quote_cookie = quote_cookie 

119 if treat_as_secure_origin is None: 

120 self._treat_as_secure_origin: frozenset[URL] = frozenset() 

121 elif isinstance(treat_as_secure_origin, URL): 

122 self._treat_as_secure_origin = frozenset({treat_as_secure_origin.origin()}) 

123 elif isinstance(treat_as_secure_origin, str): 

124 self._treat_as_secure_origin = frozenset( 

125 {URL(treat_as_secure_origin).origin()} 

126 ) 

127 else: 

128 self._treat_as_secure_origin = frozenset( 

129 { 

130 URL(url).origin() if isinstance(url, str) else url.origin() 

131 for url in treat_as_secure_origin 

132 } 

133 ) 

134 self._expire_heap: list[tuple[float, _CookieKey]] = [] 

135 self._expirations: dict[_CookieKey, float] = {} 

136 self._access_generations: dict[_CookieKey, int] = {} 

137 self._next_access_generation = 0 

138 self._domain_counts: dict[str, int] = {} 

139 self._cookie_count = 0 

140 

141 @property 

142 def unsafe(self) -> bool: 

143 return self._unsafe 

144 

145 @property 

146 def quote_cookie(self) -> bool: 

147 return self._quote_cookie 

148 

149 @property 

150 def treat_as_secure_origin(self) -> frozenset[URL]: 

151 """Return origins considered secure even over cleartext connections.""" 

152 return self._treat_as_secure_origin 

153 

154 @property 

155 def cookies(self) -> MappingProxyType[tuple[str, str], SimpleCookie]: 

156 """Return the cookies stored in this jar.""" 

157 return MappingProxyType(self._cookies) 

158 

159 @property 

160 def host_only_cookies(self) -> frozenset[tuple[str, str, str]]: 

161 """Return the host-only cookies stored in this jar.""" 

162 return frozenset(self._host_only_cookies) 

163 

164 def save(self, file_path: PathLike) -> None: 

165 """Save cookies to a file using JSON format. 

166 

167 :param file_path: Path to file where cookies will be serialized, 

168 :class:`str` or :class:`pathlib.Path` instance. 

169 """ 

170 file_path = pathlib.Path(file_path) 

171 data: dict[str, dict[str, dict[str, str | bool | float]]] = {} 

172 for (domain, path), cookie in self._cookies.items(): 

173 key = f"{domain}|{path}" 

174 data[key] = {} 

175 for name, morsel in cookie.items(): 

176 morsel_data: dict[str, str | bool | float] = { 

177 "key": morsel.key, 

178 "value": morsel.value, 

179 "coded_value": morsel.coded_value, 

180 } 

181 # Skip relative expiry; the absolute deadline is saved below. 

182 for attr in morsel._reserved: # type: ignore[attr-defined] 

183 if attr in _RELATIVE_EXPIRY_ATTRS: 

184 continue 

185 attr_val = morsel[attr] 

186 if attr_val: 

187 morsel_data[attr] = attr_val 

188 # Persist or it reloads as a domain cookie and leaks to subdomains. 

189 if (domain, path, name) in self._host_only_cookies: 

190 morsel_data["host_only"] = True 

191 if (exp := self._expirations.get((domain, path, name))) is not None: 

192 morsel_data["expires_timestamp"] = exp 

193 data[key][name] = morsel_data 

194 

195 # Cookie persistence may include authentication/session tokens. 

196 # Use 0o600 at creation time to avoid umask-dependent overexposure 

197 # and enforce least-privilege access to sensitive credential data. 

198 with open( 

199 file_path, 

200 mode="w", 

201 encoding="utf-8", 

202 opener=lambda path, flags: os.open(path, flags, 0o600), 

203 ) as f: 

204 json.dump(data, f, indent=2) 

205 

206 def load(self, file_path: PathLike) -> None: 

207 """Load cookies from a JSON file. 

208 

209 Replaces the current jar contents; loaded cookies pass through the 

210 same acceptance rules as :meth:`update_cookies`. A malformed file 

211 raises and leaves the jar unchanged. 

212 

213 :param file_path: Path to file from where cookies will be 

214 imported, :class:`str` or :class:`pathlib.Path` instance. 

215 """ 

216 file_path = pathlib.Path(file_path) 

217 with file_path.open(mode="r", encoding="utf-8") as f: 

218 data = json.load(f) 

219 self._load_json_data(data) 

220 

221 def _load_json_data( 

222 self, data: dict[str, dict[str, dict[str, str | bool | float]]] 

223 ) -> None: 

224 """Replace contents, routing cookies through _update_cookies().""" 

225 # Fill a new jar and adopt its state only once every record is in, 

226 # so a bad file leaves this one unchanged. 

227 loaded = CookieJar( 

228 unsafe=self._unsafe, 

229 quote_cookie=self._quote_cookie, 

230 treat_as_secure_origin=self._treat_as_secure_origin, 

231 ) 

232 for compound_key, cookie_data in data.items(): 

233 domain, path = compound_key.split("|", 1) 

234 for name, morsel_data in cookie_data.items(): 

235 expiration: float | None = None 

236 if (raw_expiration := morsel_data.get("expires_timestamp")) is not None: 

237 try: 

238 expiration = float(raw_expiration) 

239 if not math.isfinite(expiration): 

240 raise ValueError(expiration) 

241 except (OverflowError, TypeError, ValueError) as exc: 

242 raise ValueError( 

243 f"Cookie {name!r} in {compound_key!r}: " 

244 "expiration timestamp must be a finite number" 

245 ) from exc 

246 morsel: Morsel[str] = Morsel() 

247 # Use __setstate__ to bypass validation, same pattern 

248 # used in _build_morsel and _cookie_helpers. 

249 morsel.__setstate__( # type: ignore[attr-defined] 

250 { 

251 "key": morsel_data["key"], 

252 "value": morsel_data["value"], 

253 "coded_value": morsel_data["coded_value"], 

254 } 

255 ) 

256 # Restore morsel attributes 

257 for attr in morsel._reserved: # type: ignore[attr-defined] 

258 if attr in morsel_data and attr not in ( 

259 "key", 

260 "value", 

261 "coded_value", 

262 ): 

263 morsel[attr] = morsel_data[attr] 

264 # Drop the domain so update_cookies() re-marks it host-only. 

265 if morsel_data.get("host_only"): 

266 morsel["domain"] = "" 

267 response_url = ( 

268 URL.build(scheme="https", host=domain) if domain else URL() 

269 ) 

270 loaded._update_cookies( 

271 {name: morsel}, 

272 response_url, 

273 copy_morsels=False, 

274 expiration=expiration, 

275 ) 

276 # Refill the existing mapping so views from the cookies property stay live. 

277 cookies = self._cookies 

278 cookies.clear() 

279 cookies.update(loaded._cookies) 

280 self.__dict__.update(loaded.__dict__) 

281 self._cookies = cookies 

282 

283 def clear(self, predicate: ClearCookiePredicate | None = None) -> None: 

284 if predicate is None: 

285 self._expire_heap.clear() 

286 self._cookies.clear() 

287 self._morsel_cache.clear() 

288 self._host_only_cookies.clear() 

289 self._expirations.clear() 

290 self._access_generations.clear() 

291 self._domain_counts.clear() 

292 self._cookie_count = 0 

293 return 

294 

295 now = time.time() 

296 to_del = [ 

297 key 

298 for (domain, path), cookie in self._cookies.items() 

299 for name, morsel in cookie.items() 

300 if ( 

301 (key := (domain, path, name)) in self._expirations 

302 and self._expirations[key] <= now 

303 ) 

304 or predicate(morsel) 

305 ] 

306 if to_del: 

307 self._delete_cookies(to_del) 

308 

309 def clear_domain(self, domain: str) -> None: 

310 self.clear(lambda x: self._is_domain_match(domain, x["domain"])) 

311 

312 def __iter__(self) -> "Iterator[Morsel[str]]": 

313 self._do_expiration() 

314 for val in self._cookies.values(): 

315 yield from val.values() 

316 

317 def __len__(self) -> int: 

318 """Return number of cookies. 

319 

320 This function does not iterate self to avoid unnecessary expiration 

321 checks. 

322 """ 

323 return self._cookie_count 

324 

325 def _do_expiration(self) -> None: 

326 """Remove expired cookies.""" 

327 if not (expire_heap_len := len(self._expire_heap)): 

328 return 

329 

330 # If the expiration heap grows larger than the number expirations 

331 # times two, we clean it up to avoid keeping expired entries in 

332 # the heap and consuming memory. We guard this with a minimum 

333 # threshold to avoid cleaning up the heap too often when there are 

334 # only a few scheduled expirations. 

335 if ( 

336 expire_heap_len > _MIN_SCHEDULED_COOKIE_EXPIRATION 

337 and expire_heap_len > len(self._expirations) * 2 

338 ): 

339 # Remove any expired entries from the expiration heap 

340 # that do not match the expiration time in the expirations 

341 # as it means the cookie has been re-added to the heap 

342 # with a different expiration time. 

343 self._expire_heap = [ 

344 entry 

345 for entry in self._expire_heap 

346 if self._expirations.get(entry[1]) == entry[0] 

347 ] 

348 heapq.heapify(self._expire_heap) 

349 

350 now = time.time() 

351 to_del: list[_CookieKey] = [] 

352 # Find any expired cookies and add them to the to-delete list 

353 while self._expire_heap: 

354 when, cookie_key = self._expire_heap[0] 

355 if when > now: 

356 break 

357 heapq.heappop(self._expire_heap) 

358 # Check if the cookie hasn't been re-added to the heap 

359 # with a different expiration time as it will be removed 

360 # later when it reaches the top of the heap and its 

361 # expiration time is met. 

362 if self._expirations.get(cookie_key) == when: 

363 to_del.append(cookie_key) 

364 

365 if to_del: 

366 self._delete_cookies(to_del) 

367 

368 def _delete_cookies(self, to_del: list[_CookieKey]) -> None: 

369 for domain, path, name in to_del: 

370 cookie_key = (domain, path, name) 

371 jar_key = (domain, path) 

372 self._host_only_cookies.discard(cookie_key) 

373 if (cookies := self._cookies.get(jar_key)) is not None: 

374 if cookies.pop(name, None) is not None: 

375 self._cookie_count -= 1 

376 if domain_count := self._domain_counts[domain] - 1: 

377 self._domain_counts[domain] = domain_count 

378 else: 

379 del self._domain_counts[domain] 

380 if not cookies: 

381 del self._cookies[jar_key] 

382 if (morsels := self._morsel_cache.get(jar_key)) is not None: 

383 morsels.pop(name, None) 

384 if not morsels: 

385 del self._morsel_cache[jar_key] 

386 self._expirations.pop(cookie_key, None) 

387 self._access_generations.pop(cookie_key, None) 

388 

389 def _make_room(self, domain: str, secure: bool) -> bool: 

390 """Evict for a new cookie like Firefox; False drops the new cookie. 

391 

392 Mirrors AddCookie and FindStaleCookies: expired, then non-secure, 

393 cookies go first, least recently used first; secure cookies are only 

394 evicted for another secure cookie. 

395 """ 

396 if self._domain_counts.get(domain, 0) >= _MAX_COOKIES_PER_DOMAIN: 

397 self._do_expiration() 

398 count = self._domain_counts.get(domain, 0) 

399 # Leave room so the domain ends at its quota. 

400 if (excess := count - _COOKIE_QUOTA_PER_DOMAIN + 1) <= 0: 

401 return True 

402 non_secure: list[tuple[int, _CookieKey]] = [] 

403 secure_keys: list[tuple[int, _CookieKey]] = [] 

404 for (cookie_domain, path), cookies in self._cookies.items(): 

405 if cookie_domain != domain: 

406 continue 

407 for name, cookie in cookies.items(): 

408 cookie_key = (domain, path, name) 

409 entry = (self._access_generations.get(cookie_key, 0), cookie_key) 

410 (secure_keys if cookie["secure"] else non_secure).append(entry) 

411 if not (candidates := non_secure): 

412 if count < _MAX_COOKIES_PER_DOMAIN: # expiry freed a slot 

413 return True 

414 if not secure: 

415 return False 

416 candidates = secure_keys 

417 candidates.sort() 

418 evicted = candidates[:excess] 

419 self._delete_cookies([key for _, key in evicted]) 

420 internal_logger.debug( 

421 "Evicted %d cookie(s) for %s over the %d-cookie domain limit", 

422 len(evicted), 

423 domain, 

424 _MAX_COOKIES_PER_DOMAIN, 

425 ) 

426 else: 

427 self._purge_cookies() 

428 return True 

429 

430 def _purge_cookies(self) -> None: 

431 """Evict expired, then least recently used, cookies down to the total.""" 

432 self._do_expiration() 

433 if (excess := self._cookie_count - _COOKIE_QUOTA_TOTAL) <= 0: 

434 return 

435 generations = self._access_generations 

436 self._delete_cookies(sorted(generations, key=generations.__getitem__)[:excess]) 

437 internal_logger.debug( 

438 "Evicted %d cookie(s) over the %d-cookie total limit", 

439 excess, 

440 _MAX_COOKIES_TOTAL, 

441 ) 

442 

443 def _expire_cookie(self, when: float, domain: str, path: str, name: str) -> None: 

444 cookie_key = (domain, path, name) 

445 if self._expirations.get(cookie_key) == when: 

446 # Avoid adding duplicates to the heap 

447 return 

448 heapq.heappush(self._expire_heap, (when, cookie_key)) 

449 self._expirations[cookie_key] = when 

450 

451 def update_cookies(self, cookies: LooseCookies, response_url: URL = URL()) -> None: 

452 """Update cookies.""" 

453 self._update_cookies(cookies, response_url, copy_morsels=True) 

454 

455 def update_cookies_from_headers( 

456 self, headers: Sequence[str], response_url: URL 

457 ) -> None: 

458 """Update cookies from raw Set-Cookie headers.""" 

459 hostname = response_url.raw_host 

460 if not self._unsafe and is_ip_address(hostname): 

461 # Reject IP-origin cookies before parsing the fields. 

462 return 

463 if headers and (cookies_to_update := parse_set_cookie_headers(headers)): 

464 # The freshly parsed Morsels are not shared with the caller, 

465 # so they can be stored and normalized without a defensive copy. 

466 # The parser only produces cookies that can be sent. 

467 self._update_cookies( 

468 cookies_to_update, 

469 response_url, 

470 copy_morsels=False, 

471 check_sendable=False, 

472 ) 

473 

474 def _update_cookies( 

475 self, 

476 cookies: LooseCookies, 

477 response_url: URL, 

478 *, 

479 copy_morsels: bool, 

480 expiration: float | None = None, 

481 check_sendable: bool = True, 

482 ) -> None: 

483 hostname = response_url.raw_host 

484 

485 if not self._unsafe and is_ip_address(hostname): 

486 # Don't accept cookies from IPs 

487 return 

488 

489 if isinstance(cookies, Mapping): 

490 cookies = cookies.items() 

491 

492 for name, cookie in cookies: 

493 if not isinstance(cookie, Morsel): 

494 tmp = SimpleCookie() 

495 tmp[name] = cookie # type: ignore[assignment] 

496 cookie = tmp[name] 

497 elif copy_morsels: 

498 # TODO(https://github.com/python/typeshed/pull/16346): Remove cast 

499 cookie = cast("Morsel[str]", cookie.copy()) 

500 

501 domain = cookie["domain"] 

502 

503 # ignore domains with trailing dots 

504 if domain and domain[-1] == ".": 

505 domain = "" 

506 del cookie["domain"] 

507 

508 if domain and domain[0] == ".": 

509 # Remove leading dot 

510 domain = domain[1:] 

511 cookie["domain"] = domain 

512 

513 if domain and hostname and not self._is_domain_match(domain, hostname): 

514 # Setting cookies for different domains is not allowed 

515 continue 

516 

517 path = cookie["path"] 

518 if not path or path[0] != "/": 

519 # Set the cookie's path to the response path 

520 path = response_url.path 

521 if not path.startswith("/"): 

522 path = "/" 

523 else: 

524 # Cut everything from the last slash to the end 

525 path = "/" + path[1 : path.rfind("/")] 

526 cookie["path"] = path 

527 path = path.rstrip("/") 

528 

529 if not domain and hostname is not None: 

530 host_only = True 

531 domain = hostname 

532 else: 

533 host_only = False 

534 

535 deadline: float | None = None 

536 if expiration is not None: 

537 # Saved jars pass an absolute deadline. 

538 deadline = expiration 

539 elif max_age := cookie["max-age"]: 

540 try: 

541 delta_seconds = int(max_age) 

542 # https://datatracker.ietf.org/doc/html/rfc6265#section-5.2.2 

543 if delta_seconds <= 0: 

544 deadline = 0.0 

545 else: 

546 # Cap first to protect against OverflowError on next line. 

547 delta_seconds = min(delta_seconds, self.MAX_TIME) 

548 deadline = min(time.time() + delta_seconds, self.MAX_TIME) 

549 except ValueError: 

550 cookie["max-age"] = "" 

551 

552 elif expires := cookie["expires"]: 

553 if expire_time := self._parse_date(expires): 

554 deadline = expire_time 

555 else: 

556 cookie["expires"] = "" 

557 

558 if check_sendable and self._limits_enabled and not self._can_send(cookie): 

559 internal_logger.debug( 

560 "Dropped cookie %r for %s: it can't be sent in a Cookie header", 

561 name, 

562 domain, 

563 ) 

564 continue 

565 

566 key = (domain, path) 

567 cookie_key = (domain, path, name) 

568 bucket = self._cookies.get(key) 

569 stored = None if bucket is None else bucket.get(name) 

570 if ( 

571 stored is None 

572 and self._limits_enabled 

573 and ( 

574 self._domain_counts.get(domain, 0) >= _MAX_COOKIES_PER_DOMAIN 

575 or self._cookie_count >= _MAX_COOKIES_TOTAL 

576 ) 

577 # Like Firefox, a cookie that only deletes never evicts others. 

578 and (deadline is None or deadline > time.time()) 

579 and not self._make_room(domain, bool(cookie["secure"])) 

580 ): 

581 internal_logger.debug( 

582 "Dropped cookie %r for %s: its domain is full of Secure cookies", 

583 name, 

584 domain, 

585 ) 

586 continue 

587 

588 if host_only: 

589 self._host_only_cookies.add(cookie_key) 

590 cookie["domain"] = domain 

591 else: 

592 # A cookie with an explicit Domain attribute replaces any 

593 # host-only cookie with the same (domain, path, name) identity. 

594 self._host_only_cookies.discard(cookie_key) 

595 if deadline is not None: 

596 self._expire_cookie(deadline, domain, path, name) 

597 elif self._expirations: 

598 # RFC 6265 5.3 step 11: a replacement doesn't inherit expiry. 

599 self._expirations.pop(cookie_key, None) 

600 

601 if stored != cookie: 

602 # Don't blow away the cache if the same 

603 # cookie gets set again 

604 if stored is None: 

605 self._cookie_count += 1 

606 self._domain_counts[domain] = self._domain_counts.get(domain, 0) + 1 

607 self._cookies[key][name] = cookie 

608 if (morsels := self._morsel_cache.get(key)) is not None: 

609 morsels.pop(name, None) 

610 if self._limits_enabled: 

611 # Like Firefox, each stored cookie gets its own access time. 

612 self._next_access_generation += 1 

613 self._access_generations[cookie_key] = self._next_access_generation 

614 

615 self._do_expiration() 

616 

617 def filter_cookies(self, request_url: URL) -> "BaseCookie[str]": 

618 """Returns this jar's cookies filtered by their attributes.""" 

619 if not isinstance(request_url, URL): 

620 warnings.warn( # type: ignore[unreachable] 

621 f"The method accepts yarl.URL instances only, got {type(request_url)}", 

622 DeprecationWarning, 

623 ) 

624 request_url = URL(request_url) 

625 # We always use BaseCookie now since all 

626 # cookies set on on filtered are fully constructed 

627 # Morsels, not just names and values. 

628 filtered: BaseCookie[str] = BaseCookie() 

629 if not self._cookies: 

630 # Skip do_expiration() if there are no cookies. 

631 return filtered 

632 self._do_expiration() 

633 if not self._cookies: 

634 # Skip rest of function if no non-expired cookies. 

635 return filtered 

636 hostname = request_url.raw_host or "" 

637 

638 is_not_secure = request_url.scheme not in ("https", "wss") 

639 if is_not_secure and self._treat_as_secure_origin: 

640 request_origin = URL() 

641 with contextlib.suppress(ValueError): 

642 request_origin = request_url.origin() 

643 is_not_secure = request_origin not in self._treat_as_secure_origin 

644 

645 # Last (most specific) matching cookie per name wins. 

646 selected: dict[str, _MorselEntry] = {} 

647 

648 # Send shared cookie 

649 key = ("", "") 

650 if key in self._cookies: 

651 morsels = self._morsel_cache[key] 

652 for name, cookie in self._cookies[key].items(): 

653 if is_not_secure and cookie["secure"]: 

654 continue 

655 if (entry := morsels.get(name)) is None: 

656 entry = morsels[name] = self._morsel_entry(key, name, cookie) 

657 selected[name] = entry 

658 

659 if not is_ip_address(hostname): 

660 # Get all the subdomains that might match a cookie (e.g. "foo.bar.com", "bar.com", "com") 

661 domains: Iterable[str] = itertools.accumulate( 

662 reversed(hostname.split(".")), _FORMAT_DOMAIN_REVERSED 

663 ) 

664 elif self._unsafe: 

665 domains = (hostname,) 

666 else: 

667 domains = () 

668 

669 # Get all the path prefixes that might match a cookie (e.g. "", "/foo", "/foo/bar") 

670 paths = itertools.accumulate(request_url.path.split("/"), _FORMAT_PATH) 

671 # Create every combination of (domain, path) pairs. 

672 pairs = itertools.product(domains, paths) 

673 

674 path_len = len(request_url.path) 

675 # Point 2: https://www.rfc-editor.org/rfc/rfc6265.html#section-5.4 

676 for p in pairs: 

677 if p == ("", "") or p not in self._cookies: 

678 continue 

679 morsels = self._morsel_cache[p] 

680 for name, cookie in self._cookies[p].items(): 

681 domain = cookie["domain"] 

682 

683 if domain != hostname and p + (name,) in self._host_only_cookies: 

684 continue 

685 

686 # Skip edge case when the cookie has a trailing slash but request doesn't. 

687 if len(cookie["path"]) > path_len: 

688 continue 

689 

690 if is_not_secure and cookie["secure"]: 

691 continue 

692 

693 if (entry := morsels.get(name)) is None: 

694 entry = morsels[name] = self._morsel_entry(p, name, cookie) 

695 # Re-insert so the order tracks specificity, for the header and 

696 # the over-limit fallback. 

697 selected.pop(name, None) 

698 selected[name] = entry 

699 

700 if self._limits_enabled: 

701 # Each cookie after the first adds a "; " separator. 

702 header_length = _COOKIE_HEADER_PREFIX_LENGTH - 2 + 2 * len(selected) 

703 header_length += sum(entry[1] for entry in selected.values()) 

704 if ( 

705 len(selected) > _MAX_COOKIES_PER_REQUEST 

706 or header_length > _MAX_COOKIE_HEADER_LENGTH 

707 ): 

708 return self._filter_cookies_within_limits(selected, hostname) 

709 self._touch_cookies([entry[2] for entry in selected.values()]) 

710 # Bypass BaseCookie.__setitem__, which stores Morsel values as-is. 

711 dict.update(filtered, {name: entry[0] for name, entry in selected.items()}) 

712 return filtered 

713 

714 def _filter_cookies_within_limits( 

715 self, selected: dict[str, _MorselEntry], hostname: str 

716 ) -> "BaseCookie[str]": 

717 """Admit the most specific cookies first so a host's own cookies win.""" 

718 header_length = _COOKIE_HEADER_PREFIX_LENGTH - 2 # no "; " before the first 

719 admitted: list[tuple[str, _MorselEntry]] = [] 

720 for name, entry in reversed(selected.items()): 

721 if len(admitted) >= _MAX_COOKIES_PER_REQUEST: 

722 break 

723 if header_length + 2 + entry[1] > _MAX_COOKIE_HEADER_LENGTH: 

724 continue 

725 header_length += 2 + entry[1] 

726 admitted.append((name, entry)) 

727 

728 filtered: BaseCookie[str] = BaseCookie() 

729 for name, entry in reversed(admitted): 

730 dict.__setitem__(filtered, name, entry[0]) 

731 self._touch_cookies([entry[2] for _, entry in admitted]) 

732 internal_logger.debug( 

733 "Omitted %d cookie(s) for %s over the %d-cookie" 

734 " or %d-octet Cookie header limit", 

735 len(selected) - len(admitted), 

736 hostname, 

737 _MAX_COOKIES_PER_REQUEST, 

738 _MAX_COOKIE_HEADER_LENGTH, 

739 ) 

740 return filtered 

741 

742 @staticmethod 

743 def _can_send(cookie: Morsel[str]) -> bool: 

744 """Return whether the cookie can be written to a Cookie header.""" 

745 # Whichever of value and coded_value is sent, CPython builds with the 

746 # CVE-2026-3644 patch also reject control characters in either. 

747 text = f"{cookie.key}={cookie.value}{cookie.coded_value}" 

748 return _COOKIE_CTL_RE.search(text) is None and _encoded_length(text) is not None 

749 

750 def _morsel_entry( 

751 self, jar_key: _JarKey, name: str, cookie: Morsel[str] 

752 ) -> _MorselEntry: 

753 """Build the morsel to send with its length in the Cookie header.""" 

754 morsel = self._build_morsel(cookie) 

755 length = _encoded_length(f"{morsel.key}={morsel.coded_value}") 

756 assert length is not None # Checked before storing. 

757 return morsel, length, jar_key + (name,) 

758 

759 def _touch_cookies(self, cookie_keys: Sequence[_CookieKey]) -> None: 

760 """Record one access generation for cookies used together, like Firefox.""" 

761 self._next_access_generation += 1 

762 generation = self._next_access_generation 

763 generations = self._access_generations 

764 for cookie_key in cookie_keys: 

765 generations[cookie_key] = generation 

766 

767 def _build_morsel(self, cookie: Morsel[str]) -> Morsel[str]: 

768 """Build a morsel for sending, respecting quote_cookie setting.""" 

769 if self._quote_cookie and cookie.coded_value and cookie.coded_value[0] == '"': 

770 return preserve_morsel_with_coded_value(cookie) 

771 morsel: Morsel[str] = Morsel() 

772 if self._quote_cookie: 

773 value, coded_value = _SIMPLE_COOKIE.value_encode(cookie.value) 

774 else: 

775 coded_value = value = cookie.value 

776 # We use __setstate__ instead of the public set() API because it allows us to 

777 # bypass validation and set already validated state. This is more stable than 

778 # setting protected attributes directly. 

779 morsel.__setstate__({"key": cookie.key, "value": value, "coded_value": coded_value}) # type: ignore[attr-defined] 

780 return morsel 

781