TLS code modernization (#7353)

* Import some stuff from service_identity directly.

* Extract _log_tls().

* Sync validating IP addresses.

* Introduce _deps_compat.

* Deprecate getCertificateOptions().

* Deprecate ScrapyClientTLSOptions.

* Add pragma: no cover.

* verify=False and trustRoot=platformTrust() are the defaults.

* Improve docstrings for TLS classes.

* Deprecate AcceptableProtocolsContextFactory.

* Switch from DefaultOpenSSLContextFactory to CertificateOptions.

* Deprecate load_context_factory_from_settings().

* Remove the outdated warning about incompatible factories.

* Simplify _load_context_factory_from_settings().

* Make CertificateOptions and Context once per factory.

* pragma: no cover
This commit is contained in:
Andrey Rakhmatullin 2026-03-25 15:42:43 +05:00 committed by GitHub
parent 03d105ac92
commit 2ce02d417a
No known key found for this signature in database
GPG Key ID: B5690EEEBB952194
10 changed files with 176 additions and 127 deletions

View File

@ -9,7 +9,6 @@ from twisted.internet.ssl import (
AcceptableCiphers, AcceptableCiphers,
CertificateOptions, CertificateOptions,
optionsForClientTLS, optionsForClientTLS,
platformTrust,
) )
from twisted.web.client import BrowserLikePolicyForHTTPS from twisted.web.client import BrowserLikePolicyForHTTPS
from twisted.web.iweb import IPolicyForHTTPS from twisted.web.iweb import IPolicyForHTTPS
@ -18,11 +17,11 @@ from zope.interface.verify import verifyObject
from scrapy.core.downloader.tls import ( from scrapy.core.downloader.tls import (
DEFAULT_CIPHERS, DEFAULT_CIPHERS,
ScrapyClientTLSOptions, _ScrapyClientTLSOptions,
openssl_methods, openssl_methods,
) )
from scrapy.exceptions import ScrapyDeprecationWarning from scrapy.exceptions import ScrapyDeprecationWarning
from scrapy.utils.deprecate import method_is_overridden from scrapy.utils.deprecate import create_deprecated_class, method_is_overridden
from scrapy.utils.misc import build_from_crawler, load_object from scrapy.utils.misc import build_from_crawler, load_object
if TYPE_CHECKING: if TYPE_CHECKING:
@ -37,14 +36,14 @@ if TYPE_CHECKING:
@implementer(IPolicyForHTTPS) @implementer(IPolicyForHTTPS)
class ScrapyClientContextFactory(BrowserLikePolicyForHTTPS): class ScrapyClientContextFactory(BrowserLikePolicyForHTTPS):
""" """Non-peer-certificate verifying HTTPS context factory.
Non-peer-certificate verifying HTTPS context factory
Default OpenSSL method is TLS_METHOD (also called SSLv23_METHOD) Default OpenSSL method is ``TLS_METHOD`` (also called ``SSLv23_METHOD``)
which allows TLS protocol negotiation which allows TLS protocol negotiation.
'A TLS/SSL connection established with [this method] may The purpose of this custom class is to provide a ``creatorForNetloc()``
understand the TLSv1, TLSv1.1 and TLSv1.2 protocols.' method that returns a ``_ScrapyClientTLSOptions`` instance configured based
on TLS settings provided to the factory.
""" """
def __init__( def __init__(
@ -63,6 +62,12 @@ class ScrapyClientContextFactory(BrowserLikePolicyForHTTPS):
self.tls_ciphers = AcceptableCiphers.fromOpenSSLCipherString(tls_ciphers) self.tls_ciphers = AcceptableCiphers.fromOpenSSLCipherString(tls_ciphers)
else: else:
self.tls_ciphers = DEFAULT_CIPHERS self.tls_ciphers = DEFAULT_CIPHERS
self._certificate_options = CertificateOptions(
method=self._ssl_method,
fixBrokenPeers=True,
acceptableCiphers=self.tls_ciphers,
)
self._ctx = self._get_context()
if method_is_overridden(type(self), ScrapyClientContextFactory, "getContext"): if method_is_overridden(type(self), ScrapyClientContextFactory, "getContext"):
warnings.warn( warnings.warn(
"Overriding ScrapyClientContextFactory.getContext() is deprecated and that method" "Overriding ScrapyClientContextFactory.getContext() is deprecated and that method"
@ -70,6 +75,15 @@ class ScrapyClientContextFactory(BrowserLikePolicyForHTTPS):
category=ScrapyDeprecationWarning, category=ScrapyDeprecationWarning,
stacklevel=2, stacklevel=2,
) )
if method_is_overridden(
type(self), ScrapyClientContextFactory, "getCertificateOptions"
): # pragma: no cover
warnings.warn(
"Overriding ScrapyClientContextFactory.getCertificateOptions() is deprecated and that method"
" will be removed in a future Scrapy version. Override creatorForNetloc() instead.",
category=ScrapyDeprecationWarning,
stacklevel=2,
)
@classmethod @classmethod
def from_crawler( def from_crawler(
@ -91,27 +105,33 @@ class ScrapyClientContextFactory(BrowserLikePolicyForHTTPS):
**kwargs, **kwargs,
) )
def getCertificateOptions(self) -> CertificateOptions: def getCertificateOptions(self) -> CertificateOptions: # pragma: no cover
# setting verify=True will require you to provide CAs warnings.warn(
# to verify against; in other words: it's not that simple "ScrapyClientContextFactory.getCertificateOptions() is deprecated.",
return CertificateOptions( ScrapyDeprecationWarning,
verify=False, stacklevel=2,
method=self._ssl_method,
fixBrokenPeers=True,
acceptableCiphers=self.tls_ciphers,
) )
return self._certificate_options
# kept for old-style HTTP/1.0 downloader context twisted calls, # kept for old-style HTTP/1.0 downloader context twisted calls,
# e.g. connectSSL() # e.g. connectSSL()
def getContext(self, hostname: Any = None, port: Any = None) -> SSL.Context: def getContext(self, hostname: Any = None, port: Any = None) -> SSL.Context:
ctx: SSL.Context = self.getCertificateOptions().getContext() warnings.warn(
"ScrapyClientContextFactory.getContext() is deprecated.",
ScrapyDeprecationWarning,
stacklevel=2,
)
return self._ctx
def _get_context(self) -> SSL.Context:
ctx = self._certificate_options.getContext()
ctx.set_options(0x4) # OP_LEGACY_SERVER_CONNECT ctx.set_options(0x4) # OP_LEGACY_SERVER_CONNECT
return ctx return ctx
def creatorForNetloc(self, hostname: bytes, port: int) -> ClientTLSOptions: def creatorForNetloc(self, hostname: bytes, port: int) -> ClientTLSOptions:
return ScrapyClientTLSOptions( return _ScrapyClientTLSOptions(
hostname.decode("ascii"), hostname.decode("ascii"),
self.getContext(), self._ctx,
verbose_logging=self.tls_verbose_logging, verbose_logging=self.tls_verbose_logging,
) )
@ -133,25 +153,28 @@ class BrowserLikeContextFactory(ScrapyClientContextFactory):
The default OpenSSL method is ``TLS_METHOD`` (also called The default OpenSSL method is ``TLS_METHOD`` (also called
``SSLv23_METHOD``) which allows TLS protocol negotiation. ``SSLv23_METHOD``) which allows TLS protocol negotiation.
As this overrides the parent ``creatorForNetloc()`` method, only
``self._ssl_method`` is used from the parent class.
""" """
def creatorForNetloc(self, hostname: bytes, port: int) -> ClientTLSOptions: def creatorForNetloc(self, hostname: bytes, port: int) -> ClientTLSOptions:
# trustRoot set to platformTrust() will use the platform's root CAs.
#
# This means that a website like https://www.cacert.org will be rejected
# by default, since CAcert.org CA certificate is seldom shipped.
return optionsForClientTLS( return optionsForClientTLS(
hostname=hostname.decode("ascii"), hostname=hostname.decode("ascii"),
trustRoot=platformTrust(),
extraCertificateOptions={"method": self._ssl_method}, extraCertificateOptions={"method": self._ssl_method},
) )
@implementer(IPolicyForHTTPS) @implementer(IPolicyForHTTPS)
class AcceptableProtocolsContextFactory: class _AcceptableProtocolsContextFactory:
"""Context factory to used to override the acceptable protocols """Context factory to used to override the acceptable protocols
to set up the [OpenSSL.SSL.Context] for doing NPN and/or ALPN to set up the :class:`OpenSSL.SSL.Context` for doing ALPN negotiation.
negotiation. It's a private class for :class:`~.H2DownloadHandler`.
This class wraps ``creatorForNetloc()`` of another factory class, setting
the acceptable protocols on the :class:`.ClientTLSOptions` instance
returned by it. It's only needed because we support custom factories via
:setting:`DOWNLOADER_CLIENTCONTEXTFACTORY`.
""" """
def __init__(self, context_factory: Any, acceptable_protocols: list[bytes]): def __init__(self, context_factory: Any, acceptable_protocols: list[bytes]):
@ -167,31 +190,36 @@ class AcceptableProtocolsContextFactory:
return options return options
AcceptableProtocolsContextFactory = create_deprecated_class(
"AcceptableProtocolsContextFactory",
_AcceptableProtocolsContextFactory,
subclass_warn_message="{old} is deprecated.",
instance_warn_message="{cls} is deprecated.",
)
def _load_context_factory_from_settings(crawler: Crawler) -> IPolicyForHTTPS:
"""Create an instance of :setting:`DOWNLOADER_CLIENTCONTEXTFACTORY`.
Also passes values of other relevant settings to the factory class.
"""
ssl_method = openssl_methods[crawler.settings.get("DOWNLOADER_CLIENT_TLS_METHOD")]
context_factory_cls = load_object(
crawler.settings["DOWNLOADER_CLIENTCONTEXTFACTORY"]
)
return build_from_crawler(
context_factory_cls,
crawler,
method=ssl_method,
)
def load_context_factory_from_settings( def load_context_factory_from_settings(
settings: BaseSettings, crawler: Crawler settings: BaseSettings, crawler: Crawler
) -> IPolicyForHTTPS: ) -> IPolicyForHTTPS: # pragma: no cover
ssl_method = openssl_methods[settings.get("DOWNLOADER_CLIENT_TLS_METHOD")] warnings.warn(
context_factory_cls = load_object(settings["DOWNLOADER_CLIENTCONTEXTFACTORY"]) "load_context_factory_from_settings() is deprecated.",
# try method-aware context factory ScrapyDeprecationWarning,
try: stacklevel=2,
context_factory = build_from_crawler( )
context_factory_cls, return _load_context_factory_from_settings(crawler)
crawler,
method=ssl_method,
)
except TypeError:
# use context factory defaults
context_factory = build_from_crawler(
context_factory_cls,
crawler,
)
msg = (
f"{settings['DOWNLOADER_CLIENTCONTEXTFACTORY']} does not accept "
"a `method` argument (type OpenSSL.SSL method, e.g. "
"OpenSSL.SSL.SSLv23_METHOD) and/or a `tls_verbose_logging` "
"argument and/or a `tls_ciphers` argument. Please, upgrade your "
"context factory class to handle them or ignore them."
)
warnings.warn(msg)
return context_factory

View File

@ -30,7 +30,7 @@ from twisted.web.iweb import UNKNOWN_LENGTH, IBodyProducer, IPolicyForHTTPS, IRe
from zope.interface import implementer from zope.interface import implementer
from scrapy import Request, signals from scrapy import Request, signals
from scrapy.core.downloader.contextfactory import load_context_factory_from_settings from scrapy.core.downloader.contextfactory import _load_context_factory_from_settings
from scrapy.exceptions import ( from scrapy.exceptions import (
DownloadCancelledError, DownloadCancelledError,
DownloadTimeoutError, DownloadTimeoutError,
@ -94,8 +94,8 @@ class HTTP11DownloadHandler(BaseHttpDownloadHandler):
) )
self._pool._factory.noisy = False self._pool._factory.noisy = False
self._contextFactory: IPolicyForHTTPS = load_context_factory_from_settings( self._contextFactory: IPolicyForHTTPS = _load_context_factory_from_settings(
crawler.settings, crawler crawler
) )
self._bind_address = crawler.settings.get("DOWNLOAD_BIND_ADDRESS") self._bind_address = crawler.settings.get("DOWNLOAD_BIND_ADDRESS")
self._disconnect_timeout: int = 1 self._disconnect_timeout: int = 1

View File

@ -6,7 +6,7 @@ from urllib.parse import urldefrag
from twisted.web.client import URI from twisted.web.client import URI
from scrapy.core.downloader.contextfactory import load_context_factory_from_settings from scrapy.core.downloader.contextfactory import _load_context_factory_from_settings
from scrapy.core.downloader.handlers.base import BaseDownloadHandler from scrapy.core.downloader.handlers.base import BaseDownloadHandler
from scrapy.core.http2.agent import H2Agent, H2ConnectionPool, ScrapyProxyH2Agent from scrapy.core.http2.agent import H2Agent, H2ConnectionPool, ScrapyProxyH2Agent
from scrapy.exceptions import DownloadTimeoutError, NotConfigured from scrapy.exceptions import DownloadTimeoutError, NotConfigured
@ -40,9 +40,7 @@ class H2DownloadHandler(BaseDownloadHandler):
from twisted.internet import reactor from twisted.internet import reactor
self._pool = H2ConnectionPool(reactor, crawler.settings) self._pool = H2ConnectionPool(reactor, crawler.settings)
self._context_factory = load_context_factory_from_settings( self._context_factory = _load_context_factory_from_settings(crawler)
crawler.settings, crawler
)
self._bind_address = crawler.settings.get("DOWNLOAD_BIND_ADDRESS") self._bind_address = crawler.settings.get("DOWNLOAD_BIND_ADDRESS")
async def download_request(self, request: Request) -> Response: async def download_request(self, request: Request) -> Response:

View File

@ -2,14 +2,14 @@ import logging
from typing import Any from typing import Any
from OpenSSL import SSL from OpenSSL import SSL
from OpenSSL.SSL import Connection
from service_identity import VerificationError
from service_identity.exceptions import CertificateError from service_identity.exceptions import CertificateError
from twisted.internet._sslverify import ( from service_identity.pyopenssl import verify_hostname, verify_ip_address
ClientTLSOptions, from twisted.internet._sslverify import ClientTLSOptions
VerificationError,
verifyHostname,
)
from twisted.internet.ssl import AcceptableCiphers from twisted.internet.ssl import AcceptableCiphers
from scrapy.utils.deprecate import create_deprecated_class
from scrapy.utils.ssl import get_temp_key_info, x509name_to_string from scrapy.utils.ssl import get_temp_key_info, x509name_to_string
logger = logging.getLogger(__name__) logger = logging.getLogger(__name__)
@ -29,15 +29,38 @@ openssl_methods: dict[str, int] = {
} }
class ScrapyClientTLSOptions(ClientTLSOptions): def _log_tls(hostname: str, connection: Connection) -> None:
logger.debug(
"SSL connection to %s using protocol %s, cipher %s",
hostname,
connection.get_protocol_version_name(),
connection.get_cipher_name(),
)
server_cert = connection.get_peer_certificate()
if server_cert:
logger.debug(
'SSL connection certificate: issuer "%s", subject "%s"',
x509name_to_string(server_cert.get_issuer()),
x509name_to_string(server_cert.get_subject()),
)
key_info = get_temp_key_info(connection._ssl)
if key_info:
logger.debug("SSL temp key: %s", key_info)
class _ScrapyClientTLSOptions(ClientTLSOptions):
""" """
SSL Client connection creator ignoring certificate verification errors SSL Client connection creator ignoring certificate verification errors
(for genuinely invalid certificates or bugs in verification code). (for genuinely invalid certificates or bugs in verification code) and
optionally logging TLS details of the connection.
Same as Twisted's private _sslverify.ClientTLSOptions, Same as Twisted's private _sslverify.ClientTLSOptions,
except that VerificationError, CertificateError and ValueError except that VerificationError, CertificateError and ValueError
exceptions are caught, so that the connection is not closed, only exceptions are caught, so that the connection is not closed, only
logging warnings. Also, HTTPS connection parameters logging is added. logging warnings. Also, HTTPS connection parameters logging is added.
Instances of this class are returned from
:class:`.ScrapyClientContextFactory`.
""" """
def __init__(self, hostname: str, ctx: SSL.Context, verbose_logging: bool = False): def __init__(self, hostname: str, ctx: SSL.Context, verbose_logging: bool = False):
@ -47,36 +70,23 @@ class ScrapyClientTLSOptions(ClientTLSOptions):
def _identityVerifyingInfoCallback( def _identityVerifyingInfoCallback(
self, connection: SSL.Connection, where: int, ret: Any self, connection: SSL.Connection, where: int, ret: Any
) -> None: ) -> None:
if where & SSL.SSL_CB_HANDSHAKE_START: if where & SSL.SSL_CB_HANDSHAKE_START and self._hostnameIsDnsName:
connection.set_tlsext_host_name(self._hostnameBytes) connection.set_tlsext_host_name(self._hostnameBytes)
elif where & SSL.SSL_CB_HANDSHAKE_DONE: elif where & SSL.SSL_CB_HANDSHAKE_DONE:
if self.verbose_logging: if self.verbose_logging:
logger.debug( _log_tls(self._hostnameASCII, connection)
"SSL connection to %s using protocol %s, cipher %s",
self._hostnameASCII,
connection.get_protocol_version_name(),
connection.get_cipher_name(),
)
server_cert = connection.get_peer_certificate()
if server_cert:
logger.debug(
'SSL connection certificate: issuer "%s", subject "%s"',
x509name_to_string(server_cert.get_issuer()),
x509name_to_string(server_cert.get_subject()),
)
key_info = get_temp_key_info(connection._ssl)
if key_info:
logger.debug("SSL temp key: %s", key_info)
try: try:
verifyHostname(connection, self._hostnameASCII) if self._hostnameIsDnsName:
verify_hostname(connection, self._hostnameASCII)
else:
verify_ip_address(connection, self._hostnameASCII)
except (CertificateError, VerificationError) as e: except (CertificateError, VerificationError) as e:
logger.warning( logger.warning(
'Remote certificate is not valid for hostname "%s"; %s', 'Remote certificate is not valid for hostname "%s"; %s',
self._hostnameASCII, self._hostnameASCII,
e, e,
) )
except ValueError as e: except ValueError as e:
logger.warning( logger.warning(
"Ignoring error while verifying certificate " "Ignoring error while verifying certificate "
@ -86,6 +96,14 @@ class ScrapyClientTLSOptions(ClientTLSOptions):
) )
ScrapyClientTLSOptions = create_deprecated_class(
"ScrapyClientTLSOptions",
_ScrapyClientTLSOptions,
subclass_warn_message="{old} is deprecated.",
instance_warn_message="{cls} is deprecated.",
)
DEFAULT_CIPHERS: AcceptableCiphers = AcceptableCiphers.fromOpenSSLCipherString( DEFAULT_CIPHERS: AcceptableCiphers = AcceptableCiphers.fromOpenSSLCipherString(
"DEFAULT" "DEFAULT"
) )

