Coverage for /pythoncovmergedfiles/medio/medio/src/aiohttp/aiohttp/cookiejar.py: 22%
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
1import calendar
2import contextlib
3import datetime
4import heapq
5import itertools
6import json
7import os
8import pathlib
9import re
10import time
11import warnings
12from collections import defaultdict
13from collections.abc import Iterable, Iterator, Mapping, Sequence
14from http.cookies import BaseCookie, Morsel, SimpleCookie
15from types import MappingProxyType
16from typing import Union, cast
18from yarl import URL
20from ._cookie_helpers import parse_set_cookie_headers, preserve_morsel_with_coded_value
21from .abc import AbstractCookieJar, ClearCookiePredicate
22from .helpers import is_ip_address
23from .typedefs import LooseCookies, PathLike, StrOrURL
25__all__ = ("CookieJar", "DummyCookieJar")
28CookieItem = Union[str, "Morsel[str]"]
30# We cache these string methods here as their use is in performance critical code.
31_FORMAT_PATH = "{}/{}".format
32_FORMAT_DOMAIN_REVERSED = "{1}.{0}".format
34# The minimum number of scheduled cookie expirations before we start cleaning up
35# the expiration heap. This is a performance optimization to avoid cleaning up the
36# heap too often when there are only a few scheduled expirations.
37_MIN_SCHEDULED_COOKIE_EXPIRATION = 100
38_SIMPLE_COOKIE = SimpleCookie()
40# Not persisted; the absolute deadline is saved instead.
41_RELATIVE_EXPIRY_ATTRS = frozenset(("max-age", "expires"))
44class CookieJar(AbstractCookieJar):
45 """Implements cookie storage adhering to RFC 6265."""
47 # https://datatracker.ietf.org/doc/html/rfc6265#section-5.1.1
48 DATE_TOKENS_RE = re.compile(
49 r"[\x09\x20-\x2F\x3B-\x40\x5B-\x60\x7B-\x7E]*"
50 r"(?P<token>[\x00-\x08\x0A-\x1F\d:a-zA-Z\x7F-\xFF]+)",
51 re.ASCII,
52 )
54 DATE_HMS_TIME_RE = re.compile(r"(\d{1,2}):(\d{1,2}):(\d{1,2})", re.ASCII)
56 DATE_DAY_OF_MONTH_RE = re.compile(r"(\d{1,2})", re.ASCII)
58 DATE_MONTH_RE = re.compile(
59 "(jan)|(feb)|(mar)|(apr)|(may)|(jun)|(jul)|(aug)|(sep)|(oct)|(nov)|(dec)",
60 re.I | re.ASCII,
61 )
63 DATE_YEAR_RE = re.compile(r"(\d{2,4})", re.ASCII)
65 # calendar.timegm() fails for timestamps after datetime.datetime.max
66 # Minus one as a loss of precision occurs when timestamp() is called.
67 MAX_TIME = (
68 int(datetime.datetime.max.replace(tzinfo=datetime.timezone.utc).timestamp()) - 1
69 )
70 try:
71 calendar.timegm(time.gmtime(MAX_TIME))
72 except OSError:
73 # Hit the maximum representable time on Windows
74 # https://learn.microsoft.com/en-us/cpp/c-runtime-library/reference/localtime-localtime32-localtime64
75 MAX_TIME = calendar.timegm((3000, 12, 31, 23, 59, 59, -1, -1, -1))
76 except OverflowError:
77 # #4515: datetime.max may not be representable on 32-bit platforms
78 MAX_TIME = 2**31 - 1
79 # Avoid minuses in the future, 3x faster
80 SUB_MAX_TIME = MAX_TIME - 1
82 def __init__(
83 self,
84 *,
85 unsafe: bool = False,
86 quote_cookie: bool = True,
87 treat_as_secure_origin: StrOrURL | Iterable[StrOrURL] | None = None,
88 ) -> None:
89 self._cookies: defaultdict[tuple[str, str], SimpleCookie] = defaultdict(
90 SimpleCookie
91 )
92 self._morsel_cache: defaultdict[tuple[str, str], dict[str, Morsel[str]]] = (
93 defaultdict(dict)
94 )
95 # Cookie identity is (domain, path, name).
96 self._host_only_cookies: set[tuple[str, str, str]] = set()
97 self._unsafe = unsafe
98 self._quote_cookie = quote_cookie
99 if treat_as_secure_origin is None:
100 self._treat_as_secure_origin: frozenset[URL] = frozenset()
101 elif isinstance(treat_as_secure_origin, URL):
102 self._treat_as_secure_origin = frozenset({treat_as_secure_origin.origin()})
103 elif isinstance(treat_as_secure_origin, str):
104 self._treat_as_secure_origin = frozenset(
105 {URL(treat_as_secure_origin).origin()}
106 )
107 else:
108 self._treat_as_secure_origin = frozenset(
109 {
110 URL(url).origin() if isinstance(url, str) else url.origin()
111 for url in treat_as_secure_origin
112 }
113 )
114 self._expire_heap: list[tuple[float, tuple[str, str, str]]] = []
115 self._expirations: dict[tuple[str, str, str], float] = {}
117 @property
118 def unsafe(self) -> bool:
119 return self._unsafe
121 @property
122 def quote_cookie(self) -> bool:
123 return self._quote_cookie
125 @property
126 def cookies(self) -> MappingProxyType[tuple[str, str], SimpleCookie]:
127 """Return the cookies stored in this jar."""
128 return MappingProxyType(self._cookies)
130 @property
131 def host_only_cookies(self) -> frozenset[tuple[str, str, str]]:
132 """Return the host-only cookies stored in this jar."""
133 return frozenset(self._host_only_cookies)
135 def save(self, file_path: PathLike) -> None:
136 """Save cookies to a file using JSON format.
138 :param file_path: Path to file where cookies will be serialized,
139 :class:`str` or :class:`pathlib.Path` instance.
140 """
141 file_path = pathlib.Path(file_path)
142 data: dict[str, dict[str, dict[str, str | bool | float]]] = {}
143 for (domain, path), cookie in self._cookies.items():
144 key = f"{domain}|{path}"
145 data[key] = {}
146 for name, morsel in cookie.items():
147 morsel_data: dict[str, str | bool | float] = {
148 "key": morsel.key,
149 "value": morsel.value,
150 "coded_value": morsel.coded_value,
151 }
152 # Skip relative expiry; the absolute deadline is saved below.
153 for attr in morsel._reserved: # type: ignore[attr-defined]
154 if attr in _RELATIVE_EXPIRY_ATTRS:
155 continue
156 attr_val = morsel[attr]
157 if attr_val:
158 morsel_data[attr] = attr_val
159 # Persist or it reloads as a domain cookie and leaks to subdomains.
160 if (domain, path, name) in self._host_only_cookies:
161 morsel_data["host_only"] = True
162 if (exp := self._expirations.get((domain, path, name))) is not None:
163 morsel_data["expires_timestamp"] = exp
164 data[key][name] = morsel_data
166 # Cookie persistence may include authentication/session tokens.
167 # Use 0o600 at creation time to avoid umask-dependent overexposure
168 # and enforce least-privilege access to sensitive credential data.
169 with open(
170 file_path,
171 mode="w",
172 encoding="utf-8",
173 opener=lambda path, flags: os.open(path, flags, 0o600),
174 ) as f:
175 json.dump(data, f, indent=2)
177 def load(self, file_path: PathLike) -> None:
178 """Load cookies from a JSON file.
180 Replaces the current jar contents; loaded cookies pass through the
181 same acceptance rules as :meth:`update_cookies`.
183 :param file_path: Path to file from where cookies will be
184 imported, :class:`str` or :class:`pathlib.Path` instance.
185 """
186 file_path = pathlib.Path(file_path)
187 with file_path.open(mode="r", encoding="utf-8") as f:
188 data = json.load(f)
189 self._load_json_data(data)
191 def _load_json_data(
192 self, data: dict[str, dict[str, dict[str, str | bool | float]]]
193 ) -> None:
194 """Replace contents, routing cookies through update_cookies()."""
195 self.clear()
196 for compound_key, cookie_data in data.items():
197 domain, path = compound_key.split("|", 1)
198 for name, morsel_data in cookie_data.items():
199 morsel: Morsel[str] = Morsel()
200 # Use __setstate__ to bypass validation, same pattern
201 # used in _build_morsel and _cookie_helpers.
202 morsel.__setstate__( # type: ignore[attr-defined]
203 {
204 "key": morsel_data["key"],
205 "value": morsel_data["value"],
206 "coded_value": morsel_data["coded_value"],
207 }
208 )
209 # Restore morsel attributes
210 for attr in morsel._reserved: # type: ignore[attr-defined]
211 if attr in morsel_data and attr not in (
212 "key",
213 "value",
214 "coded_value",
215 ):
216 morsel[attr] = morsel_data[attr]
217 # Drop the domain so update_cookies() re-marks it host-only.
218 if morsel_data.get("host_only"):
219 morsel["domain"] = ""
220 response_url = (
221 URL.build(scheme="https", host=domain) if domain else URL()
222 )
223 self.update_cookies({name: morsel}, response_url)
224 # Restore the absolute deadline; update_cookies() schedules none.
225 if (exp := morsel_data.get("expires_timestamp")) is not None:
226 self._expire_cookie(float(exp), domain, path, name)
227 self._do_expiration()
229 def clear(self, predicate: ClearCookiePredicate | None = None) -> None:
230 if predicate is None:
231 self._expire_heap.clear()
232 self._cookies.clear()
233 self._morsel_cache.clear()
234 self._host_only_cookies.clear()
235 self._expirations.clear()
236 return
238 now = time.time()
239 to_del = [
240 key
241 for (domain, path), cookie in self._cookies.items()
242 for name, morsel in cookie.items()
243 if (
244 (key := (domain, path, name)) in self._expirations
245 and self._expirations[key] <= now
246 )
247 or predicate(morsel)
248 ]
249 if to_del:
250 self._delete_cookies(to_del)
252 def clear_domain(self, domain: str) -> None:
253 self.clear(lambda x: self._is_domain_match(domain, x["domain"]))
255 def __iter__(self) -> "Iterator[Morsel[str]]":
256 self._do_expiration()
257 for val in self._cookies.values():
258 yield from val.values()
260 def __len__(self) -> int:
261 """Return number of cookies.
263 This function does not iterate self to avoid unnecessary expiration
264 checks.
265 """
266 return sum(len(cookie.values()) for cookie in self._cookies.values())
268 def _do_expiration(self) -> None:
269 """Remove expired cookies."""
270 if not (expire_heap_len := len(self._expire_heap)):
271 return
273 # If the expiration heap grows larger than the number expirations
274 # times two, we clean it up to avoid keeping expired entries in
275 # the heap and consuming memory. We guard this with a minimum
276 # threshold to avoid cleaning up the heap too often when there are
277 # only a few scheduled expirations.
278 if (
279 expire_heap_len > _MIN_SCHEDULED_COOKIE_EXPIRATION
280 and expire_heap_len > len(self._expirations) * 2
281 ):
282 # Remove any expired entries from the expiration heap
283 # that do not match the expiration time in the expirations
284 # as it means the cookie has been re-added to the heap
285 # with a different expiration time.
286 self._expire_heap = [
287 entry
288 for entry in self._expire_heap
289 if self._expirations.get(entry[1]) == entry[0]
290 ]
291 heapq.heapify(self._expire_heap)
293 now = time.time()
294 to_del: list[tuple[str, str, str]] = []
295 # Find any expired cookies and add them to the to-delete list
296 while self._expire_heap:
297 when, cookie_key = self._expire_heap[0]
298 if when > now:
299 break
300 heapq.heappop(self._expire_heap)
301 # Check if the cookie hasn't been re-added to the heap
302 # with a different expiration time as it will be removed
303 # later when it reaches the top of the heap and its
304 # expiration time is met.
305 if self._expirations.get(cookie_key) == when:
306 to_del.append(cookie_key)
308 if to_del:
309 self._delete_cookies(to_del)
311 def _delete_cookies(self, to_del: list[tuple[str, str, str]]) -> None:
312 for domain, path, name in to_del:
313 self._host_only_cookies.discard((domain, path, name))
314 self._cookies[(domain, path)].pop(name, None)
315 self._morsel_cache[(domain, path)].pop(name, None)
316 self._expirations.pop((domain, path, name), None)
318 def _expire_cookie(self, when: float, domain: str, path: str, name: str) -> None:
319 cookie_key = (domain, path, name)
320 if self._expirations.get(cookie_key) == when:
321 # Avoid adding duplicates to the heap
322 return
323 heapq.heappush(self._expire_heap, (when, cookie_key))
324 self._expirations[cookie_key] = when
326 def update_cookies(self, cookies: LooseCookies, response_url: URL = URL()) -> None:
327 """Update cookies."""
328 self._update_cookies(cookies, response_url, copy_morsels=True)
330 def update_cookies_from_headers(
331 self, headers: Sequence[str], response_url: URL
332 ) -> None:
333 """Update cookies from raw Set-Cookie headers."""
334 if headers and (cookies_to_update := parse_set_cookie_headers(headers)):
335 # The freshly parsed Morsels are not shared with the caller,
336 # so they can be stored and normalized without a defensive copy.
337 self._update_cookies(cookies_to_update, response_url, copy_morsels=False)
339 def _update_cookies(
340 self, cookies: LooseCookies, response_url: URL, *, copy_morsels: bool
341 ) -> None:
342 hostname = response_url.raw_host
344 if not self._unsafe and is_ip_address(hostname):
345 # Don't accept cookies from IPs
346 return
348 if isinstance(cookies, Mapping):
349 cookies = cookies.items()
351 for name, cookie in cookies:
352 if not isinstance(cookie, Morsel):
353 tmp = SimpleCookie()
354 tmp[name] = cookie # type: ignore[assignment]
355 cookie = tmp[name]
356 elif copy_morsels:
357 # TODO(https://github.com/python/typeshed/pull/16346): Remove cast
358 cookie = cast("Morsel[str]", cookie.copy())
360 domain = cookie["domain"]
362 # ignore domains with trailing dots
363 if domain and domain[-1] == ".":
364 domain = ""
365 del cookie["domain"]
367 if domain and domain[0] == ".":
368 # Remove leading dot
369 domain = domain[1:]
370 cookie["domain"] = domain
372 if domain and hostname and not self._is_domain_match(domain, hostname):
373 # Setting cookies for different domains is not allowed
374 continue
376 path = cookie["path"]
377 if not path or path[0] != "/":
378 # Set the cookie's path to the response path
379 path = response_url.path
380 if not path.startswith("/"):
381 path = "/"
382 else:
383 # Cut everything from the last slash to the end
384 path = "/" + path[1 : path.rfind("/")]
385 cookie["path"] = path
386 path = path.rstrip("/")
388 if not domain and hostname is not None:
389 self._host_only_cookies.add((hostname, path, name))
390 domain = cookie["domain"] = hostname
391 else:
392 # A cookie with an explicit Domain attribute replaces any
393 # host-only cookie with the same (domain, path, name) identity.
394 self._host_only_cookies.discard((domain, path, name))
396 if max_age := cookie["max-age"]:
397 try:
398 delta_seconds = int(max_age)
399 # https://datatracker.ietf.org/doc/html/rfc6265#section-5.2.2
400 if delta_seconds <= 0:
401 max_age_expiration = 0.0
402 else:
403 # Cap first to protect against OverflowError on next line.
404 delta_seconds = min(delta_seconds, self.MAX_TIME)
405 max_age_expiration = min(
406 time.time() + delta_seconds, self.MAX_TIME
407 )
408 self._expire_cookie(max_age_expiration, domain, path, name)
409 except ValueError:
410 cookie["max-age"] = ""
412 elif expires := cookie["expires"]:
413 if expire_time := self._parse_date(expires):
414 self._expire_cookie(expire_time, domain, path, name)
415 else:
416 cookie["expires"] = ""
418 key = (domain, path)
419 if self._cookies[key].get(name) != cookie:
420 # Don't blow away the cache if the same
421 # cookie gets set again
422 self._cookies[key][name] = cookie
423 self._morsel_cache[key].pop(name, None)
425 self._do_expiration()
427 def filter_cookies(self, request_url: URL) -> "BaseCookie[str]":
428 """Returns this jar's cookies filtered by their attributes."""
429 if not isinstance(request_url, URL):
430 warnings.warn( # type: ignore[unreachable]
431 f"The method accepts yarl.URL instances only, got {type(request_url)}",
432 DeprecationWarning,
433 )
434 request_url = URL(request_url)
435 # We always use BaseCookie now since all
436 # cookies set on on filtered are fully constructed
437 # Morsels, not just names and values.
438 filtered: BaseCookie[str] = BaseCookie()
439 if not self._cookies:
440 # Skip do_expiration() if there are no cookies.
441 return filtered
442 self._do_expiration()
443 if not self._cookies:
444 # Skip rest of function if no non-expired cookies.
445 return filtered
446 hostname = request_url.raw_host or ""
448 is_not_secure = request_url.scheme not in ("https", "wss")
449 if is_not_secure and self._treat_as_secure_origin:
450 request_origin = URL()
451 with contextlib.suppress(ValueError):
452 request_origin = request_url.origin()
453 is_not_secure = request_origin not in self._treat_as_secure_origin
455 # Send shared cookie
456 key = ("", "")
457 for c in self._cookies[key].values():
458 # Check cache first
459 if c.key in self._morsel_cache[key]:
460 filtered[c.key] = self._morsel_cache[key][c.key]
461 continue
463 # Build and cache the morsel
464 mrsl_val = self._build_morsel(c)
465 self._morsel_cache[key][c.key] = mrsl_val
466 filtered[c.key] = mrsl_val
468 if is_ip_address(hostname):
469 if not self._unsafe:
470 return filtered
471 domains: Iterable[str] = (hostname,)
472 else:
473 # Get all the subdomains that might match a cookie (e.g. "foo.bar.com", "bar.com", "com")
474 domains = itertools.accumulate(
475 reversed(hostname.split(".")), _FORMAT_DOMAIN_REVERSED
476 )
478 # Get all the path prefixes that might match a cookie (e.g. "", "/foo", "/foo/bar")
479 paths = itertools.accumulate(request_url.path.split("/"), _FORMAT_PATH)
480 # Create every combination of (domain, path) pairs.
481 pairs = itertools.product(domains, paths)
483 path_len = len(request_url.path)
484 # Point 2: https://www.rfc-editor.org/rfc/rfc6265.html#section-5.4
485 for p in pairs:
486 if p not in self._cookies:
487 continue
488 for name, cookie in self._cookies[p].items():
489 domain = cookie["domain"]
491 if domain != hostname and p + (name,) in self._host_only_cookies:
492 continue
494 # Skip edge case when the cookie has a trailing slash but request doesn't.
495 if len(cookie["path"]) > path_len:
496 continue
498 if is_not_secure and cookie["secure"]:
499 continue
501 # We already built the Morsel so reuse it here
502 if name in self._morsel_cache[p]:
503 filtered[name] = self._morsel_cache[p][name]
504 continue
506 # Build and cache the morsel
507 mrsl_val = self._build_morsel(cookie)
508 self._morsel_cache[p][name] = mrsl_val
509 filtered[name] = mrsl_val
511 return filtered
513 def _build_morsel(self, cookie: Morsel[str]) -> Morsel[str]:
514 """Build a morsel for sending, respecting quote_cookie setting."""
515 if self._quote_cookie and cookie.coded_value and cookie.coded_value[0] == '"':
516 return preserve_morsel_with_coded_value(cookie)
517 morsel: Morsel[str] = Morsel()
518 if self._quote_cookie:
519 value, coded_value = _SIMPLE_COOKIE.value_encode(cookie.value)
520 else:
521 coded_value = value = cookie.value
522 # We use __setstate__ instead of the public set() API because it allows us to
523 # bypass validation and set already validated state. This is more stable than
524 # setting protected attributes directly.
525 morsel.__setstate__({"key": cookie.key, "value": value, "coded_value": coded_value}) # type: ignore[attr-defined]
526 return morsel
528 @staticmethod
529 def _is_domain_match(domain: str, hostname: str) -> bool:
530 """Implements domain matching adhering to RFC 6265."""
531 if hostname == domain:
532 return True
534 if not hostname.endswith(domain):
535 return False
537 non_matching = hostname[: -len(domain)]
539 if not non_matching.endswith("."):
540 return False
542 return not is_ip_address(hostname)
544 @classmethod
545 def _parse_date(cls, date_str: str) -> int | None:
546 """Implements date string parsing adhering to RFC 6265."""
547 if not date_str:
548 return None
550 found_time = False
551 found_day = False
552 found_month = False
553 found_year = False
555 hour = minute = second = 0
556 day = 0
557 month = 0
558 year = 0
560 for token_match in cls.DATE_TOKENS_RE.finditer(date_str):
561 token = token_match.group("token")
563 if not found_time:
564 time_match = cls.DATE_HMS_TIME_RE.match(token)
565 if time_match:
566 found_time = True
567 hour, minute, second = (int(s) for s in time_match.groups())
568 continue
570 if not found_day:
571 day_match = cls.DATE_DAY_OF_MONTH_RE.match(token)
572 if day_match:
573 found_day = True
574 day = int(day_match.group())
575 continue
577 if not found_month:
578 month_match = cls.DATE_MONTH_RE.match(token)
579 if month_match:
580 found_month = True
581 assert month_match.lastindex is not None
582 month = month_match.lastindex
583 continue
585 if not found_year:
586 year_match = cls.DATE_YEAR_RE.match(token)
587 if year_match:
588 found_year = True
589 year = int(year_match.group())
591 if 70 <= year <= 99:
592 year += 1900
593 elif 0 <= year <= 69:
594 year += 2000
596 if False in (found_day, found_month, found_year, found_time):
597 return None
599 if not 1 <= day <= 31:
600 return None
602 if year < 1601 or hour > 23 or minute > 59 or second > 59:
603 return None
605 return calendar.timegm((year, month, day, hour, minute, second, -1, -1, -1))
608class DummyCookieJar(AbstractCookieJar):
609 """Implements a dummy cookie storage.
611 It can be used with the ClientSession when no cookie processing is needed.
613 """
615 def __iter__(self) -> "Iterator[Morsel[str]]":
616 while False:
617 yield None # type: ignore[unreachable]
619 def __len__(self) -> int:
620 return 0
622 @property
623 def unsafe(self) -> bool:
624 return False
626 @property
627 def quote_cookie(self) -> bool:
628 return True
630 @property
631 def cookies(self) -> MappingProxyType[tuple[str, str], SimpleCookie]:
632 """Return an empty mapping."""
633 return MappingProxyType({})
635 @property
636 def host_only_cookies(self) -> frozenset[tuple[str, str, str]]:
637 """Return an empty frozenset."""
638 return frozenset()
640 def clear(self, predicate: ClearCookiePredicate | None = None) -> None:
641 pass
643 def clear_domain(self, domain: str) -> None:
644 pass
646 def update_cookies(self, cookies: LooseCookies, response_url: URL = URL()) -> None:
647 pass
649 def filter_cookies(self, request_url: URL) -> "BaseCookie[str]":
650 return SimpleCookie()