RomanNum3ral
/
scrapy
mirror of https://github.com/scrapy/scrapy.git
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

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().
This commit is contained in:
Andrey Rakhmatullin 2026-06-29 21:49:09 +05:00 committed by GitHub
parent 6591cb756c
commit 52147017b4
No known key found for this signature in database
GPG Key ID: B5690EEEBB952194
5 changed files with 116 additions and 116 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

32
scrapy/downloadermiddlewares/cookies.py
View File

@ -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))

2
scrapy/http/request/__init__.py
View File

@ -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")

73
scrapy/utils/request.py
View File

@ -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())

81
tests/test_downloadermiddleware_cookies.py
View File

@ -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 <GET http://example.org/1>:"
" {'value': 'bar', 'secure': False} ('name' is missing)",
),
(
"scrapy.downloadermiddlewares.cookies",
"scrapy.utils.request",
"WARNING",
"Invalid cookie found in request <GET http://example.org/2>:"
" {'name': 'foo', 'secure': False} ('value' is missing)",
),
(
"scrapy.downloadermiddlewares.cookies",
"scrapy.utils.request",
"WARNING",
"Invalid cookie found in request <GET http://example.org/3>:"
" {'name': 'foo', 'value': None, 'secure': False} ('value' is missing)",

44
tests/test_utils_request.py
View File

@ -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)