Coverage for /pythoncovmergedfiles/medio/medio/usr/local/lib/python3.11/site-packages/aiohttp/cookiejar.py: 33%

1import asyncio 

2import calendar 

3import contextlib 

4import datetime 

5import heapq 

6import itertools 

7import json 

8import math 

9import os 

10import pathlib 

11import pickle 

12import re 

13import time 

14import warnings 

15from collections import defaultdict 

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

17from http.cookies import BaseCookie, Morsel, SimpleCookie 

18from types import MappingProxyType 

19from typing import Union, cast 

20 

21from yarl import URL 

22 

23from ._cookie_helpers import ( 

24 _COOKIE_CTL_RE, 

25 _encoded_length, 

26 parse_set_cookie_headers, 

27 preserve_morsel_with_coded_value, 

28) 

29from .abc import AbstractCookieJar, ClearCookiePredicate 

30from .helpers import is_ip_address 

31from .log import internal_logger 

32from .typedefs import LooseCookies, PathLike, StrOrURL 

33 

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

35 

36 

37CookieItem = Union[str, "Morsel[str]"] 

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

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

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

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

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

43 

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

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

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

47 

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

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

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

51_MIN_SCHEDULED_COOKIE_EXPIRATION = 100 

52_SIMPLE_COOKIE = SimpleCookie() 

53 

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

55# Eviction follows Firefox's CookieStorage. 

56_MAX_COOKIES_PER_DOMAIN = 180 

57_COOKIE_QUOTA_PER_DOMAIN = 150 

58_MAX_COOKIES_TOTAL = 3300 

59_COOKIE_QUOTA_TOTAL = 3000 

60_MAX_COOKIES_PER_REQUEST = 150 

61_MAX_COOKIE_HEADER_LENGTH = 8190 

62_COOKIE_HEADER_PREFIX_LENGTH = len(b"Cookie: ") 

63 

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

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

66 

67 

68class _RestrictedCookieUnpickler(pickle._Unpickler): 

69 """A restricted unpickler that only allows cookie-related types. 

70 

71 This prevents arbitrary code execution when loading pickled cookie data 

72 from untrusted sources. Only types that are expected in a serialized 

73 CookieJar are permitted. 

74 

75 Subclasses :class:`pickle._Unpickler` (the pure-Python implementation) 

76 rather than :class:`pickle.Unpickler` because the accelerated unpickler 

77 on some implementations (notably PyPy) does not dispatch through 

78 :meth:`find_class` overrides. 

79 

80 See: https://docs.python.org/3/library/pickle.html#restricting-globals 

81 """ 

82 

83 _ALLOWED_CLASSES: frozenset[tuple[str, str]] = frozenset( 

84 { 

85 # Core cookie types 

86 ("http.cookies", "SimpleCookie"), 

87 ("http.cookies", "Morsel"), 

88 # Container types used by CookieJar._cookies 

89 ("collections", "defaultdict"), 

90 # builtins that pickle uses for reconstruction 

91 ("builtins", "tuple"), 

92 ("builtins", "set"), 

93 ("builtins", "frozenset"), 

94 ("builtins", "dict"), 

95 } 

96 ) 

97 

98 def find_class(self, module: str, name: str) -> type: 

99 if (module, name) not in self._ALLOWED_CLASSES: 

100 raise pickle.UnpicklingError( 

101 f"Forbidden class: {module}.{name}. " 

102 "CookieJar.load() only allows cookie-related types for security. " 

103 "See https://docs.python.org/3/library/pickle.html#restricting-globals" 

104 ) 

105 return super().find_class(module, name) # type: ignore[no-any-return] 

106 

107 

108class CookieJar(AbstractCookieJar): 

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

110 

111 _limits_enabled = True 

112 

113 DATE_TOKENS_RE = re.compile( 

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

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

116 ) 

117 

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

119 

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

121 

122 DATE_MONTH_RE = re.compile( 

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

124 re.I, 

125 ) 

