From 52147017b43baff1f9b4668e46ab11341aa8f400 Mon Sep 17 00:00:00 2001 From: Andrey Rakhmatullin Date: Mon, 29 Jun 2026 21:49:09 +0500 Subject: [PATCH] Cleanup cookie handling in request_to_curl() (#7684) * Adjust CookiesT. * Drop list of plain cookie dicts support from request_to_curl(). * Extract _decode_cookie(). * Unify logging. * Extract _to_verbose_cookies(). * Sync and type hint _cookie_to_set_cookie_value() in tests. * Add tests for bytes in Request.cookies. * Type hint assertCookieValEqual(). --- scrapy/downloadermiddlewares/cookies.py | 32 ++------- scrapy/http/request/__init__.py | 2 +- scrapy/utils/request.py | 73 +++++++++++++------ tests/test_downloadermiddleware_cookies.py | 81 +++++++++++++--------- tests/test_utils_request.py | 44 +++--------- 5 files changed, 116 insertions(+), 116 deletions(-) diff --git a/scrapy/downloadermiddlewares/cookies.py b/scrapy/downloadermiddlewares/cookies.py index cd8c2abca..34d071fd5 100644 --- a/scrapy/downloadermiddlewares/cookies.py +++ b/scrapy/downloadermiddlewares/cookies.py @@ -12,6 +12,7 @@ from scrapy.http.cookies import CookieJar from scrapy.utils.decorators import _warn_spider_arg from scrapy.utils.httpobj import urlparse_cached from scrapy.utils.python import to_unicode +from scrapy.utils.request import _decode_cookie, _to_verbose_cookies if TYPE_CHECKING: from collections.abc import Iterable, Sequence @@ -134,29 +135,10 @@ class CookiesMiddleware: Given a dict consisting of cookie components, return its string representation. Decode from bytes if necessary. """ - decoded = {} + decoded = _decode_cookie(cookie, request) + if decoded is None: + return None flags = set() - for key in ("name", "value", "path", "domain"): - value = cookie.get(key) - if value is None: - if key in {"name", "value"}: - msg = f"Invalid cookie found in request {request}: {cookie} ('{key}' is missing)" - logger.warning(msg) - return None - continue - if isinstance(value, (bool, float, int, str)): - decoded[key] = str(value) - else: - assert isinstance(value, bytes) - try: - decoded[key] = value.decode("utf8") - except UnicodeDecodeError: - logger.warning( - "Non UTF-8 encoded cookie found in request %s: %s", - request, - cookie, - ) - decoded[key] = value.decode("latin1", errors="replace") for flag in ("secure",): value = cookie.get(flag, _UNSET) if value is _UNSET or not value: @@ -177,11 +159,7 @@ class CookiesMiddleware: """ if not request.cookies: return () - cookies: Iterable[VerboseCookie] - if isinstance(request.cookies, dict): - cookies = tuple({"name": k, "value": v} for k, v in request.cookies.items()) - else: - cookies = request.cookies + cookies: Iterable[VerboseCookie] = _to_verbose_cookies(request.cookies) for cookie in cookies: cookie.setdefault("secure", urlparse_cached(request).scheme == "https") formatted = filter(None, (self._format_cookie(c, request) for c in cookies)) diff --git a/scrapy/http/request/__init__.py b/scrapy/http/request/__init__.py index 7db648a45..73c2e7dd4 100644 --- a/scrapy/http/request/__init__.py +++ b/scrapy/http/request/__init__.py @@ -51,7 +51,7 @@ class VerboseCookie(TypedDict): secure: NotRequired[bool] -CookiesT: TypeAlias = dict[str, str] | list[VerboseCookie] +CookiesT: TypeAlias = dict[str | bytes, str | bytes] | list[VerboseCookie] RequestTypeVar = TypeVar("RequestTypeVar", bound="Request") diff --git a/scrapy/utils/request.py b/scrapy/utils/request.py index 4a85526c0..b2fd9a834 100644 --- a/scrapy/utils/request.py +++ b/scrapy/utils/request.py @@ -7,6 +7,7 @@ from __future__ import annotations import hashlib import json +import logging from typing import TYPE_CHECKING, Any, Protocol from urllib.parse import urlunparse from weakref import WeakKeyDictionary @@ -25,6 +26,9 @@ if TYPE_CHECKING: from typing_extensions import Self from scrapy.crawler import Crawler + from scrapy.http.request import CookiesT, VerboseCookie + +logger = logging.getLogger(__name__) _fingerprint_cache: WeakKeyDictionary[ @@ -179,17 +183,52 @@ def _get_method(obj: Any, name: Any) -> Any: raise ValueError(f"Method {name!r} not found in: {obj}") from None -def _cookie_value_to_unicode(value: str | bytes | float) -> str: - if isinstance(value, bytes): - return value.decode() - return str(value) +def _to_verbose_cookies(cookies: CookiesT) -> list[VerboseCookie]: + """Return a list of verbose cookies from ``request.cookies``. + + The list of dicts form is returned as is, the dict one is converted first. + """ + if isinstance(cookies, dict): + return [{"name": k, "value": v} for k, v in cookies.items()] + return cookies + + +def _decode_cookie(cookie: VerboseCookie, request: Request) -> dict[str, str] | None: + """Return a dict with non-flag verbose cookie values converted to strings. + + ``name``, ``value``, ``path``, ``domain`` are included, ``secure`` isn't. + """ + + decoded = {} + for key in ("name", "value", "path", "domain"): + value = cookie.get(key) + if value is None: + if key in {"name", "value"}: + logger.warning( + f"Invalid cookie found in request {request}:" + f" {cookie} ('{key}' is missing)" + ) + return None + continue + if isinstance(value, (bool, float, int, str)): + decoded[key] = str(value) + else: + assert isinstance(value, bytes) + try: + decoded[key] = value.decode("utf8") + except UnicodeDecodeError: + logger.warning( + f"Non UTF-8 encoded cookie found in request {request}: {cookie}", + ) + decoded[key] = value.decode("latin1", errors="replace") + return decoded def request_to_curl(request: Request) -> str: """ Converts a :class:`~scrapy.Request` object to a curl command. - :param :class:`~scrapy.Request`: Request object to be converted + :param request: Request object to be converted :return: string containing the curl command """ method = request.method @@ -201,22 +240,14 @@ def request_to_curl(request: Request) -> str: ) url = request.url - cookies = "" - if request.cookies: - if isinstance(request.cookies, dict): - cookie = "; ".join( - f"{_cookie_value_to_unicode(k)}={_cookie_value_to_unicode(v)}" - for k, v in request.cookies.items() - ) - cookies = f"--cookie '{cookie}'" - elif isinstance(request.cookies, list): - cookie = "; ".join( - f"{_cookie_value_to_unicode(c['name'])}={_cookie_value_to_unicode(c['value'])}" - if "name" in c and "value" in c - else f"{_cookie_value_to_unicode(next(iter(c.keys())))}={_cookie_value_to_unicode(next(iter(c.values())))}" - for c in request.cookies - ) - cookies = f"--cookie '{cookie}'" + + cookie_list: list[VerboseCookie] = _to_verbose_cookies(request.cookies) + pairs = [ + f"{decoded['name']}={decoded['value']}" + for c in cookie_list + if (decoded := _decode_cookie(c, request)) is not None + ] + cookies = f"--cookie '{'; '.join(pairs)}'" if pairs else "" curl_cmd = f"curl -X {method} {url} {data} {headers} {cookies}".strip() return " ".join(curl_cmd.split()) diff --git a/tests/test_downloadermiddleware_cookies.py b/tests/test_downloadermiddleware_cookies.py index f79591020..b4419dc67 100644 --- a/tests/test_downloadermiddleware_cookies.py +++ b/tests/test_downloadermiddleware_cookies.py @@ -1,4 +1,5 @@ import logging +from collections.abc import Iterable import pytest from testfixtures import LogCapture @@ -8,30 +9,34 @@ from scrapy.downloadermiddlewares.defaultheaders import DefaultHeadersMiddleware from scrapy.downloadermiddlewares.redirect import RedirectMiddleware from scrapy.exceptions import NotConfigured from scrapy.http import Request, Response +from scrapy.http.request import CookiesT, VerboseCookie from scrapy.utils.python import to_bytes +from scrapy.utils.request import _to_verbose_cookies from scrapy.utils.spider import DefaultSpider from scrapy.utils.test import get_crawler UNSET = object() -def _cookie_to_set_cookie_value(cookie): +def _cookie_to_set_cookie_value(cookie: VerboseCookie) -> str | None: """Given a cookie defined as a dictionary with name and value keys, and optional path and domain keys, return the equivalent string that can be associated to a ``Set-Cookie`` header.""" decoded = {} for key in ("name", "value", "path", "domain"): - if cookie.get(key) is None: - if key in ("name", "value"): + value = cookie.get(key) + if value is None: + if key in {"name", "value"}: return None continue - if isinstance(cookie[key], (bool, float, int, str)): - decoded[key] = str(cookie[key]) + if isinstance(value, (bool, float, int, str)): + decoded[key] = str(value) else: + assert isinstance(value, bytes) try: - decoded[key] = cookie[key].decode("utf8") + decoded[key] = value.decode("utf8") except UnicodeDecodeError: - decoded[key] = cookie[key].decode("latin1", errors="replace") + decoded[key] = value.decode("latin1", errors="replace") cookie_str = f"{decoded.pop('name')}={decoded.pop('value')}" for key, value in decoded.items(): # path, domain @@ -39,24 +44,30 @@ def _cookie_to_set_cookie_value(cookie): return cookie_str -def _cookies_to_set_cookie_list(cookies): +def _cookies_to_set_cookie_list(cookies: CookiesT) -> Iterable[str]: """Given a group of cookie defined either as a dictionary or as a list of dictionaries (i.e. in a format supported by the cookies parameter of Request), return the equivalent list of strings that can be associated to a ``Set-Cookie`` header.""" if not cookies: return [] - if isinstance(cookies, dict): - cookies = ({"name": k, "value": v} for k, v in cookies.items()) - return filter(None, (_cookie_to_set_cookie_value(cookie) for cookie in cookies)) + return filter( + None, + ( + _cookie_to_set_cookie_value(cookie) + for cookie in _to_verbose_cookies(cookies) + ), + ) class TestCookiesMiddleware: - def assertCookieValEqual(self, first, second, msg=None): - def split_cookies(cookies): + @staticmethod + def assertCookieValEqual(first: bytes | str | None, second: bytes | str) -> None: + def split_cookies(cookies: bytes | str) -> list[bytes]: return sorted([s.strip() for s in to_bytes(cookies).split(b";")]) - assert split_cookies(first) == split_cookies(second), msg + assert first is not None + assert split_cookies(first) == split_cookies(second) def setup_method(self): crawler = get_crawler(DefaultSpider) @@ -372,21 +383,25 @@ class TestCookiesMiddleware: assert self.mw.process_request(req3) is None self.assertCookieValEqual(req3.headers["Cookie"], "a=new; c=d; e=f") - def test_request_cookies_encoding(self): - # 1) UTF8-encoded bytes - req1 = Request("http://example.org", cookies={"a": "á".encode()}) - assert self.mw.process_request(req1) is None - self.assertCookieValEqual(req1.headers["Cookie"], b"a=\xc3\xa1") - - # 2) Non UTF8-encoded bytes - req2 = Request("http://example.org", cookies={"a": "á".encode("latin1")}) - assert self.mw.process_request(req2) is None - self.assertCookieValEqual(req2.headers["Cookie"], b"a=\xc3\xa1") - - # 3) String - req3 = Request("http://example.org", cookies={"a": "á"}) - assert self.mw.process_request(req3) is None - self.assertCookieValEqual(req3.headers["Cookie"], b"a=\xc3\xa1") + @pytest.mark.parametrize( + "cookies", + [ + # UTF8-encoded bytes + {"a": "á".encode()}, + # non UTF8-encoded bytes + {"a": "á".encode("latin1")}, + # string + {"a": "á"}, + # key as bytes + {b"a": "á"}, + # key and value as bytes + {b"a": "á".encode()}, + ], + ) + def test_request_cookies_encoding(self, cookies: CookiesT) -> None: + req = Request("http://example.org", cookies=cookies) + assert self.mw.process_request(req) is None + self.assertCookieValEqual(req.headers["Cookie"], b"a=\xc3\xa1") @pytest.mark.xfail(reason="Cookie header is not currently being processed") def test_request_headers_cookie_encoding(self): @@ -410,7 +425,7 @@ class TestCookiesMiddleware: Invalid cookies are logged as warnings and discarded """ with LogCapture( - "scrapy.downloadermiddlewares.cookies", + "scrapy.utils.request", propagate=False, level=logging.INFO, ) as lc: @@ -425,19 +440,19 @@ class TestCookiesMiddleware: assert self.mw.process_request(req3) is None lc.check( ( - "scrapy.downloadermiddlewares.cookies", + "scrapy.utils.request", "WARNING", "Invalid cookie found in request :" " {'value': 'bar', 'secure': False} ('name' is missing)", ), ( - "scrapy.downloadermiddlewares.cookies", + "scrapy.utils.request", "WARNING", "Invalid cookie found in request :" " {'name': 'foo', 'secure': False} ('value' is missing)", ), ( - "scrapy.downloadermiddlewares.cookies", + "scrapy.utils.request", "WARNING", "Invalid cookie found in request :" " {'name': 'foo', 'value': None, 'secure': False} ('value' is missing)", diff --git a/tests/test_utils_request.py b/tests/test_utils_request.py index 63926e9d2..1642a932b 100644 --- a/tests/test_utils_request.py +++ b/tests/test_utils_request.py @@ -354,16 +354,18 @@ class TestCustomRequestFingerprinter: class TestRequestToCurl: - def _test_request(self, request_object, expected_curl_command): + def _test_request( + self, request_object: Request, expected_curl_command: str + ) -> None: curl_command = request_to_curl(request_object) assert curl_command == expected_curl_command - def test_get(self): + def test_get(self) -> None: request_object = Request("https://www.example.com") expected_curl_command = "curl -X GET https://www.example.com" self._test_request(request_object, expected_curl_command) - def test_post(self): + def test_post(self) -> None: request_object = Request( "https://www.httpbin.org/post", method="POST", @@ -374,7 +376,7 @@ class TestRequestToCurl: ) self._test_request(request_object, expected_curl_command) - def test_headers(self): + def test_headers(self) -> None: request_object = Request( "https://www.httpbin.org/post", method="POST", @@ -388,7 +390,7 @@ class TestRequestToCurl: ) self._test_request(request_object, expected_curl_command) - def test_cookies_dict(self): + def test_cookies_dict(self) -> None: request_object = Request( "https://www.httpbin.org/post", method="POST", @@ -401,7 +403,7 @@ class TestRequestToCurl: ) self._test_request(request_object, expected_curl_command) - def test_cookies_dict_bytes(self): + def test_cookies_dict_bytes(self) -> None: request_object = Request( "https://www.httpbin.org/post", method="POST", @@ -414,20 +416,7 @@ class TestRequestToCurl: ) self._test_request(request_object, expected_curl_command) - def test_cookies_list(self): - request_object = Request( - "https://www.httpbin.org/post", - method="POST", - cookies=[{"foo": "bar"}], - body=json.dumps({"foo": "bar"}), - ) - expected_curl_command = ( - "curl -X POST https://www.httpbin.org/post" - " --data-raw '{\"foo\": \"bar\"}' --cookie 'foo=bar'" - ) - self._test_request(request_object, expected_curl_command) - - def test_cookies_list_verbose(self): + def test_cookies_list_verbose(self) -> None: request_object = Request( "https://www.httpbin.org/post", method="POST", @@ -448,7 +437,7 @@ class TestRequestToCurl: ) self._test_request(request_object, expected_curl_command) - def test_cookies_list_verbose_non_string_value(self): + def test_cookies_list_verbose_non_string_value(self) -> None: request_object = Request( "https://www.httpbin.org/post", method="POST", @@ -468,16 +457,3 @@ class TestRequestToCurl: " --data-raw '{\"foo\": \"bar\"}' --cookie 'foo=1'" ) self._test_request(request_object, expected_curl_command) - - def test_cookies_list_bytes_nonstandard_key(self): - request_object = Request( - "https://www.httpbin.org/post", - method="POST", - cookies=[{b"foo": b"bar"}], - body=json.dumps({"foo": "bar"}), - ) - expected_curl_command = ( - "curl -X POST https://www.httpbin.org/post" - " --data-raw '{\"foo\": \"bar\"}' --cookie 'foo=bar'" - ) - self._test_request(request_object, expected_curl_command)