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

Merge 73369e6833 into e28e56aa61

This commit is contained in:
Andrey Rakhmatullin 2026-08-15 11:16:48 -05:00 committed by GitHub
parent e28e56aa61 73369e6833
commit 442bfbe889
No known key found for this signature in database
GPG Key ID: B5690EEEBB952194
15 changed files with 1334 additions and 20 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

1
docs/conf.py
View File

@ -164,6 +164,7 @@ scrapy_intersphinx_enable = [
"form2request",
"itemloaders",
"parsel",
"platformdirs",
"pytest",
"pypug",
"scrapy-lint",

2
docs/requirements.in
View File

@ -6,4 +6,4 @@ sphinx-notfound-page
sphinx-reredirects
sphinx-rtd-theme
sphinx-rtd-dark-mode
sphinx-scrapy @ git+https://github.com/scrapy/sphinx-scrapy.git@0.8.11
sphinx-scrapy @ git+https://github.com/scrapy/sphinx-scrapy.git@0.8.12

5
docs/requirements.txt
View File

@ -1,5 +1,5 @@
# This file was autogenerated by uv via the following command:
# uv pip compile -p 3.13 docs/requirements.in -o docs/requirements.txt
# uv pip compile -p 3.14 docs/requirements.in -o docs/requirements.txt
alabaster==1.0.0
# via sphinx
annotated-types==0.7.0
@ -36,6 +36,7 @@ docutils==0.22.4
# sphinx
# sphinx-markdown-builder
# sphinx-rtd-theme
# sphinx-scrapy
filelock==3.25.2
# via tldextract
h2==4.3.0
@ -156,7 +157,7 @@ sphinx-rtd-theme==3.1.0
# via
# -r docs/requirements.in
# sphinx-rtd-dark-mode
sphinx-scrapy @ git+https://github.com/scrapy/sphinx-scrapy.git@6f8e5e0bbd171a857da480f7188f2a205041cb60
sphinx-scrapy @ git+https://github.com/scrapy/sphinx-scrapy.git@3581b4148e62f34f0a835cdd06eee5a6e0f5f843
# via -r docs/requirements.in
sphinx-sitemap==2.9.0
# via sphinx-scrapy

108
docs/topics/extensions.rst
View File

@ -149,23 +149,6 @@ Log Count extension
.. autoclass:: LogCount
.. _topics-extensions-ref-telnetconsole:
Telnet console extension
~~~~~~~~~~~~~~~~~~~~~~~~
.. module:: scrapy.extensions.telnet
:synopsis: Telnet console
.. class:: TelnetConsole
Provides a telnet console for getting into a Python interpreter inside the
currently running Scrapy process, which can be very useful for debugging.
The telnet console must be enabled by the :setting:`TELNETCONSOLE_ENABLED`
setting, and the server will listen in the port specified in
:setting:`TELNETCONSOLE_PORT`.
.. _topics-extensions-ref-memusage:
Memory usage extension
@ -495,3 +478,94 @@ signal is received. After the debugger is exited, the Scrapy process continues
running normally.
This extension only works on POSIX-compliant platforms (i.e. not Windows).
.. _topics-extensions-ref-telnetconsole:
Telnet console extension
~~~~~~~~~~~~~~~~~~~~~~~~
.. module:: scrapy.extensions.telnet
:synopsis: Telnet console
.. class:: TelnetConsole
Provides a telnet console for getting into a Python interpreter inside the
currently running Scrapy process, which can be very useful for debugging.
The telnet console must be enabled by the :setting:`TELNETCONSOLE_ENABLED`
setting, and the server will listen in the port specified in
:setting:`TELNETCONSOLE_PORT`.
Remote control extension
~~~~~~~~~~~~~~~~~~~~~~~~
.. module:: scrapy.extensions.remote_control
:synopsis: Remote control extension
.. autoclass:: RemoteControl
.. setting:: REMOTE_CONTROL_ENABLED
REMOTE_CONTROL_ENABLED
""""""""""""""""""""""
Default: ``True``
Whether to enable the :class:`RemoteControl` extension.
.. setting:: REMOTE_CONTROL_JOBS_DIR
REMOTE_CONTROL_JOBS_DIR
"""""""""""""""""""""""
Default: ``None``
The directory for storing :class:`RemoteControl` job files. When this is set to
``None``, a ``scrapy/jobfiles`` subdirectory in
:func:`platformdirs.user_state_dir` is used.
As job files contain authentication tokens necessary to connect to Scrapy
processes, this directory should not be exposed to untrusted environments.
.. setting:: REMOTE_CONTROL_TIMEOUT_DEFAULT
REMOTE_CONTROL_TIMEOUT_DEFAULT
""""""""""""""""""""""""""""""
Default: ``30.0``
The default timeout in seconds for running a single code snippet sent to the
:class:`RemoteControl` ``/execute`` endpoint. You can override it for a single
request via the ``timeout_sec`` request field.
.. setting:: REMOTE_CONTROL_TIMEOUT_MAX
REMOTE_CONTROL_TIMEOUT_MAX
""""""""""""""""""""""""""
Default: ``600.0``
The maximum allowed value for the ``timeout_sec`` field of
:class:`RemoteControl` ``/execute`` endpoint requests. Higher values will be
clamped to this value.
.. setting:: REMOTE_CONTROL_OUTPUT_MAX_BYTES
REMOTE_CONTROL_OUTPUT_MAX_BYTES
"""""""""""""""""""""""""""""""
Default: ``65536``
The maximum size of the ``output`` field in responses of :class:`RemoteControl`
``/execute`` endpoint requests. Longer ones will be truncated.
.. setting:: REMOTE_CONTROL_TRACEBACK_MAX_BYTES
REMOTE_CONTROL_TRACEBACK_MAX_BYTES
""""""""""""""""""""""""""""""""""
Default: ``16384``
The maximum size of the ``traceback`` field in responses of
:class:`RemoteControl` ``/execute`` endpoint requests. Longer ones will be
truncated.

33
docs/topics/security.rst
View File

@ -223,6 +223,39 @@ More generally, if you crawl URLs from untrusted sources, consider validating
their schemes (and, where applicable, their hosts) before scheduling requests,
to avoid server-side request forgery (SSRF) and similar issues.
.. _security-remote-control:
Remote control server
=====================
Scrapy enables the remote control HTTP server
(:class:`scrapy.extensions.remote_control.RemoteControl`) by default
(:setting:`REMOTE_CONTROL_ENABLED`). Its purpose is to run arbitrary code
inside the Scrapy process, so anyone who can connect to it can do that.
The server listens on a random localhost port and requires a token for
authentication. This token is stored in a job file (see
:setting:`REMOTE_CONTROL_JOBS_DIR` for the location of these files), so you
should protect these files from unauthorized access. On Linux and macOS systems
Scrapy sets file system permissions for job files and the directory containing
them to be accessible only by the owner.
.. note::
The server doesn't use HTTPS so it's possible to sniff the traffic or
tamper with it, but it requires the attacker to get access to the loopback
traffic.
If you do not use this feature, disable it entirely:
.. code-block:: python
REMOTE_CONTROL_ENABLED = False
* **Pro:** removes a local code-execution surface and one less listening port.
* **Con:** you can no longer inspect and control a running crawler through it.
.. _security-telnet:
Telnet console

1
docs/topics/settings.rst
View File

@ -1376,6 +1376,7 @@ Default:
"scrapy.extensions.logstats.LogStats": 0,
"scrapy.extensions.spiderstate.SpiderState": 0,
"scrapy.extensions.throttle.AutoThrottle": 0,
"scrapy.extensions.remote_control.RemoteControl": 0,
}
A dict containing the extensions available by default in Scrapy, and their

2
pyproject.toml
View File

@ -8,6 +8,7 @@ dynamic = ["version"]
description = "A high-level Web Crawling and Web Scraping framework"
dependencies = [
"Twisted>=21.7.0",
"aiohttp>=3.13.3",
"cryptography>=37.0.0",
"cssselect>=0.9.1",
"defusedxml>=0.7.1",
@ -16,6 +17,7 @@ dependencies = [
"lxml>=4.6.4",
"packaging",
"parsel>=1.5.0",
"platformdirs>=2.0.0",
"protego>=0.1.15",
"pyOpenSSL>=22.0.0",
"queuelib>=1.6.1",

1
scrapy/commands/shell.py
View File

@ -28,6 +28,7 @@ class Command(ScrapyCommand):
default_settings: ClassVar[dict[str, Any]] = {
"DUPEFILTER_CLASS": "scrapy.dupefilters.BaseDupeFilter",
"LOGSTATS_INTERVAL": 0,
"REMOTE_CONTROL_ENABLED": False,
}
def syntax(self) -> str:

336
scrapy/extensions/remote_control.py Normal file
View File

@ -0,0 +1,336 @@
from __future__ import annotations
import ast
import asyncio
import contextlib
import hmac
import inspect
import io
import logging
import os
import secrets
import time
import traceback
from types import CodeType
from typing import TYPE_CHECKING, Any, Literal, cast
from aiohttp import web
import scrapy
from scrapy import signals
from scrapy.exceptions import NotConfigured
from scrapy.utils._remote_control import (
ExecuteResult,
StatusResult,
job_files_dir,
new_job_file_name,
write_job_file,
)
from scrapy.utils.asyncio import is_asyncio_available
if TYPE_CHECKING:
from datetime import datetime
from pathlib import Path
# typing.Self requires Python 3.11
from typing_extensions import Self
from scrapy.crawler import Crawler
logger = logging.getLogger(__name__)
STOP_TIMEOUT = 2.0
class RemoteControl:
"""Provides an HTTP server that can run Python code passed to it in HTTP requests
and return the output in responses.
The code runs inside the Scrapy process and has access to the
:class:`~scrapy.crawler.Crawler` instance in the ``crawler`` variable and
to a persistent dictionary in the ``stash`` variable.
This extension can be disabled by setting the
:setting:`REMOTE_CONTROL_ENABLED` setting to ``False``. It requires
:ref:`asyncio support <using-asyncio>` and will be disabled without it.
The HTTP server listens on a random port on ``localhost`` and requires a
``Bearer`` token for authentication. The token and port are written to a
job file in the user's profile directory so that other processes can
discover and use them to connect to the server.
Available endpoints:
- ``/status``: expects a ``GET`` request and returns a JSON object with the
following keys:
- ``pid`` (number): the process ID of the Scrapy process.
- ``spider`` (string): the name of the currently running spider.
- ``project`` (string): the name of the Scrapy project.
- ``scrapy_version`` (string): the version of Scrapy.
- ``start_time`` (number or null): the start time of the Scrapy process
as a UNIX timestamp.
- ``/execute``: expects a ``POST`` request with a JSON object containing
the following keys:
- ``code`` (string): Python code to execute.
- ``timeout_sec`` (number, optional): the maximum number of seconds
to allow the code to run.
The response is a JSON object with the following keys:
- ``status`` (string): one of ``"ok"``, ``"compile_error"``,
``"error"``, or ``"timeout"``.
- ``output`` (string): the output of the code.
- ``traceback`` (string or null): the traceback if an exception was
raised.
- ``elapsed_sec`` (number): the number of seconds the code took to run.
- ``output_truncated`` (boolean, optional): whether the output was
truncated (omitted if ``false``).
- ``traceback_truncated`` (boolean, optional): whether the traceback
was truncated (omitted if ``false``).
.. versionadded:: VERSION
"""
def __init__(self, crawler: Crawler):
if not crawler.settings.getbool("REMOTE_CONTROL_ENABLED"):
raise NotConfigured
if (
crawler.settings.getbool("TWISTED_REACTOR_ENABLED")
and not is_asyncio_available()
):
raise NotConfigured(
f"{type(self).__name__} requires the asyncio support."
f" You can set the REMOTE_CONTROL_ENABLED setting to False to remove this warning."
)
self.crawler: Crawler = crawler
self._default_timeout: float = crawler.settings.getfloat(
"REMOTE_CONTROL_TIMEOUT_DEFAULT"
)
self._max_timeout: float = crawler.settings.getfloat(
"REMOTE_CONTROL_TIMEOUT_MAX"
)
self._output_max_bytes: int = crawler.settings.getint(
"REMOTE_CONTROL_OUTPUT_MAX_BYTES"
)
self._traceback_max_bytes: int = crawler.settings.getint(
"REMOTE_CONTROL_TRACEBACK_MAX_BYTES"
)
if self._default_timeout <= 0 or self._max_timeout <= 0:
raise NotConfigured("REMOTE_CONTROL_TIMEOUT_* must be positive")
self._stash: dict[str, Any] = {}
self._auth_token: str | None = None
self._runner: web.AppRunner | None = None
self._job_file_path: Path | None = None
crawler.signals.connect(self.start, signal=signals.engine_started)
crawler.signals.connect(self.stop, signal=signals.engine_stopped)
@classmethod
def from_crawler(cls, crawler: Crawler) -> Self:
return cls(crawler)
def _make_namespace(self, buf: io.StringIO) -> dict[str, Any]:
def _print(*args: Any, **kwargs: Any) -> None:
kwargs.setdefault("file", buf)
print(*args, **kwargs)
return {"crawler": self.crawler, "stash": self._stash, "print": _print}
async def start(self) -> None:
"""Start the HTTP server."""
try:
self._auth_token = secrets.token_urlsafe(32)
app = web.Application()
app.router.add_get("/status", self._handle_status, allow_head=False)
app.router.add_post("/execute", self._handle_execute)
self._runner = web.AppRunner(
app, access_log=None, shutdown_timeout=STOP_TIMEOUT
)
await self._runner.setup()
site = web.TCPSite(self._runner, "127.0.0.1", 0)
await site.start()
port = self._runner.addresses[0][1]
job_path = job_files_dir(self.crawler.settings) / new_job_file_name()
assert self.crawler.spider
# we create the job file after starting the HTTP server
write_job_file(
job_path,
spider=self.crawler.spider.name,
project=self.crawler.settings.get("BOT_NAME"),
scrapy_version=scrapy.__version__,
port=port,
token=self._auth_token,
)
self._job_file_path = job_path
logger.info(
f"Remote control HTTP server listening on"
f" port {port} (job {job_path.stem})",
extra={"crawler": self.crawler},
)
except Exception:
logger.exception(
"Remote control HTTP server failed to start",
extra={"crawler": self.crawler},
)
await self.stop()
async def stop(self) -> None:
"""Stop the HTTP server and remove the job file."""
if self._job_file_path is not None:
# we remove the job file before stopping the HTTP server
with contextlib.suppress(OSError):
self._job_file_path.unlink(missing_ok=True)
self._job_file_path = None
if self._runner is None:
return
try:
await self._runner.cleanup()
except Exception:
logger.exception(
"Error stopping the remote control HTTP server",
extra={"crawler": self.crawler},
)
finally:
self._stash.clear()
self._runner = None
self._auth_token = None
async def _handle_status(self, request: web.Request) -> web.Response:
"""An aiohttp request handler for the ``/status`` endpoint."""
if not self._is_authenticated(request):
return web.json_response({"error": "unauthorized"}, status=401)
return web.json_response(self._get_status())
async def _handle_execute(self, request: web.Request) -> web.Response:
"""An aiohttp request handler for the ``/execute`` endpoint."""
if not self._is_authenticated(request):
return web.json_response({"error": "unauthorized"}, status=401)
try:
body = await request.json()
except Exception:
return web.json_response({"error": "invalid JSON body"}, status=400)
if not isinstance(body, dict) or not isinstance(body.get("code"), str):
return web.json_response(
{"error": "Missing or invalid 'code' value"}, status=400
)
requested_timeout = body.get("timeout_sec")
if requested_timeout is not None and not isinstance(
requested_timeout, (int, float)
):
return web.json_response(
{"error": "Invalid 'timeout_sec' value"}, status=400
)
timeout = _effective_timeout(
requested_timeout, self._default_timeout, self._max_timeout
)
compiled = _compile(body["code"])
result: ExecuteResult
if isinstance(compiled, CodeType):
result = await self._run_code(compiled, timeout)
else:
result = {
"status": "compile_error",
"output": "",
"traceback": compiled,
"elapsed_sec": 0.0,
}
return web.json_response(result)
def _is_authenticated(self, request: web.Request) -> bool:
"""Check if the request is authenticated with the correct Bearer token."""
token = request.headers.get("Authorization", "").removeprefix("Bearer ")
return (
self._auth_token is not None
and token.isascii()
and hmac.compare_digest(token, self._auth_token)
)
async def _run_code(self, code_obj: CodeType, timeout: float) -> ExecuteResult:
"""Run a compiled code object with a timeout and capture its output."""
buf = io.StringIO()
ns = self._make_namespace(buf)
status: Literal["ok", "error", "timeout"] = "ok"
tb: str | None = None
start_time = time.perf_counter()
try:
# eval() returns a coroutine if and only if the source used a top-level await,
# else it runs synchronously and returns None.
eval_result = eval(code_obj, ns) # noqa: S307 - arbitrary code by design
if inspect.iscoroutine(eval_result):
try:
await asyncio.wait_for(eval_result, timeout)
except asyncio.TimeoutError:
# wait_for cancelled the coroutine at an await point.
status = "timeout"
except Exception:
# intentionally doesn't catch asyncio.CancelledError, which is a BaseException
status = "error"
tb = traceback.format_exc()
elapsed = round(time.perf_counter() - start_time, 3)
output, out_was_truncated = _cap(buf.getvalue(), self._output_max_bytes)
if tb is not None:
tb, tb_was_truncated = _cap(tb, self._traceback_max_bytes)
else:
tb_was_truncated = False
result: ExecuteResult = {
"status": status,
"output": output,
"traceback": tb,
"elapsed_sec": elapsed,
}
if out_was_truncated:
result["output_truncated"] = True
if tb_was_truncated:
result["traceback_truncated"] = True
return result
def _get_status(self) -> StatusResult:
"""Return the data for the ``/status`` response."""
assert self.crawler.spider
start_time: datetime | None = self.crawler.stats.get_value("start_time")
return {
"pid": os.getpid(),
"spider": self.crawler.spider.name,
"project": self.crawler.settings.get("BOT_NAME"),
"scrapy_version": scrapy.__version__,
"start_time": start_time.timestamp() if start_time is not None else None,
}
def _cap(s: str, limit: int) -> tuple[str, bool]:
"""Cap a string to ``limit`` bytes, appending an inline truncation marker."""
b = s.encode("utf-8")
if len(b) <= limit:
return s, False
extra_kb = (len(b) - limit) // 1024 + 1
head = b[:limit].decode("utf-8", "ignore")
return f"{head}…[truncated, +{extra_kb}KB]", True
def _compile(src: str) -> CodeType | str:
"""Compile with top-level await support.
:return: compiled code or a string with the traceback of the compile error.
"""
try:
return cast(
"CodeType",
compile(src, "<execute>", "exec", flags=ast.PyCF_ALLOW_TOP_LEVEL_AWAIT),
)
except (SyntaxError, ValueError):
return traceback.format_exc()
def _effective_timeout(
requested: float | None, default: float, maximum: float
) -> float:
"""Clamp a client-requested timeout to ``maximum``."""
if requested is None or not requested > 0:
requested = default
return min(requested, maximum)

14
scrapy/settings/default_settings.py
View File

@ -176,6 +176,12 @@ __all__ = [
"REFERER_ENABLED",
"REFERRER_POLICIES",
"REFERRER_POLICY",
"REMOTE_CONTROL_ENABLED",
"REMOTE_CONTROL_JOBS_DIR",
"REMOTE_CONTROL_OUTPUT_MAX_BYTES",
"REMOTE_CONTROL_TIMEOUT_DEFAULT",
"REMOTE_CONTROL_TIMEOUT_MAX",
"REMOTE_CONTROL_TRACEBACK_MAX_BYTES",
"REQUEST_FINGERPRINTER_CLASS",
"RETRY_ENABLED",
"RETRY_EXCEPTIONS",
@ -354,6 +360,7 @@ EXTENSIONS_BASE = {
"scrapy.extensions.logstats.LogStats": 0,
"scrapy.extensions.spiderstate.SpiderState": 0,
"scrapy.extensions.throttle.AutoThrottle": 0,
"scrapy.extensions.remote_control.RemoteControl": 0,
}
FEEDS = {}
@ -504,6 +511,13 @@ REFERER_ENABLED = True
REFERRER_POLICY = "scrapy.spidermiddlewares.referer.DefaultReferrerPolicy"
REFERRER_POLICIES = {}
REMOTE_CONTROL_ENABLED = True
REMOTE_CONTROL_JOBS_DIR = None
REMOTE_CONTROL_TIMEOUT_DEFAULT = 30.0
REMOTE_CONTROL_TIMEOUT_MAX = 600.0
REMOTE_CONTROL_OUTPUT_MAX_BYTES = 64 * 1024
REMOTE_CONTROL_TRACEBACK_MAX_BYTES = 16 * 1024
REQUEST_FINGERPRINTER_CLASS = "scrapy.utils.request.RequestFingerprinter"
RETRY_ENABLED = True

96
scrapy/utils/_remote_control.py Normal file
View File

@ -0,0 +1,96 @@
from __future__ import annotations
import contextlib
import json
import logging
import os
import time
import uuid
from pathlib import Path
from typing import TYPE_CHECKING, Literal, TypedDict
from platformdirs import user_state_dir
if TYPE_CHECKING:
# typing.NotRequired requires Python 3.11
from typing_extensions import NotRequired
from scrapy.settings import BaseSettings
logger = logging.getLogger(__name__)
# On-disk format of the job files
JOB_FILE_VERSION = 1
class StatusResult(TypedDict):
"""The result of a ``/status`` call."""
pid: int
spider: str
project: str | None
scrapy_version: str
start_time: float | None
class ExecuteResult(TypedDict):
"""The result of an ``/execute`` call."""
status: Literal["ok", "compile_error", "error", "timeout"]
output: str
traceback: str | None
elapsed_sec: float
output_truncated: NotRequired[bool]
traceback_truncated: NotRequired[bool]
def job_files_dir(settings: BaseSettings) -> Path:
"""Return the directory used for job files."""
setting = settings.get("REMOTE_CONTROL_JOBS_DIR")
if setting:
return Path(setting)
return Path(user_state_dir("scrapy", appauthor=False), "job_files")
def new_job_file_name() -> str:
"""Return the name of a new job file."""
return f"{os.getpid()}-{uuid.uuid4().hex}.json"
def write_job_file(
path: Path,
*,
spider: str,
project: str | None,
scrapy_version: str,
port: int,
token: str,
) -> None:
"""Write the job file that makes a crawl discoverable.
The file content is sensitive information as it includes the auth token.
"""
data = {
"version": JOB_FILE_VERSION,
"pid": os.getpid(),
"port": port,
"token": token,
"spider": spider,
"project": project,
"scrapy_version": scrapy_version,
"start_time": time.time(),
}
path.parent.mkdir(parents=True, exist_ok=True)
path.parent.chmod(0o700)
# Atomic write so the file is never world-readable mid-write.
tmp = path.with_name(f".{path.name}.tmp")
fd = os.open(tmp, os.O_WRONLY | os.O_CREAT | os.O_EXCL, 0o600)
try:
with os.fdopen(fd, "w") as f:
json.dump(data, f)
tmp.chmod(0o600)
tmp.replace(path)
except BaseException:
with contextlib.suppress(OSError):
tmp.unlink()
raise

1
scrapy/utils/test.py
View File

@ -69,6 +69,7 @@ def get_crawler(
# When needed, useful settings can be added here, e.g. ones that prevent
# deprecation warnings.
settings: dict[str, Any] = {
"REMOTE_CONTROL_ENABLED": False,
"TELNETCONSOLE_ENABLED": False,
**get_reactor_settings(),
**(settings_dict or {}),

633
tests/test_extension_remote_control.py Normal file
View File

@ -0,0 +1,633 @@
from __future__ import annotations
import asyncio
import contextlib
import json
import logging
import os
from contextlib import asynccontextmanager
from datetime import datetime, timezone
from pathlib import Path
from typing import TYPE_CHECKING, Any
import aiohttp
import pytest
from aiohttp import web
import scrapy
from scrapy.exceptions import NotConfigured
from scrapy.extensions import remote_control
from scrapy.extensions.remote_control import (
RemoteControl,
_cap,
_compile,
_effective_timeout,
)
from scrapy.settings import default_settings
from scrapy.utils.test import get_crawler
from tests.utils.decorators import coroutine_test
if TYPE_CHECKING:
from collections.abc import AsyncGenerator
from types import CodeType
pytestmark = pytest.mark.only_asyncio
def _get_extension(settings: dict[str, Any] | None = None) -> RemoteControl:
crawler = get_crawler(
settings_dict={"REMOTE_CONTROL_ENABLED": True, **(settings or {})}
)
crawler.spider = crawler._create_spider()
return RemoteControl(crawler)
def compile_or_fail(source: str) -> CodeType:
compiled = _compile(source)
assert not isinstance(compiled, str)
return compiled
@asynccontextmanager
async def _started_extension(
jobs_dir: Path, settings: dict[str, Any] | None = None
) -> AsyncGenerator[RemoteControl]:
extension = _get_extension(
{"REMOTE_CONTROL_JOBS_DIR": str(jobs_dir), **(settings or {})}
)
await extension.start()
assert extension._runner is not None, "the server did not start"
try:
yield extension
finally:
await extension.stop()
async def _request(
extension: RemoteControl, method: str, path: str, **kwargs: Any
) -> tuple[int, Any]:
assert extension._runner
host, port = extension._runner.addresses[0]
url = f"http://{host}:{port}{path}"
async with (
aiohttp.ClientSession() as session,
session.request(method, url, **kwargs) as response,
):
try:
return response.status, await response.json(content_type=None)
except json.JSONDecodeError:
return response.status, await response.text()
async def _request_execute(extension: RemoteControl, **kwargs: Any) -> tuple[int, Any]:
return await _request(extension, "POST", "/execute", **kwargs)
async def _request_status(extension: RemoteControl, **kwargs: Any) -> tuple[int, Any]:
return await _request(extension, "GET", "/status", **kwargs)
def _auth(extension: RemoteControl) -> dict[str, str]:
return {"Authorization": f"Bearer {extension._auth_token}"}
BAD_AUTH_HEADERS = [
{},
{"Authorization": "Bearer nope"},
{"Authorization": "Bearer ünicode"},
{"Authorization": "Basic nope"},
]
@coroutine_test
async def test_ok_output() -> None:
extension = _get_extension()
result = await extension._run_code(compile_or_fail("print('hello')"), 5)
assert result["status"] == "ok"
assert result["output"] == "hello\n"
assert result["traceback"] is None
assert "output_truncated" not in result
assert "traceback_truncated" not in result
@coroutine_test
async def test_top_level_await() -> None:
extension = _get_extension()
result = await extension._run_code(
compile_or_fail("import asyncio\nawait asyncio.sleep(0)\nprint('done')"), 5
)
assert result["status"] == "ok"
assert result["output"] == "done\n"
@coroutine_test
async def test_sync_code_runs() -> None:
extension = _get_extension()
result = await extension._run_code(
compile_or_fail("x = sum(range(10))\nprint(x)"), 5
)
assert result["status"] == "ok"
assert result["output"] == "45\n"
@coroutine_test
async def test_crawler_is_in_the_namespace() -> None:
extension = _get_extension()
result = await extension._run_code(
compile_or_fail("print(crawler.spidercls.name)"), 5
)
assert result["output"] == f"{extension.crawler.spidercls.name}\n"
@coroutine_test
async def test_runtime_error_keeps_partial_output() -> None:
extension = _get_extension()
result = await extension._run_code(
compile_or_fail("print('before')\nraise ValueError('boom')"), 5
)
assert result["status"] == "error"
assert result["traceback"] is not None
assert "ValueError: boom" in result["traceback"]
assert "before" in result["output"]
@coroutine_test
async def test_timeout() -> None:
extension = _get_extension()
result = await extension._run_code(
compile_or_fail("import asyncio\nawait asyncio.sleep(10)"), 0.05
)
assert result["status"] == "timeout"
assert result["traceback"] is None
@coroutine_test
async def test_stash_persists_across_calls() -> None:
extension = _get_extension()
await extension._run_code(compile_or_fail("stash['x'] = 42"), 5)
result = await extension._run_code(compile_or_fail("print(stash['x'])"), 5)
assert result["output"] == "42\n"
assert extension._stash == {"x": 42}
@coroutine_test
async def test_namespace_is_fresh_each_call() -> None:
extension = _get_extension()
await extension._run_code(compile_or_fail("y = 99"), 5)
result = await extension._run_code(compile_or_fail("print('y' in dir())"), 5)
assert result["status"] == "ok"
assert result["output"] == "False\n"
@coroutine_test
async def test_concurrent_calls_are_not_serialized() -> None:
# A slow, awaiting call must not block a quick one: the quick call finishes
# first even though it was started second, and both share `stash`.
extension = _get_extension()
extension._stash["order"] = []
slow = extension._run_code(
compile_or_fail(
"import asyncio\nawait asyncio.sleep(0.3)\nstash['order'].append('slow')"
),
5,
)
quick = extension._run_code(compile_or_fail("stash['order'].append('quick')"), 5)
await asyncio.gather(slow, quick)
assert extension._stash["order"] == ["quick", "slow"]
@coroutine_test
async def test_output_truncation() -> None:
extension = _get_extension({"REMOTE_CONTROL_OUTPUT_MAX_BYTES": 10})
result = await extension._run_code(compile_or_fail("print('x' * 100)"), 5)
assert result["output_truncated"] is True
assert "truncated" in result["output"]
@coroutine_test
async def test_traceback_truncation() -> None:
extension = _get_extension({"REMOTE_CONTROL_TRACEBACK_MAX_BYTES": 10})
result = await extension._run_code(
compile_or_fail("raise ValueError('boom' * 100)"), 5
)
assert result["traceback_truncated"] is True
assert result["traceback"] is not None
assert "truncated" in result["traceback"]
@pytest.mark.parametrize("source", ["def (:", "print('a')\x00"])
def test_compile_error(source: str) -> None:
rendered = _compile(source)
assert isinstance(rendered, str)
# The docs only say "This function raises SyntaxError or ValueError if the
# compiled source is invalid." and it depends on the Python version.
assert "SyntaxError" in rendered or "ValueError" in rendered
@pytest.mark.parametrize(
("requested", "expected"),
[
(None, 30.0),
(50, 50.0),
(1000, 600.0),
(0, 30.0),
(-1, 30.0),
(float("nan"), 30.0),
(float("inf"), 600.0),
],
)
def test_effective_timeout(requested: float | None, expected: float) -> None:
assert _effective_timeout(requested, 30.0, 600.0) == expected
def test_effective_timeout_caps_the_default_too() -> None:
assert _effective_timeout(None, 1000.0, 600.0) == 600.0
def test_cap() -> None:
assert _cap("hello", 10) == ("hello", False)
capped, truncated = _cap("x" * 2048, 10)
assert truncated is True
assert capped.startswith("x" * 10)
assert capped.endswith("…[truncated, +2KB]") # 2038 bytes dropped
def test_cap_does_not_split_a_character() -> None:
# "ä" takes two bytes, so a 5 byte cap must drop the third one entirely.
capped, truncated = _cap("ä" * 10, 5)
assert truncated is True
assert capped.startswith("ää")
assert "truncated" in capped
def test_disabled_by_setting() -> None:
with pytest.raises(NotConfigured):
_get_extension({"REMOTE_CONTROL_ENABLED": False})
@pytest.mark.parametrize(
"settings",
[
{"REMOTE_CONTROL_TIMEOUT_DEFAULT": 0},
{"REMOTE_CONTROL_TIMEOUT_DEFAULT": -1},
{"REMOTE_CONTROL_TIMEOUT_MAX": 0},
{"REMOTE_CONTROL_TIMEOUT_MAX": -1},
],
)
def test_non_positive_timeouts_rejected(settings: dict[str, Any]) -> None:
with pytest.raises(NotConfigured):
_get_extension(settings)
def test_default_timeouts() -> None:
extension = _get_extension()
assert extension._default_timeout == default_settings.REMOTE_CONTROL_TIMEOUT_DEFAULT
assert extension._max_timeout == default_settings.REMOTE_CONTROL_TIMEOUT_MAX
def test_settings_are_applied() -> None:
extension = _get_extension(
{
"REMOTE_CONTROL_TIMEOUT_DEFAULT": 5.0,
"REMOTE_CONTROL_TIMEOUT_MAX": 999.0,
"REMOTE_CONTROL_OUTPUT_MAX_BYTES": 11,
"REMOTE_CONTROL_TRACEBACK_MAX_BYTES": 22,
}
)
assert extension._default_timeout == 5.0
assert extension._max_timeout == 999.0
assert extension._output_max_bytes == 11
assert extension._traceback_max_bytes == 22
@coroutine_test
async def test_status_ok(tmp_path: Path) -> None:
async with _started_extension(tmp_path) as extension:
status, result = await _request_execute(
extension, json={"code": "print(6 * 7)"}, headers=_auth(extension)
)
assert status == 200
assert result["status"] == "ok"
assert result["output"] == "42\n"
@coroutine_test
async def test_status_compile_error(tmp_path: Path) -> None:
async with _started_extension(tmp_path) as extension:
status, result = await _request_execute(
extension, json={"code": "def (:"}, headers=_auth(extension)
)
assert status == 200
assert result["status"] == "compile_error"
assert "SyntaxError" in result["traceback"]
@coroutine_test
async def test_status_error(tmp_path: Path) -> None:
async with _started_extension(tmp_path) as extension:
status, result = await _request_execute(
extension,
json={"code": "raise ValueError('boom')"},
headers=_auth(extension),
)
assert status == 200
assert result["status"] == "error"
assert "boom" in result["traceback"]
@coroutine_test
async def test_crawler_var(tmp_path: Path) -> None:
async with _started_extension(tmp_path) as extension:
status, result = await _request_execute(
extension,
json={"code": "print(type(crawler).__name__, crawler.crawling)"},
headers=_auth(extension),
)
assert status == 200
assert result["status"] == "ok"
assert result["output"] == "Crawler False\n"
@pytest.mark.parametrize("headers", BAD_AUTH_HEADERS)
@coroutine_test
async def test_execute_unauthorized(tmp_path: Path, headers: dict[str, str]) -> None:
async with _started_extension(tmp_path) as extension:
status, body = await _request_execute(
extension, json={"code": "print(1)"}, headers=headers
)
assert status == 401
assert body == {"error": "unauthorized"}
@coroutine_test
async def test_execute_rejects_other_methods(tmp_path: Path) -> None:
async with _started_extension(tmp_path) as extension:
results = [
await _request(extension, method, "/execute", headers=_auth(extension))
for method in ("HEAD", "GET", "PUT", "DELETE")
]
assert [status for status, _ in results] == [405] * 4
@pytest.mark.parametrize(
("kwargs", "error"),
[
({"data": "not json"}, "invalid JSON body"),
({"json": {"nope": 1}}, "Missing or invalid 'code' value"),
({"json": {"code": 1}}, "Missing or invalid 'code' value"),
({"json": [1, 2]}, "Missing or invalid 'code' value"),
],
)
@coroutine_test
async def test_execute_bad_requests(
tmp_path: Path, kwargs: dict[str, Any], error: str
) -> None:
async with _started_extension(tmp_path) as extension:
status, body = await _request_execute(
extension, headers=_auth(extension), **kwargs
)
assert status == 400
assert body == {"error": error}
@pytest.mark.parametrize("timeout_sec", ["abc", [1]])
@coroutine_test
async def test_execute_bad_timeout_type(tmp_path: Path, timeout_sec: Any) -> None:
async with _started_extension(tmp_path) as extension:
status, body = await _request_execute(
extension,
json={"code": "print(1)", "timeout_sec": timeout_sec},
headers=_auth(extension),
)
assert status == 400
assert body == {"error": "Invalid 'timeout_sec' value"}
@pytest.mark.parametrize("timeout_sec", [None, 0, -1])
@coroutine_test
async def test_execute_unset_timeout_is_accepted(
tmp_path: Path, timeout_sec: Any
) -> None:
async with _started_extension(tmp_path) as extension:
status, envelope = await _request_execute(
extension,
json={"code": "print(1)", "timeout_sec": timeout_sec},
headers=_auth(extension),
)
assert status == 200
assert envelope["status"] == "ok"
def test_get_status_fields() -> None:
extension = _get_extension()
assert extension._get_status() == {
"pid": os.getpid(),
"spider": extension.crawler.spidercls.name,
"project": "scrapybot",
"scrapy_version": scrapy.__version__,
"start_time": None,
}
def test_get_status_project() -> None:
extension = _get_extension({"BOT_NAME": "my_project"})
assert extension._get_status()["project"] == "my_project"
def test_get_status_project_unset() -> None:
extension = _get_extension({"BOT_NAME": None})
assert extension._get_status()["project"] is None
def test_get_status_start_time() -> None:
extension = _get_extension()
start_time = datetime(2026, 1, 2, 3, 4, 5, tzinfo=timezone.utc)
extension.crawler.stats.set_value("start_time", start_time)
assert extension._get_status()["start_time"] == start_time.timestamp()
@coroutine_test
async def test_status(tmp_path: Path) -> None:
async with _started_extension(tmp_path) as extension:
extension.crawler.stats.set_value(
"start_time", datetime(2026, 1, 2, 3, 4, 5, tzinfo=timezone.utc)
)
status, body = await _request_status(extension, headers=_auth(extension))
assert status == 200
assert body == {
"pid": os.getpid(),
"spider": extension.crawler.spidercls.name,
"project": "scrapybot",
"scrapy_version": scrapy.__version__,
"start_time": datetime(2026, 1, 2, 3, 4, 5, tzinfo=timezone.utc).timestamp(),
}
@coroutine_test
async def test_compare_status_with_job_file(tmp_path: Path) -> None:
async with _started_extension(tmp_path) as extension:
status, body = await _request_status(extension, headers=_auth(extension))
(job_file,) = tmp_path.glob("*.json")
record = json.loads(job_file.read_text(encoding="utf-8"))
assert status == 200
for key in ("pid", "spider", "project", "scrapy_version"):
assert body[key] == record[key]
@pytest.mark.parametrize("headers", BAD_AUTH_HEADERS)
@coroutine_test
async def test_status_unauthorized(tmp_path: Path, headers: dict[str, str]) -> None:
async with _started_extension(tmp_path) as extension:
status, body = await _request_status(extension, headers=headers)
assert status == 401
assert body == {"error": "unauthorized"}
@coroutine_test
async def test_status_rejects_other_methods(tmp_path: Path) -> None:
async with _started_extension(tmp_path) as extension:
results = [
await _request(extension, method, "/status", headers=_auth(extension))
for method in ("HEAD", "POST", "PUT", "DELETE")
]
assert [status for status, _ in results] == [405] * 4
@coroutine_test
async def test_status_answers_while_code_is_running(tmp_path: Path) -> None:
async with _started_extension(tmp_path) as extension:
request = asyncio.ensure_future(
_request_execute(
extension,
json={"code": "import asyncio\nawait asyncio.sleep(30)"},
headers=_auth(extension),
)
)
await asyncio.sleep(0.1) # let the request reach the handler
status, body = await asyncio.wait_for(
_request_status(extension, headers=_auth(extension)), 10
)
request.cancel()
with contextlib.suppress(asyncio.CancelledError, aiohttp.ClientError):
await request
assert status == 200
assert body["pid"] == os.getpid()
@coroutine_test
async def test_job_file_written_and_removed(tmp_path: Path) -> None:
async with _started_extension(tmp_path) as extension:
files = list(tmp_path.glob("*.json"))
assert len(files) == 1
record = json.loads(files[0].read_text(encoding="utf-8"))
assert record["token"] == extension._auth_token
assert record["spider"] == extension.crawler.spidercls.name
assert list(tmp_path.glob("*.json")) == []
@coroutine_test
async def test_token_is_never_logged(
tmp_path: Path, caplog: pytest.LogCaptureFixture
) -> None:
with caplog.at_level(logging.DEBUG):
async with _started_extension(tmp_path) as extension:
token = extension._auth_token
assert token
assert token not in caplog.text
@coroutine_test
async def test_stop_without_start() -> None:
extension = _get_extension()
await extension.stop()
assert extension._runner is None
@coroutine_test
async def test_stop_is_idempotent(tmp_path: Path) -> None:
async with _started_extension(tmp_path) as extension:
pass
await extension.stop()
assert extension._runner is None
assert list(tmp_path.glob("*.json")) == []
@coroutine_test
async def test_start_failure_disables_the_extension(
tmp_path: Path, caplog: pytest.LogCaptureFixture
) -> None:
# A file where the job file directory is expected makes writing the job
# file fail, after the HTTP server has already started.
jobs_dir = tmp_path / "jobs"
jobs_dir.write_text("", encoding="utf-8")
extension = _get_extension({"REMOTE_CONTROL_JOBS_DIR": str(jobs_dir)})
await extension.start()
assert "Remote control HTTP server failed to start" in caplog.text
assert "FileExistsError" in caplog.text
assert extension._runner is None
assert extension._auth_token is None
assert extension._job_file_path is None
@coroutine_test
async def test_stop_logs_a_cleanup_failure(
tmp_path: Path, caplog: pytest.LogCaptureFixture, monkeypatch: pytest.MonkeyPatch
) -> None:
extension = _get_extension({"REMOTE_CONTROL_JOBS_DIR": str(tmp_path)})
await extension.start()
runner = extension._runner
assert runner is not None
extension._stash["x"] = 42
real_cleanup = web.AppRunner.cleanup
async def raise_runtime_error(self: web.AppRunner) -> None:
raise RuntimeError("boom")
monkeypatch.setattr(web.AppRunner, "cleanup", raise_runtime_error)
await extension.stop()
assert "Error stopping the remote control HTTP server" in caplog.text
assert "RuntimeError: boom" in caplog.text
assert extension._runner is None
assert extension._auth_token is None
assert extension._stash == {}
assert list(tmp_path.glob("*.json")) == []
await real_cleanup(runner)
@coroutine_test
async def test_stop_ignores_a_failure_to_remove_the_job_file(
tmp_path: Path, monkeypatch: pytest.MonkeyPatch
) -> None:
def raise_os_error(*args: object, **kwargs: object) -> None:
raise OSError("cannot remove")
extension = _get_extension({"REMOTE_CONTROL_JOBS_DIR": str(tmp_path)})
await extension.start()
monkeypatch.setattr(Path, "unlink", raise_os_error)
await extension.stop()
assert extension._runner is None
assert extension._job_file_path is None
assert len(list(tmp_path.glob("*.json"))) == 1 # could not be removed
@coroutine_test
async def test_stop_does_not_wait_for_a_running_snippet(
tmp_path: Path, monkeypatch: pytest.MonkeyPatch
) -> None:
monkeypatch.setattr(remote_control, "STOP_TIMEOUT", 0.1)
extension = _get_extension({"REMOTE_CONTROL_JOBS_DIR": str(tmp_path)})
extension.crawler.spider = extension.crawler.spidercls()
await extension.start()
request = asyncio.ensure_future(
_request_execute(
extension,
json={"code": "import asyncio\nawait asyncio.sleep(30)"},
headers=_auth(extension),
)
)
await asyncio.sleep(0.1) # let the request reach the handler
await asyncio.wait_for(extension.stop(), 10)
assert extension._runner is None
request.cancel()
with contextlib.suppress(asyncio.CancelledError, aiohttp.ClientError):
await request

115
tests/test_utils_remote_control.py Normal file
View File

@ -0,0 +1,115 @@
from __future__ import annotations
import json
import os
import stat
import sys
from pathlib import Path
import pytest
from scrapy.settings import Settings
from scrapy.utils._remote_control import (
JOB_FILE_VERSION,
job_files_dir,
new_job_file_name,
write_job_file,
)
def _write_job_file(path: Path) -> None:
write_job_file(
path,
spider="dummy",
project="testbot",
scrapy_version="2.17.0",
port=12345,
token="secret",
)
def test_write_job_file(tmp_path: Path) -> None:
path = tmp_path / "jobs" / f"{os.getpid()}-abc.json"
_write_job_file(path)
if sys.platform != "win32":
assert stat.S_IMODE(path.stat().st_mode) == 0o600
assert stat.S_IMODE(path.parent.stat().st_mode) == 0o700
record = json.loads(path.read_text(encoding="utf-8"))
assert record["version"] == JOB_FILE_VERSION
assert record["pid"] == os.getpid()
assert record["port"] == 12345
assert record["token"] == "secret"
assert record["spider"] == "dummy"
assert record["project"] == "testbot"
assert record["scrapy_version"] == "2.17.0"
assert isinstance(record["start_time"], float)
def test_write_job_file_leaves_no_temporary_file(tmp_path: Path) -> None:
name = f"{os.getpid()}-abc.json"
_write_job_file(tmp_path / name)
assert [path.name for path in tmp_path.iterdir()] == [name]
def test_new_job_file_name() -> None:
name = new_job_file_name()
assert name.endswith(".json")
pid, _, rest = name.removeprefix(".").partition("-")
assert rest
assert pid.isdigit()
assert int(pid) == os.getpid()
def test_write_job_file_refuses_an_existing_temporary_file(tmp_path: Path) -> None:
path = tmp_path / new_job_file_name()
planted = path.with_name(f".{path.name}.tmp")
planted.write_text("", encoding="utf-8")
with pytest.raises(FileExistsError):
_write_job_file(path)
assert not path.exists()
assert planted.read_text(encoding="utf-8") == "" # left untouched
def test_write_job_file_removes_the_temporary_file_after_a_failure(
tmp_path: Path,
) -> None:
# A directory in the way makes the final rename fail, after the temporary
# file has already been created.
path = tmp_path / new_job_file_name()
path.mkdir()
# the specific exception differs between platforms
with pytest.raises(OSError): # noqa: PT011
_write_job_file(path)
assert list(tmp_path.iterdir()) == [path]
def test_write_job_file_ignores_a_failure_to_remove_the_temporary_file(
tmp_path: Path, monkeypatch: pytest.MonkeyPatch
) -> None:
def raise_runtime_error(*args: object, **kwargs: object) -> None:
raise RuntimeError("boom")
def raise_os_error(*args: object, **kwargs: object) -> None:
raise OSError("cannot remove")
monkeypatch.setattr(json, "dump", raise_runtime_error)
monkeypatch.setattr(Path, "unlink", raise_os_error)
path = tmp_path / new_job_file_name()
# the original error wins over the cleanup one
with pytest.raises(RuntimeError, match="boom"):
_write_job_file(path)
assert not path.exists()
assert path.with_name(f".{path.name}.tmp").exists() # could not be removed
def test_jobs_dir(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None:
assert (
job_files_dir(Settings({"REMOTE_CONTROL_JOBS_DIR": str(tmp_path)})) == tmp_path
)
# platformdirs does not determine the user state folder from environment
# variables on every platform, hence the patching.
monkeypatch.setattr(
"scrapy.utils._remote_control.user_state_dir",
lambda *args, **kwargs: str(tmp_path),
)
assert job_files_dir(Settings()) == tmp_path / "job_files"

6
tox.ini
View File

@ -74,12 +74,14 @@ deps =
Pillow==12.3.0
Protego==0.6.2
Twisted==26.4.0
aiohttp==3.14.3
attrs==26.1.0
boto3-stubs[s3]==1.43.41
botocore-stubs==1.43.14
h2==4.3.0
httpx2==2.7.0
itemadapter==0.13.1
platformdirs==4.6.0
ptpython==3.0.32
# newer ones require newer Python
ipython==8.39.0
@ -138,6 +140,7 @@ deps =
pytest==8.4.0
Protego==0.1.15
Twisted==21.7.0
aiohttp==3.13.3
brotli==1.2.0; implementation_name != "pypy"
brotlicffi==1.2.0.0; implementation_name == "pypy"
cryptography==37.0.0
@ -146,6 +149,7 @@ deps =
itemadapter==0.1.0
lxml==4.6.4
parsel==1.5.0
platformdirs==2.0.0
pyOpenSSL==22.0.0
queuelib==1.6.1
service_identity==23.1.0
@ -296,12 +300,14 @@ basepython = pypy3.11
deps =
PyPyDispatcher==2.1.0
{[test-requirements]deps}
aiohttp==3.13.3
pytest==8.4.0
Protego==0.1.15
Twisted==21.7.0
brotlicffi==1.2.0.0
cryptography==44.0.2
cssselect==0.9.1
httpx2==2.0.0
itemadapter==0.1.0
lxml==5.3.2
parsel==1.5.0