Coverage for /pythoncovmergedfiles/medio/medio/src/aiohttp/aiohttp/cookiejar.py: 19%
255 statements
« prev ^ index » next coverage.py v7.3.1, created at 2023-09-27 06:09 +0000
1import contextlib
2import datetime
3import os # noqa
4import pathlib
5import pickle
6import re
7import warnings
8from collections import defaultdict
9from http.cookies import BaseCookie, Morsel, SimpleCookie
10from typing import ( # noqa
11 DefaultDict,
12 Dict,
13 Iterable,
14 Iterator,
15 List,
16 Mapping,
17 Optional,
18 Set,
19 Tuple,
20 Union,
21 cast,
22)
24from yarl import URL
26from .abc import AbstractCookieJar, ClearCookiePredicate
27from .helpers import is_ip_address, next_whole_second
28from .typedefs import LooseCookies, PathLike, StrOrURL
30__all__ = ("CookieJar", "DummyCookieJar")
33CookieItem = Union[str, "Morsel[str]"]
36class CookieJar(AbstractCookieJar):
37 """Implements cookie storage adhering to RFC 6265."""
39 DATE_TOKENS_RE = re.compile(
40 r"[\x09\x20-\x2F\x3B-\x40\x5B-\x60\x7B-\x7E]*"
41 r"(?P<token>[\x00-\x08\x0A-\x1F\d:a-zA-Z\x7F-\xFF]+)"
42 )
44 DATE_HMS_TIME_RE = re.compile(r"(\d{1,2}):(\d{1,2}):(\d{1,2})")
46 DATE_DAY_OF_MONTH_RE = re.compile(r"(\d{1,2})")
48 DATE_MONTH_RE = re.compile(
49 "(jan)|(feb)|(mar)|(apr)|(may)|(jun)|(jul)|" "(aug)|(sep)|(oct)|(nov)|(dec)",
50 re.I,
51 )
53 DATE_YEAR_RE = re.compile(r"(\d{2,4})")
55 MAX_TIME = datetime.datetime.max.replace(tzinfo=datetime.timezone.utc)
57 MAX_32BIT_TIME = datetime.datetime.fromtimestamp(2**31 - 1, datetime.timezone.utc)
59 def __init__(
60 self,
61 *,
62 unsafe: bool = False,
63 quote_cookie: bool = True,
64 treat_as_secure_origin: Union[StrOrURL, List[StrOrURL], None] = None
65 ) -> None:
66 self._cookies: DefaultDict[Tuple[str, str], SimpleCookie[str]] = defaultdict(
67 SimpleCookie
68 )
69 self._host_only_cookies: Set[Tuple[str, str]] = set()
70 self._unsafe = unsafe
71 self._quote_cookie = quote_cookie
72 if treat_as_secure_origin is None:
73 treat_as_secure_origin = []
74 elif isinstance(treat_as_secure_origin, URL):
75 treat_as_secure_origin = [treat_as_secure_origin.origin()]
76 elif isinstance(treat_as_secure_origin, str):
77 treat_as_secure_origin = [URL(treat_as_secure_origin).origin()]
78 else:
79 treat_as_secure_origin = [
80 URL(url).origin() if isinstance(url, str) else url.origin()
81 for url in treat_as_secure_origin
82 ]
83 self._treat_as_secure_origin = treat_as_secure_origin
84 self._next_expiration = next_whole_second()
85 self._expirations: Dict[Tuple[str, str, str], datetime.datetime] = {}
86 # #4515: datetime.max may not be representable on 32-bit platforms
87 self._max_time = self.MAX_TIME
88 try:
89 self._max_time.timestamp()
90 except OverflowError:
91 self._max_time = self.MAX_32BIT_TIME
93 def save(self, file_path: PathLike) -> None:
94 file_path = pathlib.Path(file_path)
95 with file_path.open(mode="wb") as f:
96 pickle.dump(self._cookies, f, pickle.HIGHEST_PROTOCOL)
98 def load(self, file_path: PathLike) -> None:
99 file_path = pathlib.Path(file_path)
100 with file_path.open(mode="rb") as f:
101 self._cookies = pickle.load(f)
103 def clear(self, predicate: Optional[ClearCookiePredicate] = None) -> None:
104 if predicate is None:
105 self._next_expiration = next_whole_second()
106 self._cookies.clear()
107 self._host_only_cookies.clear()
108 self._expirations.clear()
109 return
111 to_del = []
112 now = datetime.datetime.now(datetime.timezone.utc)
113 for (domain, path), cookie in self._cookies.items():
114 for name, morsel in cookie.items():
115 key = (domain, path, name)
116 if (
117 key in self._expirations and self._expirations[key] <= now
118 ) or predicate(morsel):
119 to_del.append(key)
121 for domain, path, name in to_del:
122 self._host_only_cookies.discard((domain, name))
123 key = (domain, path, name)
124 if key in self._expirations:
125 del self._expirations[(domain, path, name)]
126 self._cookies[(domain, path)].pop(name, None)
128 next_expiration = min(self._expirations.values(), default=self._max_time)
129 try:
130 self._next_expiration = next_expiration.replace(
131 microsecond=0
132 ) + datetime.timedelta(seconds=1)
133 except OverflowError:
134 self._next_expiration = self._max_time
136 def clear_domain(self, domain: str) -> None:
137 self.clear(lambda x: self._is_domain_match(domain, x["domain"]))
139 def __iter__(self) -> "Iterator[Morsel[str]]":
140 self._do_expiration()
141 for val in self._cookies.values():
142 yield from val.values()
144 def __len__(self) -> int:
145 return sum(1 for i in self)
147 def _do_expiration(self) -> None:
148 self.clear(lambda x: False)
150 def _expire_cookie(
151 self, when: datetime.datetime, domain: str, path: str, name: str
152 ) -> None:
153 self._next_expiration = min(self._next_expiration, when)
154 self._expirations[(domain, path, name)] = when
156 def update_cookies(self, cookies: LooseCookies, response_url: URL = URL()) -> None:
157 """Update cookies."""
158 hostname = response_url.raw_host
160 if not self._unsafe and is_ip_address(hostname):
161 # Don't accept cookies from IPs
162 return
164 if isinstance(cookies, Mapping):
165 cookies = cookies.items()
167 for name, cookie in cookies:
168 if not isinstance(cookie, Morsel):
169 tmp: SimpleCookie[str] = SimpleCookie()
170 tmp[name] = cookie # type: ignore[assignment]
171 cookie = tmp[name]
173 domain = cookie["domain"]
175 # ignore domains with trailing dots
176 if domain.endswith("."):
177 domain = ""
178 del cookie["domain"]
180 if not domain and hostname is not None:
181 # Set the cookie's domain to the response hostname
182 # and set its host-only-flag
183 self._host_only_cookies.add((hostname, name))
184 domain = cookie["domain"] = hostname
186 if domain.startswith("."):
187 # Remove leading dot
188 domain = domain[1:]
189 cookie["domain"] = domain
191 if hostname and not self._is_domain_match(domain, hostname):
192 # Setting cookies for different domains is not allowed
193 continue
195 path = cookie["path"]
196 if not path or not path.startswith("/"):
197 # Set the cookie's path to the response path
198 path = response_url.path
199 if not path.startswith("/"):
200 path = "/"
201 else:
202 # Cut everything from the last slash to the end
203 path = "/" + path[1 : path.rfind("/")]
204 cookie["path"] = path
206 max_age = cookie["max-age"]
207 if max_age:
208 try:
209 delta_seconds = int(max_age)
210 try:
211 max_age_expiration = datetime.datetime.now(
212 datetime.timezone.utc
213 ) + datetime.timedelta(seconds=delta_seconds)
214 except OverflowError:
215 max_age_expiration = self._max_time
216 self._expire_cookie(max_age_expiration, domain, path, name)
217 except ValueError:
218 cookie["max-age"] = ""
220 else:
221 expires = cookie["expires"]
222 if expires:
223 expire_time = self._parse_date(expires)
224 if expire_time:
225 self._expire_cookie(expire_time, domain, path, name)
226 else:
227 cookie["expires"] = ""
229 self._cookies[(domain, path)][name] = cookie
231 self._do_expiration()
233 def filter_cookies(
234 self, request_url: URL = URL()
235 ) -> Union["BaseCookie[str]", "SimpleCookie[str]"]:
236 """Returns this jar's cookies filtered by their attributes."""
237 self._do_expiration()
238 if not isinstance(request_url, URL):
239 warnings.warn(
240 "The method accepts yarl.URL instances only, got {}".format(
241 type(request_url)
242 ),
243 DeprecationWarning,
244 )
245 request_url = URL(request_url)
246 filtered: Union["SimpleCookie[str]", "BaseCookie[str]"] = (
247 SimpleCookie() if self._quote_cookie else BaseCookie()
248 )
249 hostname = request_url.raw_host or ""
250 request_origin = URL()
251 with contextlib.suppress(ValueError):
252 request_origin = request_url.origin()
254 is_not_secure = (
255 request_url.scheme not in ("https", "wss")
256 and request_origin not in self._treat_as_secure_origin
257 )
259 # Point 2: https://www.rfc-editor.org/rfc/rfc6265.html#section-5.4
260 for cookie in sorted(self, key=lambda c: len(c["path"])):
261 name = cookie.key
262 domain = cookie["domain"]
264 # Send shared cookies
265 if not domain:
266 filtered[name] = cookie.value
267 continue
269 if not self._unsafe and is_ip_address(hostname):
270 continue
272 if (domain, name) in self._host_only_cookies:
273 if domain != hostname:
274 continue
275 elif not self._is_domain_match(domain, hostname):
276 continue
278 if not self._is_path_match(request_url.path, cookie["path"]):
279 continue
281 if is_not_secure and cookie["secure"]:
282 continue
284 # It's critical we use the Morsel so the coded_value
285 # (based on cookie version) is preserved
286 mrsl_val = cast("Morsel[str]", cookie.get(cookie.key, Morsel()))
287 mrsl_val.set(cookie.key, cookie.value, cookie.coded_value)
288 filtered[name] = mrsl_val
290 return filtered
292 @staticmethod
293 def _is_domain_match(domain: str, hostname: str) -> bool:
294 """Implements domain matching adhering to RFC 6265."""
295 if hostname == domain:
296 return True
298 if not hostname.endswith(domain):
299 return False
301 non_matching = hostname[: -len(domain)]
303 if not non_matching.endswith("."):
304 return False
306 return not is_ip_address(hostname)
308 @staticmethod
309 def _is_path_match(req_path: str, cookie_path: str) -> bool:
310 """Implements path matching adhering to RFC 6265."""
311 if not req_path.startswith("/"):
312 req_path = "/"
314 if req_path == cookie_path:
315 return True
317 if not req_path.startswith(cookie_path):
318 return False
320 if cookie_path.endswith("/"):
321 return True
323 non_matching = req_path[len(cookie_path) :]
325 return non_matching.startswith("/")
327 @classmethod
328 def _parse_date(cls, date_str: str) -> Optional[datetime.datetime]:
329 """Implements date string parsing adhering to RFC 6265."""
330 if not date_str:
331 return None
333 found_time = False
334 found_day = False
335 found_month = False
336 found_year = False
338 hour = minute = second = 0
339 day = 0
340 month = 0
341 year = 0
343 for token_match in cls.DATE_TOKENS_RE.finditer(date_str):
344 token = token_match.group("token")
346 if not found_time:
347 time_match = cls.DATE_HMS_TIME_RE.match(token)
348 if time_match:
349 found_time = True
350 hour, minute, second = (int(s) for s in time_match.groups())
351 continue
353 if not found_day:
354 day_match = cls.DATE_DAY_OF_MONTH_RE.match(token)
355 if day_match:
356 found_day = True
357 day = int(day_match.group())
358 continue
360 if not found_month:
361 month_match = cls.DATE_MONTH_RE.match(token)
362 if month_match:
363 found_month = True
364 assert month_match.lastindex is not None
365 month = month_match.lastindex
366 continue
368 if not found_year:
369 year_match = cls.DATE_YEAR_RE.match(token)
370 if year_match:
371 found_year = True
372 year = int(year_match.group())
374 if 70 <= year <= 99:
375 year += 1900
376 elif 0 <= year <= 69:
377 year += 2000
379 if False in (found_day, found_month, found_year, found_time):
380 return None
382 if not 1 <= day <= 31:
383 return None
385 if year < 1601 or hour > 23 or minute > 59 or second > 59:
386 return None
388 return datetime.datetime(
389 year, month, day, hour, minute, second, tzinfo=datetime.timezone.utc
390 )
393class DummyCookieJar(AbstractCookieJar):
394 """Implements a dummy cookie storage.
396 It can be used with the ClientSession when no cookie processing is needed.
398 """
400 def __iter__(self) -> "Iterator[Morsel[str]]":
401 while False:
402 yield None
404 def __len__(self) -> int:
405 return 0
407 def clear(self, predicate: Optional[ClearCookiePredicate] = None) -> None:
408 pass
410 def clear_domain(self, domain: str) -> None:
411 pass
413 def update_cookies(self, cookies: LooseCookies, response_url: URL = URL()) -> None:
414 pass
416 def filter_cookies(self, request_url: URL) -> "BaseCookie[str]":
417 return SimpleCookie()