From dcf40f30637e8752c76781cf6703b5a236749a00 Mon Sep 17 00:00:00 2001 From: "patchback[bot]" <45432694+patchback[bot]@users.noreply.github.com> Date: Sun, 22 Feb 2026 15:53:43 +0000 Subject: [PATCH] [PR #12091/8a631e74 backport][3.14] Restrict pickle deserialization in CookieJar.load() (#12105) **This is a backport of PR #12091 as merged into master (8a631e74c1d266499dbc6bcdbc83c60f4ea3ee3c).** --------- Co-authored-by: Yuval Elbar <41901908+YuvalElbar6@users.noreply.github.com> Co-authored-by: Sam Bull --- CHANGES/12091.bugfix.rst | 6 ++ aiohttp/cookiejar.py | 114 +++++++++++++++++++++- docs/client_reference.rst | 18 +++- docs/spelling_wordlist.txt | 7 +- tests/conftest.py | 5 + tests/test_cookiejar.py | 189 +++++++++++++++++++++++++++++++++++++ 6 files changed, 331 insertions(+), 8 deletions(-) create mode 100644 CHANGES/12091.bugfix.rst Index: aiohttp-3.13.5/aiohttp/cookiejar.py =================================================================== --- aiohttp-3.13.5.orig/aiohttp/cookiejar.py +++ aiohttp-3.13.5/aiohttp/cookiejar.py @@ -4,6 +4,7 @@ import contextlib import datetime import heapq import itertools +import json import os # noqa import pathlib import pickle @@ -48,6 +49,41 @@ _MIN_SCHEDULED_COOKIE_EXPIRATION = 100 _SIMPLE_COOKIE = SimpleCookie() +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. + + 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.""" @@ -123,14 +159,84 @@ class CookieJar(AbstractCookieJar): return self._quote_cookie 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) - with file_path.open(mode="wb") as f: - pickle.dump(self._cookies, f, pickle.HIGHEST_PROTOCOL) + data: dict[str, dict[str, dict[str, str | bool]]] = {} + 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] = { + "key": morsel.key, + "value": morsel.value, + "coded_value": morsel.coded_value, + } + # Save all morsel attributes that have values + for attr in morsel._reserved: # type: ignore[attr-defined] + attr_val = morsel[attr] + if attr_val: + morsel_data[attr] = attr_val + data[key][name] = morsel_data + with file_path.open(mode="w", encoding="utf-8") 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. + + :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) - with file_path.open(mode="rb") as f: - self._cookies = pickle.load(f) + # Try JSON format first + try: + with file_path.open(mode="r", encoding="utf-8") as f: + data = json.load(f) + self._cookies = self._load_json_data(data) + except (json.JSONDecodeError, UnicodeDecodeError, ValueError): + # Fall back to legacy pickle format with restricted unpickler + with file_path.open(mode="rb") as f: + self._cookies = _RestrictedCookieUnpickler(f).load() + + def _load_json_data( + self, data: dict[str, dict[str, dict[str, str | bool]]] + ) -> defaultdict[tuple[str, str], SimpleCookie]: + """Load cookies from parsed JSON data.""" + cookies: defaultdict[tuple[str, str], SimpleCookie] = defaultdict(SimpleCookie) + for compound_key, cookie_data in data.items(): + domain, path = compound_key.split("|", 1) + key = (domain, path) + for name, morsel_data in cookie_data.items(): + morsel: Morsel[str] = Morsel() + morsel_key = morsel_data["key"] + morsel_value = morsel_data["value"] + morsel_coded_value = morsel_data["coded_value"] + # Use __setstate__ to bypass validation, same pattern + # used in _build_morsel and _cookie_helpers. + morsel.__setstate__( # type: ignore[attr-defined] + { + "key": morsel_key, + "value": morsel_value, + "coded_value": morsel_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] + cookies[key][name] = morsel + return cookies def clear(self, predicate: Optional[ClearCookiePredicate] = None) -> None: if predicate is None: Index: aiohttp-3.13.5/tests/conftest.py =================================================================== --- aiohttp-3.13.5.orig/tests/conftest.py +++ aiohttp-3.13.5/tests/conftest.py @@ -95,6 +95,11 @@ def blockbuster(request: pytest.FixtureR bb.functions[func].can_block_in( "aiohttp/web_urldispatcher.py", "add_static" ) + # save/load is not async, so we must allow this: + for func in ("io.TextIOWrapper.read", "io.BufferedReader.read"): + bb.functions[func].can_block_in("aiohttp/cookiejar.py", "load") + for func in ("io.TextIOWrapper.write", "io.BufferedWriter.write"): + bb.functions[func].can_block_in("aiohttp/cookiejar.py", "save") # Note: coverage.py uses locking internally which can cause false positives # in blockbuster when it instruments code. This is particularly problematic # on Windows where it can lead to flaky test failures. Index: aiohttp-3.13.5/tests/test_cookiejar.py =================================================================== --- aiohttp-3.13.5.orig/tests/test_cookiejar.py +++ aiohttp-3.13.5/tests/test_cookiejar.py @@ -9,6 +9,7 @@ import sys import unittest from http.cookies import BaseCookie, Morsel, SimpleCookie from operator import not_ +from pathlib import Path from typing import List, Set from unittest import mock @@ -1621,3 +1622,191 @@ async def test_shared_cookie_with_multip # Verify cache is reused efficiently assert ("", "") in jar._morsel_cache assert "universal" in jar._morsel_cache[("", "")] + + +# === Security tests for restricted unpickler and JSON save/load === + + +async def test_load_rejects_malicious_pickle(tmp_path: Path) -> None: + """Verify CookieJar.load() blocks arbitrary code execution via pickle. + + A crafted pickle payload using os.system (or any non-cookie class) + must be rejected by the restricted unpickler. + """ + import os + + file_path = tmp_path / "malicious.pkl" + + class RCEPayload: + def __reduce__(self) -> tuple[object, ...]: + return (os.system, ("echo PWNED",)) + + with open(file_path, "wb") as f: + pickle.dump(RCEPayload(), f, pickle.HIGHEST_PROTOCOL) + + jar = CookieJar() + with pytest.raises(pickle.UnpicklingError, match="Forbidden class"): + jar.load(file_path) + + +async def test_load_rejects_eval_payload(tmp_path: Path) -> None: + """Verify CookieJar.load() blocks eval-based pickle payloads.""" + file_path = tmp_path / "eval_payload.pkl" + + class EvalPayload: + def __reduce__(self) -> tuple[object, ...]: + return (eval, ("__import__('os').system('echo PWNED')",)) + + with open(file_path, "wb") as f: + pickle.dump(EvalPayload(), f, pickle.HIGHEST_PROTOCOL) + + jar = CookieJar() + with pytest.raises(pickle.UnpicklingError, match="Forbidden class"): + jar.load(file_path) + + +async def test_load_rejects_subprocess_payload(tmp_path: Path) -> None: + """Verify CookieJar.load() blocks subprocess-based pickle payloads.""" + import subprocess + + file_path = tmp_path / "subprocess_payload.pkl" + + class SubprocessPayload: + def __reduce__(self) -> tuple[object, ...]: + return (subprocess.call, (["echo", "PWNED"],)) + + with open(file_path, "wb") as f: + pickle.dump(SubprocessPayload(), f, pickle.HIGHEST_PROTOCOL) + + jar = CookieJar() + with pytest.raises(pickle.UnpicklingError, match="Forbidden class"): + jar.load(file_path) + + +async def test_load_falls_back_to_pickle( + tmp_path: Path, + cookies_to_receive: SimpleCookie, +) -> None: + """Verify load() falls back to restricted pickle for legacy cookie files. + + Existing cookie files saved with older versions of aiohttp used pickle. + load() should detect that the file is not JSON and fall back to the + restricted pickle unpickler for backward compatibility. + """ + file_path = tmp_path / "legit.pkl" + + # Write a legacy pickle file directly (as old aiohttp save() would) + jar_save = CookieJar() + jar_save.update_cookies(cookies_to_receive) + with file_path.open(mode="wb") as f: + pickle.dump(jar_save._cookies, f, pickle.HIGHEST_PROTOCOL) + + jar_load = CookieJar() + jar_load.load(file_path=file_path) + + jar_test = SimpleCookie() + for cookie in jar_load: + jar_test[cookie.key] = cookie + + assert jar_test == cookies_to_receive + + +async def test_save_load_json_roundtrip( + tmp_path: Path, + cookies_to_receive: SimpleCookie, +) -> None: + """Verify save/load roundtrip preserves cookies via JSON format.""" + file_path = tmp_path / "cookies.json" + + jar_save = CookieJar() + jar_save.update_cookies(cookies_to_receive) + jar_save.save(file_path=file_path) + + jar_load = CookieJar() + jar_load.load(file_path=file_path) + + saved_cookies = SimpleCookie() + for cookie in jar_save: + saved_cookies[cookie.key] = cookie + + loaded_cookies = SimpleCookie() + for cookie in jar_load: + loaded_cookies[cookie.key] = cookie + + assert saved_cookies == loaded_cookies + + +async def test_save_load_json_partitioned_cookies(tmp_path: Path) -> None: + """Verify save/load roundtrip works with partitioned cookies.""" + file_path = tmp_path / "partitioned.json" + + jar_save = CookieJar() + jar_save.update_cookies_from_headers( + ["session=cookie; Partitioned"], URL("https://example.com/") + ) + jar_save.save(file_path=file_path) + + jar_load = CookieJar() + jar_load.load(file_path=file_path) + + # Compare individual cookie values (same approach as test_save_load_partitioned_cookies) + saved = list(jar_save) + loaded = list(jar_load) + assert len(saved) == len(loaded) + for s, lo in zip(saved, loaded): + assert s.key == lo.key + assert s.value == lo.value + assert s["domain"] == lo["domain"] + assert s["path"] == lo["path"] + + +async def test_json_format_is_safe(tmp_path: Path) -> None: + """Verify the JSON file format cannot execute code on load.""" + import json + + file_path = tmp_path / "safe.json" + + # Write something that might look dangerous but is just data + malicious_data = { + "evil.com|/": { + "session": { + "key": "session", + "value": "__import__('os').system('echo PWNED')", + "coded_value": "__import__('os').system('echo PWNED')", + } + } + } + with open(file_path, "w") as f: + json.dump(malicious_data, f) + + jar = CookieJar() + jar.load(file_path=file_path) + + # The "malicious" string is just a cookie value, not executed code + cookies = list(jar) + assert len(cookies) == 1 + assert cookies[0].value == "__import__('os').system('echo PWNED')" + + +async def test_save_load_json_secure_cookies(tmp_path: Path) -> None: + """Verify save/load preserves Secure and HttpOnly flags.""" + file_path = tmp_path / "secure.json" + + jar_save = CookieJar() + jar_save.update_cookies_from_headers( + ["token=abc123; Secure; HttpOnly; Path=/; Domain=example.com"], + URL("https://example.com/"), + ) + jar_save.save(file_path=file_path) + + jar_load = CookieJar() + jar_load.load(file_path=file_path) + + loaded_cookies = list(jar_load) + assert len(loaded_cookies) == 1 + cookie = loaded_cookies[0] + assert cookie.key == "token" + assert cookie.value == "abc123" + assert cookie["secure"] is True + assert cookie["httponly"] is True + assert cookie["domain"] == "example.com"