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

Merge remote-tracking branch 'scrapy/2.11' into 2.11-port

This commit is contained in:
Adrián Chaves 2024-02-15 06:39:50 +01:00
parent 1c9d308acc 2f1d345e74
commit 6bd45bb6d9
25 changed files with 1019 additions and 117 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

2
.bumpversion.cfg
View File

@ -1,5 +1,5 @@
[bumpversion]
current_version = 2.11.0
current_version = 2.11.1
commit = True
tag = True
tag_name = {new_version}

10
docs/faq.rst
View File

@ -297,9 +297,13 @@ build the DOM of the entire feed in memory, and this can be quite slow and
consume a lot of memory.
In order to avoid parsing all the entire feed at once in memory, you can use
the functions ``xmliter`` and ``csviter`` from ``scrapy.utils.iterators``
module. In fact, this is what the feed spiders (see :ref:`topics-spiders`) use
under the cover.
the :func:`~scrapy.utils.iterators.xmliter_lxml` and
:func:`~scrapy.utils.iterators.csviter` functions. In fact, this is what
:class:`~scrapy.spiders.XMLFeedSpider` uses.
.. autofunction:: scrapy.utils.iterators.xmliter_lxml
.. autofunction:: scrapy.utils.iterators.csviter
Does Scrapy manage cookies automatically?
-----------------------------------------

131
docs/news.rst
View File