126 

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

128 

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

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

131 MAX_TIME = ( 

132 int(datetime.datetime.max.replace(tzinfo=datetime.timezone.utc).timestamp()) - 1 

133 ) 

134 try: 

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

136 except OSError: 

137 # Hit the maximum representable time on Windows 

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

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

140 except OverflowError: 

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

142 MAX_TIME = 2**31 - 1 

143 # Avoid minuses in the future, 3x faster 

144 SUB_MAX_TIME = MAX_TIME - 1 

145 

146 def __init__( 

147 self, 

148 *, 

149 unsafe: bool = False, 

150 quote_cookie: bool = True, 

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

152 loop: asyncio.AbstractEventLoop | None = None, 

153 ) -> None: 

154 super().__init__(loop=loop) 

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

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

157 dict 

158 ) 

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

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

161 self._unsafe = unsafe 

162 self._quote_cookie = quote_cookie 

163 if treat_as_secure_origin is None: 

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

165 elif isinstance(treat_as_secure_origin, URL): 

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

167 elif isinstance(treat_as_secure_origin, str): 

168 self._treat_as_secure_origin = frozenset( 

169 {URL(treat_as_secure_origin).origin()} 

170 ) 

171 else: 

172 self._treat_as_secure_origin = frozenset( 

173 { 

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

175 for url in treat_as_secure_origin 

176 } 

177 ) 

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

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

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

181 self._next_access_generation = 0 

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

183 self._cookie_count = 0 

184 

185 @property 

186 def unsafe(self) -> bool: 

187 return self._unsafe 

188 

189 @property 

190 def quote_cookie(self) -> bool: 

191 return self._quote_cookie 

192 

193 @property 

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

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

196 return self._treat_as_secure_origin 

197 

198 @property 

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

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

201 return MappingProxyType(self._cookies) 

202 

203 @property 

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

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

206 return frozenset(self._host_only_cookies) 

207 

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

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

210 

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

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

213 """ 

214 file_path = pathlib.Path(file_path) 

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

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

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

218 data[key] = {} 

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

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

221 "key": morsel.key, 

222 "value": morsel.value, 

223 "coded_value": morsel.coded_value, 

224 } 

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

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

227 if attr in _RELATIVE_EXPIRY_ATTRS: 

228 continue 

229 attr_val = morsel[attr] 

230 if attr_val: 

231 morsel_data[attr] = attr_val 

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

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

234 morsel_data["host_only"] = True 

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

236 morsel_data["expires_timestamp"] = exp 

237 data[key][name] = morsel_data 

238 

239 # Cookie persistence may include authentication/session tokens. 

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

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

242 with open( 

243 file_path, 

244 mode="w", 

245 encoding="utf-8", 

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

247 ) as f: 

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

249 

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

251 """Load cookies from a file. 

252 

253 Tries to load JSON format first. Falls back to loading legacy 

254 pickle format (using a restricted unpickler) for backward 

255 compatibility with existing cookie files. 

256 

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

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

259 raises and leaves the jar unchanged. 

260 

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

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

