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]
[docs]
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<token>[\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)
[docs]
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)
[docs]
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)
[docs]
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)
[docs]
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
[docs]
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()
[docs]
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
[docs]
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()