@ -3,6 +3,105 @@
Release notes
=============
.. _release-2.11.1:
Scrapy 2.11.1 (2024-02-14)
--------------------------
Highlights:
- Security bug fixes.
- Support for Twisted >= 23.8.0.
- Documentation improvements.
Security bug fixes
~~~~~~~~~~~~~~~~~~
- Addressed `ReDoS vulnerabilities`_:
- ``scrapy.utils.iterators.xmliter`` is now deprecated in favor of
:func:`~scrapy.utils.iterators.xmliter_lxml`, which
:class:`~scrapy.spiders.XMLFeedSpider` now uses.
To minimize the impact of this change on existing code,
:func:`~scrapy.utils.iterators.xmliter_lxml` now supports indicating
the node namespace with a prefix in the node name, and big files with
highly nested trees when using libxml2 2.7+.
- Fixed regular expressions in the implementation of the
:func:`~scrapy.utils.response.open_in_browser` function.
Please, see the `cc65-xxvf-f7r9 security advisory`_ for more information.
.. _ReDoS vulnerabilities: https://owasp.org/www-community/attacks/Regular_expression_Denial_of_Service_-_ReDoS
.. _cc65-xxvf-f7r9 security advisory: https://github.com/scrapy/scrapy/security/advisories/GHSA-cc65-xxvf-f7r9
- :setting:`DOWNLOAD_MAXSIZE` and :setting:`DOWNLOAD_WARNSIZE` now also apply
to the decompressed response body. Please, see the `7j7m-v7m3-jqm7 security
advisory`_ for more information.
.. _7j7m-v7m3-jqm7 security advisory: https://github.com/scrapy/scrapy/security/advisories/GHSA-7j7m-v7m3-jqm7
- Also in relation with the `7j7m-v7m3-jqm7 security advisory`_, the
deprecated ``scrapy.downloadermiddlewares.decompression`` module has been
removed.
- The ``Authorization`` header is now dropped on redirects to a different
domain. Please, see the `cw9j-q3vf-hrrv security advisory`_ for more
information.
.. _cw9j-q3vf-hrrv security advisory: https://github.com/scrapy/scrapy/security/advisories/GHSA-cw9j-q3vf-hrrv
Modified requirements
~~~~~~~~~~~~~~~~~~~~~
- The Twisted dependency is no longer restricted to < 23.8.0. (:issue:`6024`,
:issue:`6064`, :issue:`6142`)
Bug fixes
~~~~~~~~~
- The OS signal handling code was refactored to no longer use private Twisted
functions. (:issue:`6024`, :issue:`6064`, :issue:`6112`)
Documentation
~~~~~~~~~~~~~
- Improved documentation for :class:`~scrapy.crawler.Crawler` initialization
changes made in the 2.11.0 release. (:issue:`6057`, :issue:`6147`)
- Extended documentation for :attr:`Request.meta <scrapy.http.Request.meta>`.
(:issue:`5565`)
- Fixed the :reqmeta:`dont_merge_cookies` documentation. (:issue:`5936`,
:issue:`6077`)
- Added a link to Zyte's export guides to the :ref:`feed exports
<topics-feed-exports>` documentation. (:issue:`6183`)
- Added a missing note about backward-incompatible changes in
:class:`~scrapy.exporters.PythonItemExporter` to the 2.11.0 release notes.
(:issue:`6060`, :issue:`6081`)
- Added a missing note about removing the deprecated
``scrapy.utils.boto.is_botocore()`` function to the 2.8.0 release notes.
(:issue:`6056`, :issue:`6061`)
- Other documentation improvements. (:issue:`6128`, :issue:`6144`,
:issue:`6163`, :issue:`6190`, :issue:`6192`)
Quality assurance
~~~~~~~~~~~~~~~~~
- Added Python 3.12 to the CI configuration, re-enabled tests that were
disabled when the pre-release support was added. (:issue:`5985`,
:issue:`6083`, :issue:`6098`)
- Fixed a test issue on PyPy 7.3.14. (:issue:`6204`, :issue:`6205`)
.. _release-2.11.0:
Scrapy 2.11.0 (2023-09-18)
@ -2877,6 +2976,38 @@ affect subclasses:
(:issue:`3884`)
.. _release-1.8.4:
Scrapy 1.8.4 (2024-02-14)
-------------------------
**Security bug fixes:**
- Due to its `ReDoS vulnerabilities`_, ``scrapy.utils.iterators.xmliter`` is
now deprecated in favor of :func:`~scrapy.utils.iterators.xmliter_lxml`,
which :class:`~scrapy.spiders.XMLFeedSpider` now uses.
To minimize the impact of this change on existing code,
:func:`~scrapy.utils.iterators.xmliter_lxml` now supports indicating
the node namespace as a prefix in the node name, and big files with highly
nested trees when using libxml2 2.7+.
Please, see the `cc65-xxvf-f7r9 security advisory`_ for more information.
- :setting:`DOWNLOAD_MAXSIZE` and :setting:`DOWNLOAD_WARNSIZE` now also apply
to the decompressed response body. Please, see the `7j7m-v7m3-jqm7 security
advisory`_ for more information.
- Also in relation with the `7j7m-v7m3-jqm7 security advisory`_, use of the
``scrapy.downloadermiddlewares.decompression`` module is discouraged and
will trigger a warning.
- The ``Authorization`` header is now dropped on redirects to a different
domain. Please, see the `cw9j-q3vf-hrrv security advisory`_ for more
information.
.. _cw9j-q3vf-hrrv security advisory: https://github.com/scrapy/scrapy/security/advisories/GHSA-cw9j-q3vf-hrrv
.. _release-1.8.3:

18
docs/topics/debug.rst
View File

@ -125,25 +125,15 @@ Fortunately, the :command:`shell` is your bread and butter in this case (see
See also: :ref:`topics-shell-inspect-response`.
Open in browser
===============
Sometimes you just want to see how a certain response looks in a browser, you
can use the ``open_in_browser`` function for that. Here is an example of how
you would use it:
can use the :func:`~scrapy.utils.response.open_in_browser` function for that:
.. code-block:: python
.. autofunction:: scrapy.utils.response.open_in_browser
from scrapy.utils.response import open_in_browser
def parse_details(self, response):
if "item name" not in response.body:
open_in_browser(response)
``open_in_browser`` will open a browser with the response received by Scrapy at
that point, adjusting the `base tag`_ so that images and styles are displayed
properly.
Logging
=======
@ -163,8 +153,6 @@ available in all future runs should they be necessary again:
For more information, check the :ref:`topics-logging` section.
.. _base tag: https://www.w3schools.com/tags/tag_base.asp
.. _debug-vscode:
Visual Studio Code

1
docs/topics/request-response.rst
View File

@ -677,6 +677,7 @@ Those are:
* :reqmeta:`download_fail_on_dataloss`
* :reqmeta:`download_latency`
* :reqmeta:`download_maxsize`
* :reqmeta:`download_warnsize`
* :reqmeta:`download_timeout`
* ``ftp_password`` (See :setting:`FTP_PASSWORD` for more info)
* ``ftp_user`` (See :setting:`FTP_USER` for more info)

36
docs/topics/settings.rst
View File

@ -873,40 +873,42 @@ The amount of time (in secs) that the downloader will wait before timing out.
Request.meta key.
.. setting:: DOWNLOAD_MAXSIZE
.. reqmeta:: download_maxsize
DOWNLOAD_MAXSIZE
----------------
Default: ``1073741824`` (1024MB)
Default: ``1073741824`` (1 GiB)
The maximum response size (in bytes) that downloader will download.
The maximum response body size (in bytes) allowed. Bigger responses are
aborted and ignored.
If you want to disable it set to 0.
This applies both before and after compression. If decompressing a response
body would exceed this limit, decompression is aborted and the response is
ignored.
.. reqmeta:: download_maxsize
Use ``0`` to disable this limit.
.. note::
This size can be set per spider using :attr:`download_maxsize`
spider attribute and per-request using :reqmeta:`download_maxsize`
Request.meta key.
This limit can be set per spider using the :attr:`download_maxsize` spider
attribute and per request using the :reqmeta:`download_maxsize` Request.meta
key.
.. setting:: DOWNLOAD_WARNSIZE
.. reqmeta:: download_warnsize
DOWNLOAD_WARNSIZE
-----------------
Default: ``33554432`` (32MB)
Default: ``33554432`` (32 MiB)
The response size (in bytes) that downloader will start to warn.
If the size of a response exceeds this value, before or after compression, a
warning will be logged about it.
If you want to disable it set to 0.
Use ``0`` to disable this limit.
.. note::
This size can be set per spider using :attr:`download_warnsize`
spider attribute and per-request using :reqmeta:`download_warnsize`
Request.meta key.
This limit can be set per spider using the :attr:`download_warnsize` spider
attribute and per request using the :reqmeta:`download_warnsize` Request.meta
key.
.. setting:: DOWNLOAD_FAIL_ON_DATALOSS

2
scrapy/VERSION
View File

@ -1 +1 @@
2.11.0
2.11.1

109
scrapy/downloadermiddlewares/httpcompression.py
View File

@ -1,50 +1,92 @@
from __future__ import annotations
import io
import zlib
import warnings
from logging import getLogger
from typing import TYPE_CHECKING, List, Optional, Union
from scrapy import Request, Spider
from scrapy import Request, Spider, signals
from scrapy.crawler import Crawler
from scrapy.exceptions import NotConfigured
from scrapy.exceptions import IgnoreRequest, NotConfigured
from scrapy.http import Response, TextResponse
from scrapy.responsetypes import responsetypes
from scrapy.statscollectors import StatsCollector
from scrapy.utils._compression import (
_DecompressionMaxSizeExceeded,
_inflate,
_unbrotli,
_unzstd,
)
from scrapy.utils.deprecate import ScrapyDeprecationWarning
from scrapy.utils.gz import gunzip
if TYPE_CHECKING:
# typing.Self requires Python 3.11
from typing_extensions import Self
logger = getLogger(__name__)
ACCEPTED_ENCODINGS: List[bytes] = [b"gzip", b"deflate"]
try:
import brotli
ACCEPTED_ENCODINGS.append(b"br")
import brotli # noqa: F401
except ImportError:
pass
else:
ACCEPTED_ENCODINGS.append(b"br")
try:
import zstandard
ACCEPTED_ENCODINGS.append(b"zstd")
import zstandard # noqa: F401
except ImportError:
pass
else:
ACCEPTED_ENCODINGS.append(b"zstd")
class HttpCompressionMiddleware:
"""This middleware allows compressed (gzip, deflate) traffic to be
sent/received from web sites"""
def __init__(self, stats: Optional[StatsCollector] = None):
self.stats = stats
def __init__(
self,
stats: Optional[StatsCollector] = None,
*,
crawler: Optional[Crawler] = None,
):
if not crawler:
self.stats = stats
self._max_size = 1073741824
self._warn_size = 33554432
return
self.stats = crawler.stats
self._max_size = crawler.settings.getint("DOWNLOAD_MAXSIZE")
self._warn_size = crawler.settings.getint("DOWNLOAD_WARNSIZE")
crawler.signals.connect(self.open_spider, signals.spider_opened)
@classmethod
def from_crawler(cls, crawler: Crawler) -> Self:
if not crawler.settings.getbool("COMPRESSION_ENABLED"):
raise NotConfigured
return cls(stats=crawler.stats)
try:
return cls(crawler=crawler)
except TypeError:
warnings.warn(
"HttpCompressionMiddleware subclasses must either modify "
"their '__init__' method to support a 'crawler' parameter or "
"reimplement their 'from_crawler' method.",
ScrapyDeprecationWarning,
)
mw = cls()
mw.stats = crawler.stats
mw._max_size = crawler.settings.getint("DOWNLOAD_MAXSIZE")
mw._warn_size = crawler.settings.getint("DOWNLOAD_WARNSIZE")
crawler.signals.connect(mw.open_spider, signals.spider_opened)
return mw
def open_spider(self, spider):
if hasattr(spider, "download_maxsize"):
self._max_size = spider.download_maxsize
if hasattr(spider, "download_warnsize"):
self._warn_size = spider.download_warnsize
def process_request(
self, request: Request, spider: Spider
@ -61,7 +103,24 @@ class HttpCompressionMiddleware:
content_encoding = response.headers.getlist("Content-Encoding")
if content_encoding:
encoding = content_encoding.pop()
decoded_body = self._decode(response.body, encoding.lower())
max_size = request.meta.get("download_maxsize", self._max_size)
warn_size = request.meta.get("download_warnsize", self._warn_size)
try:
decoded_body = self._decode(
response.body, encoding.lower(), max_size
)
except _DecompressionMaxSizeExceeded:
raise IgnoreRequest(
f"Ignored response {response} because its body "
f"({len(response.body)} B) exceeded DOWNLOAD_MAXSIZE "
f"({max_size} B) during decompression."
)
if len(response.body) < warn_size <= len(decoded_body):
logger.warning(
f"{response} body size after decompression "
f"({len(decoded_body)} B) is larger than the "
f"download warning size ({warn_size} B)."
)
if self.stats:
self.stats.inc_value(
"httpcompression/response_bytes",
@ -85,25 +144,13 @@ class HttpCompressionMiddleware:
return response
def _decode(self, body: bytes, encoding: bytes) -> bytes:
def _decode(self, body: bytes, encoding: bytes, max_size: int) -> bytes:
if encoding == b"gzip" or encoding == b"x-gzip":
body = gunzip(body)
return gunzip(body, max_size=max_size)
if encoding == b"deflate":
try:
body = zlib.decompress(body)
except zlib.error:
# ugly hack to work with raw deflate content that may
# be sent by microsoft servers. For more information, see:
# http://carsten.codimi.de/gzip.yaws/
# http://www.port80software.com/200ok/archive/2005/10/31/868.aspx
# http://www.gzip.org/zlib/zlib_faq.html#faq38
body = zlib.decompress(body, -15)
return _inflate(body, max_size=max_size)
if encoding == b"br" and b"br" in ACCEPTED_ENCODINGS:
body = brotli.decompress(body)
return _unbrotli(body, max_size=max_size)
if encoding == b"zstd" and b"zstd" in ACCEPTED_ENCODINGS:
# Using its streaming API since its simple API could handle only cases
# where there is content size data embedded in the frame
reader = zstandard.ZstdDecompressor().stream_reader(io.BytesIO(body))
body = reader.read()
return _unzstd(body, max_size=max_size)
return body

10
scrapy/downloadermiddlewares/redirect.py
View File

@ -29,11 +29,17 @@ def _build_redirect_request(
**kwargs,
cookies=None,
)
if "Cookie" in redirect_request.headers:
has_cookie_header = "Cookie" in redirect_request.headers
has_authorization_header = "Authorization" in redirect_request.headers
if has_cookie_header or has_authorization_header:
source_request_netloc = urlparse_cached(source_request).netloc
redirect_request_netloc = urlparse_cached(redirect_request).netloc
if source_request_netloc != redirect_request_netloc:
del redirect_request.headers["Cookie"]
if has_cookie_header:
del redirect_request.headers["Cookie"]
# https://fetch.spec.whatwg.org/#ref-for-cors-non-wildcard-request-header-name
if has_authorization_header:
del redirect_request.headers["Authorization"]
return redirect_request

4
scrapy/spiders/feed.py
View File

@ -7,7 +7,7 @@ See documentation in docs/topics/spiders.rst
from scrapy.exceptions import NotConfigured, NotSupported
from scrapy.selector import Selector
from scrapy.spiders import Spider
from scrapy.utils.iterators import csviter, xmliter
from scrapy.utils.iterators import csviter, xmliter_lxml
from scrapy.utils.spider import iterate_spider_output
@ -83,7 +83,7 @@ class XMLFeedSpider(Spider):
return self.parse_nodes(response, nodes)
def _iternodes(self, response):
for node in xmliter(response, self.itertag):
for node in xmliter_lxml(response, self.itertag):
self._register_namespaces(node)
yield node

35
scrapy/spiders/sitemap.py
View File

@ -1,11 +1,19 @@
import logging
import re
from typing import TYPE_CHECKING, Any
from scrapy.http import Request, XmlResponse
from scrapy.spiders import Spider
from scrapy.utils._compression import _DecompressionMaxSizeExceeded
from scrapy.utils.gz import gunzip, gzip_magic_number
from scrapy.utils.sitemap import Sitemap, sitemap_urls_from_robots
if TYPE_CHECKING:
# typing.Self requires Python 3.11
from typing_extensions import Self
from scrapy.crawler import Crawler
logger = logging.getLogger(__name__)
@ -14,6 +22,19 @@ class SitemapSpider(Spider):
sitemap_rules = [("", "parse")]
sitemap_follow = [""]
sitemap_alternate_links = False
_max_size: int
_warn_size: int
@classmethod
def from_crawler(cls, crawler: "Crawler", *args: Any, **kwargs: Any) -> "Self":
spider = super().from_crawler(crawler, *args, **kwargs)
spider._max_size = getattr(
spider, "download_maxsize", spider.settings.getint("DOWNLOAD_MAXSIZE")
)
spider._warn_size = getattr(
spider, "download_warnsize", spider.settings.getint("DOWNLOAD_WARNSIZE")
)
return spider
def __init__(self, *a, **kw):
super().__init__(*a, **kw)
@ -70,7 +91,19 @@ class SitemapSpider(Spider):
if isinstance(response, XmlResponse):
return response.body
if gzip_magic_number(response):
return gunzip(response.body)
uncompressed_size = len(response.body)
max_size = response.meta.get("download_maxsize", self._max_size)
warn_size = response.meta.get("download_warnsize", self._warn_size)
try:
body = gunzip(response.body, max_size=max_size)
except _DecompressionMaxSizeExceeded:
return None
if uncompressed_size < warn_size <= len(body):
logger.warning(
f"{response} body size after decompression ({len(body)} B) "
f"is larger than the download warning size ({warn_size} B)."
)
return body
# actual gzipped sitemap files are decompressed above ;
# if we are here (response body is not gzipped)
# and have a response for .xml.gz,

94
scrapy/utils/_compression.py Normal file
View File

@ -0,0 +1,94 @@
import zlib
from io import BytesIO
try:
import brotli
except ImportError:
pass
try:
import zstandard
except ImportError:
pass
_CHUNK_SIZE = 65536 # 64 KiB
class _DecompressionMaxSizeExceeded(ValueError):
pass
def _inflate(data: bytes, *, max_size: int = 0) -> bytes:
decompressor = zlib.decompressobj()
raw_decompressor = zlib.decompressobj(wbits=-15)
input_stream = BytesIO(data)
output_stream = BytesIO()
output_chunk = b"."
decompressed_size = 0
while output_chunk:
input_chunk = input_stream.read(_CHUNK_SIZE)
try:
output_chunk = decompressor.decompress(input_chunk)
except zlib.error:
if decompressor != raw_decompressor:
# ugly hack to work with raw deflate content that may
# be sent by microsoft servers. For more information, see:
# http://carsten.codimi.de/gzip.yaws/
# http://www.port80software.com/200ok/archive/2005/10/31/868.aspx
# http://www.gzip.org/zlib/zlib_faq.html#faq38
decompressor = raw_decompressor
output_chunk = decompressor.decompress(input_chunk)
else:
raise
decompressed_size += len(output_chunk)
if max_size and decompressed_size > max_size:
raise _DecompressionMaxSizeExceeded(
f"The number of bytes decompressed so far "
f"({decompressed_size} B) exceed the specified maximum "
f"({max_size} B)."
)
output_stream.write(output_chunk)
output_stream.seek(0)
return output_stream.read()
def _unbrotli(data: bytes, *, max_size: int = 0) -> bytes:
decompressor = brotli.Decompressor()
input_stream = BytesIO(data)
output_stream = BytesIO()
output_chunk = b"."
decompressed_size = 0
while output_chunk:
input_chunk = input_stream.read(_CHUNK_SIZE)
output_chunk = decompressor.process(input_chunk)
decompressed_size += len(output_chunk)
if max_size and decompressed_size > max_size:
raise _DecompressionMaxSizeExceeded(
f"The number of bytes decompressed so far "
f"({decompressed_size} B) exceed the specified maximum "
f"({max_size} B)."
)
output_stream.write(output_chunk)
output_stream.seek(0)
return output_stream.read()
def _unzstd(data: bytes, *, max_size: int = 0) -> bytes:
decompressor = zstandard.ZstdDecompressor()
stream_reader = decompressor.stream_reader(BytesIO(data))
output_stream = BytesIO()
output_chunk = b"."
decompressed_size = 0
while output_chunk:
output_chunk = stream_reader.read(_CHUNK_SIZE)
decompressed_size += len(output_chunk)
if max_size and decompressed_size > max_size:
raise _DecompressionMaxSizeExceeded(
f"The number of bytes decompressed so far "
f"({decompressed_size} B) exceed the specified maximum "
f"({max_size} B)."
)
output_stream.write(output_chunk)
output_stream.seek(0)
return output_stream.read()

26
scrapy/utils/gz.py
View File

@ -1,31 +1,41 @@
import struct
from gzip import GzipFile
from io import BytesIO
from typing import List
from scrapy.http import Response
from ._compression import _CHUNK_SIZE, _DecompressionMaxSizeExceeded
def gunzip(data: bytes) -> bytes:
def gunzip(data: bytes, *, max_size: int = 0) -> bytes:
"""Gunzip the given data and return as much data as possible.
This is resilient to CRC checksum errors.
"""
f = GzipFile(fileobj=BytesIO(data))
output_list: List[bytes] = []
output_stream = BytesIO()
chunk = b"."
decompressed_size = 0
while chunk:
try:
chunk = f.read1(8196)
output_list.append(chunk)
chunk = f.read1(_CHUNK_SIZE)
except (OSError, EOFError, struct.error):
# complete only if there is some data, otherwise re-raise
# see issue 87 about catching struct.error
# some pages are quite small so output_list is empty
if output_list:
# some pages are quite small so output_stream is empty
if output_stream.getbuffer().nbytes > 0:
break
raise
return b"".join(output_list)
decompressed_size += len(chunk)
if max_size and decompressed_size > max_size:
raise _DecompressionMaxSizeExceeded(
f"The number of bytes decompressed so far "
f"({decompressed_size} B) exceed the specified maximum "
f"({max_size} B)."
)
output_stream.write(chunk)
output_stream.seek(0)
return output_stream.read()
def gzip_magic_number(response: Response) -> bool:

41
scrapy/utils/iterators.py
View File

@ -16,7 +16,11 @@ from typing import (
cast,
overload,
)
from warnings import warn
from lxml import etree
from scrapy.exceptions import ScrapyDeprecationWarning
from scrapy.http import Response, TextResponse
from scrapy.selector import Selector
from scrapy.utils.python import re_rsearch, to_unicode
@ -38,6 +42,16 @@ def xmliter(
- a unicode string
- a string encoded as utf-8
"""
warn(
(
"xmliter is deprecated and its use strongly discouraged because "
"it is vulnerable to ReDoS attacks. Use xmliter_lxml instead. See "
"https://github.com/scrapy/scrapy/security/advisories/GHSA-cc65-xxvf-f7r9"
),
ScrapyDeprecationWarning,
stacklevel=2,
)
nodename_patt = re.escape(nodename)
DOCUMENT_HEADER_RE = re.compile(r"<\?xml[^>]+>\s*", re.S)
@ -81,15 +95,34 @@ def xmliter_lxml(
namespace: Optional[str] = None,
prefix: str = "x",
) -> Generator[Selector, Any, None]:
from lxml import etree
reader = _StreamReader(obj)
tag = f"{{{namespace}}}{nodename}" if namespace else nodename
iterable = etree.iterparse(
cast("SupportsReadClose[bytes]", reader), tag=tag, encoding=reader.encoding
cast("SupportsReadClose[bytes]", reader),
encoding=reader.encoding,
events=("end", "start-ns"),
huge_tree=True,
)
selxpath = "//" + (f"{prefix}:{nodename}" if namespace else nodename)
for _, node in iterable:
needs_namespace_resolution = not namespace and ":" in nodename
if needs_namespace_resolution:
prefix, nodename = nodename.split(":", maxsplit=1)
for event, data in iterable:
if event == "start-ns":
assert isinstance(data, tuple)
if needs_namespace_resolution:
_prefix, _namespace = data
if _prefix != prefix:
continue
namespace = _namespace
needs_namespace_resolution = False
selxpath = f"//{prefix}:{nodename}"
tag = f"{{{namespace}}}{nodename}"
continue
assert isinstance(data, etree._Element)
node = data
if node.tag != tag:
continue
nodetext = etree.tostring(node, encoding="unicode")
node.clear()
xs = Selector(text=nodetext, type="xml")

35
scrapy/utils/response.py
View File

@ -54,6 +54,18 @@ def response_status_message(status: Union[bytes, float, int, str]) -> str:
return f"{status_int} {to_unicode(message)}"
def _remove_html_comments(body):
start = body.find(b"<!--")
while start != -1:
end = body.find(b"-->", start + 1)
if end == -1:
return body[:start]
else:
body = body[:start] + body[end + 3 :]
start = body.find(b"<!--")
return body
def open_in_browser(
response: Union[
"scrapy.http.response.html.HtmlResponse",
@ -61,8 +73,21 @@ def open_in_browser(
],
_openfunc: Callable[[str], Any] = webbrowser.open,
) -> Any:
"""Open the given response in a local web browser, populating the <base>
tag for external links to work
"""Open *response* in a local web browser, adjusting the `base tag`_ for
external links to work, e.g. so that images and styles are displayed.
.. _base tag: https://www.w3schools.com/tags/tag_base.asp
For example:
.. code-block:: python
from scrapy.utils.response import open_in_browser
def parse_details(self, response):
if "item name" not in response.body:
open_in_browser(response)
"""
from scrapy.http import HtmlResponse, TextResponse
@ -70,9 +95,9 @@ def open_in_browser(
body = response.body
if isinstance(response, HtmlResponse):
if b"<base" not in body:
repl = rf'\1<base href="{response.url}">'
body = re.sub(b"<!--.*?-->", b"", body, flags=re.DOTALL)
body = re.sub(rb"(<head(?:>|\s.*?>))", to_bytes(repl), body)
_remove_html_comments(body)
repl = rf'\0<base href="{response.url}">'
body = re.sub(rb"<head(?:[^<>]*?>)", to_bytes(repl), body, count=1)
ext = ".html"
elif isinstance(response, TextResponse):
ext = ".txt"

2
tests/sample_data/compressed/bomb-br.bin Normal file
View File

@ -0,0 +1,2 @@
<EFBFBD>;§¯ר<C2AF>”n<E2809D>×Vp SmoY2·ו
ן(ה)-׀´=_o

BIN
tests/sample_data/compressed/bomb-deflate.bin Normal file
View File

Binary file not shown.

BIN
tests/sample_data/compressed/bomb-gzip.bin Normal file
View File

Binary file not shown.

BIN
tests/sample_data/compressed/bomb-zstd.bin Normal file
View File

Binary file not shown.

302
tests/test_downloadermiddleware_httpcompression.py
View File

@ -1,15 +1,18 @@
from gzip import GzipFile
from io import BytesIO
from logging import WARNING
from pathlib import Path
from unittest import SkipTest, TestCase
from warnings import catch_warnings
from testfixtures import LogCapture
from w3lib.encoding import resolve_encoding
from scrapy.downloadermiddlewares.httpcompression import (
ACCEPTED_ENCODINGS,
HttpCompressionMiddleware,
)
from scrapy.exceptions import NotConfigured
from scrapy.exceptions import IgnoreRequest, NotConfigured, ScrapyDeprecationWarning
from scrapy.http import HtmlResponse, Request, Response
from scrapy.responsetypes import responsetypes
from scrapy.spiders import Spider
@ -34,6 +37,15 @@ FORMAT = {
"html-zstd-streaming-no-content-size.bin",
"zstd",
),
**{
f"bomb-{format_id}": (f"bomb-{format_id}.bin", format_id)
for format_id in (
"br", # 34 → 11 511 612
"deflate", # 27 968 → 11 511 612
"gzip", # 27 988 → 11 511 612
"zstd", # 1 096 → 11 511 612
)
},
}
@ -114,18 +126,6 @@ class HttpCompressionTest(TestCase):
self.assertStatsEqual("httpcompression/response_count", 1)
self.assertStatsEqual("httpcompression/response_bytes", 74837)
def test_process_response_gzip_no_stats(self):
mw = HttpCompressionMiddleware()
response = self._getresponse("gzip")
request = response.request
self.assertEqual(response.headers["Content-Encoding"], b"gzip")
newresponse = mw.process_response(request, response, self.spider)
self.assertEqual(mw.stats, None)
assert newresponse is not response
assert newresponse.body.startswith(b"<!DOCTYPE")
assert "Content-Encoding" not in newresponse.headers
def test_process_response_br(self):
try:
import brotli # noqa: F401
@ -371,3 +371,279 @@ class HttpCompressionTest(TestCase):
self.assertEqual(response.body, b"")
self.assertStatsEqual("httpcompression/response_count", None)
self.assertStatsEqual("httpcompression/response_bytes", None)
def _test_compression_bomb_setting(self, compression_id):
settings = {"DOWNLOAD_MAXSIZE": 10_000_000}
crawler = get_crawler(Spider, settings_dict=settings)
spider = crawler._create_spider("scrapytest.org")
mw = HttpCompressionMiddleware.from_crawler(crawler)
mw.open_spider(spider)
response = self._getresponse(f"bomb-{compression_id}")
self.assertRaises(
IgnoreRequest,
mw.process_response,
response.request,
response,
spider,
)
def test_compression_bomb_setting_br(self):
try:
import brotli # noqa: F401
except ImportError:
raise SkipTest("no brotli")
self._test_compression_bomb_setting("br")
def test_compression_bomb_setting_deflate(self):
self._test_compression_bomb_setting("deflate")
def test_compression_bomb_setting_gzip(self):
self._test_compression_bomb_setting("gzip")
def test_compression_bomb_setting_zstd(self):
try:
import zstandard # noqa: F401
except ImportError:
raise SkipTest("no zstd support (zstandard)")
self._test_compression_bomb_setting("zstd")
def _test_compression_bomb_spider_attr(self, compression_id):
class DownloadMaxSizeSpider(Spider):
download_maxsize = 10_000_000
crawler = get_crawler(DownloadMaxSizeSpider)
spider = crawler._create_spider("scrapytest.org")
mw = HttpCompressionMiddleware.from_crawler(crawler)
mw.open_spider(spider)
response = self._getresponse(f"bomb-{compression_id}")
self.assertRaises(
IgnoreRequest,
mw.process_response,
response.request,
response,
spider,
)
def test_compression_bomb_spider_attr_br(self):
try:
import brotli # noqa: F401
except ImportError:
raise SkipTest("no brotli")
self._test_compression_bomb_spider_attr("br")
def test_compression_bomb_spider_attr_deflate(self):
self._test_compression_bomb_spider_attr("deflate")
def test_compression_bomb_spider_attr_gzip(self):
self._test_compression_bomb_spider_attr("gzip")
def test_compression_bomb_spider_attr_zstd(self):
try:
import zstandard # noqa: F401
except ImportError:
raise SkipTest("no zstd support (zstandard)")
self._test_compression_bomb_spider_attr("zstd")
def _test_compression_bomb_request_meta(self, compression_id):
crawler = get_crawler(Spider)
spider = crawler._create_spider("scrapytest.org")
mw = HttpCompressionMiddleware.from_crawler(crawler)
mw.open_spider(spider)
response = self._getresponse(f"bomb-{compression_id}")
response.meta["download_maxsize"] = 10_000_000
self.assertRaises(
IgnoreRequest,
mw.process_response,
response.request,
response,
spider,
)
def test_compression_bomb_request_meta_br(self):
try:
import brotli # noqa: F401
except ImportError:
raise SkipTest("no brotli")
self._test_compression_bomb_request_meta("br")
def test_compression_bomb_request_meta_deflate(self):
self._test_compression_bomb_request_meta("deflate")
def test_compression_bomb_request_meta_gzip(self):
self._test_compression_bomb_request_meta("gzip")
def test_compression_bomb_request_meta_zstd(self):
try:
import zstandard # noqa: F401
except ImportError:
raise SkipTest("no zstd support (zstandard)")
self._test_compression_bomb_request_meta("zstd")
def _test_download_warnsize_setting(self, compression_id):
settings = {"DOWNLOAD_WARNSIZE": 10_000_000}
crawler = get_crawler(Spider, settings_dict=settings)
spider = crawler._create_spider("scrapytest.org")
mw = HttpCompressionMiddleware.from_crawler(crawler)
mw.open_spider(spider)
response = self._getresponse(f"bomb-{compression_id}")
with LogCapture(
"scrapy.downloadermiddlewares.httpcompression",
propagate=False,
level=WARNING,
) as log:
mw.process_response(response.request, response, spider)
log.check(
(
"scrapy.downloadermiddlewares.httpcompression",
"WARNING",
(
"<200 http://scrapytest.org/> body size after "
"decompression (11511612 B) is larger than the download "
"warning size (10000000 B)."
),
),
)
def test_download_warnsize_setting_br(self):
try:
import brotli # noqa: F401
except ImportError:
raise SkipTest("no brotli")
self._test_download_warnsize_setting("br")
def test_download_warnsize_setting_deflate(self):
self._test_download_warnsize_setting("deflate")
def test_download_warnsize_setting_gzip(self):
self._test_download_warnsize_setting("gzip")
def test_download_warnsize_setting_zstd(self):
try:
import zstandard # noqa: F401
except ImportError:
raise SkipTest("no zstd support (zstandard)")
self._test_download_warnsize_setting("zstd")
def _test_download_warnsize_spider_attr(self, compression_id):
class DownloadWarnSizeSpider(Spider):
download_warnsize = 10_000_000
crawler = get_crawler(DownloadWarnSizeSpider)
spider = crawler._create_spider("scrapytest.org")
mw = HttpCompressionMiddleware.from_crawler(crawler)
mw.open_spider(spider)
response = self._getresponse(f"bomb-{compression_id}")
with LogCapture(
"scrapy.downloadermiddlewares.httpcompression",
propagate=False,
level=WARNING,
) as log:
mw.process_response(response.request, response, spider)
log.check(
(
"scrapy.downloadermiddlewares.httpcompression",
"WARNING",
(
"<200 http://scrapytest.org/> body size after "
"decompression (11511612 B) is larger than the download "
"warning size (10000000 B)."
),
),
)
def test_download_warnsize_spider_attr_br(self):
try:
import brotli # noqa: F401
except ImportError:
raise SkipTest("no brotli")
self._test_download_warnsize_spider_attr("br")
def test_download_warnsize_spider_attr_deflate(self):
self._test_download_warnsize_spider_attr("deflate")
def test_download_warnsize_spider_attr_gzip(self):
self._test_download_warnsize_spider_attr("gzip")
def test_download_warnsize_spider_attr_zstd(self):
try:
import zstandard # noqa: F401
except ImportError:
raise SkipTest("no zstd support (zstandard)")
self._test_download_warnsize_spider_attr("zstd")
def _test_download_warnsize_request_meta(self, compression_id):
crawler = get_crawler(Spider)
spider = crawler._create_spider("scrapytest.org")
mw = HttpCompressionMiddleware.from_crawler(crawler)
mw.open_spider(spider)
response = self._getresponse(f"bomb-{compression_id}")
response.meta["download_warnsize"] = 10_000_000
with LogCapture(
"scrapy.downloadermiddlewares.httpcompression",
propagate=False,
level=WARNING,
) as log:
mw.process_response(response.request, response, spider)
log.check(
(
"scrapy.downloadermiddlewares.httpcompression",
"WARNING",
(
"<200 http://scrapytest.org/> body size after "
"decompression (11511612 B) is larger than the download "
"warning size (10000000 B)."
),
),
)
def test_download_warnsize_request_meta_br(self):
try:
import brotli # noqa: F401
except ImportError:
raise SkipTest("no brotli")
self._test_download_warnsize_request_meta("br")
def test_download_warnsize_request_meta_deflate(self):
self._test_download_warnsize_request_meta("deflate")
def test_download_warnsize_request_meta_gzip(self):
self._test_download_warnsize_request_meta("gzip")
def test_download_warnsize_request_meta_zstd(self):
try:
import zstandard # noqa: F401
except ImportError:
raise SkipTest("no zstd support (zstandard)")
self._test_download_warnsize_request_meta("zstd")
class HttpCompressionSubclassTest(TestCase):
def test_init_missing_stats(self):
class HttpCompressionMiddlewareSubclass(HttpCompressionMiddleware):
def __init__(self):
super().__init__()
crawler = get_crawler(Spider)
with catch_warnings(record=True) as caught_warnings:
HttpCompressionMiddlewareSubclass.from_crawler(crawler)
messages = tuple(
str(warning.message)
for warning in caught_warnings
if warning.category is ScrapyDeprecationWarning
)
self.assertEqual(
messages,
(
(
"HttpCompressionMiddleware subclasses must either modify "
"their '__init__' method to support a 'crawler' parameter "
"or reimplement their 'from_crawler' method."
),
),
)

31
tests/test_downloadermiddleware_redirect.py
View File

@ -247,6 +247,37 @@ class RedirectMiddlewareTest(unittest.TestCase):
perc_encoded_utf8_url = "http://scrapytest.org/a%C3%A7%C3%A3o"
self.assertEqual(perc_encoded_utf8_url, req_result.url)
def test_cross_domain_header_dropping(self):
safe_headers = {"A": "B"}
original_request = Request(
"https://example.com",
headers={"Cookie": "a=b", "Authorization": "a", **safe_headers},
)
internal_response = Response(
"https://example.com",
headers={"Location": "https://example.com/a"},
status=301,
)
internal_redirect_request = self.mw.process_response(
original_request, internal_response, self.spider
)
self.assertIsInstance(internal_redirect_request, Request)
self.assertEqual(original_request.headers, internal_redirect_request.headers)
external_response = Response(
"https://example.com",
headers={"Location": "https://example.org/a"},
status=301,
)
external_redirect_request = self.mw.process_response(
original_request, external_response, self.spider
)
self.assertIsInstance(external_redirect_request, Request)
self.assertEqual(
safe_headers, external_redirect_request.headers.to_unicode_dict()
)
class MetaRefreshMiddlewareTest(unittest.TestCase):
def setUp(self):

128
tests/test_spider.py
View File

@ -2,6 +2,8 @@ import gzip
import inspect
import warnings
from io import BytesIO
from logging import WARNING
from pathlib import Path
from typing import Any
from unittest import mock
@ -25,7 +27,7 @@ from scrapy.spiders import (
)
from scrapy.spiders.init import InitSpider
from scrapy.utils.test import get_crawler
from tests import get_testdata
from tests import get_testdata, tests_datadir
class SpiderTest(unittest.TestCase):
@ -149,10 +151,10 @@ class XMLFeedSpiderTest(SpiderTest):
body = b"""<?xml version="1.0" encoding="UTF-8"?>
<urlset xmlns:x="http://www.google.com/schemas/sitemap/0.84"
xmlns:y="http://www.example.com/schemas/extras/1.0">
<url><x:loc>http://www.example.com/Special-Offers.html</loc><y:updated>2009-08-16</updated>
<url><x:loc>http://www.example.com/Special-Offers.html</x:loc><y:updated>2009-08-16</y:updated>
<other value="bar" y:custom="fuu"/>
</url>
<url><loc>http://www.example.com/</loc><y:updated>2009-08-16</updated><other value="foo"/></url>
<url><loc>http://www.example.com/</loc><y:updated>2009-08-16</y:updated><other value="foo"/></url>
</urlset>"""
response = XmlResponse(url="http://example.com/sitemap.xml", body=body)
@ -488,7 +490,8 @@ class SitemapSpiderTest(SpiderTest):
GZBODY = f.getvalue()
def assertSitemapBody(self, response, body):
spider = self.spider_class("example.com")
crawler = get_crawler()
spider = self.spider_class.from_crawler(crawler, "example.com")
self.assertEqual(spider._get_sitemap_body(response), body)
def test_get_sitemap_body(self):
@ -506,6 +509,7 @@ class SitemapSpiderTest(SpiderTest):
url="http://www.example.com/sitemap",
body=self.GZBODY,
headers={"content-type": "application/gzip"},
request=Request("http://www.example.com/sitemap"),
)
self.assertSitemapBody(r, self.BODY)
@ -514,7 +518,11 @@ class SitemapSpiderTest(SpiderTest):
self.assertSitemapBody(r, self.BODY)
def test_get_sitemap_body_xml_url_compressed(self):
r = Response(url="http://www.example.com/sitemap.xml.gz", body=self.GZBODY)
r = Response(
url="http://www.example.com/sitemap.xml.gz",
body=self.GZBODY,
request=Request("http://www.example.com/sitemap"),
)
self.assertSitemapBody(r, self.BODY)
# .xml.gz but body decoded by HttpCompression middleware already
@ -691,6 +699,116 @@ Sitemap: /sitemap-relative-url.xml
["http://www.example.com/sitemap2.xml"],
)
def test_compression_bomb_setting(self):
settings = {"DOWNLOAD_MAXSIZE": 10_000_000}
crawler = get_crawler(settings_dict=settings)
spider = self.spider_class.from_crawler(crawler, "example.com")
body_path = Path(tests_datadir, "compressed", "bomb-gzip.bin")
body = body_path.read_bytes()
request = Request(url="https://example.com")
response = Response(url="https://example.com", body=body, request=request)
self.assertIsNone(spider._get_sitemap_body(response))
def test_compression_bomb_spider_attr(self):
class DownloadMaxSizeSpider(self.spider_class):
download_maxsize = 10_000_000
crawler = get_crawler()
spider = DownloadMaxSizeSpider.from_crawler(crawler, "example.com")
body_path = Path(tests_datadir, "compressed", "bomb-gzip.bin")
body = body_path.read_bytes()
request = Request(url="https://example.com")
response = Response(url="https://example.com", body=body, request=request)
self.assertIsNone(spider._get_sitemap_body(response))
def test_compression_bomb_request_meta(self):
crawler = get_crawler()
spider = self.spider_class.from_crawler(crawler, "example.com")
body_path = Path(tests_datadir, "compressed", "bomb-gzip.bin")
body = body_path.read_bytes()
request = Request(
url="https://example.com", meta={"download_maxsize": 10_000_000}
)
response = Response(url="https://example.com", body=body, request=request)
self.assertIsNone(spider._get_sitemap_body(response))
def test_download_warnsize_setting(self):
settings = {"DOWNLOAD_WARNSIZE": 10_000_000}
crawler = get_crawler(settings_dict=settings)
spider = self.spider_class.from_crawler(crawler, "example.com")
body_path = Path(tests_datadir, "compressed", "bomb-gzip.bin")
body = body_path.read_bytes()
request = Request(url="https://example.com")
response = Response(url="https://example.com", body=body, request=request)
with LogCapture(
"scrapy.spiders.sitemap", propagate=False, level=WARNING
) as log:
spider._get_sitemap_body(response)
log.check(
(
"scrapy.spiders.sitemap",
"WARNING",
(
"<200 https://example.com> body size after decompression "
"(11511612 B) is larger than the download warning size "
"(10000000 B)."
),
),
)
def test_download_warnsize_spider_attr(self):
class DownloadWarnSizeSpider(self.spider_class):
download_warnsize = 10_000_000
crawler = get_crawler()
spider = DownloadWarnSizeSpider.from_crawler(crawler, "example.com")
body_path = Path(tests_datadir, "compressed", "bomb-gzip.bin")
body = body_path.read_bytes()
request = Request(
url="https://example.com", meta={"download_warnsize": 10_000_000}
)
response = Response(url="https://example.com", body=body, request=request)
with LogCapture(
"scrapy.spiders.sitemap", propagate=False, level=WARNING
) as log:
spider._get_sitemap_body(response)
log.check(
(
"scrapy.spiders.sitemap",
"WARNING",
(
"<200 https://example.com> body size after decompression "
"(11511612 B) is larger than the download warning size "
"(10000000 B)."
),
),
)
def test_download_warnsize_request_meta(self):
crawler = get_crawler()
spider = self.spider_class.from_crawler(crawler, "example.com")
body_path = Path(tests_datadir, "compressed", "bomb-gzip.bin")
body = body_path.read_bytes()
request = Request(
url="https://example.com", meta={"download_warnsize": 10_000_000}
)
response = Response(url="https://example.com", body=body, request=request)
with LogCapture(
"scrapy.spiders.sitemap", propagate=False, level=WARNING
) as log:
spider._get_sitemap_body(response)
log.check(
(
"scrapy.spiders.sitemap",
"WARNING",
(
"<200 https://example.com> body size after decompression "
"(11511612 B) is larger than the download warning size "
"(10000000 B)."
),
),
)
class DeprecationTest(unittest.TestCase):
def test_crawl_spider(self):

40
tests/test_utils_iterators.py
View File

@ -1,14 +1,14 @@
from pytest import mark
import pytest
from twisted.trial import unittest
from scrapy.exceptions import ScrapyDeprecationWarning
from scrapy.http import Response, TextResponse, XmlResponse
from scrapy.utils.iterators import _body_or_str, csviter, xmliter, xmliter_lxml
from tests import get_testdata
class XmliterTestCase(unittest.TestCase):
xmliter = staticmethod(xmliter)
class XmliterBaseTestCase:
@pytest.mark.filterwarnings("ignore::scrapy.exceptions.ScrapyDeprecationWarning")
def test_xmliter(self):
body = b"""
<?xml version="1.0" encoding="UTF-8"?>
@ -40,6 +40,7 @@ class XmliterTestCase(unittest.TestCase):
attrs, [("001", ["Name 1"], ["Type 1"]), ("002", ["Name 2"], ["Type 2"])]
)
@pytest.mark.filterwarnings("ignore::scrapy.exceptions.ScrapyDeprecationWarning")
def test_xmliter_unusual_node(self):
body = b"""<?xml version="1.0" encoding="UTF-8"?>
<root>
@ -53,6 +54,7 @@ class XmliterTestCase(unittest.TestCase):
]
self.assertEqual(nodenames, [["matchme..."]])
@pytest.mark.filterwarnings("ignore::scrapy.exceptions.ScrapyDeprecationWarning")
def test_xmliter_unicode(self):
# example taken from https://github.com/scrapy/scrapy/issues/1665
body = """<?xml version="1.0" encoding="UTF-8"?>
@ -112,6 +114,7 @@ class XmliterTestCase(unittest.TestCase):
[("26", ["-"], ["80"]), ("21", ["Ab"], ["76"]), ("27", ["A"], ["27"])],
)
@pytest.mark.filterwarnings("ignore::scrapy.exceptions.ScrapyDeprecationWarning")
def test_xmliter_text(self):
body = (
'<?xml version="1.0" encoding="UTF-8"?>'
@ -123,6 +126,7 @@ class XmliterTestCase(unittest.TestCase):
[["one"], ["two"]],
)
@pytest.mark.filterwarnings("ignore::scrapy.exceptions.ScrapyDeprecationWarning")
def test_xmliter_namespaces(self):
body = b"""
<?xml version="1.0" encoding="UTF-8"?>
@ -162,6 +166,7 @@ class XmliterTestCase(unittest.TestCase):
self.assertEqual(node.xpath("id/text()").getall(), [])
self.assertEqual(node.xpath("price/text()").getall(), [])
@pytest.mark.filterwarnings("ignore::scrapy.exceptions.ScrapyDeprecationWarning")
def test_xmliter_namespaced_nodename(self):
body = b"""
<?xml version="1.0" encoding="UTF-8"?>
@ -190,6 +195,7 @@ class XmliterTestCase(unittest.TestCase):
["http://www.mydummycompany.com/images/item1.jpg"],
)
@pytest.mark.filterwarnings("ignore::scrapy.exceptions.ScrapyDeprecationWarning")
def test_xmliter_namespaced_nodename_missing(self):
body = b"""
<?xml version="1.0" encoding="UTF-8"?>
@ -214,6 +220,7 @@ class XmliterTestCase(unittest.TestCase):
with self.assertRaises(StopIteration):
next(my_iter)
@pytest.mark.filterwarnings("ignore::scrapy.exceptions.ScrapyDeprecationWarning")
def test_xmliter_exception(self):
body = (
'<?xml version="1.0" encoding="UTF-8"?>'
@ -226,10 +233,12 @@ class XmliterTestCase(unittest.TestCase):
self.assertRaises(StopIteration, next, iter)
@pytest.mark.filterwarnings("ignore::scrapy.exceptions.ScrapyDeprecationWarning")
def test_xmliter_objtype_exception(self):
i = self.xmliter(42, "product")
self.assertRaises(TypeError, next, i)
@pytest.mark.filterwarnings("ignore::scrapy.exceptions.ScrapyDeprecationWarning")
def test_xmliter_encoding(self):
body = (
b'<?xml version="1.0" encoding="ISO-8859-9"?>\n'
@ -244,12 +253,25 @@ class XmliterTestCase(unittest.TestCase):
)
class LxmlXmliterTestCase(XmliterTestCase):
xmliter = staticmethod(xmliter_lxml)
class XmliterTestCase(XmliterBaseTestCase, unittest.TestCase):
xmliter = staticmethod(xmliter)
@mark.xfail(reason="known bug of the current implementation")
def test_xmliter_namespaced_nodename(self):
super().test_xmliter_namespaced_nodename()
def test_deprecation(self):
body = b"""
<?xml version="1.0" encoding="UTF-8"?>
<products>
<product></product>
</products>
"""
with pytest.warns(
ScrapyDeprecationWarning,
match="xmliter",
):
next(self.xmliter(body, "product"))
class LxmlXmliterTestCase(XmliterBaseTestCase, unittest.TestCase):
xmliter = staticmethod(xmliter_lxml)
def test_xmliter_iterate_namespace(self):
body = b"""

77
tests/test_utils_response.py
View File

@ -1,10 +1,14 @@
import unittest
from pathlib import Path
from time import process_time
from urllib.parse import urlparse
import pytest
from scrapy.http import HtmlResponse, Response, TextResponse
from scrapy.utils.python import to_bytes
from scrapy.utils.response import (
_remove_html_comments,
get_base_url,
get_meta_refresh,
open_in_browser,
@ -166,3 +170,76 @@ class ResponseUtilsTest(unittest.TestCase):
assert open_in_browser(
r5, _openfunc=check_base_url
), "Inject unique base url with conditional comment"
def test_open_in_browser_redos_comment(self):
MAX_CPU_TIME = 0.02
# Exploit input from
# https://makenowjust-labs.github.io/recheck/playground/
# for /<!--.*?-->/ (old pattern to remove comments).
body = b"-><!--\x00" * 25_000 + b"->\n<!---->"
response = HtmlResponse("https://example.com", body=body)
start_time = process_time()
open_in_browser(response, lambda url: True)
end_time = process_time()
self.assertLess(end_time - start_time, MAX_CPU_TIME)
def test_open_in_browser_redos_head(self):
MAX_CPU_TIME = 0.02
# Exploit input from
# https://makenowjust-labs.github.io/recheck/playground/
# for /(<head(?:>|\s.*?>))/ (old pattern to find the head element).
body = b"<head\t" * 8_000
response = HtmlResponse("https://example.com", body=body)
start_time = process_time()
open_in_browser(response, lambda url: True)
end_time = process_time()
self.assertLess(end_time - start_time, MAX_CPU_TIME)
@pytest.mark.parametrize(
"input_body,output_body",
(
(
b"a<!--",
b"a",
),
(
b"a<!---->b",
b"ab",
),
(
b"a<!--b-->c",
b"ac",
),
(
b"a<!--b-->c<!--",
b"ac",
),
(
b"a<!--b-->c<!--d",
b"ac",
),
(
b"a<!--b-->c<!---->d",
b"acd",
),
(
b"a<!--b--><!--c-->d",
b"ad",
),
),
)
def test_remove_html_comments(input_body, output_body):
assert (
_remove_html_comments(input_body) == output_body
), f"{_remove_html_comments(input_body)=} == {output_body=}"

2
tox.ini
View File

@ -124,6 +124,8 @@ deps =
robotexclusionrulesparser
Pillow
Twisted[http2]
brotli
zstandard
[testenv:extra-deps-pinned]
basepython = python3.8