263 """ 

264 file_path = pathlib.Path(file_path) 

265 # Try JSON format first; only a file that isn't JSON is a legacy pickle. 

266 try: 

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

268 data = json.load(f) 

269 except (json.JSONDecodeError, UnicodeDecodeError, ValueError): 

270 pass 

271 else: 

272 self._load_json_data(data) 

273 return 

274 # Fall back to legacy pickle format with restricted unpickler 

275 with file_path.open(mode="rb") as f: 

276 pickled = _RestrictedCookieUnpickler(f).load() 

277 loaded = self._new_loading_jar() 

278 for (domain, _), cookies in pickled.items(): 

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

280 for name, morsel in cookies.items(): 

281 loaded._update_cookies({name: morsel}, response_url, copy_morsels=False) 

282 self._replace_contents(loaded) 

283 

284 def _new_loading_jar(self) -> "CookieJar": 

285 return CookieJar( 

286 unsafe=self._unsafe, 

287 quote_cookie=self._quote_cookie, 

288 treat_as_secure_origin=self._treat_as_secure_origin, 

289 loop=self._loop, 

290 ) 

291 

292 def _replace_contents(self, loaded: "CookieJar") -> None: 

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

294 cookies = self._cookies 

295 cookies.clear() 

296 cookies.update(loaded._cookies) 

297 self.__dict__.update(loaded.__dict__) 

298 self._cookies = cookies 

299 

300 def _load_json_data( 

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

302 ) -> None: 

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

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

305 # so a bad file leaves this one unchanged. 

306 loaded = self._new_loading_jar() 

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

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

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

310 expiration: float | None = None 

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

312 try: 

313 expiration = float(raw_expiration) 

314 if not math.isfinite(expiration): 

315 raise ValueError(expiration) 

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

317 raise ValueError( 

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

319 "expiration timestamp must be a finite number" 

320 ) from exc 

321 morsel: Morsel[str] = Morsel() 

322 # Use __setstate__ to bypass validation, same pattern 

323 # used in _build_morsel and _cookie_helpers. 

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

325 { 

326 "key": morsel_data["key"], 

327 "value": morsel_data["value"], 

328 "coded_value": morsel_data["coded_value"], 

329 } 

330 ) 

331 # Restore morsel attributes 

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

333 if attr in morsel_data and attr not in ( 

334 "key", 

335 "value", 

336 "coded_value", 

337 ): 

338 morsel[attr] = morsel_data[attr] 

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

340 if morsel_data.get("host_only"): 

341 morsel["domain"] = "" 

342 response_url = ( 

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

344 ) 

345 loaded._update_cookies( 

346 {name: morsel}, 

347 response_url, 

348 copy_morsels=False, 

349 expiration=expiration, 

350 ) 

351 self._replace_contents(loaded) 

352 

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

354 if predicate is None: 

355 self._expire_heap.clear() 

356 self._cookies.clear() 

357 self._morsel_cache.clear() 

358 self._host_only_cookies.clear() 

359 self._expirations.clear() 

360 self._access_generations.clear() 

361 self._domain_counts.clear() 

362 self._cookie_count = 0 

363 return 

364 

365 now = time.time() 

366 to_del = [ 

367 key 

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

369 for name, morsel in cookie.items() 

370 if ( 

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

372 and self._expirations[key] <= now 

373 ) 

374 or predicate(morsel) 

375 ] 

376 if to_del: 

377 self._delete_cookies(to_del) 

378 

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

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

381 

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

383 self._do_expiration() 

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

385 yield from val.values() 

386 

387 def __len__(self) -> int: 

388 """Return number of cookies. 

389 

390 This function does not iterate self to avoid unnecessary expiration 

391 checks. 

392 """ 

393 return self._cookie_count 

394 

395 def _do_expiration(self) -> None: 

396 """Remove expired cookies.""" 

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

398 return 

399 

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

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

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

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

404 # only a few scheduled expirations. 

405 if ( 

406 expire_heap_len > _MIN_SCHEDULED_COOKIE_EXPIRATION 

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

408 ): 

409 # Remove any expired entries from the expiration heap 

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

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

412 # with a different expiration time. 

413 self._expire_heap = [ 

414 entry 

415 for entry in self._expire_heap 

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

417 ] 

418 heapq.heapify(self._expire_heap) 

419 

420 now = time.time() 

421 to_del: list[_CookieKey] = [] 

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

423 while self._expire_heap: 

424 when, cookie_key = self._expire_heap[0] 

425 if when > now: 

426 break 

427 heapq.heappop(self._expire_heap) 

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

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

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

431 # expiration time is met. 

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

433 to_del.append(cookie_key) 

434 

435 if to_del: 

436 self._delete_cookies(to_del) 

437 

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

439 for domain, path, name in to_del: 

440 cookie_key = (domain, path, name) 

441 jar_key = (domain, path) 

442 self._host_only_cookies.discard(cookie_key) 

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

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

445 self._cookie_count -= 1 

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

447 self._domain_counts[domain] = domain_count 

448 else: 

449 del self._domain_counts[domain] 

450 if not cookies: 

451 del self._cookies[jar_key] 

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

453 morsels.pop(name, None) 

454 if not morsels: 

455 del self._morsel_cache[jar_key] 

456 self._expirations.pop(cookie_key, None) 

457 self._access_generations.pop(cookie_key, None) 

458 

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

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

461 

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

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

464 evicted for another secure cookie. 

465 """ 

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

