Add SSLKEYLOGFILE support

docs/topics/debug.rst

@@ -153,6 +153,26 @@ available in all future runs should they be necessary again:
 
 For more information, check the :ref:`topics-logging` section.
 
+.. _debug-tls:
+
+Decrypting TLS traffic
+======================
+
+Scrapy writes the session keys of its HTTPS connections to the file that the
+``SSLKEYLOGFILE`` environment variable points to, using the `NSS key log
+format`_ that traffic analysis tools such as Wireshark understand.
+
+.. versionadded:: VERSION
+
+.. code-block:: shell
+
+    SSLKEYLOGFILE=/tmp/sslkeylog scrapy crawl myspider
+
+.. _NSS key log format: https://firefox-source-docs.mozilla.org/security/nss/legacy/key_log_format/index.html
+
+.. warning:: Anyone who can read the key log file can decrypt the traffic of
+    the connections recorded in it, including any credentials that they carry.
+
 .. _debug-vscode:
 
 Visual Studio Code

scrapy/core/downloader/contextfactory.py

@@ -24,7 +24,12 @@ from scrapy.exceptions import ScrapyDeprecationWarning
 from scrapy.utils._deps_compat import TWISTED_TLS_NEW_IMPL
 from scrapy.utils.deprecate import create_deprecated_class
 from scrapy.utils.misc import build_from_crawler, load_object
-from scrapy.utils.ssl import _get_cert_options_version_kwargs, _get_tls_version_limits
+from scrapy.utils.ssl import (
+    _get_cert_options_version_kwargs,
+    _get_keylog_filename,
+    _get_tls_version_limits,
+    _set_keylog_callback,
+)
 
 if TYPE_CHECKING:
     from twisted.internet._sslverify import ClientTLSOptions
@@ -154,10 +159,22 @@ class _ScrapyClientContextFactory(BrowserLikePolicyForHTTPS):
                 self._get_context(),  # type: ignore[arg-type]
             )
         # Otherwise use the normal Twisted function.
