fix(scrapy)!: serialize requests and HTTP cache as JSON instead of pickle

Scrapy requests stored in the Apify request queue and responses stored in the
Scrapy HTTP cache were serialized with pickle. Those storages hold JSON, and
pickle reconstructs a Python object graph from the stored bytes. Serialize them
as JSON instead, via a single shared serializer (`_serialization.py`) used by
both the request converter and the HTTP cache.

Only the known binary fields (`body`, `headers`) are base64-encoded; pydantic
models are dumped via `model_dump`. A non-JSON-serializable `meta`/`cb_kwargs`
is reported and the request skipped rather than silently corrupted. A `_class`
entry is only honored when already imported as a `scrapy.Request` subclass, and
malformed or legacy (pickle-format) cache entries are treated as a cache miss
so reads do not crash after the upgrade.

BREAKING CHANGE: requests and HTTP cache entries are now stored as JSON, not
pickle. Entries written by older versions are ignored (re-fetched), not read.

Author: vdusek

src/apify/scrapy/_serialization.py

@@ -0,0 +1,119 @@
+from __future__ import annotations
+
+import base64
+import json
+from typing import Any
+
+from pydantic import BaseModel
+
+# Scrapy persists requests and cached responses by serializing the dict produced by
+# `Request.to_dict()` (and a small response payload for the HTTP cache). The Apify integration
+# stores that payload inside the request queue and the key-value store and reads it back later.
+# Those storages hold JSON, so the payload is serialized as JSON here rather than as a pickled
+# Python object graph: JSON is a plain, portable, interoperable data format and carries no
+# executable object state.
+#
+# JSON cannot represent everything Scrapy emits. Of the values that actually appear, only two are
+# not natively JSON-serializable, and both sit at fixed, known keys with a known type:
+#   - `body`: `bytes`
+#   - `headers`: a `{bytes: [bytes]}` mapping (header name -> list of values)
+# These are base64-encoded in place. Pydantic models (e.g. Crawlee's `UserData`, which the Apify
+# integration injects into `meta['userData']`) are converted via `model_dump()`. Everything else —
+# notably the user-controlled `meta` and `cb_kwargs` — must already be JSON-serializable; if it is
+# not, serialization fails with a clear error rather than silently dropping the request. No in-band
+# sentinel is used for user data, so no legitimate value can collide with the encoding scheme.
+
+
+def encode_to_json(data: dict[str, Any]) -> str:
+    """Serialize a Scrapy request/response dict to a JSON string.
+
+    The binary `body` and `headers` fields are base64-encoded in place. All other fields must be
+    JSON-serializable; pydantic models are dumped to plain dicts. A clear `TypeError` is raised if
+    any remaining value (typically something in `meta` or `cb_kwargs`) cannot be JSON-encoded.
+
+    Args:
+        data: The dict to serialize, e.g. the output of `scrapy.Request.to_dict()`.
+
+    Returns:
+        The JSON-encoded string.
+    """
+    if not isinstance(data, dict):
+        raise TypeError(f'Expected a dict to serialize, got {type(data)}')
+
+    safe = dict(data)
+
+    if isinstance(safe.get('body'), bytes):
+        safe['body'] = base64.b64encode(safe['body']).decode('ascii')
+
+    if isinstance(safe.get('headers'), dict):
+        safe['headers'] = _encode_headers(safe['headers'])
+
+    try:
+        return json.dumps(safe, default=_json_default)
+    except TypeError as exc:
+        raise TypeError(
+            'Failed to JSON-serialize a Scrapy request/response for storage on the Apify platform. '
+            'All values in `meta` and `cb_kwargs` must be JSON-serializable (str, int, float, bool, None, '
+            'list, dict, or a pydantic model).'
+        ) from exc
+
+
+def decode_from_json(text: str) -> Any:
+    """Reconstruct a Scrapy request/response dict from a string produced by `encode_to_json`.
+
+    The base64-encoded `body` and `headers` fields are decoded back to their `bytes` representation.
+
+    Args:
+        text: The JSON-encoded string.
+
+    Returns:
+        The decoded object (a dict for valid request/response payloads).
+    """
+    data = json.loads(text)
+    if not isinstance(data, dict):
+        return data
+
+    if isinstance(data.get('body'), str):
+        data['body'] = base64.b64decode(data['body'])
+
+    if isinstance(data.get('headers'), dict):
+        data['headers'] = _decode_headers(data['headers'])
+
+    return data
+
+
+def _json_default(obj: Any) -> Any:
+    """Fallback for values `json.dumps` cannot serialize on its own.
+
+    Only pydantic models are accepted (and dumped to plain dicts); anything else raises, which
+    `encode_to_json` turns into an actionable error.
+    """
+    if isinstance(obj, BaseModel):
+        return obj.model_dump(by_alias=True)
+    raise TypeError(f'Object of type {type(obj).__name__} is not JSON-serializable')
+
+
+def _encode_headers(headers: dict[Any, Any]) -> dict[str, list[str]]:
+    """Encode a Scrapy `{bytes: [bytes]}` headers mapping to a JSON-safe `{str: [base64-str]}`."""
+    encoded: dict[str, list[str]] = {}
+    for key, value in headers.items():
+        str_key = key.decode('latin-1') if isinstance(key, bytes) else key
+        values = value if isinstance(value, (list, tuple)) else [value]
+        encoded[str_key] = [_b64encode_value(item) for item in values]
+    return encoded
+
+
+def _decode_headers(headers: dict[str, Any]) -> dict[bytes, list[bytes]]:
+    """Reverse `_encode_headers`, restoring the `{bytes: [bytes]}` mapping Scrapy expects."""
+    decoded: dict[bytes, list[bytes]] = {}
+    for key, value in headers.items():
+        bytes_key = key.encode('latin-1') if isinstance(key, str) else key
+        values = value if isinstance(value, list) else [value]
+        decoded[bytes_key] = [base64.b64decode(item) for item in values]
+    return decoded
+
+
+def _b64encode_value(value: Any) -> str:
+    """Base64-encode a single header value, coercing non-bytes values to bytes first."""
+    raw = value if isinstance(value, bytes) else str(value).encode('utf-8')
+    return base64.b64encode(raw).decode('ascii')