467 self._do_expiration() 

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

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

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

471 return True 

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

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

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

475 if cookie_domain != domain: 

476 continue 

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

478 cookie_key = (domain, path, name) 

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

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

481 if not (candidates := non_secure): 

482 if count < _MAX_COOKIES_PER_DOMAIN: # expiry freed a slot 

483 return True 

484 if not secure: 

485 return False 

486 candidates = secure_keys 

487 candidates.sort() 

488 evicted = candidates[:excess] 

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

490 internal_logger.debug( 

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

492 len(evicted), 

493 domain, 

494 _MAX_COOKIES_PER_DOMAIN, 

495 ) 

496 else: 

497 self._purge_cookies() 

498 return True 

499 

500 def _purge_cookies(self) -> None: 

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

502 self._do_expiration() 

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

504 return 

505 generations = self._access_generations 

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

507 internal_logger.debug( 

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

509 excess, 

510 _MAX_COOKIES_TOTAL, 

511 ) 

512 

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

514 cookie_key = (domain, path, name) 

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

516 # Avoid adding duplicates to the heap 

517 return 

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

519 self._expirations[cookie_key] = when 

520 

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

522 """Update cookies.""" 

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

524 

525 def update_cookies_from_headers( 

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

527 ) -> None: 

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

529 hostname = response_url.raw_host 

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

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

532 return 

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

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

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

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

537 self._update_cookies( 

538 cookies_to_update, 

539 response_url, 

540 copy_morsels=False, 

541 check_sendable=False, 

542 ) 

543 

544 def _update_cookies( 

545 self, 

546 cookies: LooseCookies, 

547 response_url: URL, 

548 *, 

549 copy_morsels: bool, 

550 expiration: float | None = None, 

551 check_sendable: bool = True, 

552 ) -> None: 

553 hostname = response_url.raw_host 

554 

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

556 # Don't accept cookies from IPs 

557 return 

558 

559 if isinstance(cookies, Mapping): 

560 cookies = cookies.items() 

561 

562 for name, cookie in cookies: 

563 if not isinstance(cookie, Morsel): 

564 tmp = SimpleCookie() 

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

566 cookie = tmp[name] 

567 elif copy_morsels: 

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

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

570 

571 domain = cookie["domain"] 

572 

573 # ignore domains with trailing dots 

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

575 domain = "" 

576 del cookie["domain"] 

577 

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

579 # Remove leading dot 

580 domain = domain[1:] 

581 cookie["domain"] = domain 

582 

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

584 # Setting cookies for different domains is not allowed 

585 continue 

586 

587 path = cookie["path"] 

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

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

590 path = response_url.path 

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

592 path = "/" 

593 else: 

594 # Cut everything from the last slash to the end 

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

596 cookie["path"] = path 

597 path = path.rstrip("/") 

598 

599 if not domain and hostname is not None: 

600 host_only = True 

601 domain = hostname 

602 else: 

603 host_only = False 

604 

605 deadline: float | None = None 

606 if expiration is not None: 

607 # Saved jars pass an absolute deadline. 

608 deadline = expiration 

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

610 try: 

611 delta_seconds = int(max_age) 

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

613 if delta_seconds <= 0: 

614 deadline = 0.0 

615 else: 

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

