import asyncio import calendar import contextlib import datetime import heapq import itertools import json import math import os import pathlib import pickle import re import time import warnings from collections import defaultdict from collections.abc import Iterable, Iterator, Mapping, Sequence from http.cookies import BaseCookie, Morsel, SimpleCookie from types import MappingProxyType from typing import Union, cast from yarl import URL from ._cookie_helpers import ( _COOKIE_CTL_RE, _encoded_length, parse_set_cookie_headers, preserve_morsel_with_coded_value, ) from .abc import AbstractCookieJar, ClearCookiePredicate from .helpers import is_ip_address from .log import internal_logger from .typedefs import LooseCookies, PathLike, StrOrURL __all__ = ("CookieJar", "DummyCookieJar") CookieItem = Union[str, "Morsel[str]"] _JarKey = tuple[str, str] # (domain, path) _CookieKey = tuple[str, str, str] # (domain, path, name) # A built morsel, its encoded name=value length and the cookie's key. _MorselEntry = tuple["Morsel[str]", int, _CookieKey] _UNLIMITED_COOKIE_KEY: _CookieKey = ("", "", "") # We cache these string methods here as their use is in performance critical code. _FORMAT_PATH = "{}/{}".format _FORMAT_DOMAIN_REVERSED = "{1}.{0}".format # The minimum number of scheduled cookie expirations before we start cleaning up # the expiration heap. This is a performance optimization to avoid cleaning up the # heap too often when there are only a few scheduled expirations. _MIN_SCHEDULED_COOKIE_EXPIRATION = 100 _SIMPLE_COOKIE = SimpleCookie() # Private, non-configurable limits for the built-in jar. # Eviction follows Firefox's CookieStorage. _MAX_COOKIES_PER_DOMAIN = 180 _COOKIE_QUOTA_PER_DOMAIN = 150 _MAX_COOKIES_TOTAL = 3300 _COOKIE_QUOTA_TOTAL = 3000 _MAX_COOKIES_PER_REQUEST = 150 _MAX_COOKIE_HEADER_LENGTH = 8190 _COOKIE_HEADER_PREFIX_LENGTH = len(b"Cookie: ") # Not persisted; the absolute deadline is saved instead. _RELATIVE_EXPIRY_ATTRS = frozenset(("max-age", "expires")) class _RestrictedCookieUnpickler(pickle._Unpickler): """A restricted unpickler that only allows cookie-related types. This prevents arbitrary code execution when loading pickled cookie data from untrusted sources. Only types that are expected in a serialized CookieJar are permitted. Subclasses :class:`pickle._Unpickler` (the pure-Python implementation) rather than :class:`pickle.Unpickler` because the accelerated unpickler on some implementations (notably PyPy) does not dispatch through :meth:`find_class` overrides. See: https://docs.python.org/3/library/pickle.html#restricting-globals """ _ALLOWED_CLASSES: frozenset[tuple[str, str]] = frozenset( { # Core cookie types ("http.cookies", "SimpleCookie"), ("http.cookies", "Morsel"), # Container types used by CookieJar._cookies ("collections", "defaultdict"), # builtins that pickle uses for reconstruction ("builtins", "tuple"), ("builtins", "set"), ("builtins", "frozenset"), ("builtins", "dict"), } ) def find_class(self, module: str, name: str) -> type: if (module, name) not in self._ALLOWED_CLASSES: raise pickle.UnpicklingError( f"Forbidden class: {module}.{name}. " "CookieJar.load() only allows cookie-related types for security. " "See https://docs.python.org/3/library/pickle.html#restricting-globals" ) return super().find_class(module, name) # type: ignore[no-any-return] class CookieJar(AbstractCookieJar): """Implements cookie storage adhering to RFC 6265.""" _limits_enabled = True DATE_TOKENS_RE = re.compile( r"[\x09\x20-\x2F\x3B-\x40\x5B-\x60\x7B-\x7E]*" r"(?P[\x00-\x08\x0A-\x1F\d:a-zA-Z\x7F-\xFF]+)" ) DATE_HMS_TIME_RE = re.compile(r"(\d{1,2}):(\d{1,2}):(\d{1,2})") DATE_DAY_OF_MONTH_RE = re.compile(r"(\d{1,2})") DATE_MONTH_RE = re.compile( "(jan)|(feb)|(mar)|(apr)|(may)|(jun)|(jul)|(aug)|(sep)|(oct)|(nov)|(dec)", re.I, ) DATE_YEAR_RE = re.compile(r"(\d{2,4})") # calendar.timegm() fails for timestamps after datetime.datetime.max # Minus one as a loss of precision occurs when timestamp() is called. MAX_TIME = ( int(datetime.datetime.max.replace(tzinfo=datetime.timezone.utc).timestamp()) - 1 ) try: calendar.timegm(time.gmtime(MAX_TIME)) except OSError: # Hit the maximum representable time on Windows # https://learn.microsoft.com/en-us/cpp/c-runtime-library/reference/localtime-localtime32-localtime64 MAX_TIME = calendar.timegm((3000, 12, 31, 23, 59, 59, -1, -1, -1)) except OverflowError: # #4515: datetime.max may not be representable on 32-bit platforms MAX_TIME = 2**31 - 1 # Avoid minuses in the future, 3x faster SUB_MAX_TIME = MAX_TIME - 1 def __init__( self, *, unsafe: bool = False, quote_cookie: bool = True, treat_as_secure_origin: StrOrURL | Iterable[StrOrURL] | None = None, loop: asyncio.AbstractEventLoop | None = None, ) -> None: super().__init__(loop=loop) self._cookies: defaultdict[_JarKey, SimpleCookie] = defaultdict(SimpleCookie) self._morsel_cache: defaultdict[_JarKey, dict[str, _MorselEntry]] = defaultdict( dict ) # Cookie identity is (domain, path, name). self._host_only_cookies: set[_CookieKey] = set() self._unsafe = unsafe self._quote_cookie = quote_cookie if treat_as_secure_origin is None: self._treat_as_secure_origin: frozenset[URL] = frozenset() elif isinstance(treat_as_secure_origin, URL): self._treat_as_secure_origin = frozenset({treat_as_secure_origin.origin()}) elif isinstance(treat_as_secure_origin, str): self._treat_as_secure_origin = frozenset( {URL(treat_as_secure_origin).origin()} ) else: self._treat_as_secure_origin = frozenset( { URL(url).origin() if isinstance(url, str) else url.origin() for url in treat_as_secure_origin } ) self._expire_heap: list[tuple[float, _CookieKey]] = [] self._expirations: dict[_CookieKey, float] = {} self._access_generations: dict[_CookieKey, int] = {} self._next_access_generation = 0 self._domain_counts: dict[str, int] = {} self._cookie_count = 0 @property def unsafe(self) -> bool: return self._unsafe @property def quote_cookie(self) -> bool: return self._quote_cookie @property def treat_as_secure_origin(self) -> frozenset[URL]: """Return origins considered secure even over cleartext connections.""" return self._treat_as_secure_origin @property def cookies(self) -> MappingProxyType[tuple[str, str], SimpleCookie]: """Return the cookies stored in this jar.""" return MappingProxyType(self._cookies) @property def host_only_cookies(self) -> frozenset[tuple[str, str, str]]: """Return the host-only cookies stored in this jar.""" return frozenset(self._host_only_cookies) def save(self, file_path: PathLike) -> None: """Save cookies to a file using JSON format. :param file_path: Path to file where cookies will be serialized, :class:`str` or :class:`pathlib.Path` instance. """ file_path = pathlib.Path(file_path) data: dict[str, dict[str, dict[str, str | bool | float]]] = {} for (domain, path), cookie in self._cookies.items(): key = f"{domain}|{path}" data[key] = {} for name, morsel in cookie.items(): morsel_data: dict[str, str | bool | float] = { "key": morsel.key, "value": morsel.value, "coded_value": morsel.coded_value, } # Skip relative expiry; the absolute deadline is saved below. for attr in morsel._reserved: # type: ignore[attr-defined] if attr in _RELATIVE_EXPIRY_ATTRS: continue attr_val = morsel[attr] if attr_val: morsel_data[attr] = attr_val # Persist or it reloads as a domain cookie and leaks to subdomains. if (domain, path, name) in self._host_only_cookies: morsel_data["host_only"] = True if (exp := self._expirations.get((domain, path, name))) is not None: morsel_data["expires_timestamp"] = exp data[key][name] = morsel_data # Cookie persistence may include authentication/session tokens. # Use 0o600 at creation time to avoid umask-dependent overexposure # and enforce least-privilege access to sensitive credential data. with open( file_path, mode="w", encoding="utf-8", opener=lambda path, flags: os.open(path, flags, 0o600), ) as f: json.dump(data, f, indent=2) def load(self, file_path: PathLike) -> None: """Load cookies from a file. Tries to load JSON format first. Falls back to loading legacy pickle format (using a restricted unpickler) for backward compatibility with existing cookie files. Replaces the current jar contents; loaded cookies pass through the same acceptance rules as :meth:`update_cookies`. A malformed file raises and leaves the jar unchanged. :param file_path: Path to file from where cookies will be imported, :class:`str` or :class:`pathlib.Path` instance. """ file_path = pathlib.Path(file_path) # Try JSON format first; only a file that isn't JSON is a legacy pickle. try: with file_path.open(mode="r", encoding="utf-8") as f: data = json.load(f) except (json.JSONDecodeError, UnicodeDecodeError, ValueError): pass else: self._load_json_data(data) return # Fall back to legacy pickle format with restricted unpickler with file_path.open(mode="rb") as f: pickled = _RestrictedCookieUnpickler(f).load() loaded = self._new_loading_jar() for (domain, _), cookies in pickled.items(): response_url = URL.build(scheme="https", host=domain) if domain else URL() for name, morsel in cookies.items(): loaded._update_cookies({name: morsel}, response_url, copy_morsels=False) self._replace_contents(loaded) def _new_loading_jar(self) -> "CookieJar": return CookieJar( unsafe=self._unsafe, quote_cookie=self._quote_cookie, treat_as_secure_origin=self._treat_as_secure_origin, loop=self._loop, ) def _replace_contents(self, loaded: "CookieJar") -> None: # Refill the existing mapping so views from the cookies property stay live. cookies = self._cookies cookies.clear() cookies.update(loaded._cookies) self.__dict__.update(loaded.__dict__) self._cookies = cookies def _load_json_data( self, data: dict[str, dict[str, dict[str, str | bool | float]]] ) -> None: """Replace contents, routing cookies through _update_cookies().""" # Fill a new jar and adopt its state only once every record is in, # so a bad file leaves this one unchanged. loaded = self._new_loading_jar() for compound_key, cookie_data in data.items(): domain, path = compound_key.split("|", 1) for name, morsel_data in cookie_data.items(): expiration: float | None = None if (raw_expiration := morsel_data.get("expires_timestamp")) is not None: try: expiration = float(raw_expiration) if not math.isfinite(expiration): raise ValueError(expiration) except (OverflowError, TypeError, ValueError) as exc: raise ValueError( f"Cookie {name!r} in {compound_key!r}: " "expiration timestamp must be a finite number" ) from exc morsel: Morsel[str] = Morsel() # Use __setstate__ to bypass validation, same pattern # used in _build_morsel and _cookie_helpers. morsel.__setstate__( # type: ignore[attr-defined] { "key": morsel_data["key"], "value": morsel_data["value"], "coded_value": morsel_data["coded_value"], } ) # Restore morsel attributes for attr in morsel._reserved: # type: ignore[attr-defined] if attr in morsel_data and attr not in ( "key", "value", "coded_value", ): morsel[attr] = morsel_data[attr] # Drop the domain so update_cookies() re-marks it host-only. if morsel_data.get("host_only"): morsel["domain"] = "" response_url = ( URL.build(scheme="https", host=domain) if domain else URL() ) loaded._update_cookies( {name: morsel}, response_url, copy_morsels=False, expiration=expiration, ) self._replace_contents(loaded) def clear(self, predicate: ClearCookiePredicate | None = None) -> None: if predicate is None: self._expire_heap.clear() self._cookies.clear() self._morsel_cache.clear() self._host_only_cookies.clear() self._expirations.clear() self._access_generations.clear() self._domain_counts.clear() self._cookie_count = 0 return now = time.time() to_del = [ key for (domain, path), cookie in self._cookies.items() for name, morsel in cookie.items() if ( (key := (domain, path, name)) in self._expirations and self._expirations[key] <= now ) or predicate(morsel) ] if to_del: self._delete_cookies(to_del) def clear_domain(self, domain: str) -> None: self.clear(lambda x: self._is_domain_match(domain, x["domain"])) def __iter__(self) -> "Iterator[Morsel[str]]": self._do_expiration() for val in self._cookies.values(): yield from val.values() def __len__(self) -> int: """Return number of cookies. This function does not iterate self to avoid unnecessary expiration checks. """ return self._cookie_count def _do_expiration(self) -> None: """Remove expired cookies.""" if not (expire_heap_len := len(self._expire_heap)): return # If the expiration heap grows larger than the number expirations # times two, we clean it up to avoid keeping expired entries in # the heap and consuming memory. We guard this with a minimum # threshold to avoid cleaning up the heap too often when there are # only a few scheduled expirations. if ( expire_heap_len > _MIN_SCHEDULED_COOKIE_EXPIRATION and expire_heap_len > len(self._expirations) * 2 ): # Remove any expired entries from the expiration heap # that do not match the expiration time in the expirations # as it means the cookie has been re-added to the heap # with a different expiration time. self._expire_heap = [ entry for entry in self._expire_heap if self._expirations.get(entry[1]) == entry[0] ] heapq.heapify(self._expire_heap) now = time.time() to_del: list[_CookieKey] = [] # Find any expired cookies and add them to the to-delete list while self._expire_heap: when, cookie_key = self._expire_heap[0] if when > now: break heapq.heappop(self._expire_heap) # Check if the cookie hasn't been re-added to the heap # with a different expiration time as it will be removed # later when it reaches the top of the heap and its # expiration time is met. if self._expirations.get(cookie_key) == when: to_del.append(cookie_key) if to_del: self._delete_cookies(to_del) def _delete_cookies(self, to_del: list[_CookieKey]) -> None: for domain, path, name in to_del: cookie_key = (domain, path, name) jar_key = (domain, path) self._host_only_cookies.discard(cookie_key) if (cookies := self._cookies.get(jar_key)) is not None: if cookies.pop(name, None) is not None: self._cookie_count -= 1 if domain_count := self._domain_counts[domain] - 1: self._domain_counts[domain] = domain_count else: del self._domain_counts[domain] if not cookies: del self._cookies[jar_key] if (morsels := self._morsel_cache.get(jar_key)) is not None: morsels.pop(name, None) if not morsels: del self._morsel_cache[jar_key] self._expirations.pop(cookie_key, None) self._access_generations.pop(cookie_key, None) def _make_room(self, domain: str, secure: bool) -> bool: """Evict for a new cookie like Firefox; False drops the new cookie. Mirrors AddCookie and FindStaleCookies: expired, then non-secure, cookies go first, least recently used first; secure cookies are only evicted for another secure cookie. """ if self._domain_counts.get(domain, 0) >= _MAX_COOKIES_PER_DOMAIN: self._do_expiration() count = self._domain_counts.get(domain, 0) # Leave room so the domain ends at its quota. if (excess := count - _COOKIE_QUOTA_PER_DOMAIN + 1) <= 0: return True non_secure: list[tuple[int, _CookieKey]] = [] secure_keys: list[tuple[int, _CookieKey]] = [] for (cookie_domain, path), cookies in self._cookies.items(): if cookie_domain != domain: continue for name, cookie in cookies.items(): cookie_key = (domain, path, name) entry = (self._access_generations.get(cookie_key, 0), cookie_key) (secure_keys if cookie["secure"] else non_secure).append(entry) if not (candidates := non_secure): if count < _MAX_COOKIES_PER_DOMAIN: # expiry freed a slot return True if not secure: return False candidates = secure_keys candidates.sort() evicted = candidates[:excess] self._delete_cookies([key for _, key in evicted]) internal_logger.debug( "Evicted %d cookie(s) for %s over the %d-cookie domain limit", len(evicted), domain, _MAX_COOKIES_PER_DOMAIN, ) else: self._purge_cookies() return True def _purge_cookies(self) -> None: """Evict expired, then least recently used, cookies down to the total.""" self._do_expiration() if (excess := self._cookie_count - _COOKIE_QUOTA_TOTAL) <= 0: return generations = self._access_generations self._delete_cookies(sorted(generations, key=generations.__getitem__)[:excess]) internal_logger.debug( "Evicted %d cookie(s) over the %d-cookie total limit", excess, _MAX_COOKIES_TOTAL, ) def _expire_cookie(self, when: float, domain: str, path: str, name: str) -> None: cookie_key = (domain, path, name) if self._expirations.get(cookie_key) == when: # Avoid adding duplicates to the heap return heapq.heappush(self._expire_heap, (when, cookie_key)) self._expirations[cookie_key] = when def update_cookies(self, cookies: LooseCookies, response_url: URL = URL()) -> None: """Update cookies.""" self._update_cookies(cookies, response_url, copy_morsels=True) def update_cookies_from_headers( self, headers: Sequence[str], response_url: URL ) -> None: """Update cookies from raw Set-Cookie headers.""" hostname = response_url.raw_host if not self._unsafe and is_ip_address(hostname): # Reject IP-origin cookies before parsing the fields. return if headers and (cookies_to_update := parse_set_cookie_headers(headers)): # The freshly parsed Morsels are not shared with the caller, # so they can be stored and normalized without a defensive copy. # The parser only produces cookies that can be sent. self._update_cookies( cookies_to_update, response_url, copy_morsels=False, check_sendable=False, ) def _update_cookies( self, cookies: LooseCookies, response_url: URL, *, copy_morsels: bool, expiration: float | None = None, check_sendable: bool = True, ) -> None: hostname = response_url.raw_host if not self._unsafe and is_ip_address(hostname): # Don't accept cookies from IPs return if isinstance(cookies, Mapping): cookies = cookies.items() for name, cookie in cookies: if not isinstance(cookie, Morsel): tmp = SimpleCookie() tmp[name] = cookie # type: ignore[assignment] cookie = tmp[name] elif copy_morsels: # TODO(https://github.com/python/typeshed/pull/16346): Remove cast cookie = cast("Morsel[str]", cookie.copy()) domain = cookie["domain"] # ignore domains with trailing dots if domain and domain[-1] == ".": domain = "" del cookie["domain"] if domain and domain[0] == ".": # Remove leading dot domain = domain[1:] cookie["domain"] = domain if domain and hostname and not self._is_domain_match(domain, hostname): # Setting cookies for different domains is not allowed continue path = cookie["path"] if not path or path[0] != "/": # Set the cookie's path to the response path path = response_url.path if not path.startswith("/"): path = "/" else: # Cut everything from the last slash to the end path = "/" + path[1 : path.rfind("/")] cookie["path"] = path path = path.rstrip("/") if not domain and hostname is not None: host_only = True domain = hostname else: host_only = False deadline: float | None = None if expiration is not None: # Saved jars pass an absolute deadline. deadline = expiration elif max_age := cookie["max-age"]: try: delta_seconds = int(max_age) # https://datatracker.ietf.org/doc/html/rfc6265#section-5.2.2 if delta_seconds <= 0: deadline = 0.0 else: # Cap first to protect against OverflowError on next line. delta_seconds = min(delta_seconds, self.MAX_TIME) deadline = min(time.time() + delta_seconds, self.MAX_TIME) except ValueError: cookie["max-age"] = "" elif expires := cookie["expires"]: if expire_time := self._parse_date(expires): deadline = expire_time else: cookie["expires"] = "" if check_sendable and self._limits_enabled and not self._can_send(cookie): internal_logger.debug( "Dropped cookie %r for %s: it can't be sent in a Cookie header", name, domain, ) continue key = (domain, path) cookie_key = (domain, path, name) bucket = self._cookies.get(key) stored = None if bucket is None else bucket.get(name) if ( stored is None and self._limits_enabled and ( self._domain_counts.get(domain, 0) >= _MAX_COOKIES_PER_DOMAIN or self._cookie_count >= _MAX_COOKIES_TOTAL ) # Like Firefox, a cookie that only deletes never evicts others. and (deadline is None or deadline > time.time()) and not self._make_room(domain, bool(cookie["secure"])) ): internal_logger.debug( "Dropped cookie %r for %s: its domain is full of Secure cookies", name, domain, ) continue if host_only: self._host_only_cookies.add(cookie_key) cookie["domain"] = domain else: # A cookie with an explicit Domain attribute replaces any # host-only cookie with the same (domain, path, name) identity. self._host_only_cookies.discard(cookie_key) if deadline is not None: self._expire_cookie(deadline, domain, path, name) elif self._expirations: # RFC 6265 5.3 step 11: a replacement doesn't inherit expiry. self._expirations.pop(cookie_key, None) if stored != cookie: # Don't blow away the cache if the same # cookie gets set again if stored is None: self._cookie_count += 1 self._domain_counts[domain] = self._domain_counts.get(domain, 0) + 1 self._cookies[key][name] = cookie if (morsels := self._morsel_cache.get(key)) is not None: morsels.pop(name, None) if self._limits_enabled: # Like Firefox, each stored cookie gets its own access time. self._next_access_generation += 1 self._access_generations[cookie_key] = self._next_access_generation self._do_expiration() def filter_cookies(self, request_url: URL = URL()) -> "BaseCookie[str]": """Returns this jar's cookies filtered by their attributes.""" # We always use BaseCookie now since all # cookies set on on filtered are fully constructed # Morsels, not just names and values. filtered: BaseCookie[str] = BaseCookie() if not self._cookies: # Skip do_expiration() if there are no cookies. return filtered self._do_expiration() if not self._cookies: # Skip rest of function if no non-expired cookies. return filtered if type(request_url) is not URL: warnings.warn( "filter_cookies expects yarl.URL instances only," f"and will stop working in 4.x, got {type(request_url)}", DeprecationWarning, stacklevel=2, ) request_url = URL(request_url) hostname = request_url.raw_host or "" is_not_secure = request_url.scheme not in ("https", "wss") if is_not_secure and self._treat_as_secure_origin: request_origin = URL() with contextlib.suppress(ValueError): request_origin = request_url.origin() is_not_secure = request_origin not in self._treat_as_secure_origin # Last (most specific) matching cookie per name wins. selected: dict[str, _MorselEntry] = {} # Send shared cookie key = ("", "") if key in self._cookies: morsels = self._morsel_cache[key] for name, cookie in self._cookies[key].items(): if is_not_secure and cookie["secure"]: continue if (entry := morsels.get(name)) is None: entry = morsels[name] = self._morsel_entry(key, name, cookie) selected[name] = entry if not is_ip_address(hostname): # Get all the subdomains that might match a cookie (e.g. "foo.bar.com", "bar.com", "com") domains: Iterable[str] = itertools.accumulate( reversed(hostname.split(".")), _FORMAT_DOMAIN_REVERSED ) elif self._unsafe: domains = (hostname,) else: domains = () # Get all the path prefixes that might match a cookie (e.g. "", "/foo", "/foo/bar") paths = itertools.accumulate(request_url.path.split("/"), _FORMAT_PATH) # Create every combination of (domain, path) pairs. pairs = itertools.product(domains, paths) path_len = len(request_url.path) # Point 2: https://www.rfc-editor.org/rfc/rfc6265.html#section-5.4 for p in pairs: if p == ("", "") or p not in self._cookies: continue morsels = self._morsel_cache[p] for name, cookie in self._cookies[p].items(): domain = cookie["domain"] if domain != hostname and p + (name,) in self._host_only_cookies: continue # Skip edge case when the cookie has a trailing slash but request doesn't. if len(cookie["path"]) > path_len: continue if is_not_secure and cookie["secure"]: continue if (entry := morsels.get(name)) is None: entry = morsels[name] = self._morsel_entry(p, name, cookie) # Re-insert so the order tracks specificity, for the header and # the over-limit fallback. selected.pop(name, None) selected[name] = entry if self._limits_enabled: # Each cookie after the first adds a "; " separator. header_length = _COOKIE_HEADER_PREFIX_LENGTH - 2 + 2 * len(selected) header_length += sum(entry[1] for entry in selected.values()) if ( len(selected) > _MAX_COOKIES_PER_REQUEST or header_length > _MAX_COOKIE_HEADER_LENGTH ): return self._filter_cookies_within_limits(selected, hostname) self._touch_cookies([entry[2] for entry in selected.values()]) # Bypass BaseCookie.__setitem__, which stores Morsel values as-is. dict.update(filtered, {name: entry[0] for name, entry in selected.items()}) return filtered def _filter_cookies_within_limits( self, selected: dict[str, _MorselEntry], hostname: str ) -> "BaseCookie[str]": """Admit the most specific cookies first so a host's own cookies win.""" header_length = _COOKIE_HEADER_PREFIX_LENGTH - 2 # no "; " before the first admitted: list[tuple[str, _MorselEntry]] = [] for name, entry in reversed(selected.items()): if len(admitted) >= _MAX_COOKIES_PER_REQUEST: break if header_length + 2 + entry[1] > _MAX_COOKIE_HEADER_LENGTH: continue header_length += 2 + entry[1] admitted.append((name, entry)) filtered: BaseCookie[str] = BaseCookie() for name, entry in reversed(admitted): dict.__setitem__(filtered, name, entry[0]) self._touch_cookies([entry[2] for _, entry in admitted]) internal_logger.debug( "Omitted %d cookie(s) for %s over the %d-cookie" " or %d-octet Cookie header limit", len(selected) - len(admitted), hostname, _MAX_COOKIES_PER_REQUEST, _MAX_COOKIE_HEADER_LENGTH, ) return filtered @staticmethod def _can_send(cookie: Morsel[str]) -> bool: """Return whether the cookie can be written to a Cookie header.""" # Whichever of value and coded_value is sent, CPython builds with the # CVE-2026-3644 patch also reject control characters in either. text = f"{cookie.key}={cookie.value}{cookie.coded_value}" return _COOKIE_CTL_RE.search(text) is None and _encoded_length(text) is not None def _morsel_entry( self, jar_key: _JarKey, name: str, cookie: Morsel[str] ) -> _MorselEntry: """Build the morsel to send with its length in the Cookie header.""" morsel = self._build_morsel(cookie) length = _encoded_length(f"{morsel.key}={morsel.coded_value}") assert length is not None # Checked before storing. return morsel, length, jar_key + (name,) def _touch_cookies(self, cookie_keys: Sequence[_CookieKey]) -> None: """Record one access generation for cookies used together, like Firefox.""" self._next_access_generation += 1 generation = self._next_access_generation generations = self._access_generations for cookie_key in cookie_keys: generations[cookie_key] = generation def _build_morsel(self, cookie: Morsel[str]) -> Morsel[str]: """Build a morsel for sending, respecting quote_cookie setting.""" if self._quote_cookie and cookie.coded_value and cookie.coded_value[0] == '"': return preserve_morsel_with_coded_value(cookie) morsel: Morsel[str] = Morsel() if self._quote_cookie: value, coded_value = _SIMPLE_COOKIE.value_encode(cookie.value) else: coded_value = value = cookie.value # We use __setstate__ instead of the public set() API because it allows us to # bypass validation and set already validated state. This is more stable than # setting protected attributes directly and unlikely to change since it would # break pickling. morsel.__setstate__({"key": cookie.key, "value": value, "coded_value": coded_value}) # type: ignore[attr-defined] return morsel @staticmethod def _is_domain_match(domain: str, hostname: str) -> bool: """Implements domain matching adhering to RFC 6265.""" if hostname == domain: return True if not hostname.endswith(domain): return False non_matching = hostname[: -len(domain)] if not non_matching.endswith("."): return False return not is_ip_address(hostname) @classmethod def _parse_date(cls, date_str: str) -> int | None: """Implements date string parsing adhering to RFC 6265.""" if not date_str: return None found_time = False found_day = False found_month = False found_year = False hour = minute = second = 0 day = 0 month = 0 year = 0 for token_match in cls.DATE_TOKENS_RE.finditer(date_str): token = token_match.group("token") if not found_time: time_match = cls.DATE_HMS_TIME_RE.match(token) if time_match: found_time = True hour, minute, second = (int(s) for s in time_match.groups()) continue if not found_day: day_match = cls.DATE_DAY_OF_MONTH_RE.match(token) if day_match: found_day = True day = int(day_match.group()) continue if not found_month: month_match = cls.DATE_MONTH_RE.match(token) if month_match: found_month = True assert month_match.lastindex is not None month = month_match.lastindex continue if not found_year: year_match = cls.DATE_YEAR_RE.match(token) if year_match: found_year = True year = int(year_match.group()) if 70 <= year <= 99: year += 1900 elif 0 <= year <= 69: year += 2000 if False in (found_day, found_month, found_year, found_time): return None if not 1 <= day <= 31: return None if year < 1601 or hour > 23 or minute > 59 or second > 59: return None return calendar.timegm((year, month, day, hour, minute, second, -1, -1, -1)) class _UnlimitedCookieJar(CookieJar): """Jar for per-request cookies, which the jar limits don't apply to.""" _limits_enabled = False def _morsel_entry( self, jar_key: _JarKey, name: str, cookie: Morsel[str] ) -> _MorselEntry: # The length and LRU key only serve the limits. These cookies skip the # storage check, so an unencodable one is left for the writer to report. return self._build_morsel(cookie), 0, _UNLIMITED_COOKIE_KEY class DummyCookieJar(AbstractCookieJar): """Implements a dummy cookie storage. It can be used with the ClientSession when no cookie processing is needed. """ def __init__(self, *, loop: asyncio.AbstractEventLoop | None = None) -> None: super().__init__(loop=loop) def __iter__(self) -> "Iterator[Morsel[str]]": while False: yield None def __len__(self) -> int: return 0 @property def unsafe(self) -> bool: return False @property def quote_cookie(self) -> bool: return True @property def cookies(self) -> MappingProxyType[tuple[str, str], SimpleCookie]: """Return an empty mapping.""" return MappingProxyType({}) @property def host_only_cookies(self) -> frozenset[tuple[str, str, str]]: """Return an empty frozenset.""" return frozenset() def clear(self, predicate: ClearCookiePredicate | None = None) -> None: pass def clear_domain(self, domain: str) -> None: pass def update_cookies(self, cookies: LooseCookies, response_url: URL = URL()) -> None: pass def update_cookies_from_headers( self, headers: Sequence[str], response_url: URL ) -> None: """Ignore raw Set-Cookie fields without parsing them.""" def filter_cookies(self, request_url: URL) -> "BaseCookie[str]": return SimpleCookie()