src/apify/scrapy/extensions/_httpcache.py

@@ -2,7 +2,6 @@
 
 import gzip
 import io
-import pickle
 import re
 import struct
 from logging import getLogger
@@ -14,6 +13,7 @@
 
 from apify import Configuration
 from apify.scrapy._async_thread import AsyncThread
+from apify.scrapy._serialization import decode_from_json, encode_to_json
 from apify.storage_clients import ApifyStorageClient
 from apify.storages import KeyValueStore
 
@@ -36,7 +36,10 @@ class ApifyCacheStorage:
     """
 
     def __init__(self, settings: BaseSettings) -> None:
-        self._expiration_max_items = 100
+        # Upper bound on how many keys the per-spider-close cleanup sweeps. Configurable because the
+        # cleanup is bounded and best-effort (see `close_spider`); the default keeps the historical
+        # behavior.
+        self._expiration_max_items: int = settings.getint('APIFY_HTTPCACHE_EXPIRATION_MAX_ITEMS', 100)
         self._expiration_secs: int = settings.getint('HTTPCACHE_EXPIRATION_SECS')
         self._spider: Spider | None = None
         self._kvs: KeyValueStore | None = None
@@ -79,8 +82,15 @@ def close_spider(self, _: Spider, current_time: int | None = None) -> None:
             async def expire_kvs() -> None:
                 if self._kvs is None:
                     raise ValueError('Key value store not initialized')
-                i = 0
+                # Best-effort, bounded cleanup: at most `_expiration_max_items` keys are swept per
+                # spider close and `iterate_keys()` order is not guaranteed, so stale entries may
+                # linger across runs. This only reclaims storage; correctness is handled at read
+                # time, where `retrieve_response` treats an expired entry as a cache miss.
+                processed = 0
                 async for item in self._kvs.iterate_keys():
+                    if processed >= self._expiration_max_items:
+                        break
+                    processed += 1
                     value = await self._kvs.get_value(item.key)
                     try:
                         gzip_time = read_gzip_time(value)
@@ -93,9 +103,6 @@ async def expire_kvs() -> None:
                             await self._kvs.set_value(item.key, None)
                         else:
                             logger.debug(f'Valid cache item {item.key}')
-                    if i == self._expiration_max_items:
-                        break
-                    i += 1
 
             self._async_thread.run_coro(expire_kvs())
 
@@ -127,11 +134,18 @@ def retrieve_response(self, _: Spider, request: Request, current_time: int | Non
 
         if current_time is None:
             current_time = int(time())
-        if 0 < self._expiration_secs < current_time - read_gzip_time(value):
-            logger.debug('Cache expired', extra={'request': request})
+
+        # A malformed or legacy (e.g. pickle-format) cache entry must not crash retrieval. Treat it
+        # as a cache miss so Scrapy re-fetches and re-stores it in the current format.
+        try:
+            if 0 < self._expiration_secs < current_time - read_gzip_time(value):
+                logger.debug('Cache expired', extra={'request': request})
+                return None
+            data = from_gzip(value)
+        except Exception as exc:
+            logger.warning(f'Ignoring malformed cache entry {key!r}: {exc}', extra={'request': request})
             return None
 
-        data = from_gzip(value)
         url = data['url']
         status = data['status']
         headers = Headers(data['headers'])
@@ -162,18 +176,26 @@ def store_response(self, _: Spider, request: Request, response: Response) -> Non
 
 
 def to_gzip(data: dict, mtime: int | None = None) -> bytes:
-    """Dump a dictionary to a gzip-compressed byte stream."""
+    """Dump a dictionary to a gzip-compressed byte stream as JSON.
+
+    Cached entries are stored in and read back from the Apify key-value store, which holds JSON, so
+    the payload is serialized as JSON rather than pickled. See `apify.scrapy._serialization` for the
+    encoding details.
+    """
+    payload = encode_to_json(data).encode('utf-8')
     with io.BytesIO() as byte_stream:
         with gzip.GzipFile(fileobj=byte_stream, mode='wb', mtime=mtime) as gzip_file:
-            pickle.dump(data, gzip_file, protocol=4)
+            gzip_file.write(payload)
         return byte_stream.getvalue()
 
 
 def from_gzip(gzip_bytes: bytes) -> dict:
-    """Load a dictionary from a gzip-compressed byte stream."""
+    """Load a dictionary from a gzip-compressed JSON byte stream."""
     with io.BytesIO(gzip_bytes) as byte_stream, gzip.GzipFile(fileobj=byte_stream, mode='rb') as gzip_file:
-        data: dict = pickle.load(gzip_file)
-        return data
+        data = decode_from_json(gzip_file.read().decode('utf-8'))
+    if not isinstance(data, dict):
+        raise TypeError(f'Expected a dict from the cached payload, got {type(data)}')
+    return data
 
 
 def read_gzip_time(gzip_bytes: bytes) -> int:

src/apify/scrapy/requests.py

@@ -1,7 +1,7 @@
 from __future__ import annotations
 
 import codecs
-import pickle
+import sys
 from logging import getLogger
 from typing import Any, cast
 
@@ -13,11 +13,41 @@
 from crawlee._request import UserData
 from crawlee._types import HttpHeaders
 
+from ._serialization import decode_from_json, encode_to_json
 from apify import Request as ApifyRequest
 
 logger = getLogger(__name__)
 
 
+def _ensure_known_request_class(request_dict: dict[str, Any]) -> None:
+    """Validate the optional `_class` entry before `request_from_dict` resolves it.
+
+    `scrapy.utils.request.request_from_dict` resolves a `_class` entry via `load_object`, which
+    imports the dotted path it is given. To keep reconstruction self-contained — importing nothing
+    that the running spider has not already imported — we only accept a `_class` that is already
+    present in `sys.modules` and is a `scrapy.Request` subclass.
+
+    A spider that reads its own requests always has its request classes imported by the time the
+    requests are reconstructed, so this does not restrict legitimate use.
+    """
+    class_path = request_dict.get('_class')
+    if class_path is None:
+        return
+
+    if not isinstance(class_path, str):
+        raise TypeError(f'Invalid scrapy_request `_class`, expected a string, got {type(class_path)}')
+
+    module_name, _, class_name = class_path.rpartition('.')
+    module = sys.modules.get(module_name) if module_name else None
+    request_cls = getattr(module, class_name, None) if module is not None else None
+
+    if not (isinstance(request_cls, type) and issubclass(request_cls, ScrapyRequest)):
+        raise TypeError(
+            f'Refusing to reconstruct a Scrapy request of type {class_path!r}: it is not an already-imported '
+            f'scrapy.Request subclass.'
+        )
+
+
 def to_apify_request(scrapy_request: ScrapyRequest, spider: Spider) -> ApifyRequest | None:
     """Convert a Scrapy request to an Apify request.
 