-        return optionsForClientTLS(  # type: ignore[no-any-return]
+        creator = optionsForClientTLS(
             hostname=hostname.decode("ascii"),
             extraCertificateOptions=self._get_cert_options_kwargs(),
         )
+        if _get_keylog_filename():
+            # This creator builds its own certificate options, so its context
+            # doesn't come from _ScrapyCertificateOptions.
+            _set_keylog_callback(_get_creator_context(creator))
+        return cast("ClientTLSOptions", creator)
+
+
+def _get_creator_context(creator: Any) -> SSL.Context:
+    """Return the context that *creator* uses for its connections."""
+    if TWISTED_TLS_NEW_IMPL:
+        return cast("SSL.Context", creator._createConnection.__self__.getContext())
+    return cast("SSL.Context", creator._ctx)
 
 
 ScrapyClientContextFactory = create_deprecated_class(
@@ -264,6 +281,7 @@ class _ScrapyCertificateOptions(CertificateOptions):
         else:
             ctx = super()._makeContext()
         ctx.set_options(0x4)  # OP_LEGACY_SERVER_CONNECT
+        _set_keylog_callback(ctx)
         return ctx
 
 

scrapy/utils/ssl.py

@@ -1,8 +1,11 @@
 from __future__ import annotations
 
 import logging
+import os
 import ssl
+import sys
 import warnings
+from pathlib import Path
 from typing import TYPE_CHECKING, Any, TypedDict, TypeVar
 
 import OpenSSL._util as pyOpenSSLutil
@@ -53,6 +56,13 @@ def _get_tls_version_limits(
     )
 
 
+def _get_keylog_filename() -> str | None:
+    """Return the path where TLS session keys should be logged, or ``None``."""
+    if sys.flags.ignore_environment:
+        return None
+    return os.environ.get("SSLKEYLOGFILE")
+
+
 # stdlib ssl module utils
 
 _STDLIB_VERSION_MAP: dict[str, ssl.TLSVersion] = {
@@ -89,6 +99,8 @@ def _make_ssl_context(settings: BaseSettings) -> ssl.SSLContext:
         ctx.maximum_version = tls_max_ver
     if ciphers_setting:
         ctx.set_ciphers(ciphers_setting)
+    if keylog_filename := _get_keylog_filename():
+        ctx.keylog_filename = keylog_filename
     return ctx
 
 
@@ -101,6 +113,8 @@ def _make_insecure_ssl_ctx() -> ssl.SSLContext:
     ctx = ssl.SSLContext(ssl.PROTOCOL_TLS_CLIENT)
     ctx.check_hostname = False
     ctx.verify_mode = ssl.CERT_NONE
+    if keylog_filename := _get_keylog_filename():
+        ctx.keylog_filename = keylog_filename
     return ctx
 
 
@@ -121,6 +135,20 @@ def _log_sslobj_debug_info(sslobj: ssl.SSLObject) -> None:
 # pyOpenSSL utils
 
 
+def _set_keylog_callback(ctx: OpenSSL.SSL.Context) -> None:
+    """Make *ctx* write TLS session keys to the key log file, if enabled."""
+    keylog_filename = _get_keylog_filename()
+    if not keylog_filename:
+        return
+    keylog_path = Path(keylog_filename)
+
+    def write_keylog_line(connection: OpenSSL.SSL.Connection, line: bytes) -> None:
+        with keylog_path.open("ab") as f:
+            f.write(line + b"\n")
+
+    ctx.set_keylog_callback(write_keylog_line)
+
+
 def _ffi_buf_to_string(buf: Any) -> str:
     return to_unicode(pyOpenSSLutil.ffi.string(buf))
 

tests/test_core_downloader.py

@@ -36,6 +36,8 @@ from tests.mockserver.utils import ssl_context_factory
 from tests.utils.decorators import coroutine_test
 
 if TYPE_CHECKING:
+    from pathlib import Path
+
     from twisted.internet.defer import Deferred
     from twisted.internet.interfaces import IListeningPort
     from twisted.web.iweb import IBodyProducer
@@ -174,6 +176,25 @@ class TestContextFactory(TestContextFactoryBase):
             )
             factory.creatorForNetloc(b"website.tld", 443)
 
+    @pytest.mark.parametrize("verify_certs", [False, True])
+    def test_keylog(
+        self,
+        monkeypatch: pytest.MonkeyPatch,
+        tmp_path: Path,
+        verify_certs: bool,
+    ) -> None:
+        """The key log callback should be set with and without verification."""
+        monkeypatch.setenv("SSLKEYLOGFILE", str(tmp_path / "keylog"))
+        crawler = get_crawler(
+            settings_dict={"DOWNLOAD_VERIFY_CERTIFICATES": verify_certs}
+        )
+        factory: _ScrapyClientContextFactory = _load_context_factory_from_settings(
+            crawler
+        )
+        creator = factory.creatorForNetloc(b"website.tld", 443)
+        conn = creator.clientConnectionForTLS(self._get_dummy_protocol())
+        assert conn.get_context()._keylog_callback is not None
+
     def test_ctx_flags(self, factory: _ScrapyClientContextFactory) -> None:
         """The context should have the expected flags set."""
         creator = factory.creatorForNetloc(b"website.tld", 443)

tests/utils/bases/download_handlers_http.py

@@ -55,6 +55,7 @@ from tests.utils.decorators import coroutine_test
 
 if TYPE_CHECKING:
     from collections.abc import AsyncGenerator, Generator
+    from pathlib import Path
 
     from scrapy.core.downloader.handlers import DownloadHandlerProtocol
     from tests.mockserver.http import MockServer
@@ -862,6 +863,28 @@ class TestHttpsBase(TestHttpBase):
         assert response.body == b"Works"
         assert self.tls_log_message in caplog.text
 
+    @coroutine_test
+    async def test_keylog(
+        self,
+        mockserver: MockServer,
+        monkeypatch: pytest.MonkeyPatch,
+        tmp_path: Path,
+    ) -> None:
+        keylog_file = tmp_path / "keylog"
+        monkeypatch.setenv("SSLKEYLOGFILE", str(keylog_file))
+        request = Request(mockserver.url("/text", is_secure=self.is_secure))
+        async with self.get_dh() as download_handler:
+            response = await download_handler.download_request(request)
+        assert response.body == b"Works"
+        # The stdlib writes a comment line at the top of the file.
+        lines = [
+            line
+            for line in keylog_file.read_text().splitlines()
+            if not line.startswith("#")
+        ]
+        assert lines
+        assert all(len(line.split()) == 3 for line in lines)
+
     @coroutine_test
     async def test_verify_certs_deprecated(self, mockserver: MockServer) -> None:
         request = Request(mockserver.url("/text", is_secure=self.is_secure))