617 delta_seconds = min(delta_seconds, self.MAX_TIME) 

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

619 except ValueError: 

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

621 

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

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

624 deadline = expire_time 

625 else: 

626 cookie["expires"] = "" 

627 

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

629 internal_logger.debug( 

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

631 name, 

632 domain, 

633 ) 

634 continue 

635 

636 key = (domain, path) 

637 cookie_key = (domain, path, name) 

638 bucket = self._cookies.get(key) 

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

640 if ( 

641 stored is None 

642 and self._limits_enabled 

643 and ( 

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

645 or self._cookie_count >= _MAX_COOKIES_TOTAL 

646 ) 

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

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

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

650 ): 

651 internal_logger.debug( 

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

653 name, 

654 domain, 

655 ) 

656 continue 

657 

658 if host_only: 

659 self._host_only_cookies.add(cookie_key) 

660 cookie["domain"] = domain 

661 else: 

662 # A cookie with an explicit Domain attribute replaces any 

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

664 self._host_only_cookies.discard(cookie_key) 

665 if deadline is not None: 

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

667 elif self._expirations: 

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

669 self._expirations.pop(cookie_key, None) 

670 

671 if stored != cookie: 

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

673 # cookie gets set again 

674 if stored is None: 

675 self._cookie_count += 1 

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

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

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

679 morsels.pop(name, None) 

680 if self._limits_enabled: 

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

682 self._next_access_generation += 1 

683 self._access_generations[cookie_key] = self._next_access_generation 

684 

685 self._do_expiration() 

686 

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

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

689 # We always use BaseCookie now since all 

690 # cookies set on on filtered are fully constructed 

691 # Morsels, not just names and values. 

692 filtered: BaseCookie[str] = BaseCookie() 

693 if not self._cookies: 

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

695 return filtered 

696 self._do_expiration() 

697 if not self._cookies: 

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

699 return filtered 

700 if type(request_url) is not URL: 

701 warnings.warn( 

702 "filter_cookies expects yarl.URL instances only," 

703 f"and will stop working in 4.x, got {type(request_url)}", 

704 DeprecationWarning, 

705 stacklevel=2, 

706 ) 

707 request_url = URL(request_url) 

708 hostname = request_url.raw_host or "" 

709 

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

711 if is_not_secure and self._treat_as_secure_origin: 

712 request_origin = URL() 

713 with contextlib.suppress(ValueError): 

714 request_origin = request_url.origin() 

715 is_not_secure = request_origin not in self._treat_as_secure_origin 

716 

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

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

719 

720 # Send shared cookie 

721 key = ("", "") 

722 if key in self._cookies: 

723 morsels = self._morsel_cache[key] 

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

725 if is_not_secure and cookie["secure"]: 

726 continue 

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

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

729 selected[name] = entry 

730 

731 if not is_ip_address(hostname): 

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

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

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

735 ) 

736 elif self._unsafe: 

737 domains = (hostname,) 

738 else: 

739 domains = () 

740 

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

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

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

744 pairs = itertools.product(domains, paths) 

745 

746 path_len = len(request_url.path) 

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

748 for p in pairs: 

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

750 continue 

751 morsels = self._morsel_cache[p] 

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

753 domain = cookie["domain"] 

754 

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

756 continue 

757 

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

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

760 continue 

761 

762 if is_not_secure and cookie["secure"]: 

763 continue 

764 

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

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

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

768 # the over-limit fallback. 

769 selected.pop(name, None) 

770 selected[name] = entry 

771 

772 if self._limits_enabled: 

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

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

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

776 if ( 

777 len(selected) > _MAX_COOKIES_PER_REQUEST 

778 or header_length > _MAX_COOKIE_HEADER_LENGTH 

779 ): 

780 return self._filter_cookies_within_limits(selected, hostname) 

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

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

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

784 return filtered 

785 

786 def _filter_cookies_within_limits( 

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

788 ) -> "BaseCookie[str]": 

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

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

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

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