@@ -35,6 +65,13 @@ def to_apify_request(scrapy_request: ScrapyRequest, spider: Spider) -> ApifyRequ
     logger.debug(f'to_apify_request was called (scrapy_request={scrapy_request})...')
 
     # Configuration to behave as similarly as possible to Scrapy's default RFPDupeFilter.
+    #
+    # `payload` carries the request body, which is used both for platform processing and for
+    # computing the extended unique key. The body is also part of the serialized Scrapy request
+    # stored further below, where it is needed to faithfully reconstruct the request. Both copies
+    # originate from `scrapy_request.body` and are kept intentionally: dropping `payload` would
+    # change deduplication, and dropping the serialized copy would couple reconstruction to the
+    # Apify payload.
     request_kwargs: dict[str, Any] = {
         'url': scrapy_request.url,
         'method': scrapy_request.method,
@@ -67,29 +104,49 @@ def to_apify_request(scrapy_request: ScrapyRequest, spider: Spider) -> ApifyRequ
 
         request_kwargs['user_data'] = user_data if isinstance(user_data, dict) else {}
 
-        # Convert Scrapy's headers to a HttpHeaders and store them in the apify_request
+        # Convert Scrapy's headers to HttpHeaders and store them on the apify_request. This is only
+        # the Apify-platform-level view of the headers; the authoritative copy, with exact bytes,
+        # travels inside the serialized scrapy_request below. `to_unicode_dict()` decodes as UTF-8
+        # and raises on non-UTF-8 header values, so it is guarded: a request with binary headers
+        # keeps them in the serialized payload instead of being dropped entirely.
         if isinstance(scrapy_request.headers, Headers):
-            headers = cast('dict[str, str]', dict(scrapy_request.headers.to_unicode_dict()))
-            request_kwargs['headers'] = HttpHeaders(headers)
+            try:
+                headers = cast('dict[str, str]', dict(scrapy_request.headers.to_unicode_dict()))
+                request_kwargs['headers'] = HttpHeaders(headers)
+            except UnicodeDecodeError:
+                logger.warning(
+                    'Could not represent Scrapy request headers as Apify request headers (non-UTF-8 values); '
+                    'they are preserved in the serialized request instead.'
+                )
         else:
             logger.warning(
                 f'Invalid scrapy_request.headers type, not scrapy.http.headers.Headers: {scrapy_request.headers}'
             )
 
         apify_request = ApifyRequest.from_url(**request_kwargs)
-
-        # Serialize the Scrapy ScrapyRequest and store it in the apify_request.
-        #   - This process involves converting the Scrapy ScrapyRequest object into a dictionary, encoding it to base64,
-        #     and storing it as 'scrapy_request' within the 'userData' dictionary of the apify_request.
-        #   - The serialization process can be referenced at: https://stackoverflow.com/questions/30469575/.
         scrapy_request_dict = scrapy_request.to_dict(spider=spider)
-        scrapy_request_dict_encoded = codecs.encode(pickle.dumps(scrapy_request_dict), 'base64').decode()
-        apify_request.user_data['scrapy_request'] = scrapy_request_dict_encoded
 
     except Exception as exc:
         logger.warning(f'Conversion of Scrapy request {scrapy_request} to Apify request failed; {exc}')
         return None
 
+    # Serialize the Scrapy request and store it (base64-encoded JSON) under 'scrapy_request' in the
+    # Apify request's user data. This is intentionally outside the broad except above so that a
+    # non-JSON-serializable value in `meta`/`cb_kwargs` is reported loudly rather than hidden as a
+    # generic warning. The failure is logged with a full traceback and the request is skipped (None
+    # is returned, honoring this function's contract) instead of crashing the whole crawl. See
+    # `_serialization` for the encoding details.
+    try:
+        scrapy_request_json = encode_to_json(scrapy_request_dict)
+    except TypeError:
+        logger.exception(
+            f'Failed to serialize Scrapy request {scrapy_request} for storage on the Apify platform; skipping it. '
+            'Ensure all values in `meta` and `cb_kwargs` are JSON-serializable.'
+        )
+        return None
+
+    apify_request.user_data['scrapy_request'] = codecs.encode(scrapy_request_json.encode('utf-8'), 'base64').decode()
+
     logger.debug(f'scrapy_request was converted to the apify_request={apify_request}')
     return apify_request
 
@@ -102,14 +159,15 @@ def to_scrapy_request(apify_request: ApifyRequest, spider: Spider) -> ScrapyRequ
         spider: The Scrapy spider that the request is associated with.
 
     Raises:
-        TypeError: If the Apify request is not an instance of the `ApifyRequest` class.
-        ValueError: If the Apify request does not contain the required keys.
+        TypeError: If `apify_request` is not an `ApifyRequest`, if the stored Scrapy request payload
+            is malformed, or if its `_class` does not refer to an already-imported `scrapy.Request`
+            subclass.
 
     Returns:
         The converted Scrapy request.
     """
     if not isinstance(cast('Any', apify_request), ApifyRequest):
-        raise TypeError('apify_request must be a crawlee.ScrapyRequest instance')
+        raise TypeError('apify_request must be an apify.Request instance')
 
     logger.debug(f'to_scrapy_request was called (apify_request={apify_request})...')
 
@@ -124,10 +182,14 @@ def to_scrapy_request(apify_request: ApifyRequest, spider: Spider) -> ScrapyRequ
         if not isinstance(scrapy_request_dict_encoded, str):
             raise TypeError('scrapy_request_dict_encoded must be a string')
 
-        scrapy_request_dict = pickle.loads(codecs.decode(scrapy_request_dict_encoded.encode(), 'base64'))
+        scrapy_request_json = codecs.decode(scrapy_request_dict_encoded.encode(), 'base64').decode('utf-8')
+        scrapy_request_dict = decode_from_json(scrapy_request_json)
         if not isinstance(scrapy_request_dict, dict):
             raise TypeError('scrapy_request_dict must be a dictionary')
 
+        # Validate any `_class` entry before request_from_dict resolves and imports it.
+        _ensure_known_request_class(scrapy_request_dict)
+
         scrapy_request = request_from_dict(scrapy_request_dict, spider=spider)
         if not isinstance(scrapy_request, ScrapyRequest):
             raise TypeError('scrapy_request must be an instance of the ScrapyRequest class')