View File

@ -14,7 +14,7 @@ from twisted.web.client import (
) )
from twisted.web.error import SchemeNotSupported from twisted.web.error import SchemeNotSupported
from scrapy.core.downloader.contextfactory import AcceptableProtocolsContextFactory from scrapy.core.downloader.contextfactory import _AcceptableProtocolsContextFactory
from scrapy.core.http2.protocol import H2ClientFactory, H2ClientProtocol from scrapy.core.http2.protocol import H2ClientFactory, H2ClientProtocol
if TYPE_CHECKING: if TYPE_CHECKING:
@ -126,7 +126,7 @@ class H2Agent:
) -> None: ) -> None:
self._reactor = reactor self._reactor = reactor
self._pool = pool self._pool = pool
self._context_factory = AcceptableProtocolsContextFactory( self._context_factory = _AcceptableProtocolsContextFactory(
context_factory, acceptable_protocols=[b"h2"] context_factory, acceptable_protocols=[b"h2"]
) )
self.endpoint_factory = _StandardEndpointFactory( self.endpoint_factory = _StandardEndpointFactory(

View File

@ -8,13 +8,12 @@ from abc import ABC, abstractmethod
from collections import defaultdict from collections import defaultdict
from typing import TYPE_CHECKING, Any, Literal, TypeAlias, TypedDict, cast from typing import TYPE_CHECKING, Any, Literal, TypeAlias, TypedDict, cast
from twisted import version as twisted_version
from twisted.internet.defer import Deferred, DeferredList from twisted.internet.defer import Deferred, DeferredList
from twisted.python.failure import Failure from twisted.python.failure import Failure
from twisted.python.versions import Version
from scrapy.exceptions import ScrapyDeprecationWarning from scrapy.exceptions import ScrapyDeprecationWarning
from scrapy.http.request import NO_CALLBACK, Request from scrapy.http.request import NO_CALLBACK, Request
from scrapy.utils._deps_compat import TWISTED_FAILURE_HAS_STACK
from scrapy.utils.asyncio import call_later, is_asyncio_available from scrapy.utils.asyncio import call_later, is_asyncio_available
from scrapy.utils.datatypes import SequenceExclude from scrapy.utils.datatypes import SequenceExclude
from scrapy.utils.decorators import _warn_spider_arg from scrapy.utils.decorators import _warn_spider_arg
@ -232,7 +231,7 @@ class MediaPipeline(ABC):
# minimize cached information for failure # minimize cached information for failure
result.cleanFailure() result.cleanFailure()
result.frames = [] result.frames = []
if twisted_version < Version("twisted", 24, 10, 0): if TWISTED_FAILURE_HAS_STACK:
result.stack = [] # type: ignore[method-assign] result.stack = [] # type: ignore[method-assign]
# This code fixes a memory leak by avoiding to keep references to # This code fixes a memory leak by avoiding to keep references to
# the Request and Response objects on the Media Pipeline cache. # the Request and Response objects on the Media Pipeline cache.

View File

@ -0,0 +1,10 @@
from OpenSSL import __version__ as PYOPENSSL_VERSION_STRING
from packaging.version import Version
from twisted import version as TWISTED_VERSION
from twisted.python.versions import Version as TxVersion
TWISTED_FAILURE_HAS_STACK = TWISTED_VERSION < TxVersion("twisted", 24, 10, 0)
PYOPENSSL_VERSION = Version(PYOPENSSL_VERSION_STRING)
# SSL.Context.use_certificate wants an X509 object, SSL.Context.use_privatekey wants a PKey object
PYOPENSSL_WANTS_X509_PKEY = PYOPENSSL_VERSION < Version("24.3.0")

View File

@ -8,9 +8,6 @@ import os
import socket import socket
from pathlib import Path from pathlib import Path
from twisted import version as TWISTED_VERSION
from twisted.python.versions import Version
# ignore system-wide proxies for tests # ignore system-wide proxies for tests
# which would send requests to a totally unsuspecting server # which would send requests to a totally unsuspecting server
# (e.g. because urllib does not fully understand the proxy spec) # (e.g. because urllib does not fully understand the proxy spec)
@ -33,6 +30,3 @@ except socket.gaierror:
def get_testdata(*paths: str) -> bytes: def get_testdata(*paths: str) -> bytes:
"""Return test data""" """Return test data"""
return Path(tests_datadir, *paths).read_bytes() return Path(tests_datadir, *paths).read_bytes()
TWISTED_KEEPS_TRACEBACKS = TWISTED_VERSION >= Version("twisted", 24, 10, 0)

View File

@ -1,25 +1,35 @@
from __future__ import annotations from __future__ import annotations
from pathlib import Path from pathlib import Path
from typing import TYPE_CHECKING
from cryptography.hazmat.primitives.serialization import load_pem_private_key
from cryptography.x509 import load_pem_x509_certificate
from OpenSSL import SSL from OpenSSL import SSL
from twisted.internet import ssl from OpenSSL.crypto import FILETYPE_PEM, load_certificate, load_privatekey
from twisted.internet.ssl import CertificateOptions, ContextFactory
from scrapy.utils._deps_compat import PYOPENSSL_WANTS_X509_PKEY
from scrapy.utils.python import to_bytes from scrapy.utils.python import to_bytes
if TYPE_CHECKING:
from twisted.internet.ssl import ContextFactory
def ssl_context_factory( def ssl_context_factory(
keyfile: str = "keys/localhost.key", keyfile: str = "keys/localhost.key",
certfile: str = "keys/localhost.crt", certfile: str = "keys/localhost.crt",
cipher_string: str | None = None, cipher_string: str | None = None,
) -> ContextFactory: ) -> ContextFactory:
factory = ssl.DefaultOpenSSLContextFactory( keyfile_path = Path(__file__).parent.parent / keyfile
str(Path(__file__).parent.parent / keyfile), certfile_path = Path(__file__).parent.parent / certfile
str(Path(__file__).parent.parent / certfile),
if not PYOPENSSL_WANTS_X509_PKEY:
cert = load_pem_x509_certificate(certfile_path.read_bytes())
key = load_pem_private_key(keyfile_path.read_bytes(), password=None)
else:
cert = load_certificate(FILETYPE_PEM, certfile_path.read_bytes()) # type: ignore[assignment]
key = load_privatekey(FILETYPE_PEM, keyfile_path.read_bytes()) # type: ignore[assignment]
factory = CertificateOptions(
privateKey=key,
certificate=cert,
) )
if cipher_string: if cipher_string:
ctx = factory.getContext() ctx = factory.getContext()

View File

@ -13,11 +13,10 @@ from twisted.web.client import Response as TxResponse
from scrapy.core.downloader import Downloader, Slot from scrapy.core.downloader import Downloader, Slot
from scrapy.core.downloader.contextfactory import ( from scrapy.core.downloader.contextfactory import (
ScrapyClientContextFactory, ScrapyClientContextFactory,
load_context_factory_from_settings, _load_context_factory_from_settings,
) )
from scrapy.core.downloader.handlers.http11 import _RequestBodyProducer from scrapy.core.downloader.handlers.http11 import _RequestBodyProducer
from scrapy.exceptions import ScrapyDeprecationWarning from scrapy.exceptions import ScrapyDeprecationWarning
from scrapy.settings import Settings
from scrapy.utils.defer import maybe_deferred_to_future from scrapy.utils.defer import maybe_deferred_to_future
from scrapy.utils.misc import build_from_crawler from scrapy.utils.misc import build_from_crawler
from scrapy.utils.python import to_bytes from scrapy.utils.python import to_bytes
@ -102,8 +101,7 @@ class TestContextFactory(TestContextFactoryBase):
async def testPayload(self, server_url: str) -> None: async def testPayload(self, server_url: str) -> None:
s = "0123456789" * 10 s = "0123456789" * 10
crawler = get_crawler() crawler = get_crawler()
settings = Settings() client_context_factory = _load_context_factory_from_settings(crawler)
client_context_factory = load_context_factory_from_settings(settings, crawler)
body = await self.get_page( body = await self.get_page(
server_url + "payload", client_context_factory, body=s server_url + "payload", client_context_factory, body=s
) )
@ -117,13 +115,11 @@ class TestContextFactory(TestContextFactoryBase):
ctx: OpenSSL.SSL.Context = super().getContext(hostname, port) ctx: OpenSSL.SSL.Context = super().getContext(hostname, port)
return ctx return ctx
with warnings.catch_warnings(record=True) as w: with pytest.warns(
ScrapyDeprecationWarning,
match=r"ScrapyClientContextFactory\.getContext\(\) is deprecated",
):
MyFactory() MyFactory()
assert len(w) == 1
assert (
"Overriding ScrapyClientContextFactory.getContext() is deprecated"
in str(w[0].message)
)
class TestContextFactoryTLSMethod(TestContextFactoryBase): class TestContextFactoryTLSMethod(TestContextFactoryBase):
@ -139,28 +135,24 @@ class TestContextFactoryTLSMethod(TestContextFactoryBase):
@coroutine_test @coroutine_test
async def test_setting_default(self, server_url: str) -> None: async def test_setting_default(self, server_url: str) -> None:
crawler = get_crawler() crawler = get_crawler()
settings = Settings() client_context_factory = _load_context_factory_from_settings(crawler)
client_context_factory = load_context_factory_from_settings(settings, crawler)
assert client_context_factory._ssl_method == OpenSSL.SSL.SSLv23_METHOD assert client_context_factory._ssl_method == OpenSSL.SSL.SSLv23_METHOD
await self._assert_factory_works(server_url, client_context_factory) await self._assert_factory_works(server_url, client_context_factory)
def test_setting_none(self): def test_setting_none(self):
crawler = get_crawler() crawler = get_crawler(settings_dict={"DOWNLOADER_CLIENT_TLS_METHOD": None})
settings = Settings({"DOWNLOADER_CLIENT_TLS_METHOD": None})
with pytest.raises(KeyError): with pytest.raises(KeyError):
load_context_factory_from_settings(settings, crawler) _load_context_factory_from_settings(crawler)
def test_setting_bad(self): def test_setting_bad(self):
crawler = get_crawler() crawler = get_crawler(settings_dict={"DOWNLOADER_CLIENT_TLS_METHOD": "bad"})
settings = Settings({"DOWNLOADER_CLIENT_TLS_METHOD": "bad"})
with pytest.raises(KeyError): with pytest.raises(KeyError):
load_context_factory_from_settings(settings, crawler) _load_context_factory_from_settings(crawler)
@coroutine_test @coroutine_test
async def test_setting_explicit(self, server_url: str) -> None: async def test_setting_explicit(self, server_url: str) -> None:
crawler = get_crawler() crawler = get_crawler(settings_dict={"DOWNLOADER_CLIENT_TLS_METHOD": "TLSv1.2"})
settings = Settings({"DOWNLOADER_CLIENT_TLS_METHOD": "TLSv1.2"}) client_context_factory = _load_context_factory_from_settings(crawler)
client_context_factory = load_context_factory_from_settings(settings, crawler)
assert client_context_factory._ssl_method == OpenSSL.SSL.TLSv1_2_METHOD assert client_context_factory._ssl_method == OpenSSL.SSL.TLSv1_2_METHOD
await self._assert_factory_works(server_url, client_context_factory) await self._assert_factory_works(server_url, client_context_factory)