Shortcuts on this page
r m x toggle line displays
j k next/prev highlighted chunk
0 (zero) top of page
1 (one) first highlighted chunk
1"""
2Module for using pyOpenSSL as a TLS backend. This module was relevant before
3the standard library ``ssl`` module supported SNI, but now that we've dropped
4support for Python 2.7 all relevant Python versions support SNI so
5**this module is no longer recommended**.
7This needs the following packages installed:
9* `pyOpenSSL`_ (tested with 19.0.0)
10* `cryptography`_ (minimum 2.3, from pyopenssl)
11* `idna`_ (minimum 2.1, from cryptography)
13However, pyOpenSSL depends on cryptography, so while we use all three directly here we
14end up having relatively few packages required.
16You can install them with the following command:
18.. code-block:: bash
20 $ python -m pip install pyopenssl cryptography idna
22To activate certificate checking, call
23:func:`~urllib3.contrib.pyopenssl.inject_into_urllib3` from your Python code
24before you begin making HTTP requests. This can be done in a ``sitecustomize``
25module, or at any other time before your application begins using ``urllib3``,
26like this:
28.. code-block:: python
30 try:
31 import urllib3.contrib.pyopenssl
32 urllib3.contrib.pyopenssl.inject_into_urllib3()
33 except ImportError:
34 pass
36.. _pyopenssl: https://www.pyopenssl.org
37.. _cryptography: https://cryptography.io
38.. _idna: https://github.com/kjd/idna
39"""
41from __future__ import annotations
43import OpenSSL.SSL
44from cryptography import x509
45from cryptography.hazmat.primitives.serialization import load_pem_private_key
46from cryptography.x509.oid import NameOID
48try:
49 from cryptography.x509 import UnsupportedExtension # type: ignore[attr-defined]
50except ImportError:
51 # UnsupportedExtension is gone in cryptography >= 2.1.0
52 class UnsupportedExtension(Exception): # type: ignore[no-redef]
53 pass
56import logging
57import ssl
58import typing
59from socket import socket as socket_cls
61from .. import util
63if typing.TYPE_CHECKING:
64 from OpenSSL.crypto import X509
67__all__ = ["inject_into_urllib3", "extract_from_urllib3"]
69# Map from urllib3 to PyOpenSSL compatible parameter-values.
70_openssl_versions: dict[int, int] = {
71 util.ssl_.PROTOCOL_TLS: OpenSSL.SSL.SSLv23_METHOD, # type: ignore[attr-defined]
72 util.ssl_.PROTOCOL_TLS_CLIENT: OpenSSL.SSL.SSLv23_METHOD, # type: ignore[attr-defined]
73}
75if hasattr(ssl, "PROTOCOL_TLSv1") and hasattr(OpenSSL.SSL, "TLSv1_METHOD"):
76 _openssl_versions[ssl.PROTOCOL_TLSv1] = OpenSSL.SSL.TLSv1_METHOD
78if hasattr(ssl, "PROTOCOL_TLSv1_1") and hasattr(OpenSSL.SSL, "TLSv1_1_METHOD"):
79 _openssl_versions[ssl.PROTOCOL_TLSv1_1] = OpenSSL.SSL.TLSv1_1_METHOD
81if hasattr(ssl, "PROTOCOL_TLSv1_2") and hasattr(OpenSSL.SSL, "TLSv1_2_METHOD"):
82 _openssl_versions[ssl.PROTOCOL_TLSv1_2] = OpenSSL.SSL.TLSv1_2_METHOD
85_stdlib_to_openssl_verify = {
86 ssl.CERT_NONE: OpenSSL.SSL.VERIFY_NONE,
87 ssl.CERT_OPTIONAL: OpenSSL.SSL.VERIFY_PEER,
88 ssl.CERT_REQUIRED: OpenSSL.SSL.VERIFY_PEER
89 + OpenSSL.SSL.VERIFY_FAIL_IF_NO_PEER_CERT,
90}
91_openssl_to_stdlib_verify = {v: k for k, v in _stdlib_to_openssl_verify.items()}
93# The SSLvX values are the most likely to be missing in the future
94# but we check them all just to be sure.
95_OP_NO_SSLv2_OR_SSLv3: int = getattr(OpenSSL.SSL, "OP_NO_SSLv2", 0) | getattr(
96 OpenSSL.SSL, "OP_NO_SSLv3", 0
97)
98_OP_NO_TLSv1: int = getattr(OpenSSL.SSL, "OP_NO_TLSv1", 0)
99_OP_NO_TLSv1_1: int = getattr(OpenSSL.SSL, "OP_NO_TLSv1_1", 0)
100_OP_NO_TLSv1_2: int = getattr(OpenSSL.SSL, "OP_NO_TLSv1_2", 0)
101_OP_NO_TLSv1_3: int = getattr(OpenSSL.SSL, "OP_NO_TLSv1_3", 0)
103_openssl_to_ssl_minimum_version: dict[int, int] = {
104 ssl.TLSVersion.MINIMUM_SUPPORTED: _OP_NO_SSLv2_OR_SSLv3,
105 ssl.TLSVersion.TLSv1: _OP_NO_SSLv2_OR_SSLv3,
106 ssl.TLSVersion.TLSv1_1: _OP_NO_SSLv2_OR_SSLv3 | _OP_NO_TLSv1,
107 ssl.TLSVersion.TLSv1_2: _OP_NO_SSLv2_OR_SSLv3 | _OP_NO_TLSv1 | _OP_NO_TLSv1_1,
108 ssl.TLSVersion.TLSv1_3: (
109 _OP_NO_SSLv2_OR_SSLv3 | _OP_NO_TLSv1 | _OP_NO_TLSv1_1 | _OP_NO_TLSv1_2
110 ),
111 ssl.TLSVersion.MAXIMUM_SUPPORTED: (
112 _OP_NO_SSLv2_OR_SSLv3 | _OP_NO_TLSv1 | _OP_NO_TLSv1_1 | _OP_NO_TLSv1_2
113 ),
114}
115_openssl_to_ssl_maximum_version: dict[int, int] = {
116 ssl.TLSVersion.MINIMUM_SUPPORTED: (
117 _OP_NO_SSLv2_OR_SSLv3
118 | _OP_NO_TLSv1
119 | _OP_NO_TLSv1_1
120 | _OP_NO_TLSv1_2
121 | _OP_NO_TLSv1_3
122 ),
123 ssl.TLSVersion.TLSv1: (
124 _OP_NO_SSLv2_OR_SSLv3 | _OP_NO_TLSv1_1 | _OP_NO_TLSv1_2 | _OP_NO_TLSv1_3
125 ),
126 ssl.TLSVersion.TLSv1_1: _OP_NO_SSLv2_OR_SSLv3 | _OP_NO_TLSv1_2 | _OP_NO_TLSv1_3,
127 ssl.TLSVersion.TLSv1_2: _OP_NO_SSLv2_OR_SSLv3 | _OP_NO_TLSv1_3,
128 ssl.TLSVersion.TLSv1_3: _OP_NO_SSLv2_OR_SSLv3,
129 ssl.TLSVersion.MAXIMUM_SUPPORTED: _OP_NO_SSLv2_OR_SSLv3,
130}
132# OpenSSL will only write 16K at a time
133SSL_WRITE_BLOCKSIZE = 16384
135orig_util_SSLContext = util.ssl_.SSLContext
138log = logging.getLogger(__name__)
141def inject_into_urllib3() -> None:
142 "Monkey-patch urllib3 with PyOpenSSL-backed SSL-support."
144 _validate_dependencies_met()
146 util.SSLContext = PyOpenSSLContext # type: ignore[assignment]
147 util.ssl_.SSLContext = PyOpenSSLContext # type: ignore[assignment]
148 util.IS_PYOPENSSL = True
149 util.ssl_.IS_PYOPENSSL = True
152def extract_from_urllib3() -> None:
153 "Undo monkey-patching by :func:`inject_into_urllib3`."
155 util.SSLContext = orig_util_SSLContext
156 util.ssl_.SSLContext = orig_util_SSLContext
157 util.IS_PYOPENSSL = False
158 util.ssl_.IS_PYOPENSSL = False
161def _validate_dependencies_met() -> None:
162 """
163 Verifies that PyOpenSSL's package-level dependencies have been met.
164 Throws `ImportError` if they are not met.
165 """
166 # Method added in `cryptography==1.1`; not available in older versions
167 from cryptography.x509.extensions import Extensions
169 if getattr(Extensions, "get_extension_for_class", None) is None:
170 raise ImportError(
171 "'cryptography' module missing required functionality. "
172 "Try upgrading to v1.3.4 or newer."
173 )
175 # pyOpenSSL 0.14 and above use cryptography for OpenSSL bindings. The _x509
176 # attribute is only present on those versions.
177 from OpenSSL.crypto import X509
179 x509 = X509()
180 if getattr(x509, "_x509", None) is None:
181 raise ImportError(
182 "'pyOpenSSL' module missing required functionality. "
183 "Try upgrading to v0.14 or newer."
184 )
187def _dnsname_to_stdlib(name: str) -> str | None:
188 """
189 Converts a dNSName SubjectAlternativeName field to the form used by the
190 standard library on the given Python version.
192 Cryptography produces a dNSName as a unicode string that was idna-decoded
193 from ASCII bytes. We need to idna-encode that string to get it back, and
194 then on Python 3 we also need to convert to unicode via UTF-8 (the stdlib
195 uses PyUnicode_FromStringAndSize on it, which decodes via UTF-8).
197 If the name cannot be idna-encoded then we return None signalling that
198 the name given should be skipped.
199 """
201 def idna_encode(name: str) -> bytes | None:
202 """
203 Borrowed wholesale from the Python Cryptography Project. It turns out
204 that we can't just safely call `idna.encode`: it can explode for
205 wildcard names. This avoids that problem.
206 """
207 import idna
209 try:
210 for prefix in ["*.", "."]:
211 if name.startswith(prefix):
212 name = name[len(prefix) :]
213 return prefix.encode("ascii") + idna.encode(name)
214 return idna.encode(name)
215 except idna.core.IDNAError:
216 return None
218 # Don't send IPv6 addresses through the IDNA encoder.
219 if ":" in name:
220 return name
222 encoded_name = idna_encode(name)
223 if encoded_name is None:
224 return None
225 return encoded_name.decode("utf-8")
228def get_subj_alt_name(peer_cert: X509) -> list[tuple[str, str]]:
229 """
230 Given an PyOpenSSL certificate, provides all the subject alternative names.
231 """
232 cert = peer_cert.to_cryptography()
234 # We want to find the SAN extension. Ask Cryptography to locate it (it's
235 # faster than looping in Python)
236 try:
237 ext = cert.extensions.get_extension_for_class(x509.SubjectAlternativeName).value
238 except x509.ExtensionNotFound:
239 # No such extension, return the empty list.
240 return []
241 except (
242 x509.DuplicateExtension,
243 UnsupportedExtension,
244 x509.UnsupportedGeneralNameType,
245 UnicodeError,
246 ) as e:
247 # A problem has been found with the quality of the certificate. Assume
248 # no SAN field is present.
249 log.warning(
250 "A problem was encountered with the certificate that prevented "
251 "urllib3 from finding the SubjectAlternativeName field. This can "
252 "affect certificate validation. The error was %s",
253 e,
254 )
255 return []
257 # We want to return dNSName and iPAddress fields. We need to cast the IPs
258 # back to strings because the match_hostname function wants them as
259 # strings.
260 # Sadly the DNS names need to be idna encoded and then, on Python 3, UTF-8
261 # decoded. This is pretty frustrating, but that's what the standard library
262 # does with certificates, and so we need to attempt to do the same.
263 # We also want to skip over names which cannot be idna encoded.
264 names = [
265 ("DNS", name)
266 for name in map(_dnsname_to_stdlib, ext.get_values_for_type(x509.DNSName))
267 if name is not None
268 ]
269 names.extend(
270 ("IP Address", str(name)) for name in ext.get_values_for_type(x509.IPAddress)
271 )
273 return names
276def _get_common_name(peer_cert: X509) -> str | None:
277 """
278 Given a pyOpenSSL certificate, return the subject's common name.
279 """
280 cert = peer_cert.to_cryptography()
281 names = cert.subject.get_attributes_for_oid(NameOID.COMMON_NAME)
282 return typing.cast(str, names[0].value) if names else None
285class WrappedSocket:
286 """API-compatibility wrapper for Python OpenSSL's Connection-class."""
288 def __init__(
289 self,
290 connection: OpenSSL.SSL.Connection,
291 socket: socket_cls,
292 suppress_ragged_eofs: bool = True,
293 ) -> None:
294 self.connection = connection
295 self.socket = socket
296 self.suppress_ragged_eofs = suppress_ragged_eofs
297 self._io_refs = 0
298 self._closed = False
300 def fileno(self) -> int:
301 return self.socket.fileno()
303 # Copy-pasted from Python 3.5 source code
304 def _decref_socketios(self) -> None:
305 if self._io_refs > 0:
306 self._io_refs -= 1
307 if self._closed:
308 self.close()
310 def recv(self, *args: typing.Any, **kwargs: typing.Any) -> bytes:
311 try:
312 data = self.connection.recv(*args, **kwargs)
313 except OpenSSL.SSL.SysCallError as e:
314 if self.suppress_ragged_eofs and e.args == (-1, "Unexpected EOF"):
315 return b""
316 else:
317 raise OSError(e.args[0], str(e)) from e
318 except OpenSSL.SSL.ZeroReturnError:
319 if self.connection.get_shutdown() == OpenSSL.SSL.RECEIVED_SHUTDOWN:
320 return b""
321 else:
322 raise
323 except OpenSSL.SSL.WantReadError as e:
324 if not util.wait_for_read(self.socket, self.socket.gettimeout()):
325 raise TimeoutError("The read operation timed out") from e
326 else:
327 return self.recv(*args, **kwargs)
329 # TLS 1.3 post-handshake authentication
330 except OpenSSL.SSL.Error as e:
331 raise ssl.SSLError(f"read error: {e!r}") from e
332 else:
333 return data
335 def recv_into(self, *args: typing.Any, **kwargs: typing.Any) -> int:
336 try:
337 return self.connection.recv_into(*args, **kwargs)
338 except OpenSSL.SSL.SysCallError as e:
339 if self.suppress_ragged_eofs and e.args == (-1, "Unexpected EOF"):
340 return 0
341 else:
342 raise OSError(e.args[0], str(e)) from e
343 except OpenSSL.SSL.ZeroReturnError:
344 if self.connection.get_shutdown() == OpenSSL.SSL.RECEIVED_SHUTDOWN:
345 return 0
346 else:
347 raise
348 except OpenSSL.SSL.WantReadError as e:
349 if not util.wait_for_read(self.socket, self.socket.gettimeout()):
350 raise TimeoutError("The read operation timed out") from e
351 else:
352 return self.recv_into(*args, **kwargs)
354 # TLS 1.3 post-handshake authentication
355 except OpenSSL.SSL.Error as e:
356 raise ssl.SSLError(f"read error: {e!r}") from e
358 def settimeout(self, timeout: float) -> None:
359 return self.socket.settimeout(timeout)
361 def _send_until_done(self, data: bytes) -> int:
362 while True:
363 try:
364 return self.connection.send(data)
365 except OpenSSL.SSL.WantWriteError as e:
366 if not util.wait_for_write(self.socket, self.socket.gettimeout()):
367 raise TimeoutError() from e
368 continue
369 except OpenSSL.SSL.SysCallError as e:
370 raise OSError(e.args[0], str(e)) from e
372 def sendall(self, data: bytes) -> None:
373 total_sent = 0
374 while total_sent < len(data):
375 sent = self._send_until_done(
376 data[total_sent : total_sent + SSL_WRITE_BLOCKSIZE]
377 )
378 total_sent += sent
380 def shutdown(self, how: int) -> None:
381 try:
382 self.connection.shutdown()
383 except OpenSSL.SSL.Error as e:
384 raise ssl.SSLError(f"shutdown error: {e!r}") from e
386 def close(self) -> None:
387 self._closed = True
388 if self._io_refs <= 0:
389 self._real_close()
391 def _real_close(self) -> None:
392 try:
393 return self.connection.close() # type: ignore[no-any-return]
394 except OpenSSL.SSL.Error:
395 return
397 @typing.overload
398 def getpeercert(
399 self, binary_form: typing.Literal[False] = False
400 ) -> dict[str, list[typing.Any]] | None: ...
402 @typing.overload
403 def getpeercert(self, binary_form: typing.Literal[True]) -> bytes | None: ...
405 def getpeercert(
406 self, binary_form: bool = False
407 ) -> dict[str, list[typing.Any]] | bytes | None:
408 x509 = self.connection.get_peer_certificate()
410 if x509 is None:
411 return None
413 if binary_form:
414 return OpenSSL.crypto.dump_certificate(OpenSSL.crypto.FILETYPE_ASN1, x509)
416 return {
417 "subject": ((("commonName", _get_common_name(x509)),),), # type: ignore[dict-item]
418 "subjectAltName": get_subj_alt_name(x509),
419 }
421 def version(self) -> str:
422 return self.connection.get_protocol_version_name()
424 def selected_alpn_protocol(self) -> str | None:
425 alpn_proto = self.connection.get_alpn_proto_negotiated()
426 return alpn_proto.decode() if alpn_proto else None
429WrappedSocket.makefile = socket_cls.makefile # type: ignore[attr-defined]
432class PyOpenSSLContext:
433 """
434 I am a wrapper class for the PyOpenSSL ``Context`` object. I am responsible
435 for translating the interface of the standard library ``SSLContext`` object
436 to calls into PyOpenSSL.
437 """
439 def __init__(self, protocol: int) -> None:
440 self.protocol = _openssl_versions[protocol]
441 self._ctx = OpenSSL.SSL.Context(self.protocol)
442 self._options = 0
443 self.check_hostname = False
444 self._minimum_version: int = ssl.TLSVersion.MINIMUM_SUPPORTED
445 self._maximum_version: int = ssl.TLSVersion.MAXIMUM_SUPPORTED
446 self._verify_flags: int = ssl.VERIFY_X509_TRUSTED_FIRST
448 @property
449 def options(self) -> int:
450 return self._options
452 @options.setter
453 def options(self, value: int) -> None:
454 self._options = value
455 self._set_ctx_options()
457 @property
458 def verify_flags(self) -> int:
459 return self._verify_flags
461 @verify_flags.setter
462 def verify_flags(self, value: int) -> None:
463 self._verify_flags = value
464 self._ctx.get_cert_store().set_flags(self._verify_flags) # type: ignore[union-attr]
466 @property
467 def verify_mode(self) -> int:
468 return _openssl_to_stdlib_verify[self._ctx.get_verify_mode()]
470 @verify_mode.setter
471 def verify_mode(self, value: ssl.VerifyMode) -> None:
472 self._ctx.set_verify(_stdlib_to_openssl_verify[value], _verify_callback)
474 def set_default_verify_paths(self) -> None:
475 self._ctx.set_default_verify_paths()
477 def set_ciphers(self, ciphers: bytes | str) -> None:
478 if isinstance(ciphers, str):
479 ciphers = ciphers.encode("utf-8")
480 self._ctx.set_cipher_list(ciphers)
482 def load_verify_locations(
483 self,
484 cafile: str | None = None,
485 capath: str | None = None,
486 cadata: bytes | None = None,
487 ) -> None:
488 if cafile is not None:
489 cafile = cafile.encode("utf-8") # type: ignore[assignment]
490 if capath is not None:
491 capath = capath.encode("utf-8") # type: ignore[assignment]
492 try:
493 self._ctx.load_verify_locations(cafile, capath)
494 if cadata is not None:
495 self._ctx.load_verify_locations(cadata)
496 except OpenSSL.SSL.Error as e:
497 raise ssl.SSLError(f"unable to load trusted certificates: {e!r}") from e
499 def load_cert_chain(
500 self,
501 certfile: str,
502 keyfile: str | None = None,
503 password: str | bytes | None = None,
504 ) -> None:
505 try:
506 self._ctx.use_certificate_chain_file(certfile)
507 if password is not None:
508 if not isinstance(password, bytes):
509 password = password.encode("utf-8")
510 # pyOpenSSL added cryptography-key support in 24.3.0.
511 # Keep using the older password-callback path until 2026's
512 # versions because set_passwd_cb() became deprecated in 26.3.0.
513 if int(OpenSSL.__version__.split(".")[0]) >= 26:
514 with open(keyfile or certfile, "rb") as key_file:
515 key_data = key_file.read()
516 try:
517 private_key = load_pem_private_key(key_data, password)
518 except TypeError:
519 # Match SSLContext by ignoring passwords supplied
520 # for unencrypted keys.
521 # Incorrect passwords for encrypted keys raise
522 # ValueError instead.
523 private_key = load_pem_private_key(key_data, None)
524 # cryptography's loader returns a wider private-key union
525 # than pyOpenSSL accepts, so we add `type: ignore` here.
526 self._ctx.use_privatekey(private_key) # type: ignore[arg-type]
527 else:
528 self._ctx.set_passwd_cb(lambda *_: password)
529 self._ctx.use_privatekey_file(keyfile or certfile)
530 else:
531 self._ctx.use_privatekey_file(keyfile or certfile)
532 except (OpenSSL.SSL.Error, TypeError, ValueError) as e:
533 raise ssl.SSLError(f"Unable to load certificate chain: {e!r}") from e
535 def set_alpn_protocols(self, protocols: list[bytes | str]) -> None:
536 protocols = [util.util.to_bytes(p, "ascii") for p in protocols]
537 return self._ctx.set_alpn_protos(protocols) # type: ignore[arg-type]
539 def wrap_socket(
540 self,
541 sock: socket_cls,
542 server_side: bool = False,
543 do_handshake_on_connect: bool = True,
544 suppress_ragged_eofs: bool = True,
545 server_hostname: bytes | str | None = None,
546 ) -> WrappedSocket:
547 cnx = OpenSSL.SSL.Connection(self._ctx, sock)
549 # If server_hostname is an IP, don't use it for SNI, per RFC6066 Section 3
550 if server_hostname and not util.ssl_.is_ipaddress(server_hostname):
551 if isinstance(server_hostname, str):
552 server_hostname = server_hostname.encode("utf-8")
553 cnx.set_tlsext_host_name(server_hostname)
555 cnx.set_connect_state()
557 while True:
558 try:
559 cnx.do_handshake()
560 except OpenSSL.SSL.WantReadError as e:
561 if not util.wait_for_read(sock, sock.gettimeout()):
562 raise TimeoutError("select timed out") from e
563 continue
564 except OpenSSL.SSL.Error as e:
565 raise ssl.SSLError(f"bad handshake: {e!r}") from e
566 break
568 return WrappedSocket(cnx, sock)
570 def _set_ctx_options(self) -> None:
571 self._ctx.set_options(
572 self._options
573 | _openssl_to_ssl_minimum_version[self._minimum_version]
574 | _openssl_to_ssl_maximum_version[self._maximum_version]
575 )
577 @property
578 def minimum_version(self) -> int:
579 return self._minimum_version
581 @minimum_version.setter
582 def minimum_version(self, minimum_version: int) -> None:
583 self._minimum_version = minimum_version
584 self._set_ctx_options()
586 @property
587 def maximum_version(self) -> int:
588 return self._maximum_version
590 @maximum_version.setter
591 def maximum_version(self, maximum_version: int) -> None:
592 self._maximum_version = maximum_version
593 self._set_ctx_options()
596def _verify_callback(
597 cnx: OpenSSL.SSL.Connection,
598 x509: X509,
599 err_no: int,
600 err_depth: int,
601 return_code: int,
602) -> bool:
603 return err_no == 0