Coverage for /pythoncovmergedfiles/medio/medio/usr/local/lib/python3.11/site-packages/OpenSSL/SSL.py: 35%
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
1from __future__ import annotations
3import os
4import socket
5import sys
6import typing
7import warnings
8from collections.abc import Sequence
9from errno import errorcode
10from functools import partial, wraps
11from itertools import chain, count
12from sys import platform
13from typing import Any, Callable, Optional, TypeVar
14from weakref import WeakValueDictionary
16if sys.version_info >= (3, 13):
17 from warnings import deprecated
18else:
19 from typing_extensions import deprecated
21from cryptography import x509
22from cryptography.hazmat.primitives.asymmetric import ec
24from OpenSSL._util import (
25 StrOrBytesPath as _StrOrBytesPath,
26)
27from OpenSSL._util import (
28 exception_from_error_queue as _exception_from_error_queue,
29)
30from OpenSSL._util import (
31 ffi as _ffi,
32)
33from OpenSSL._util import (
34 lib as _lib,
35)
36from OpenSSL._util import (
37 make_assert as _make_assert,
38)
39from OpenSSL._util import (
40 no_zero_allocator as _no_zero_allocator,
41)
42from OpenSSL._util import (
43 path_bytes as _path_bytes,
44)
45from OpenSSL._util import (
46 text_to_bytes_and_warn as _text_to_bytes_and_warn,
47)
48from OpenSSL.crypto import (
49 FILETYPE_PEM,
50 X509,
51 PKey,
52 X509Name,
53 X509Store,
54 _EllipticCurve,
55 _PassphraseHelper,
56 _PrivateKey,
57)
59__all__ = [
60 "DTLS_CLIENT_METHOD",
61 "DTLS_METHOD",
62 "DTLS_SERVER_METHOD",
63 "MODE_RELEASE_BUFFERS",
64 "NO_OVERLAPPING_PROTOCOLS",
65 "OPENSSL_BUILT_ON",
66 "OPENSSL_CFLAGS",
67 "OPENSSL_DIR",
68 "OPENSSL_PLATFORM",
69 "OPENSSL_VERSION",
70 "OPENSSL_VERSION_NUMBER",
71 "OP_ALL",
72 "OP_CIPHER_SERVER_PREFERENCE",
73 "OP_DONT_INSERT_EMPTY_FRAGMENTS",
74 "OP_EPHEMERAL_RSA",
75 "OP_MICROSOFT_BIG_SSLV3_BUFFER",
76 "OP_MICROSOFT_SESS_ID_BUG",
77 "OP_MSIE_SSLV2_RSA_PADDING",
78 "OP_NETSCAPE_CA_DN_BUG",
79 "OP_NETSCAPE_CHALLENGE_BUG",
80 "OP_NETSCAPE_DEMO_CIPHER_CHANGE_BUG",
81 "OP_NETSCAPE_REUSE_CIPHER_CHANGE_BUG",
82 "OP_NO_COMPRESSION",
83 "OP_NO_QUERY_MTU",
84 "OP_NO_TICKET",
85 "OP_PKCS1_CHECK_1",
86 "OP_PKCS1_CHECK_2",
87 "OP_SINGLE_DH_USE",
88 "OP_SINGLE_ECDH_USE",
89 "OP_SSLEAY_080_CLIENT_DH_BUG",
90 "OP_SSLREF2_REUSE_CERT_TYPE_BUG",
91 "OP_TLS_BLOCK_PADDING_BUG",
92 "OP_TLS_D5_BUG",
93 "OP_TLS_ROLLBACK_BUG",
94 "RECEIVED_SHUTDOWN",
95 "SENT_SHUTDOWN",
96 "SESS_CACHE_BOTH",
97 "SESS_CACHE_CLIENT",
98 "SESS_CACHE_NO_AUTO_CLEAR",
99 "SESS_CACHE_NO_INTERNAL",
100 "SESS_CACHE_NO_INTERNAL_LOOKUP",
101 "SESS_CACHE_NO_INTERNAL_STORE",
102 "SESS_CACHE_OFF",
103 "SESS_CACHE_SERVER",
104 "SSL3_VERSION",
105 "SSLEAY_BUILT_ON",
106 "SSLEAY_CFLAGS",
107 "SSLEAY_DIR",
108 "SSLEAY_PLATFORM",
109 "SSLEAY_VERSION",
110 "SSL_CB_ACCEPT_EXIT",
111 "SSL_CB_ACCEPT_LOOP",
112 "SSL_CB_ALERT",
113 "SSL_CB_CONNECT_EXIT",
114 "SSL_CB_CONNECT_LOOP",
115 "SSL_CB_EXIT",
116 "SSL_CB_HANDSHAKE_DONE",
117 "SSL_CB_HANDSHAKE_START",
118 "SSL_CB_LOOP",
119 "SSL_CB_READ",
120 "SSL_CB_READ_ALERT",
121 "SSL_CB_WRITE",
122 "SSL_CB_WRITE_ALERT",
123 "SSL_ST_ACCEPT",
124 "SSL_ST_CONNECT",
125 "SSL_ST_MASK",
126 "TLS1_1_VERSION",
127 "TLS1_2_VERSION",
128 "TLS1_3_VERSION",
129 "TLS1_VERSION",
130 "TLS_CLIENT_METHOD",
131 "TLS_METHOD",
132 "TLS_SERVER_METHOD",
133 "VERIFY_CLIENT_ONCE",
134 "VERIFY_FAIL_IF_NO_PEER_CERT",
135 "VERIFY_NONE",
136 "VERIFY_PEER",
137 "Connection",
138 "Context",
139 "Error",
140 "OP_NO_SSLv2",
141 "OP_NO_SSLv3",
142 "OP_NO_TLSv1",
143 "OP_NO_TLSv1_1",
144 "OP_NO_TLSv1_2",
145 "OP_NO_TLSv1_3",
146 "SSLeay_version",
147 "SSLv23_METHOD",
148 "Session",
149 "SysCallError",
150 "TLSv1_1_METHOD",
151 "TLSv1_2_METHOD",
152 "TLSv1_METHOD",
153 "WantReadError",
154 "WantWriteError",
155 "WantX509LookupError",
156 "X509VerificationCodes",
157 "ZeroReturnError",
158]
161OPENSSL_VERSION_NUMBER: int = _lib.OPENSSL_VERSION_NUMBER
162OPENSSL_VERSION: int = _lib.OPENSSL_VERSION
163OPENSSL_CFLAGS: int = _lib.OPENSSL_CFLAGS
164OPENSSL_PLATFORM: int = _lib.OPENSSL_PLATFORM
165OPENSSL_DIR: int = _lib.OPENSSL_DIR
166OPENSSL_BUILT_ON: int = _lib.OPENSSL_BUILT_ON
168SSLEAY_VERSION = OPENSSL_VERSION
169SSLEAY_CFLAGS = OPENSSL_CFLAGS
170SSLEAY_PLATFORM = OPENSSL_PLATFORM
171SSLEAY_DIR = OPENSSL_DIR
172SSLEAY_BUILT_ON = OPENSSL_BUILT_ON
174SENT_SHUTDOWN = _lib.SSL_SENT_SHUTDOWN
175RECEIVED_SHUTDOWN = _lib.SSL_RECEIVED_SHUTDOWN
177SSLv23_METHOD = 3
178TLSv1_METHOD = 4
179TLSv1_1_METHOD = 5
180TLSv1_2_METHOD = 6
181TLS_METHOD = 7
182TLS_SERVER_METHOD = 8
183TLS_CLIENT_METHOD = 9
184DTLS_METHOD = 10
185DTLS_SERVER_METHOD = 11
186DTLS_CLIENT_METHOD = 12
188SSL3_VERSION: int = _lib.SSL3_VERSION
189TLS1_VERSION: int = _lib.TLS1_VERSION
190TLS1_1_VERSION: int = _lib.TLS1_1_VERSION
191TLS1_2_VERSION: int = _lib.TLS1_2_VERSION
192TLS1_3_VERSION: int = _lib.TLS1_3_VERSION
194OP_NO_SSLv2: int = _lib.SSL_OP_NO_SSLv2
195OP_NO_SSLv3: int = _lib.SSL_OP_NO_SSLv3
196OP_NO_TLSv1: int = _lib.SSL_OP_NO_TLSv1
197OP_NO_TLSv1_1: int = _lib.SSL_OP_NO_TLSv1_1
198OP_NO_TLSv1_2: int = _lib.SSL_OP_NO_TLSv1_2
199OP_NO_TLSv1_3: int = _lib.SSL_OP_NO_TLSv1_3
201MODE_RELEASE_BUFFERS: int = _lib.SSL_MODE_RELEASE_BUFFERS
203OP_SINGLE_DH_USE: int = _lib.SSL_OP_SINGLE_DH_USE
204OP_SINGLE_ECDH_USE: int = _lib.SSL_OP_SINGLE_ECDH_USE
205OP_EPHEMERAL_RSA: int = _lib.SSL_OP_EPHEMERAL_RSA
206OP_MICROSOFT_SESS_ID_BUG: int = _lib.SSL_OP_MICROSOFT_SESS_ID_BUG
207OP_NETSCAPE_CHALLENGE_BUG: int = _lib.SSL_OP_NETSCAPE_CHALLENGE_BUG
208OP_NETSCAPE_REUSE_CIPHER_CHANGE_BUG: int = (
209 _lib.SSL_OP_NETSCAPE_REUSE_CIPHER_CHANGE_BUG
210)
211OP_SSLREF2_REUSE_CERT_TYPE_BUG: int = _lib.SSL_OP_SSLREF2_REUSE_CERT_TYPE_BUG
212OP_MICROSOFT_BIG_SSLV3_BUFFER: int = _lib.SSL_OP_MICROSOFT_BIG_SSLV3_BUFFER
213OP_MSIE_SSLV2_RSA_PADDING: int = _lib.SSL_OP_MSIE_SSLV2_RSA_PADDING
214OP_SSLEAY_080_CLIENT_DH_BUG: int = _lib.SSL_OP_SSLEAY_080_CLIENT_DH_BUG
215OP_TLS_D5_BUG: int = _lib.SSL_OP_TLS_D5_BUG
216OP_TLS_BLOCK_PADDING_BUG: int = _lib.SSL_OP_TLS_BLOCK_PADDING_BUG
217OP_DONT_INSERT_EMPTY_FRAGMENTS: int = _lib.SSL_OP_DONT_INSERT_EMPTY_FRAGMENTS
218OP_CIPHER_SERVER_PREFERENCE: int = _lib.SSL_OP_CIPHER_SERVER_PREFERENCE
219OP_TLS_ROLLBACK_BUG: int = _lib.SSL_OP_TLS_ROLLBACK_BUG
220OP_PKCS1_CHECK_1 = _lib.SSL_OP_PKCS1_CHECK_1
221OP_PKCS1_CHECK_2: int = _lib.SSL_OP_PKCS1_CHECK_2
222OP_NETSCAPE_CA_DN_BUG: int = _lib.SSL_OP_NETSCAPE_CA_DN_BUG
223OP_NETSCAPE_DEMO_CIPHER_CHANGE_BUG: int = (
224 _lib.SSL_OP_NETSCAPE_DEMO_CIPHER_CHANGE_BUG
225)
226OP_NO_COMPRESSION: int = _lib.SSL_OP_NO_COMPRESSION
228OP_NO_QUERY_MTU: int = _lib.SSL_OP_NO_QUERY_MTU
229try:
230 OP_COOKIE_EXCHANGE: int | None = _lib.SSL_OP_COOKIE_EXCHANGE
231 __all__.append("OP_COOKIE_EXCHANGE")
232except AttributeError:
233 OP_COOKIE_EXCHANGE = None
234OP_NO_TICKET: int = _lib.SSL_OP_NO_TICKET
236try:
237 OP_NO_RENEGOTIATION: int = _lib.SSL_OP_NO_RENEGOTIATION
238 __all__.append("OP_NO_RENEGOTIATION")
239except AttributeError:
240 pass
242try:
243 OP_IGNORE_UNEXPECTED_EOF: int = _lib.SSL_OP_IGNORE_UNEXPECTED_EOF
244 __all__.append("OP_IGNORE_UNEXPECTED_EOF")
245except AttributeError:
246 pass
248try:
249 OP_LEGACY_SERVER_CONNECT: int = _lib.SSL_OP_LEGACY_SERVER_CONNECT
250 __all__.append("OP_LEGACY_SERVER_CONNECT")
251except AttributeError:
252 pass
254OP_ALL: int = _lib.SSL_OP_ALL
256VERIFY_PEER: int = _lib.SSL_VERIFY_PEER
257VERIFY_FAIL_IF_NO_PEER_CERT: int = _lib.SSL_VERIFY_FAIL_IF_NO_PEER_CERT
258VERIFY_CLIENT_ONCE: int = _lib.SSL_VERIFY_CLIENT_ONCE
259VERIFY_NONE: int = _lib.SSL_VERIFY_NONE
261SESS_CACHE_OFF: int = _lib.SSL_SESS_CACHE_OFF
262SESS_CACHE_CLIENT: int = _lib.SSL_SESS_CACHE_CLIENT
263SESS_CACHE_SERVER: int = _lib.SSL_SESS_CACHE_SERVER
264SESS_CACHE_BOTH: int = _lib.SSL_SESS_CACHE_BOTH
265SESS_CACHE_NO_AUTO_CLEAR: int = _lib.SSL_SESS_CACHE_NO_AUTO_CLEAR
266SESS_CACHE_NO_INTERNAL_LOOKUP: int = _lib.SSL_SESS_CACHE_NO_INTERNAL_LOOKUP
267SESS_CACHE_NO_INTERNAL_STORE: int = _lib.SSL_SESS_CACHE_NO_INTERNAL_STORE
268SESS_CACHE_NO_INTERNAL: int = _lib.SSL_SESS_CACHE_NO_INTERNAL
270SSL_ST_CONNECT: int = _lib.SSL_ST_CONNECT
271SSL_ST_ACCEPT: int = _lib.SSL_ST_ACCEPT
272SSL_ST_MASK: int = _lib.SSL_ST_MASK
274SSL_CB_LOOP: int = _lib.SSL_CB_LOOP
275SSL_CB_EXIT: int = _lib.SSL_CB_EXIT
276SSL_CB_READ: int = _lib.SSL_CB_READ
277SSL_CB_WRITE: int = _lib.SSL_CB_WRITE
278SSL_CB_ALERT: int = _lib.SSL_CB_ALERT
279SSL_CB_READ_ALERT: int = _lib.SSL_CB_READ_ALERT
280SSL_CB_WRITE_ALERT: int = _lib.SSL_CB_WRITE_ALERT
281SSL_CB_ACCEPT_LOOP: int = _lib.SSL_CB_ACCEPT_LOOP
282SSL_CB_ACCEPT_EXIT: int = _lib.SSL_CB_ACCEPT_EXIT
283SSL_CB_CONNECT_LOOP: int = _lib.SSL_CB_CONNECT_LOOP
284SSL_CB_CONNECT_EXIT: int = _lib.SSL_CB_CONNECT_EXIT
285SSL_CB_HANDSHAKE_START: int = _lib.SSL_CB_HANDSHAKE_START
286SSL_CB_HANDSHAKE_DONE: int = _lib.SSL_CB_HANDSHAKE_DONE
288_Buffer = typing.Union[bytes, bytearray, memoryview]
289_T = TypeVar("_T")
292class _NoOverlappingProtocols:
293 pass
296NO_OVERLAPPING_PROTOCOLS = _NoOverlappingProtocols()
298# Callback types.
299_ALPNSelectCallback = Callable[
300 [
301 "Connection",
302 typing.List[bytes],
303 ],
304 typing.Union[bytes, _NoOverlappingProtocols],
305]
306_CookieGenerateCallback = Callable[["Connection"], bytes]
307_CookieVerifyCallback = Callable[["Connection", bytes], bool]
308_OCSPClientCallback = Callable[["Connection", bytes, Optional[_T]], bool]
309_OCSPServerCallback = Callable[["Connection", Optional[_T]], bytes]
310_PassphraseCallback = Callable[[int, bool, Optional[_T]], bytes]
311_VerifyCallback = Callable[["Connection", X509, int, int, int], bool]
314class X509VerificationCodes:
315 """
316 Success and error codes for X509 verification, as returned by the
317 underlying ``X509_STORE_CTX_get_error()`` function and passed by pyOpenSSL
318 to verification callback functions.
320 See `OpenSSL Verification Errors
321 <https://www.openssl.org/docs/manmaster/man3/X509_verify_cert_error_string.html#ERROR-CODES>`_
322 for details.
323 """
325 OK = _lib.X509_V_OK
326 ERR_UNABLE_TO_GET_ISSUER_CERT = _lib.X509_V_ERR_UNABLE_TO_GET_ISSUER_CERT
327 ERR_UNABLE_TO_GET_CRL = _lib.X509_V_ERR_UNABLE_TO_GET_CRL
328 ERR_UNABLE_TO_DECRYPT_CERT_SIGNATURE = (
329 _lib.X509_V_ERR_UNABLE_TO_DECRYPT_CERT_SIGNATURE
330 )
331 ERR_UNABLE_TO_DECRYPT_CRL_SIGNATURE = (
332 _lib.X509_V_ERR_UNABLE_TO_DECRYPT_CRL_SIGNATURE
333 )
334 ERR_UNABLE_TO_DECODE_ISSUER_PUBLIC_KEY = (
335 _lib.X509_V_ERR_UNABLE_TO_DECODE_ISSUER_PUBLIC_KEY
336 )
337 ERR_CERT_SIGNATURE_FAILURE = _lib.X509_V_ERR_CERT_SIGNATURE_FAILURE
338 ERR_CRL_SIGNATURE_FAILURE = _lib.X509_V_ERR_CRL_SIGNATURE_FAILURE
339 ERR_CERT_NOT_YET_VALID = _lib.X509_V_ERR_CERT_NOT_YET_VALID
340 ERR_CERT_HAS_EXPIRED = _lib.X509_V_ERR_CERT_HAS_EXPIRED
341 ERR_CRL_NOT_YET_VALID = _lib.X509_V_ERR_CRL_NOT_YET_VALID
342 ERR_CRL_HAS_EXPIRED = _lib.X509_V_ERR_CRL_HAS_EXPIRED
343 ERR_ERROR_IN_CERT_NOT_BEFORE_FIELD = (
344 _lib.X509_V_ERR_ERROR_IN_CERT_NOT_BEFORE_FIELD
345 )
346 ERR_ERROR_IN_CERT_NOT_AFTER_FIELD = (
347 _lib.X509_V_ERR_ERROR_IN_CERT_NOT_AFTER_FIELD
348 )
349 ERR_ERROR_IN_CRL_LAST_UPDATE_FIELD = (
350 _lib.X509_V_ERR_ERROR_IN_CRL_LAST_UPDATE_FIELD
351 )
352 ERR_ERROR_IN_CRL_NEXT_UPDATE_FIELD = (
353 _lib.X509_V_ERR_ERROR_IN_CRL_NEXT_UPDATE_FIELD
354 )
355 ERR_OUT_OF_MEM = _lib.X509_V_ERR_OUT_OF_MEM
356 ERR_DEPTH_ZERO_SELF_SIGNED_CERT = (
357 _lib.X509_V_ERR_DEPTH_ZERO_SELF_SIGNED_CERT
358 )
359 ERR_SELF_SIGNED_CERT_IN_CHAIN = _lib.X509_V_ERR_SELF_SIGNED_CERT_IN_CHAIN
360 ERR_UNABLE_TO_GET_ISSUER_CERT_LOCALLY = (
361 _lib.X509_V_ERR_UNABLE_TO_GET_ISSUER_CERT_LOCALLY
362 )
363 ERR_UNABLE_TO_VERIFY_LEAF_SIGNATURE = (
364 _lib.X509_V_ERR_UNABLE_TO_VERIFY_LEAF_SIGNATURE
365 )
366 ERR_CERT_CHAIN_TOO_LONG = _lib.X509_V_ERR_CERT_CHAIN_TOO_LONG
367 ERR_CERT_REVOKED = _lib.X509_V_ERR_CERT_REVOKED
368 ERR_INVALID_CA = _lib.X509_V_ERR_INVALID_CA
369 ERR_PATH_LENGTH_EXCEEDED = _lib.X509_V_ERR_PATH_LENGTH_EXCEEDED
370 ERR_INVALID_PURPOSE = _lib.X509_V_ERR_INVALID_PURPOSE
371 ERR_CERT_UNTRUSTED = _lib.X509_V_ERR_CERT_UNTRUSTED
372 ERR_CERT_REJECTED = _lib.X509_V_ERR_CERT_REJECTED
373 ERR_SUBJECT_ISSUER_MISMATCH = _lib.X509_V_ERR_SUBJECT_ISSUER_MISMATCH
374 ERR_AKID_SKID_MISMATCH = _lib.X509_V_ERR_AKID_SKID_MISMATCH
375 ERR_AKID_ISSUER_SERIAL_MISMATCH = (
376 _lib.X509_V_ERR_AKID_ISSUER_SERIAL_MISMATCH
377 )
378 ERR_KEYUSAGE_NO_CERTSIGN = _lib.X509_V_ERR_KEYUSAGE_NO_CERTSIGN
379 ERR_UNABLE_TO_GET_CRL_ISSUER = _lib.X509_V_ERR_UNABLE_TO_GET_CRL_ISSUER
380 ERR_UNHANDLED_CRITICAL_EXTENSION = (
381 _lib.X509_V_ERR_UNHANDLED_CRITICAL_EXTENSION
382 )
383 ERR_KEYUSAGE_NO_CRL_SIGN = _lib.X509_V_ERR_KEYUSAGE_NO_CRL_SIGN
384 ERR_UNHANDLED_CRITICAL_CRL_EXTENSION = (
385 _lib.X509_V_ERR_UNHANDLED_CRITICAL_CRL_EXTENSION
386 )
387 ERR_INVALID_NON_CA = _lib.X509_V_ERR_INVALID_NON_CA
388 ERR_PROXY_PATH_LENGTH_EXCEEDED = _lib.X509_V_ERR_PROXY_PATH_LENGTH_EXCEEDED
389 ERR_KEYUSAGE_NO_DIGITAL_SIGNATURE = (
390 _lib.X509_V_ERR_KEYUSAGE_NO_DIGITAL_SIGNATURE
391 )
392 ERR_PROXY_CERTIFICATES_NOT_ALLOWED = (
393 _lib.X509_V_ERR_PROXY_CERTIFICATES_NOT_ALLOWED
394 )
395 ERR_INVALID_EXTENSION = _lib.X509_V_ERR_INVALID_EXTENSION
396 ERR_INVALID_POLICY_EXTENSION = _lib.X509_V_ERR_INVALID_POLICY_EXTENSION
397 ERR_NO_EXPLICIT_POLICY = _lib.X509_V_ERR_NO_EXPLICIT_POLICY
398 ERR_DIFFERENT_CRL_SCOPE = _lib.X509_V_ERR_DIFFERENT_CRL_SCOPE
399 ERR_UNSUPPORTED_EXTENSION_FEATURE = (
400 _lib.X509_V_ERR_UNSUPPORTED_EXTENSION_FEATURE
401 )
402 ERR_UNNESTED_RESOURCE = _lib.X509_V_ERR_UNNESTED_RESOURCE
403 ERR_PERMITTED_VIOLATION = _lib.X509_V_ERR_PERMITTED_VIOLATION
404 ERR_EXCLUDED_VIOLATION = _lib.X509_V_ERR_EXCLUDED_VIOLATION
405 ERR_SUBTREE_MINMAX = _lib.X509_V_ERR_SUBTREE_MINMAX
406 ERR_UNSUPPORTED_CONSTRAINT_TYPE = (
407 _lib.X509_V_ERR_UNSUPPORTED_CONSTRAINT_TYPE
408 )
409 ERR_UNSUPPORTED_CONSTRAINT_SYNTAX = (
410 _lib.X509_V_ERR_UNSUPPORTED_CONSTRAINT_SYNTAX
411 )
412 ERR_UNSUPPORTED_NAME_SYNTAX = _lib.X509_V_ERR_UNSUPPORTED_NAME_SYNTAX
413 ERR_CRL_PATH_VALIDATION_ERROR = _lib.X509_V_ERR_CRL_PATH_VALIDATION_ERROR
414 ERR_HOSTNAME_MISMATCH = _lib.X509_V_ERR_HOSTNAME_MISMATCH
415 ERR_EMAIL_MISMATCH = _lib.X509_V_ERR_EMAIL_MISMATCH
416 ERR_IP_ADDRESS_MISMATCH = _lib.X509_V_ERR_IP_ADDRESS_MISMATCH
417 ERR_APPLICATION_VERIFICATION = _lib.X509_V_ERR_APPLICATION_VERIFICATION
420# Taken from https://golang.org/src/crypto/x509/root_linux.go
421_CERTIFICATE_FILE_LOCATIONS = [
422 "/etc/ssl/certs/ca-certificates.crt", # Debian/Ubuntu/Gentoo etc.
423 "/etc/pki/tls/certs/ca-bundle.crt", # Fedora/RHEL 6
424 "/etc/ssl/ca-bundle.pem", # OpenSUSE
425 "/etc/pki/tls/cacert.pem", # OpenELEC
426 "/etc/pki/ca-trust/extracted/pem/tls-ca-bundle.pem", # CentOS/RHEL 7
427]
429_CERTIFICATE_PATH_LOCATIONS = [
430 "/etc/ssl/certs", # SLES10/SLES11
431]
433# These values are compared to output from cffi's ffi.string so they must be
434# byte strings.
435_CRYPTOGRAPHY_MANYLINUX_CA_DIR = b"/opt/pyca/cryptography/openssl/certs"
436_CRYPTOGRAPHY_MANYLINUX_CA_FILE = b"/opt/pyca/cryptography/openssl/cert.pem"
439class Error(Exception):
440 """
441 An error occurred in an `OpenSSL.SSL` API.
442 """
445_raise_current_error = partial(_exception_from_error_queue, Error)
446_openssl_assert = _make_assert(Error)
449class WantReadError(Error):
450 pass
453class WantWriteError(Error):
454 pass
457class WantX509LookupError(Error):
458 pass
461class ZeroReturnError(Error):
462 pass
465class SysCallError(Error):
466 pass
469class _CallbackExceptionHelper:
470 """
471 A base class for wrapper classes that allow for intelligent exception
472 handling in OpenSSL callbacks.
474 :ivar list _problems: Any exceptions that occurred while executing in a
475 context where they could not be raised in the normal way. Typically
476 this is because OpenSSL has called into some Python code and requires a
477 return value. The exceptions are saved to be raised later when it is
478 possible to do so.
479 """
481 def __init__(self) -> None:
482 self._problems: list[Exception] = []
484 def raise_if_problem(self) -> None:
485 """
486 Raise an exception from the OpenSSL error queue or that was previously
487 captured whe running a callback.
488 """
489 if self._problems:
490 try:
491 _raise_current_error()
492 except Error:
493 pass
494 raise self._problems.pop(0)
497class _VerifyHelper(_CallbackExceptionHelper):
498 """
499 Wrap a callback such that it can be used as a certificate verification
500 callback.
501 """
503 def __init__(self, callback: _VerifyCallback) -> None:
504 _CallbackExceptionHelper.__init__(self)
506 @wraps(callback)
507 def wrapper(ok, store_ctx): # type: ignore[no-untyped-def]
508 x509 = _lib.X509_STORE_CTX_get_current_cert(store_ctx)
509 _lib.X509_up_ref(x509)
510 cert = X509._from_raw_x509_ptr(x509)
511 error_number = _lib.X509_STORE_CTX_get_error(store_ctx)
512 error_depth = _lib.X509_STORE_CTX_get_error_depth(store_ctx)
514 index = _lib.SSL_get_ex_data_X509_STORE_CTX_idx()
515 ssl = _lib.X509_STORE_CTX_get_ex_data(store_ctx, index)
516 connection = Connection._reverse_mapping[ssl]
518 try:
519 result = callback(
520 connection, cert, error_number, error_depth, ok
521 )
522 except Exception as e:
523 self._problems.append(e)
524 return 0
525 else:
526 if result:
527 _lib.X509_STORE_CTX_set_error(store_ctx, _lib.X509_V_OK)
528 return 1
529 else:
530 return 0
532 self.callback = _ffi.callback(
533 "int (*)(int, X509_STORE_CTX *)", wrapper
534 )
537class _ALPNSelectHelper(_CallbackExceptionHelper):
538 """
539 Wrap a callback such that it can be used as an ALPN selection callback.
540 """
542 def __init__(self, callback: _ALPNSelectCallback) -> None:
543 _CallbackExceptionHelper.__init__(self)
545 @wraps(callback)
546 def wrapper(ssl, out, outlen, in_, inlen, arg): # type: ignore[no-untyped-def]
547 try:
548 conn = Connection._reverse_mapping[ssl]
550 # The string passed to us is made up of multiple
551 # length-prefixed bytestrings. We need to split that into a
552 # list.
553 instr = _ffi.buffer(in_, inlen)[:]
554 protolist = []
555 while instr:
556 encoded_len = instr[0]
557 proto = instr[1 : encoded_len + 1]
558 protolist.append(proto)
559 instr = instr[encoded_len + 1 :]
561 # Call the callback
562 outbytes = callback(conn, protolist)
563 any_accepted = True
564 if outbytes is NO_OVERLAPPING_PROTOCOLS:
565 outbytes = b""
566 any_accepted = False
567 elif not isinstance(outbytes, bytes):
568 raise TypeError(
569 "ALPN callback must return a bytestring or the "
570 "special NO_OVERLAPPING_PROTOCOLS sentinel value."
571 )
573 # Save our callback arguments on the connection object to make
574 # sure that they don't get freed before OpenSSL can use them.
575 # Then, return them in the appropriate output parameters.
576 conn._alpn_select_callback_args = [
577 _ffi.new("unsigned char *", len(outbytes)),
578 _ffi.new("unsigned char[]", outbytes),
579 ]
580 outlen[0] = conn._alpn_select_callback_args[0][0]
581 out[0] = conn._alpn_select_callback_args[1]
582 if not any_accepted:
583 return _lib.SSL_TLSEXT_ERR_NOACK
584 return _lib.SSL_TLSEXT_ERR_OK
585 except Exception as e:
586 self._problems.append(e)
587 return _lib.SSL_TLSEXT_ERR_ALERT_FATAL
589 self.callback = _ffi.callback(
590 (
591 "int (*)(SSL *, unsigned char **, unsigned char *, "
592 "const unsigned char *, unsigned int, void *)"
593 ),
594 wrapper,
595 )
598class _OCSPServerCallbackHelper(_CallbackExceptionHelper):
599 """
600 Wrap a callback such that it can be used as an OCSP callback for the server
601 side.
603 Annoyingly, OpenSSL defines one OCSP callback but uses it in two different
604 ways. For servers, that callback is expected to retrieve some OCSP data and
605 hand it to OpenSSL, and may return only SSL_TLSEXT_ERR_OK,
606 SSL_TLSEXT_ERR_FATAL, and SSL_TLSEXT_ERR_NOACK. For clients, that callback
607 is expected to check the OCSP data, and returns a negative value on error,
608 0 if the response is not acceptable, or positive if it is. These are
609 mutually exclusive return code behaviours, and they mean that we need two
610 helpers so that we always return an appropriate error code if the user's
611 code throws an exception.
613 Given that we have to have two helpers anyway, these helpers are a bit more
614 helpery than most: specifically, they hide a few more of the OpenSSL
615 functions so that the user has an easier time writing these callbacks.
617 This helper implements the server side.
618 """
620 def __init__(self, callback: _OCSPServerCallback[Any]) -> None:
621 _CallbackExceptionHelper.__init__(self)
623 @wraps(callback)
624 def wrapper(ssl, cdata): # type: ignore[no-untyped-def]
625 try:
626 conn = Connection._reverse_mapping[ssl]
628 # Extract the data if any was provided.
629 if cdata != _ffi.NULL:
630 data = _ffi.from_handle(cdata)
631 else:
632 data = None
634 # Call the callback.
635 ocsp_data = callback(conn, data)
637 if not isinstance(ocsp_data, bytes):
638 raise TypeError("OCSP callback must return a bytestring.")
640 # If the OCSP data was provided, we will pass it to OpenSSL.
641 # However, we have an early exit here: if no OCSP data was
642 # provided we will just exit out and tell OpenSSL that there
643 # is nothing to do.
644 if not ocsp_data:
645 return 3 # SSL_TLSEXT_ERR_NOACK
647 # OpenSSL takes ownership of this data and expects it to have
648 # been allocated by OPENSSL_malloc.
649 ocsp_data_length = len(ocsp_data)
650 data_ptr = _lib.OPENSSL_malloc(ocsp_data_length)
651 _ffi.buffer(data_ptr, ocsp_data_length)[:] = ocsp_data
653 _lib.SSL_set_tlsext_status_ocsp_resp(
654 ssl, data_ptr, ocsp_data_length
655 )
657 return 0
658 except Exception as e:
659 self._problems.append(e)
660 return 2 # SSL_TLSEXT_ERR_ALERT_FATAL
662 self.callback = _ffi.callback("int (*)(SSL *, void *)", wrapper)
665class _OCSPClientCallbackHelper(_CallbackExceptionHelper):
666 """
667 Wrap a callback such that it can be used as an OCSP callback for the client
668 side.
670 Annoyingly, OpenSSL defines one OCSP callback but uses it in two different
671 ways. For servers, that callback is expected to retrieve some OCSP data and
672 hand it to OpenSSL, and may return only SSL_TLSEXT_ERR_OK,
673 SSL_TLSEXT_ERR_FATAL, and SSL_TLSEXT_ERR_NOACK. For clients, that callback
674 is expected to check the OCSP data, and returns a negative value on error,
675 0 if the response is not acceptable, or positive if it is. These are
676 mutually exclusive return code behaviours, and they mean that we need two
677 helpers so that we always return an appropriate error code if the user's
678 code throws an exception.
680 Given that we have to have two helpers anyway, these helpers are a bit more
681 helpery than most: specifically, they hide a few more of the OpenSSL
682 functions so that the user has an easier time writing these callbacks.
684 This helper implements the client side.
685 """
687 def __init__(self, callback: _OCSPClientCallback[Any]) -> None:
688 _CallbackExceptionHelper.__init__(self)
690 @wraps(callback)
691 def wrapper(ssl, cdata): # type: ignore[no-untyped-def]
692 try:
693 conn = Connection._reverse_mapping[ssl]
695 # Extract the data if any was provided.
696 if cdata != _ffi.NULL:
697 data = _ffi.from_handle(cdata)
698 else:
699 data = None
701 # Get the OCSP data.
702 ocsp_ptr = _ffi.new("unsigned char **")
703 ocsp_len = _lib.SSL_get_tlsext_status_ocsp_resp(ssl, ocsp_ptr)
704 if ocsp_len < 0:
705 # No OCSP data.
706 ocsp_data = b""
707 else:
708 # Copy the OCSP data, then pass it to the callback.
709 ocsp_data = _ffi.buffer(ocsp_ptr[0], ocsp_len)[:]
711 valid = callback(conn, ocsp_data, data)
713 # Return 1 on success or 0 on error.
714 return int(bool(valid))
716 except Exception as e:
717 self._problems.append(e)
718 # Return negative value if an exception is hit.
719 return -1
721 self.callback = _ffi.callback("int (*)(SSL *, void *)", wrapper)
724class _CookieGenerateCallbackHelper(_CallbackExceptionHelper):
725 def __init__(self, callback: _CookieGenerateCallback) -> None:
726 _CallbackExceptionHelper.__init__(self)
728 max_cookie_len = getattr(_lib, "DTLS1_COOKIE_LENGTH", 255)
730 @wraps(callback)
731 def wrapper(ssl, out, outlen): # type: ignore[no-untyped-def]
732 try:
733 conn = Connection._reverse_mapping[ssl]
734 cookie = callback(conn)
735 if len(cookie) > max_cookie_len:
736 raise ValueError(
737 f"Cookie too long (got {len(cookie)} bytes, "
738 f"max {max_cookie_len})"
739 )
740 out[0 : len(cookie)] = cookie
741 outlen[0] = len(cookie)
742 return 1
743 except Exception as e:
744 self._problems.append(e)
745 # "a zero return value can be used to abort the handshake"
746 return 0
748 self.callback = _ffi.callback(
749 "int (*)(SSL *, unsigned char *, unsigned int *)",
750 wrapper,
751 )
754class _CookieVerifyCallbackHelper(_CallbackExceptionHelper):
755 def __init__(self, callback: _CookieVerifyCallback) -> None:
756 _CallbackExceptionHelper.__init__(self)
758 @wraps(callback)
759 def wrapper(ssl, c_cookie, cookie_len): # type: ignore[no-untyped-def]
760 try:
761 conn = Connection._reverse_mapping[ssl]
762 return callback(conn, bytes(c_cookie[0:cookie_len]))
763 except Exception as e:
764 self._problems.append(e)
765 return 0
767 self.callback = _ffi.callback(
768 "int (*)(SSL *, unsigned char *, unsigned int)",
769 wrapper,
770 )
773def _asFileDescriptor(obj: Any) -> int:
774 fd = None
775 if not isinstance(obj, int):
776 meth = getattr(obj, "fileno", None)
777 if meth is not None:
778 obj = meth()
780 if isinstance(obj, int):
781 fd = obj
783 if not isinstance(fd, int):
784 raise TypeError("argument must be an int, or have a fileno() method.")
785 elif fd < 0:
786 raise ValueError(
787 f"file descriptor cannot be a negative integer ({fd:i})"
788 )
790 return fd
793def OpenSSL_version(type: int) -> bytes:
794 """
795 Return a string describing the version of OpenSSL in use.
797 :param type: One of the :const:`OPENSSL_` constants defined in this module.
798 """
799 return _ffi.string(_lib.OpenSSL_version(type))
802SSLeay_version = OpenSSL_version
805def _make_requires(flag: int, error: str) -> Callable[[_T], _T]:
806 """
807 Builds a decorator that ensures that functions that rely on OpenSSL
808 functions that are not present in this build raise NotImplementedError,
809 rather than AttributeError coming out of cryptography.
811 :param flag: A cryptography flag that guards the functions, e.g.
812 ``Cryptography_HAS_NEXTPROTONEG``.
813 :param error: The string to be used in the exception if the flag is false.
814 """
816 def _requires_decorator(func): # type: ignore[no-untyped-def]
817 if not flag:
819 @wraps(func)
820 def explode(*args, **kwargs): # type: ignore[no-untyped-def]
821 raise NotImplementedError(error)
823 return explode
824 else:
825 return func
827 return _requires_decorator
830_requires_keylog = _make_requires(
831 getattr(_lib, "Cryptography_HAS_KEYLOG", 0), "Key logging not available"
832)
834_requires_ssl_get0_group_name = _make_requires(
835 getattr(_lib, "Cryptography_HAS_SSL_GET0_GROUP_NAME", 0),
836 "Getting group name is not supported by the linked OpenSSL version",
837)
839_requires_ssl_cookie = _make_requires(
840 getattr(_lib, "Cryptography_HAS_SSL_COOKIE", 0),
841 "DTLS cookie support is not available",
842)
845class Session:
846 """
847 A class representing an SSL session. A session defines certain connection
848 parameters which may be re-used to speed up the setup of subsequent
849 connections.
851 .. versionadded:: 0.14
852 """
854 _session: Any
855 # The Context the Connection this Session came from was using. OpenSSL
856 # requires that a session only be re-used with a compatible SSL_CTX, but
857 # doesn't verify it, so we pin the Context here and enforce identity in
858 # Connection.set_session.
859 _context: Context
862F = TypeVar("F", bound=Callable[..., Any])
865def _require_not_used(f: F) -> F:
866 @wraps(f)
867 def inner(self: Context, *args: Any, **kwargs: Any) -> Any:
868 if self._used:
869 raise ValueError(
870 "Context has already been used to create a Connection, it "
871 "cannot be mutated again"
872 )
873 return f(self, *args, **kwargs)
875 return typing.cast(F, inner)
878class Context:
879 """
880 :class:`OpenSSL.SSL.Context` instances define the parameters for setting
881 up new SSL connections.
883 :param method: One of TLS_METHOD, TLS_CLIENT_METHOD, TLS_SERVER_METHOD,
884 DTLS_METHOD, DTLS_CLIENT_METHOD, or DTLS_SERVER_METHOD.
885 SSLv23_METHOD, TLSv1_METHOD, etc. are deprecated and should
886 not be used.
887 """
889 _methods: typing.ClassVar[
890 dict[int, tuple[Callable[[], Any], int | None]]
891 ] = {
892 SSLv23_METHOD: (_lib.TLS_method, None),
893 TLSv1_METHOD: (_lib.TLS_method, TLS1_VERSION),
894 TLSv1_1_METHOD: (_lib.TLS_method, TLS1_1_VERSION),
895 TLSv1_2_METHOD: (_lib.TLS_method, TLS1_2_VERSION),
896 TLS_METHOD: (_lib.TLS_method, None),
897 TLS_SERVER_METHOD: (_lib.TLS_server_method, None),
898 TLS_CLIENT_METHOD: (_lib.TLS_client_method, None),
899 DTLS_METHOD: (_lib.DTLS_method, None),
900 DTLS_SERVER_METHOD: (_lib.DTLS_server_method, None),
901 DTLS_CLIENT_METHOD: (_lib.DTLS_client_method, None),
902 }
904 def __init__(self, method: int) -> None:
905 if not isinstance(method, int):
906 raise TypeError("method must be an integer")
908 try:
909 method_func, version = self._methods[method]
910 except KeyError:
911 raise ValueError("No such protocol")
913 method_obj = method_func()
914 _openssl_assert(method_obj != _ffi.NULL)
916 context = _lib.SSL_CTX_new(method_obj)
917 _openssl_assert(context != _ffi.NULL)
918 context = _ffi.gc(context, _lib.SSL_CTX_free)
920 self._context = context
921 self._used = False
922 self._passphrase_helper: _PassphraseHelper | None = None
923 self._passphrase_callback: _PassphraseCallback[Any] | None = None
924 self._passphrase_userdata: Any | None = None
925 self._verify_helper: _VerifyHelper | None = None
926 self._verify_callback: _VerifyCallback | None = None
927 self._info_callback = None
928 self._keylog_callback = None
929 self._tlsext_servername_callback = None
930 self._app_data = None
931 self._alpn_select_helper: _ALPNSelectHelper | None = None
932 self._alpn_select_callback: _ALPNSelectCallback | None = None
933 self._ocsp_helper: (
934 _OCSPClientCallbackHelper | _OCSPServerCallbackHelper | None
935 ) = None
936 self._ocsp_callback: (
937 _OCSPClientCallback[Any] | _OCSPServerCallback[Any] | None
938 ) = None
939 self._ocsp_data: Any | None = None
940 self._cookie_generate_helper: _CookieGenerateCallbackHelper | None = (
941 None
942 )
943 self._cookie_verify_helper: _CookieVerifyCallbackHelper | None = None
945 self.set_mode(
946 _lib.SSL_MODE_ENABLE_PARTIAL_WRITE
947 | _lib.SSL_MODE_ACCEPT_MOVING_WRITE_BUFFER
948 )
949 if version is not None:
950 self.set_min_proto_version(version)
951 self.set_max_proto_version(version)
953 @_require_not_used
954 def set_min_proto_version(self, version: int) -> None:
955 """
956 Set the minimum supported protocol version. Setting the minimum
957 version to 0 will enable protocol versions down to the lowest version
958 supported by the library.
960 If the underlying OpenSSL build is missing support for the selected
961 version, this method will raise an exception.
962 """
963 _openssl_assert(
964 _lib.SSL_CTX_set_min_proto_version(self._context, version) == 1
965 )
967 @_require_not_used
968 def set_max_proto_version(self, version: int) -> None:
969 """
970 Set the maximum supported protocol version. Setting the maximum
971 version to 0 will enable protocol versions up to the highest version
972 supported by the library.
974 If the underlying OpenSSL build is missing support for the selected
975 version, this method will raise an exception.
976 """
977 _openssl_assert(
978 _lib.SSL_CTX_set_max_proto_version(self._context, version) == 1
979 )
981 @_require_not_used
982 def load_verify_locations(
983 self,
984 cafile: _StrOrBytesPath | None,
985 capath: _StrOrBytesPath | None = None,
986 ) -> None:
987 """
988 Let SSL know where we can find trusted certificates for the certificate
989 chain. Note that the certificates have to be in PEM format.
991 If capath is passed, it must be a directory prepared using the
992 ``c_rehash`` tool included with OpenSSL. Either, but not both, of
993 *pemfile* or *capath* may be :data:`None`.
995 :param cafile: In which file we can find the certificates (``bytes`` or
996 ``str``).
997 :param capath: In which directory we can find the certificates
998 (``bytes`` or ``str``).
1000 :return: None
1001 """
1002 if cafile is None:
1003 cafile = _ffi.NULL
1004 else:
1005 cafile = _path_bytes(cafile)
1007 if capath is None:
1008 capath = _ffi.NULL
1009 else:
1010 capath = _path_bytes(capath)
1012 load_result = _lib.SSL_CTX_load_verify_locations(
1013 self._context, cafile, capath
1014 )
1015 if not load_result:
1016 _raise_current_error()
1018 def _wrap_callback(
1019 self, callback: _PassphraseCallback[_T]
1020 ) -> _PassphraseHelper:
1021 @wraps(callback)
1022 def wrapper(size: int, verify: bool, userdata: Any) -> bytes:
1023 return callback(size, verify, self._passphrase_userdata)
1025 return _PassphraseHelper(
1026 FILETYPE_PEM, wrapper, more_args=True, truncate=True
1027 )
1029 @deprecated(
1030 "Context.set_passwd_cb is deprecated. You should decrypt and load "
1031 "your private key yourself, with cryptography's key loading APIs, "
1032 "and then use Context.use_privatekey instead."
1033 )
1034 @_require_not_used
1035 def set_passwd_cb(
1036 self,
1037 callback: _PassphraseCallback[_T],
1038 userdata: _T | None = None,
1039 ) -> None:
1040 """
1041 Set the passphrase callback. This function will be called
1042 when a private key with a passphrase is loaded.
1044 :param callback: The Python callback to use. This must accept three
1045 positional arguments. First, an integer giving the maximum length
1046 of the passphrase it may return. If the returned passphrase is
1047 longer than this, it will be truncated. Second, a boolean value
1048 which will be true if the user should be prompted for the
1049 passphrase twice and the callback should verify that the two values
1050 supplied are equal. Third, the value given as the *userdata*
1051 parameter to :meth:`set_passwd_cb`. The *callback* must return
1052 a byte string. If an error occurs, *callback* should return a false
1053 value (e.g. an empty string).
1054 :param userdata: (optional) A Python object which will be given as
1055 argument to the callback
1056 :return: None
1057 """
1058 if not callable(callback):
1059 raise TypeError("callback must be callable")
1061 self._passphrase_helper = self._wrap_callback(callback)
1062 self._passphrase_callback = self._passphrase_helper.callback
1063 _lib.SSL_CTX_set_default_passwd_cb(
1064 self._context, self._passphrase_callback
1065 )
1066 self._passphrase_userdata = userdata
1068 @_require_not_used
1069 def set_default_verify_paths(self) -> None:
1070 """
1071 Specify that the platform provided CA certificates are to be used for
1072 verification purposes. This method has some caveats related to the
1073 binary wheels that cryptography (pyOpenSSL's primary dependency) ships:
1075 * macOS will only load certificates using this method if the user has
1076 the ``openssl@3`` `Homebrew <https://brew.sh>`_ formula installed
1077 in the default location.
1078 * Windows will not work.
1079 * manylinux cryptography wheels will work on most common Linux
1080 distributions in pyOpenSSL 17.1.0 and above. pyOpenSSL detects the
1081 manylinux wheel and attempts to load roots via a fallback path.
1083 :return: None
1084 """
1085 # SSL_CTX_set_default_verify_paths will attempt to load certs from
1086 # both a cafile and capath that are set at compile time. However,
1087 # it will first check environment variables and, if present, load
1088 # those paths instead
1089 set_result = _lib.SSL_CTX_set_default_verify_paths(self._context)
1090 _openssl_assert(set_result == 1)
1091 # After attempting to set default_verify_paths we need to know whether
1092 # to go down the fallback path.
1093 # First we'll check to see if any env vars have been set. If so,
1094 # we won't try to do anything else because the user has set the path
1095 # themselves.
1096 if not self._check_env_vars_set("SSL_CERT_DIR", "SSL_CERT_FILE"):
1097 default_dir = _ffi.string(_lib.X509_get_default_cert_dir())
1098 default_file = _ffi.string(_lib.X509_get_default_cert_file())
1099 # Now we check to see if the default_dir and default_file are set
1100 # to the exact values we use in our manylinux builds. If they are
1101 # then we know to load the fallbacks
1102 if (
1103 default_dir == _CRYPTOGRAPHY_MANYLINUX_CA_DIR
1104 and default_file == _CRYPTOGRAPHY_MANYLINUX_CA_FILE
1105 ):
1106 # This is manylinux, let's load our fallback paths
1107 self._fallback_default_verify_paths(
1108 _CERTIFICATE_FILE_LOCATIONS, _CERTIFICATE_PATH_LOCATIONS
1109 )
1111 def _check_env_vars_set(self, dir_env_var: str, file_env_var: str) -> bool:
1112 """
1113 Check to see if the default cert dir/file environment vars are present.
1115 :return: bool
1116 """
1117 return (
1118 os.environ.get(file_env_var) is not None
1119 or os.environ.get(dir_env_var) is not None
1120 )
1122 def _fallback_default_verify_paths(
1123 self, file_path: list[str], dir_path: list[str]
1124 ) -> None:
1125 """
1126 Default verify paths are based on the compiled version of OpenSSL.
1127 However, when pyca/cryptography is compiled as a manylinux wheel
1128 that compiled location can potentially be wrong. So, like Go, we
1129 will try a predefined set of paths and attempt to load roots
1130 from there.
1132 :return: None
1133 """
1134 for cafile in file_path:
1135 if os.path.isfile(cafile):
1136 self.load_verify_locations(cafile)
1137 break
1139 for capath in dir_path:
1140 if os.path.isdir(capath):
1141 self.load_verify_locations(None, capath)
1142 break
1144 @_require_not_used
1145 def use_certificate_chain_file(self, certfile: _StrOrBytesPath) -> None:
1146 """
1147 Load a certificate chain from a file.
1149 :param certfile: The name of the certificate chain file (``bytes`` or
1150 ``str``). Must be PEM encoded.
1152 :return: None
1153 """
1154 certfile = _path_bytes(certfile)
1156 result = _lib.SSL_CTX_use_certificate_chain_file(
1157 self._context, certfile
1158 )
1159 if not result:
1160 _raise_current_error()
1162 @_require_not_used
1163 def use_certificate_file(
1164 self, certfile: _StrOrBytesPath, filetype: int = FILETYPE_PEM
1165 ) -> None:
1166 """
1167 Load a certificate from a file
1169 :param certfile: The name of the certificate file (``bytes`` or
1170 ``str``).
1171 :param filetype: (optional) The encoding of the file, which is either
1172 :const:`FILETYPE_PEM` or :const:`FILETYPE_ASN1`. The default is
1173 :const:`FILETYPE_PEM`.
1175 :return: None
1176 """
1177 certfile = _path_bytes(certfile)
1178 if not isinstance(filetype, int):
1179 raise TypeError("filetype must be an integer")
1181 use_result = _lib.SSL_CTX_use_certificate_file(
1182 self._context, certfile, filetype
1183 )
1184 if not use_result:
1185 _raise_current_error()
1187 @_require_not_used
1188 def use_certificate(self, cert: X509 | x509.Certificate) -> None:
1189 """
1190 Load a certificate from a X509 object
1192 :param cert: The X509 object
1193 :return: None
1194 """
1195 # Mirrored at Connection.use_certificate
1196 if not isinstance(cert, X509):
1197 cert = X509.from_cryptography(cert)
1198 else:
1199 warnings.warn(
1200 (
1201 "Passing pyOpenSSL X509 objects is deprecated. You "
1202 "should use a cryptography.x509.Certificate instead."
1203 ),
1204 DeprecationWarning,
1205 stacklevel=2,
1206 )
1208 use_result = _lib.SSL_CTX_use_certificate(self._context, cert._x509)
1209 if not use_result:
1210 _raise_current_error()
1212 @_require_not_used
1213 def add_extra_chain_cert(self, certobj: X509 | x509.Certificate) -> None:
1214 """
1215 Add certificate to chain
1217 :param certobj: The X509 certificate object to add to the chain
1218 :return: None
1219 """
1220 if not isinstance(certobj, X509):
1221 certobj = X509.from_cryptography(certobj)
1222 else:
1223 warnings.warn(
1224 (
1225 "Passing pyOpenSSL X509 objects is deprecated. You "
1226 "should use a cryptography.x509.Certificate instead."
1227 ),
1228 DeprecationWarning,
1229 stacklevel=2,
1230 )
1232 copy = _lib.X509_dup(certobj._x509)
1233 add_result = _lib.SSL_CTX_add_extra_chain_cert(self._context, copy)
1234 if not add_result:
1235 # TODO: This is untested.
1236 _lib.X509_free(copy)
1237 _raise_current_error()
1239 def _raise_passphrase_exception(self) -> None:
1240 if self._passphrase_helper is not None:
1241 self._passphrase_helper.raise_if_problem(Error)
1243 _raise_current_error()
1245 @_require_not_used
1246 def use_privatekey_file(
1247 self, keyfile: _StrOrBytesPath, filetype: int = FILETYPE_PEM
1248 ) -> None:
1249 """
1250 Load a private key from a file
1252 :param keyfile: The name of the key file (``bytes`` or ``str``)
1253 :param filetype: (optional) The encoding of the file, which is either
1254 :const:`FILETYPE_PEM` or :const:`FILETYPE_ASN1`. The default is
1255 :const:`FILETYPE_PEM`.
1257 :return: None
1258 """
1259 keyfile = _path_bytes(keyfile)
1261 if not isinstance(filetype, int):
1262 raise TypeError("filetype must be an integer")
1264 use_result = _lib.SSL_CTX_use_PrivateKey_file(
1265 self._context, keyfile, filetype
1266 )
1267 if not use_result:
1268 self._raise_passphrase_exception()
1270 @_require_not_used
1271 def use_privatekey(self, pkey: _PrivateKey | PKey) -> None:
1272 """
1273 Load a private key from a PKey object
1275 :param pkey: The PKey object
1276 :return: None
1277 """
1278 # Mirrored at Connection.use_privatekey
1279 if not isinstance(pkey, PKey):
1280 pkey = PKey.from_cryptography_key(pkey)
1281 else:
1282 warnings.warn(
1283 (
1284 "Passing pyOpenSSL PKey objects is deprecated. You "
1285 "should use a cryptography private key instead."
1286 ),
1287 DeprecationWarning,
1288 stacklevel=2,
1289 )
1291 use_result = _lib.SSL_CTX_use_PrivateKey(self._context, pkey._pkey)
1292 if not use_result:
1293 self._raise_passphrase_exception()
1295 def check_privatekey(self) -> None:
1296 """
1297 Check if the private key (loaded with :meth:`use_privatekey`) matches
1298 the certificate (loaded with :meth:`use_certificate`)
1300 :return: :data:`None` (raises :exc:`Error` if something's wrong)
1301 """
1302 if not _lib.SSL_CTX_check_private_key(self._context):
1303 _raise_current_error()
1305 @_require_not_used
1306 def load_client_ca(self, cafile: bytes) -> None:
1307 """
1308 Load the trusted certificates that will be sent to the client. Does
1309 not actually imply any of the certificates are trusted; that must be
1310 configured separately.
1312 :param bytes cafile: The path to a certificates file in PEM format.
1313 :return: None
1314 """
1315 ca_list = _lib.SSL_load_client_CA_file(
1316 _text_to_bytes_and_warn("cafile", cafile)
1317 )
1318 _openssl_assert(ca_list != _ffi.NULL)
1319 _lib.SSL_CTX_set_client_CA_list(self._context, ca_list)
1321 @_require_not_used
1322 def set_session_id(self, buf: bytes) -> None:
1323 """
1324 Set the session id to *buf* within which a session can be reused for
1325 this Context object. This is needed when doing session resumption,
1326 because there is no way for a stored session to know which Context
1327 object it is associated with.
1329 :param bytes buf: The session id.
1331 :returns: None
1332 """
1333 buf = _text_to_bytes_and_warn("buf", buf)
1334 _openssl_assert(
1335 _lib.SSL_CTX_set_session_id_context(self._context, buf, len(buf))
1336 == 1
1337 )
1339 @_require_not_used
1340 def set_session_cache_mode(self, mode: int) -> int:
1341 """
1342 Set the behavior of the session cache used by all connections using
1343 this Context. The previously set mode is returned. See
1344 :const:`SESS_CACHE_*` for details about particular modes.
1346 :param mode: One or more of the SESS_CACHE_* flags (combine using
1347 bitwise or)
1348 :returns: The previously set caching mode.
1350 .. versionadded:: 0.14
1351 """
1352 if not isinstance(mode, int):
1353 raise TypeError("mode must be an integer")
1355 return _lib.SSL_CTX_set_session_cache_mode(self._context, mode)
1357 def get_session_cache_mode(self) -> int:
1358 """
1359 Get the current session cache mode.
1361 :returns: The currently used cache mode.
1363 .. versionadded:: 0.14
1364 """
1365 return _lib.SSL_CTX_get_session_cache_mode(self._context)
1367 @_require_not_used
1368 def set_verify(
1369 self, mode: int, callback: _VerifyCallback | None = None
1370 ) -> None:
1371 """
1372 Set the verification flags for this Context object to *mode* and
1373 specify that *callback* should be used for verification callbacks.
1375 :param mode: The verify mode, this should be one of
1376 :const:`VERIFY_NONE` and :const:`VERIFY_PEER`. If
1377 :const:`VERIFY_PEER` is used, *mode* can be OR:ed with
1378 :const:`VERIFY_FAIL_IF_NO_PEER_CERT` and
1379 :const:`VERIFY_CLIENT_ONCE` to further control the behaviour.
1380 :param callback: The optional Python verification callback to use.
1381 This should take five arguments: A Connection object, an X509
1382 object, and three integer variables, which are in turn potential
1383 error number, error depth and return code. *callback* should
1384 return True if verification passes and False otherwise.
1385 If omitted, OpenSSL's default verification is used.
1386 :return: None
1388 See SSL_CTX_set_verify(3SSL) for further details.
1389 """
1390 if not isinstance(mode, int):
1391 raise TypeError("mode must be an integer")
1393 if callback is None:
1394 self._verify_helper = None
1395 self._verify_callback = None
1396 _lib.SSL_CTX_set_verify(self._context, mode, _ffi.NULL)
1397 else:
1398 if not callable(callback):
1399 raise TypeError("callback must be callable")
1401 self._verify_helper = _VerifyHelper(callback)
1402 self._verify_callback = self._verify_helper.callback
1403 _lib.SSL_CTX_set_verify(self._context, mode, self._verify_callback)
1405 @_require_not_used
1406 def set_verify_depth(self, depth: int) -> None:
1407 """
1408 Set the maximum depth for the certificate chain verification that shall
1409 be allowed for this Context object.
1411 :param depth: An integer specifying the verify depth
1412 :return: None
1413 """
1414 if not isinstance(depth, int):
1415 raise TypeError("depth must be an integer")
1417 _lib.SSL_CTX_set_verify_depth(self._context, depth)
1419 def get_verify_mode(self) -> int:
1420 """
1421 Retrieve the Context object's verify mode, as set by
1422 :meth:`set_verify`.
1424 :return: The verify mode
1425 """
1426 return _lib.SSL_CTX_get_verify_mode(self._context)
1428 def get_verify_depth(self) -> int:
1429 """
1430 Retrieve the Context object's verify depth, as set by
1431 :meth:`set_verify_depth`.
1433 :return: The verify depth
1434 """
1435 return _lib.SSL_CTX_get_verify_depth(self._context)
1437 @_require_not_used
1438 def load_tmp_dh(self, dhfile: _StrOrBytesPath) -> None:
1439 """
1440 Load parameters for Ephemeral Diffie-Hellman
1442 :param dhfile: The file to load EDH parameters from (``bytes`` or
1443 ``str``).
1445 :return: None
1446 """
1447 dhfile = _path_bytes(dhfile)
1449 bio = _lib.BIO_new_file(dhfile, b"r")
1450 if bio == _ffi.NULL:
1451 _raise_current_error()
1452 bio = _ffi.gc(bio, _lib.BIO_free)
1454 dh = _lib.PEM_read_bio_DHparams(bio, _ffi.NULL, _ffi.NULL, _ffi.NULL)
1455 dh = _ffi.gc(dh, _lib.DH_free)
1456 res = _lib.SSL_CTX_set_tmp_dh(self._context, dh)
1457 _openssl_assert(res == 1)
1459 @_require_not_used
1460 def set_tmp_ecdh(self, curve: _EllipticCurve | ec.EllipticCurve) -> None:
1461 """
1462 Select a curve to use for ECDHE key exchange.
1464 :param curve: A curve instance from cryptography
1465 (:class:`~cryptogragraphy.hazmat.primitives.asymmetric.ec.EllipticCurve`).
1466 Alternatively (deprecated) a curve object from either
1467 :meth:`OpenSSL.crypto.get_elliptic_curve` or
1468 :meth:`OpenSSL.crypto.get_elliptic_curves`.
1470 :return: None
1471 """
1473 if isinstance(curve, _EllipticCurve):
1474 warnings.warn(
1475 (
1476 "Passing pyOpenSSL elliptic curves to set_tmp_ecdh is "
1477 "deprecated. You should use cryptography's elliptic curve "
1478 "types instead."
1479 ),
1480 DeprecationWarning,
1481 stacklevel=2,
1482 )
1483 _lib.SSL_CTX_set_tmp_ecdh(self._context, curve._to_EC_KEY())
1484 else:
1485 name = curve.name
1486 if name == "secp192r1":
1487 name = "prime192v1"
1488 elif name == "secp256r1":
1489 name = "prime256v1"
1490 nid = _lib.OBJ_txt2nid(name.encode())
1491 if nid == _lib.NID_undef:
1492 _raise_current_error()
1494 ec = _lib.EC_KEY_new_by_curve_name(nid)
1495 _openssl_assert(ec != _ffi.NULL)
1496 ec = _ffi.gc(ec, _lib.EC_KEY_free)
1497 _lib.SSL_CTX_set_tmp_ecdh(self._context, ec)
1499 @_require_not_used
1500 def set_cipher_list(self, cipher_list: bytes) -> None:
1501 """
1502 Set the list of ciphers to be used in this context.
1504 See the OpenSSL manual for more information (e.g.
1505 :manpage:`ciphers(1)`).
1507 Note this API does not change the cipher suites used in TLS 1.3
1508 Use `set_tls13_ciphersuites` for that.
1510 :param bytes cipher_list: An OpenSSL cipher string.
1511 :return: None
1512 """
1513 cipher_list = _text_to_bytes_and_warn("cipher_list", cipher_list)
1515 if not isinstance(cipher_list, bytes):
1516 raise TypeError("cipher_list must be a byte string.")
1518 _openssl_assert(
1519 _lib.SSL_CTX_set_cipher_list(self._context, cipher_list) == 1
1520 )
1522 @_require_not_used
1523 def set_tls13_ciphersuites(self, ciphersuites: bytes) -> None:
1524 """
1525 Set the list of TLS 1.3 ciphers to be used in this context.
1526 OpenSSL maintains a separate list of TLS 1.3+ ciphers to
1527 ciphers for TLS 1.2 and lowers.
1529 See the OpenSSL manual for more information (e.g.
1530 :manpage:`ciphers(1)`).
1532 :param bytes ciphersuites: An OpenSSL cipher string containing
1533 TLS 1.3+ ciphersuites.
1534 :return: None
1536 .. versionadded:: 25.2.0
1537 """
1538 if not isinstance(ciphersuites, bytes):
1539 raise TypeError("ciphersuites must be a byte string.")
1541 _openssl_assert(
1542 _lib.SSL_CTX_set_ciphersuites(self._context, ciphersuites) == 1
1543 )
1545 @deprecated(
1546 "Context.set_client_ca_list is deprecated. X509Name support in "
1547 "pyOpenSSL is deprecated."
1548 )
1549 @_require_not_used
1550 def set_client_ca_list(
1551 self, certificate_authorities: Sequence[X509Name]
1552 ) -> None:
1553 """
1554 Set the list of preferred client certificate signers for this server
1555 context.
1557 This list of certificate authorities will be sent to the client when
1558 the server requests a client certificate.
1560 :param certificate_authorities: a sequence of X509Names.
1561 :return: None
1563 .. versionadded:: 0.10
1564 """
1565 name_stack = _lib.sk_X509_NAME_new_null()
1566 _openssl_assert(name_stack != _ffi.NULL)
1568 try:
1569 for ca_name in certificate_authorities:
1570 if not isinstance(ca_name, X509Name):
1571 raise TypeError(
1572 f"client CAs must be X509Name objects, not "
1573 f"{type(ca_name).__name__} objects"
1574 )
1575 copy = _lib.X509_NAME_dup(ca_name._name)
1576 _openssl_assert(copy != _ffi.NULL)
1577 push_result = _lib.sk_X509_NAME_push(name_stack, copy)
1578 if not push_result:
1579 _lib.X509_NAME_free(copy)
1580 _raise_current_error()
1581 except Exception:
1582 _lib.sk_X509_NAME_free(name_stack)
1583 raise
1585 _lib.SSL_CTX_set_client_CA_list(self._context, name_stack)
1587 @_require_not_used
1588 def add_client_ca(
1589 self, certificate_authority: X509 | x509.Certificate
1590 ) -> None:
1591 """
1592 Add the CA certificate to the list of preferred signers for this
1593 context.
1595 The list of certificate authorities will be sent to the client when the
1596 server requests a client certificate.
1598 :param certificate_authority: certificate authority's X509 certificate.
1599 :return: None
1601 .. versionadded:: 0.10
1602 """
1603 if not isinstance(certificate_authority, X509):
1604 certificate_authority = X509.from_cryptography(
1605 certificate_authority
1606 )
1607 else:
1608 warnings.warn(
1609 (
1610 "Passing pyOpenSSL X509 objects is deprecated. You "
1611 "should use a cryptography.x509.Certificate instead."
1612 ),
1613 DeprecationWarning,
1614 stacklevel=2,
1615 )
1617 add_result = _lib.SSL_CTX_add_client_CA(
1618 self._context, certificate_authority._x509
1619 )
1620 _openssl_assert(add_result == 1)
1622 @_require_not_used
1623 def set_timeout(self, timeout: int) -> None:
1624 """
1625 Set the timeout for newly created sessions for this Context object to
1626 *timeout*. The default value is 300 seconds. See the OpenSSL manual
1627 for more information (e.g. :manpage:`SSL_CTX_set_timeout(3)`).
1629 :param timeout: The timeout in (whole) seconds
1630 :return: The previous session timeout
1631 """
1632 if not isinstance(timeout, int):
1633 raise TypeError("timeout must be an integer")
1635 return _lib.SSL_CTX_set_timeout(self._context, timeout)
1637 def get_timeout(self) -> int:
1638 """
1639 Retrieve session timeout, as set by :meth:`set_timeout`. The default
1640 is 300 seconds.
1642 :return: The session timeout
1643 """
1644 return _lib.SSL_CTX_get_timeout(self._context)
1646 @_require_not_used
1647 def set_info_callback(
1648 self, callback: Callable[[Connection, int, int], None]
1649 ) -> None:
1650 """
1651 Set the information callback to *callback*. This function will be
1652 called from time to time during SSL handshakes.
1654 :param callback: The Python callback to use. This should take three
1655 arguments: a Connection object and two integers. The first integer
1656 specifies where in the SSL handshake the function was called, and
1657 the other the return code from a (possibly failed) internal
1658 function call.
1659 :return: None
1660 """
1662 @wraps(callback)
1663 def wrapper(ssl, where, return_code): # type: ignore[no-untyped-def]
1664 callback(Connection._reverse_mapping[ssl], where, return_code)
1666 self._info_callback = _ffi.callback(
1667 "void (*)(const SSL *, int, int)", wrapper
1668 )
1669 _lib.SSL_CTX_set_info_callback(self._context, self._info_callback)
1671 @_requires_keylog
1672 @_require_not_used
1673 def set_keylog_callback(
1674 self, callback: Callable[[Connection, bytes], None]
1675 ) -> None:
1676 """
1677 Set the TLS key logging callback to *callback*. This function will be
1678 called whenever TLS key material is generated or received, in order
1679 to allow applications to store this keying material for debugging
1680 purposes.
1682 :param callback: The Python callback to use. This should take two
1683 arguments: a Connection object and a bytestring that contains
1684 the key material in the format used by NSS for its SSLKEYLOGFILE
1685 debugging output.
1686 :return: None
1687 """
1689 @wraps(callback)
1690 def wrapper(ssl, line): # type: ignore[no-untyped-def]
1691 line = _ffi.string(line)
1692 callback(Connection._reverse_mapping[ssl], line)
1694 self._keylog_callback = _ffi.callback(
1695 "void (*)(const SSL *, const char *)", wrapper
1696 )
1697 _lib.SSL_CTX_set_keylog_callback(self._context, self._keylog_callback)
1699 def get_app_data(self) -> Any:
1700 """
1701 Get the application data (supplied via :meth:`set_app_data()`)
1703 :return: The application data
1704 """
1705 return self._app_data
1707 @_require_not_used
1708 def set_app_data(self, data: Any) -> None:
1709 """
1710 Set the application data (will be returned from get_app_data())
1712 :param data: Any Python object
1713 :return: None
1714 """
1715 self._app_data = data
1717 def get_cert_store(self) -> X509Store | None:
1718 """
1719 Get the certificate store for the context. This can be used to add
1720 "trusted" certificates without using the
1721 :meth:`load_verify_locations` method.
1723 :return: A X509Store object or None if it does not have one.
1724 """
1725 store = _lib.SSL_CTX_get_cert_store(self._context)
1726 if store == _ffi.NULL:
1727 # TODO: This is untested.
1728 return None
1730 pystore = X509Store.__new__(X509Store)
1731 pystore._store = store
1732 return pystore
1734 @_require_not_used
1735 def set_options(self, options: int) -> int:
1736 """
1737 Add options. Options set before are not cleared!
1738 This method should be used with the :const:`OP_*` constants.
1740 :param options: The options to add.
1741 :return: The new option bitmask.
1742 """
1743 if not isinstance(options, int):
1744 raise TypeError("options must be an integer")
1746 return _lib.SSL_CTX_set_options(self._context, options)
1748 @_require_not_used
1749 def set_mode(self, mode: int) -> int:
1750 """
1751 Add modes via bitmask. Modes set before are not cleared! This method
1752 should be used with the :const:`MODE_*` constants.
1754 :param mode: The mode to add.
1755 :return: The new mode bitmask.
1756 """
1757 if not isinstance(mode, int):
1758 raise TypeError("mode must be an integer")
1760 return _lib.SSL_CTX_set_mode(self._context, mode)
1762 @_require_not_used
1763 def clear_mode(self, mode_to_clear: int) -> int:
1764 """
1765 Modes previously set cannot be overwritten without being
1766 cleared first. This method should be used to clear existing modes.
1767 """
1768 return _lib.SSL_CTX_clear_mode(self._context, mode_to_clear)
1770 @_require_not_used
1771 def set_tlsext_servername_callback(
1772 self, callback: Callable[[Connection], None]
1773 ) -> None:
1774 """
1775 Specify a callback function to be called when clients specify a server
1776 name.
1778 :param callback: The callback function. It will be invoked with one
1779 argument, the Connection instance.
1781 .. versionadded:: 0.13
1782 """
1784 @wraps(callback)
1785 def wrapper(ssl, alert, arg): # type: ignore[no-untyped-def]
1786 try:
1787 callback(Connection._reverse_mapping[ssl])
1788 except Exception:
1789 sys.excepthook(*sys.exc_info())
1790 return _lib.SSL_TLSEXT_ERR_ALERT_FATAL
1791 return 0
1793 self._tlsext_servername_callback = _ffi.callback(
1794 "int (*)(SSL *, int *, void *)", wrapper
1795 )
1796 _lib.SSL_CTX_set_tlsext_servername_callback(
1797 self._context, self._tlsext_servername_callback
1798 )
1800 @_require_not_used
1801 def set_tlsext_use_srtp(self, profiles: bytes) -> None:
1802 """
1803 Enable support for negotiating SRTP keying material.
1805 :param bytes profiles: A colon delimited list of protection profile
1806 names, like ``b'SRTP_AES128_CM_SHA1_80:SRTP_AES128_CM_SHA1_32'``.
1807 :return: None
1808 """
1809 if not isinstance(profiles, bytes):
1810 raise TypeError("profiles must be a byte string.")
1812 _openssl_assert(
1813 _lib.SSL_CTX_set_tlsext_use_srtp(self._context, profiles) == 0
1814 )
1816 @_require_not_used
1817 def set_alpn_protos(self, protos: list[bytes]) -> None:
1818 """
1819 Specify the protocols that the client is prepared to speak after the
1820 TLS connection has been negotiated using Application Layer Protocol
1821 Negotiation.
1823 :param protos: A list of the protocols to be offered to the server.
1824 This list should be a Python list of bytestrings representing the
1825 protocols to offer, e.g. ``[b'http/1.1', b'spdy/2']``.
1826 """
1827 # Different versions of OpenSSL are inconsistent about how they handle
1828 # empty proto lists (see #1043), so we avoid the problem entirely by
1829 # rejecting them ourselves.
1830 if not protos:
1831 raise ValueError("at least one protocol must be specified")
1833 # Take the list of protocols and join them together, prefixing them
1834 # with their lengths.
1835 protostr = b"".join(
1836 chain.from_iterable((bytes((len(p),)), p) for p in protos)
1837 )
1839 # Build a C string from the list. We don't need to save this off
1840 # because OpenSSL immediately copies the data out.
1841 input_str = _ffi.new("unsigned char[]", protostr)
1843 # https://www.openssl.org/docs/man1.1.0/man3/SSL_CTX_set_alpn_protos.html:
1844 # SSL_CTX_set_alpn_protos() and SSL_set_alpn_protos()
1845 # return 0 on success, and non-0 on failure.
1846 # WARNING: these functions reverse the return value convention.
1847 _openssl_assert(
1848 _lib.SSL_CTX_set_alpn_protos(
1849 self._context, input_str, len(protostr)
1850 )
1851 == 0
1852 )
1854 @_require_not_used
1855 def set_alpn_select_callback(self, callback: _ALPNSelectCallback) -> None:
1856 """
1857 Specify a callback function that will be called on the server when a
1858 client offers protocols using ALPN.
1860 :param callback: The callback function. It will be invoked with two
1861 arguments: the Connection, and a list of offered protocols as
1862 bytestrings, e.g ``[b'http/1.1', b'spdy/2']``. It can return
1863 one of those bytestrings to indicate the chosen protocol, the
1864 empty bytestring to terminate the TLS connection, or the
1865 :py:obj:`NO_OVERLAPPING_PROTOCOLS` to indicate that no offered
1866 protocol was selected, but that the connection should not be
1867 aborted.
1868 """
1869 self._alpn_select_helper = _ALPNSelectHelper(callback)
1870 self._alpn_select_callback = self._alpn_select_helper.callback
1871 _lib.SSL_CTX_set_alpn_select_cb(
1872 self._context, self._alpn_select_callback, _ffi.NULL
1873 )
1875 def _set_ocsp_callback(
1876 self,
1877 helper: _OCSPClientCallbackHelper | _OCSPServerCallbackHelper,
1878 data: Any | None,
1879 ) -> None:
1880 """
1881 This internal helper does the common work for
1882 ``set_ocsp_server_callback`` and ``set_ocsp_client_callback``, which is
1883 almost all of it.
1884 """
1885 self._ocsp_helper = helper
1886 self._ocsp_callback = helper.callback
1887 if data is None:
1888 self._ocsp_data = _ffi.NULL
1889 else:
1890 self._ocsp_data = _ffi.new_handle(data)
1892 rc = _lib.SSL_CTX_set_tlsext_status_cb(
1893 self._context, self._ocsp_callback
1894 )
1895 _openssl_assert(rc == 1)
1896 rc = _lib.SSL_CTX_set_tlsext_status_arg(self._context, self._ocsp_data)
1897 _openssl_assert(rc == 1)
1899 @_require_not_used
1900 def set_ocsp_server_callback(
1901 self,
1902 callback: _OCSPServerCallback[_T],
1903 data: _T | None = None,
1904 ) -> None:
1905 """
1906 Set a callback to provide OCSP data to be stapled to the TLS handshake
1907 on the server side.
1909 :param callback: The callback function. It will be invoked with two
1910 arguments: the Connection, and the optional arbitrary data you have
1911 provided. The callback must return a bytestring that contains the
1912 OCSP data to staple to the handshake. If no OCSP data is available
1913 for this connection, return the empty bytestring.
1914 :param data: Some opaque data that will be passed into the callback
1915 function when called. This can be used to avoid needing to do
1916 complex data lookups or to keep track of what context is being
1917 used. This parameter is optional.
1918 """
1919 helper = _OCSPServerCallbackHelper(callback)
1920 self._set_ocsp_callback(helper, data)
1922 @_require_not_used
1923 def set_ocsp_client_callback(
1924 self,
1925 callback: _OCSPClientCallback[_T],
1926 data: _T | None = None,
1927 ) -> None:
1928 """
1929 Set a callback to validate OCSP data stapled to the TLS handshake on
1930 the client side.
1932 :param callback: The callback function. It will be invoked with three
1933 arguments: the Connection, a bytestring containing the stapled OCSP
1934 assertion, and the optional arbitrary data you have provided. The
1935 callback must return a boolean that indicates the result of
1936 validating the OCSP data: ``True`` if the OCSP data is valid and
1937 the certificate can be trusted, or ``False`` if either the OCSP
1938 data is invalid or the certificate has been revoked.
1939 :param data: Some opaque data that will be passed into the callback
1940 function when called. This can be used to avoid needing to do
1941 complex data lookups or to keep track of what context is being
1942 used. This parameter is optional.
1943 """
1944 helper = _OCSPClientCallbackHelper(callback)
1945 self._set_ocsp_callback(helper, data)
1947 @_require_not_used
1948 @_requires_ssl_cookie
1949 def set_cookie_generate_callback(
1950 self, callback: _CookieGenerateCallback
1951 ) -> None:
1952 self._cookie_generate_helper = _CookieGenerateCallbackHelper(callback)
1953 _lib.SSL_CTX_set_cookie_generate_cb(
1954 self._context,
1955 self._cookie_generate_helper.callback,
1956 )
1958 @_require_not_used
1959 @_requires_ssl_cookie
1960 def set_cookie_verify_callback(
1961 self, callback: _CookieVerifyCallback
1962 ) -> None:
1963 self._cookie_verify_helper = _CookieVerifyCallbackHelper(callback)
1964 _lib.SSL_CTX_set_cookie_verify_cb(
1965 self._context,
1966 self._cookie_verify_helper.callback,
1967 )
1970class Connection:
1971 _reverse_mapping: typing.MutableMapping[Any, Connection] = (
1972 WeakValueDictionary()
1973 )
1975 def __init__(
1976 self, context: Context, socket: socket.socket | None = None
1977 ) -> None:
1978 """
1979 Create a new Connection object, using the given OpenSSL.SSL.Context
1980 instance and socket.
1982 :param context: An SSL Context to use for this connection
1983 :param socket: The socket to use for transport layer
1984 """
1985 if not isinstance(context, Context):
1986 raise TypeError("context must be a Context instance")
1988 context._used = True
1990 ssl = _lib.SSL_new(context._context)
1991 self._ssl = _ffi.gc(ssl, _lib.SSL_free)
1992 # We set SSL_MODE_AUTO_RETRY to handle situations where OpenSSL returns
1993 # an SSL_ERROR_WANT_READ when processing a non-application data packet
1994 # even though there is still data on the underlying transport.
1995 # See https://github.com/openssl/openssl/issues/6234 for more details.
1996 _lib.SSL_set_mode(self._ssl, _lib.SSL_MODE_AUTO_RETRY)
1997 self._context = context
1998 self._app_data = None
2000 # References to strings used for Application Layer Protocol
2001 # Negotiation. These strings get copied at some point but it's well
2002 # after the callback returns, so we have to hang them somewhere to
2003 # avoid them getting freed.
2004 self._alpn_select_callback_args: Any = None
2006 # Reference the verify_callback of the Context. This ensures that if
2007 # set_verify is called again after the SSL object has been created we
2008 # do not point to a dangling reference
2009 self._verify_helper = context._verify_helper
2010 self._verify_callback = context._verify_callback
2012 # And likewise for the cookie callbacks
2013 self._cookie_generate_helper = context._cookie_generate_helper
2014 self._cookie_verify_helper = context._cookie_verify_helper
2016 self._reverse_mapping[self._ssl] = self
2018 if socket is None:
2019 self._socket = None
2020 # Don't set up any gc for these, SSL_free will take care of them.
2021 self._into_ssl = _lib.BIO_new(_lib.BIO_s_mem())
2022 _openssl_assert(self._into_ssl != _ffi.NULL)
2024 self._from_ssl = _lib.BIO_new(_lib.BIO_s_mem())
2025 _openssl_assert(self._from_ssl != _ffi.NULL)
2027 _lib.SSL_set_bio(self._ssl, self._into_ssl, self._from_ssl)
2028 else:
2029 self._into_ssl = None
2030 self._from_ssl = None
2031 self._socket = socket
2032 set_result = _lib.SSL_set_fd(
2033 self._ssl, _asFileDescriptor(self._socket)
2034 )
2035 _openssl_assert(set_result == 1)
2037 def __getattr__(self, name: str) -> Any:
2038 """
2039 Look up attributes on the wrapped socket object if they are not found
2040 on the Connection object.
2041 """
2042 if self._socket is None:
2043 raise AttributeError(
2044 f"'{self.__class__.__name__}' object has no attribute '{name}'"
2045 )
2046 else:
2047 return getattr(self._socket, name)
2049 def _raise_ssl_error(self, ssl: Any, result: int) -> None:
2050 if self._context._verify_helper is not None:
2051 self._context._verify_helper.raise_if_problem()
2052 if self._context._alpn_select_helper is not None:
2053 self._context._alpn_select_helper.raise_if_problem()
2054 if self._context._ocsp_helper is not None:
2055 self._context._ocsp_helper.raise_if_problem()
2057 error = _lib.SSL_get_error(ssl, result)
2058 if error == _lib.SSL_ERROR_WANT_READ:
2059 raise WantReadError()
2060 elif error == _lib.SSL_ERROR_WANT_WRITE:
2061 raise WantWriteError()
2062 elif error == _lib.SSL_ERROR_ZERO_RETURN:
2063 raise ZeroReturnError()
2064 elif error == _lib.SSL_ERROR_WANT_X509_LOOKUP:
2065 # TODO: This is untested.
2066 raise WantX509LookupError()
2067 elif error == _lib.SSL_ERROR_SYSCALL:
2068 if platform == "win32":
2069 errno = _ffi.getwinerror()[0]
2070 else:
2071 errno = _ffi.errno
2072 if _lib.ERR_peek_error() == 0 or errno != 0:
2073 if result < 0 and errno != 0:
2074 raise SysCallError(errno, errorcode.get(errno))
2075 raise SysCallError(-1, "Unexpected EOF")
2076 else:
2077 # TODO: This is untested, but I think twisted hits it?
2078 _raise_current_error()
2079 elif error == _lib.SSL_ERROR_SSL and _lib.ERR_peek_error() != 0:
2080 # In 3.0.x an unexpected EOF no longer triggers syscall error
2081 # but we want to maintain compatibility so we check here and
2082 # raise syscall if it is an EOF. Since we're not actually sure
2083 # what else could raise SSL_ERROR_SSL we check for the presence
2084 # of the OpenSSL 3 constant SSL_R_UNEXPECTED_EOF_WHILE_READING
2085 # and if it's not present we just raise an error, which matches
2086 # the behavior before we added this elif section
2087 peeked_error = _lib.ERR_peek_error()
2088 reason = _lib.ERR_GET_REASON(peeked_error)
2089 if _lib.Cryptography_HAS_UNEXPECTED_EOF_WHILE_READING:
2090 _openssl_assert(
2091 reason == _lib.SSL_R_UNEXPECTED_EOF_WHILE_READING
2092 )
2093 _lib.ERR_clear_error()
2094 raise SysCallError(-1, "Unexpected EOF")
2095 else:
2096 _raise_current_error()
2097 elif error == _lib.SSL_ERROR_NONE:
2098 pass
2099 else:
2100 _raise_current_error()
2102 def get_context(self) -> Context:
2103 """
2104 Retrieve the :class:`Context` object associated with this
2105 :class:`Connection`.
2106 """
2107 return self._context
2109 def set_context(self, context: Context) -> None:
2110 """
2111 Switch this connection to a new session context.
2113 :param context: A :class:`Context` instance giving the new session
2114 context to use.
2115 """
2116 if not isinstance(context, Context):
2117 raise TypeError("context must be a Context instance")
2119 _lib.SSL_set_SSL_CTX(self._ssl, context._context)
2120 self._context = context
2121 self._context._used = True
2123 def set_options(self, options: int) -> int:
2124 """
2125 Add options. Options set before are not cleared!
2126 This method should be used with the :const:`OP_*` constants.
2128 :param options: The options to add.
2129 :return: The new option bitmask.
2130 """
2131 if not isinstance(options, int):
2132 raise TypeError("options must be an integer")
2134 return _lib.SSL_set_options(self._ssl, options)
2136 def get_servername(self) -> bytes | None:
2137 """
2138 Retrieve the servername extension value if provided in the client hello
2139 message, or None if there wasn't one.
2141 :return: A byte string giving the server name or :data:`None`.
2143 .. versionadded:: 0.13
2144 """
2145 name = _lib.SSL_get_servername(
2146 self._ssl, _lib.TLSEXT_NAMETYPE_host_name
2147 )
2148 if name == _ffi.NULL:
2149 return None
2151 return _ffi.string(name)
2153 def set_verify(
2154 self, mode: int, callback: _VerifyCallback | None = None
2155 ) -> None:
2156 """
2157 Override the Context object's verification flags for this specific
2158 connection. See :py:meth:`Context.set_verify` for details.
2159 """
2160 if not isinstance(mode, int):
2161 raise TypeError("mode must be an integer")
2163 if callback is None:
2164 self._verify_helper = None
2165 self._verify_callback = None
2166 _lib.SSL_set_verify(self._ssl, mode, _ffi.NULL)
2167 else:
2168 if not callable(callback):
2169 raise TypeError("callback must be callable")
2171 self._verify_helper = _VerifyHelper(callback)
2172 self._verify_callback = self._verify_helper.callback
2173 _lib.SSL_set_verify(self._ssl, mode, self._verify_callback)
2175 def get_verify_mode(self) -> int:
2176 """
2177 Retrieve the Connection object's verify mode, as set by
2178 :meth:`set_verify`.
2180 :return: The verify mode
2181 """
2182 return _lib.SSL_get_verify_mode(self._ssl)
2184 def use_certificate(self, cert: X509 | x509.Certificate) -> None:
2185 """
2186 Load a certificate from a X509 object
2188 :param cert: The X509 object
2189 :return: None
2190 """
2191 # Mirrored from Context.use_certificate
2192 if not isinstance(cert, X509):
2193 cert = X509.from_cryptography(cert)
2194 else:
2195 warnings.warn(
2196 (
2197 "Passing pyOpenSSL X509 objects is deprecated. You "
2198 "should use a cryptography.x509.Certificate instead."
2199 ),
2200 DeprecationWarning,
2201 stacklevel=2,
2202 )
2204 use_result = _lib.SSL_use_certificate(self._ssl, cert._x509)
2205 if not use_result:
2206 _raise_current_error()
2208 def use_privatekey(self, pkey: _PrivateKey | PKey) -> None:
2209 """
2210 Load a private key from a PKey object
2212 :param pkey: The PKey object
2213 :return: None
2214 """
2215 # Mirrored from Context.use_privatekey
2216 if not isinstance(pkey, PKey):
2217 pkey = PKey.from_cryptography_key(pkey)
2218 else:
2219 warnings.warn(
2220 (
2221 "Passing pyOpenSSL PKey objects is deprecated. You "
2222 "should use a cryptography private key instead."
2223 ),
2224 DeprecationWarning,
2225 stacklevel=2,
2226 )
2228 use_result = _lib.SSL_use_PrivateKey(self._ssl, pkey._pkey)
2229 if not use_result:
2230 self._context._raise_passphrase_exception()
2232 def set_ciphertext_mtu(self, mtu: int) -> None:
2233 """
2234 For DTLS, set the maximum UDP payload size (*not* including IP/UDP
2235 overhead).
2237 Note that you might have to set :data:`OP_NO_QUERY_MTU` to prevent
2238 OpenSSL from spontaneously clearing this.
2240 :param mtu: An integer giving the maximum transmission unit.
2242 .. versionadded:: 21.1
2243 """
2244 _lib.SSL_set_mtu(self._ssl, mtu)
2246 def get_cleartext_mtu(self) -> int:
2247 """
2248 For DTLS, get the maximum size of unencrypted data you can pass to
2249 :meth:`write` without exceeding the MTU (as passed to
2250 :meth:`set_ciphertext_mtu`).
2252 :return: The effective MTU as an integer.
2254 .. versionadded:: 21.1
2255 """
2257 if not hasattr(_lib, "DTLS_get_data_mtu"):
2258 raise NotImplementedError("requires OpenSSL 1.1.1 or better")
2259 return _lib.DTLS_get_data_mtu(self._ssl)
2261 def set_tlsext_host_name(self, name: bytes) -> None:
2262 """
2263 Set the value of the servername extension to send in the client hello.
2265 :param name: A byte string giving the name.
2267 .. versionadded:: 0.13
2268 """
2269 if not isinstance(name, bytes):
2270 raise TypeError("name must be a byte string")
2271 elif b"\0" in name:
2272 raise TypeError("name must not contain NUL byte")
2274 # XXX I guess this can fail sometimes?
2275 _lib.SSL_set_tlsext_host_name(self._ssl, name)
2277 def pending(self) -> int:
2278 """
2279 Get the number of bytes that can be safely read from the SSL buffer
2280 (**not** the underlying transport buffer).
2282 :return: The number of bytes available in the receive buffer.
2283 """
2284 return _lib.SSL_pending(self._ssl)
2286 def send(self, buf: _Buffer, flags: int = 0) -> int:
2287 """
2288 Send data on the connection. NOTE: If you get one of the WantRead,
2289 WantWrite or WantX509Lookup exceptions on this, you have to call the
2290 method again with the SAME buffer.
2292 :param buf: The string, buffer or memoryview to send
2293 :param flags: (optional) Included for compatibility with the socket
2294 API, the value is ignored
2295 :return: The number of bytes written
2296 """
2297 # Backward compatibility
2298 buf = _text_to_bytes_and_warn("buf", buf)
2300 with _ffi.from_buffer(buf) as data:
2301 # check len(buf) instead of len(data) for testability
2302 if len(buf) > 2147483647:
2303 raise ValueError(
2304 "Cannot send more than 2**31-1 bytes at once."
2305 )
2307 result = _lib.SSL_write(self._ssl, data, len(data))
2308 self._raise_ssl_error(self._ssl, result)
2310 return result
2312 write = send
2314 def sendall(self, buf: _Buffer, flags: int = 0) -> int:
2315 """
2316 Send "all" data on the connection. This calls send() repeatedly until
2317 all data is sent. If an error occurs, it's impossible to tell how much
2318 data has been sent.
2320 :param buf: The string, buffer or memoryview to send
2321 :param flags: (optional) Included for compatibility with the socket
2322 API, the value is ignored
2323 :return: The number of bytes written
2324 """
2325 buf = _text_to_bytes_and_warn("buf", buf)
2327 with _ffi.from_buffer(buf) as data:
2328 left_to_send = len(buf)
2329 total_sent = 0
2331 while left_to_send:
2332 # SSL_write's num arg is an int,
2333 # so we cannot send more than 2**31-1 bytes at once.
2334 result = _lib.SSL_write(
2335 self._ssl, data + total_sent, min(left_to_send, 2147483647)
2336 )
2337 self._raise_ssl_error(self._ssl, result)
2338 total_sent += result
2339 left_to_send -= result
2341 return total_sent
2343 def recv(self, bufsiz: int, flags: int | None = None) -> bytes:
2344 """
2345 Receive data on the connection.
2347 :param bufsiz: The maximum number of bytes to read
2348 :param flags: (optional) The only supported flag is ``MSG_PEEK``,
2349 all other flags are ignored.
2350 :return: The string read from the Connection
2351 """
2352 buf = _no_zero_allocator("char[]", bufsiz)
2353 if flags is not None and flags & socket.MSG_PEEK:
2354 result = _lib.SSL_peek(self._ssl, buf, bufsiz)
2355 else:
2356 result = _lib.SSL_read(self._ssl, buf, bufsiz)
2357 self._raise_ssl_error(self._ssl, result)
2358 return _ffi.buffer(buf, result)[:]
2360 read = recv
2362 def recv_into(
2363 self,
2364 buffer: Any, # collections.abc.Buffer once we use Python 3.12+
2365 nbytes: int | None = None,
2366 flags: int | None = None,
2367 ) -> int:
2368 """
2369 Receive data on the connection and copy it directly into the provided
2370 buffer, rather than creating a new string.
2372 :param buffer: The buffer to copy into.
2373 :param nbytes: (optional) The maximum number of bytes to read into the
2374 buffer. If not present, defaults to the size of the buffer. If
2375 larger than the size of the buffer, is reduced to the size of the
2376 buffer.
2377 :param flags: (optional) The only supported flag is ``MSG_PEEK``,
2378 all other flags are ignored.
2379 :return: The number of bytes read into the buffer.
2380 """
2381 if nbytes is None:
2382 nbytes = len(buffer)
2383 else:
2384 nbytes = min(nbytes, len(buffer))
2386 # We need to create a temporary buffer. This is annoying, it would be
2387 # better if we could pass memoryviews straight into the SSL_read call,
2388 # but right now we can't. Revisit this if CFFI gets that ability.
2389 buf = _no_zero_allocator("char[]", nbytes)
2390 if flags is not None and flags & socket.MSG_PEEK:
2391 result = _lib.SSL_peek(self._ssl, buf, nbytes)
2392 else:
2393 result = _lib.SSL_read(self._ssl, buf, nbytes)
2394 self._raise_ssl_error(self._ssl, result)
2396 # This strange line is all to avoid a memory copy. The buffer protocol
2397 # should allow us to assign a CFFI buffer to the LHS of this line, but
2398 # on CPython 3.3+ that segfaults. As a workaround, we can temporarily
2399 # wrap it in a memoryview.
2400 buffer[:result] = memoryview(_ffi.buffer(buf, result))
2402 return result
2404 def _handle_bio_errors(self, bio: Any, result: int) -> typing.NoReturn:
2405 if _lib.BIO_should_retry(bio):
2406 if _lib.BIO_should_read(bio):
2407 raise WantReadError()
2408 elif _lib.BIO_should_write(bio):
2409 # TODO: This is untested.
2410 raise WantWriteError()
2411 elif _lib.BIO_should_io_special(bio):
2412 # TODO: This is untested. I think io_special means the socket
2413 # BIO has a not-yet connected socket.
2414 raise ValueError("BIO_should_io_special")
2415 else:
2416 # TODO: This is untested.
2417 raise ValueError("unknown bio failure")
2418 else:
2419 # TODO: This is untested.
2420 _raise_current_error()
2422 def bio_read(self, bufsiz: int) -> bytes:
2423 """
2424 If the Connection was created with a memory BIO, this method can be
2425 used to read bytes from the write end of that memory BIO. Many
2426 Connection methods will add bytes which must be read in this manner or
2427 the buffer will eventually fill up and the Connection will be able to
2428 take no further actions.
2430 :param bufsiz: The maximum number of bytes to read
2431 :return: The string read.
2432 """
2433 if self._from_ssl is None:
2434 raise TypeError("Connection sock was not None")
2436 if not isinstance(bufsiz, int):
2437 raise TypeError("bufsiz must be an integer")
2439 buf = _no_zero_allocator("char[]", bufsiz)
2440 result = _lib.BIO_read(self._from_ssl, buf, bufsiz)
2441 if result <= 0:
2442 self._handle_bio_errors(self._from_ssl, result)
2444 return _ffi.buffer(buf, result)[:]
2446 def bio_write(self, buf: _Buffer) -> int:
2447 """
2448 If the Connection was created with a memory BIO, this method can be
2449 used to add bytes to the read end of that memory BIO. The Connection
2450 can then read the bytes (for example, in response to a call to
2451 :meth:`recv`).
2453 :param buf: The string to put into the memory BIO.
2454 :return: The number of bytes written
2455 """
2456 buf = _text_to_bytes_and_warn("buf", buf)
2458 if self._into_ssl is None:
2459 raise TypeError("Connection sock was not None")
2461 with _ffi.from_buffer(buf) as data:
2462 result = _lib.BIO_write(self._into_ssl, data, len(data))
2463 if result <= 0:
2464 self._handle_bio_errors(self._into_ssl, result)
2465 return result
2467 def renegotiate(self) -> bool:
2468 """
2469 Renegotiate the session.
2471 :return: True if the renegotiation can be started, False otherwise
2472 """
2473 if not self.renegotiate_pending():
2474 _openssl_assert(_lib.SSL_renegotiate(self._ssl) == 1)
2475 return True
2476 return False
2478 def do_handshake(self) -> None:
2479 """
2480 Perform an SSL handshake (usually called after :meth:`renegotiate` or
2481 one of :meth:`set_accept_state` or :meth:`set_connect_state`). This can
2482 raise the same exceptions as :meth:`send` and :meth:`recv`.
2484 :return: None.
2485 """
2486 result = _lib.SSL_do_handshake(self._ssl)
2487 self._raise_ssl_error(self._ssl, result)
2489 def renegotiate_pending(self) -> bool:
2490 """
2491 Check if there's a renegotiation in progress, it will return False once
2492 a renegotiation is finished.
2494 :return: Whether there's a renegotiation in progress
2495 """
2496 return _lib.SSL_renegotiate_pending(self._ssl) == 1
2498 def total_renegotiations(self) -> int:
2499 """
2500 Find out the total number of renegotiations.
2502 :return: The number of renegotiations.
2503 """
2504 return _lib.SSL_total_renegotiations(self._ssl)
2506 def connect(self, addr: Any) -> None:
2507 """
2508 Call the :meth:`connect` method of the underlying socket and set up SSL
2509 on the socket, using the :class:`Context` object supplied to this
2510 :class:`Connection` object at creation.
2512 :param addr: A remote address
2513 :return: What the socket's connect method returns
2514 """
2515 _lib.SSL_set_connect_state(self._ssl)
2516 return self._socket.connect(addr) # type: ignore[return-value, union-attr]
2518 def connect_ex(self, addr: Any) -> int:
2519 """
2520 Call the :meth:`connect_ex` method of the underlying socket and set up
2521 SSL on the socket, using the Context object supplied to this Connection
2522 object at creation. Note that if the :meth:`connect_ex` method of the
2523 socket doesn't return 0, SSL won't be initialized.
2525 :param addr: A remove address
2526 :return: What the socket's connect_ex method returns
2527 """
2528 connect_ex = self._socket.connect_ex # type: ignore[union-attr]
2529 self.set_connect_state()
2530 return connect_ex(addr)
2532 def accept(self) -> tuple[Connection, Any]:
2533 """
2534 Call the :meth:`accept` method of the underlying socket and set up SSL
2535 on the returned socket, using the Context object supplied to this
2536 :class:`Connection` object at creation.
2538 :return: A *(conn, addr)* pair where *conn* is the new
2539 :class:`Connection` object created, and *address* is as returned by
2540 the socket's :meth:`accept`.
2541 """
2542 client, addr = self._socket.accept() # type: ignore[union-attr]
2543 conn = Connection(self._context, client)
2544 conn.set_accept_state()
2545 return (conn, addr)
2547 def DTLSv1_listen(self) -> None:
2548 """
2549 Call the OpenSSL function DTLSv1_listen on this connection. See the
2550 OpenSSL manual for more details.
2552 :return: None
2553 """
2554 # Possible future extension: return the BIO_ADDR in some form.
2555 bio_addr = _lib.BIO_ADDR_new()
2556 try:
2557 result = _lib.DTLSv1_listen(self._ssl, bio_addr)
2558 finally:
2559 _lib.BIO_ADDR_free(bio_addr)
2560 # DTLSv1_listen is weird. A zero return value means 'didn't find a
2561 # ClientHello with valid cookie, but keep trying'. So basically
2562 # WantReadError. But it doesn't work correctly with _raise_ssl_error.
2563 # So we raise it manually instead.
2564 if self._cookie_generate_helper is not None:
2565 self._cookie_generate_helper.raise_if_problem()
2566 if self._cookie_verify_helper is not None:
2567 self._cookie_verify_helper.raise_if_problem()
2568 if result == 0:
2569 raise WantReadError()
2570 if result < 0:
2571 self._raise_ssl_error(self._ssl, result)
2573 def DTLSv1_get_timeout(self) -> int | None:
2574 """
2575 Determine when the DTLS SSL object next needs to perform internal
2576 processing due to the passage of time.
2578 When the returned number of seconds have passed, the
2579 :meth:`DTLSv1_handle_timeout` method needs to be called.
2581 :return: The time left in seconds before the next timeout or `None`
2582 if no timeout is currently active.
2583 """
2584 ptv_sec = _ffi.new("time_t *")
2585 ptv_usec = _ffi.new("long *")
2586 if _lib.Cryptography_DTLSv1_get_timeout(self._ssl, ptv_sec, ptv_usec):
2587 return ptv_sec[0] + (ptv_usec[0] / 1000000)
2588 else:
2589 return None
2591 def DTLSv1_handle_timeout(self) -> bool:
2592 """
2593 Handles any timeout events which have become pending on a DTLS SSL
2594 object.
2596 :return: `True` if there was a pending timeout, `False` otherwise.
2597 """
2598 result = _lib.DTLSv1_handle_timeout(self._ssl)
2599 if result < 0:
2600 self._raise_ssl_error(self._ssl, result)
2601 assert False, "unreachable"
2602 else:
2603 return bool(result)
2605 def bio_shutdown(self) -> None:
2606 """
2607 If the Connection was created with a memory BIO, this method can be
2608 used to indicate that *end of file* has been reached on the read end of
2609 that memory BIO.
2611 :return: None
2612 """
2613 if self._from_ssl is None:
2614 raise TypeError("Connection sock was not None")
2616 _lib.BIO_set_mem_eof_return(self._into_ssl, 0)
2618 def shutdown(self) -> bool:
2619 """
2620 Send the shutdown message to the Connection.
2622 :return: True if the shutdown completed successfully (i.e. both sides
2623 have sent closure alerts), False otherwise (in which case you
2624 call :meth:`recv` or :meth:`send` when the connection becomes
2625 readable/writeable).
2626 """
2627 result = _lib.SSL_shutdown(self._ssl)
2628 if result < 0:
2629 self._raise_ssl_error(self._ssl, result)
2630 assert False, "unreachable"
2631 elif result > 0:
2632 return True
2633 else:
2634 return False
2636 def get_cipher_list(self) -> list[str]:
2637 """
2638 Retrieve the list of ciphers used by the Connection object.
2640 :return: A list of native cipher strings.
2641 """
2642 ciphers = []
2643 for i in count():
2644 result = _lib.SSL_get_cipher_list(self._ssl, i)
2645 if result == _ffi.NULL:
2646 break
2647 ciphers.append(_ffi.string(result).decode("utf-8"))
2648 return ciphers
2650 @typing.overload
2651 def get_client_ca_list(
2652 self, *, as_cryptography: typing.Literal[True]
2653 ) -> list[x509.Name]:
2654 pass
2656 @typing.overload
2657 def get_client_ca_list(
2658 self, *, as_cryptography: typing.Literal[False] = False
2659 ) -> list[X509Name]:
2660 pass
2662 def get_client_ca_list(
2663 self,
2664 *,
2665 as_cryptography: typing.Literal[True] | typing.Literal[False] = False,
2666 ) -> list[X509Name] | list[x509.Name]:
2667 """
2668 Get CAs whose certificates are suggested for client authentication.
2670 :param bool as_cryptography: Controls whether a list of
2671 ``cryptography.x509.Name`` or ``OpenSSL.crypto.X509Name``
2672 objects should be returned.
2674 :return: If this is a server connection, the list of certificate
2675 authorities that will be sent or has been sent to the client, as
2676 controlled by this :class:`Connection`'s :class:`Context`.
2678 If this is a client connection, the list will be empty until the
2679 connection with the server is established.
2681 .. versionadded:: 0.10
2682 """
2683 ca_names = _lib.SSL_get_client_CA_list(self._ssl)
2684 if ca_names == _ffi.NULL:
2685 # TODO: This is untested.
2686 return []
2688 if as_cryptography:
2689 names = []
2690 for i in range(_lib.sk_X509_NAME_num(ca_names)):
2691 name = _lib.sk_X509_NAME_value(ca_names, i)
2692 result_buffer = _ffi.new("unsigned char**")
2693 encode_result = _lib.i2d_X509_NAME(name, result_buffer)
2694 _openssl_assert(encode_result >= 0)
2695 der = _ffi.buffer(result_buffer[0], encode_result)[:]
2696 _lib.OPENSSL_free(result_buffer[0])
2698 names.append(x509.Name.from_bytes(der))
2699 return names
2701 result = []
2702 for i in range(_lib.sk_X509_NAME_num(ca_names)):
2703 name = _lib.sk_X509_NAME_value(ca_names, i)
2704 copy = _lib.X509_NAME_dup(name)
2705 _openssl_assert(copy != _ffi.NULL)
2707 # Bypass X509Name.__new__, which warns that X509Name is
2708 # deprecated -- this method is not itself deprecated.
2709 pyname = object.__new__(X509Name)
2710 pyname._name = _ffi.gc(copy, _lib.X509_NAME_free)
2711 result.append(pyname)
2712 return result
2714 def makefile(self, *args: Any, **kwargs: Any) -> typing.NoReturn:
2715 """
2716 The makefile() method is not implemented, since there is no dup
2717 semantics for SSL connections
2719 :raise: NotImplementedError
2720 """
2721 raise NotImplementedError(
2722 "Cannot make file object of OpenSSL.SSL.Connection"
2723 )
2725 def get_app_data(self) -> Any:
2726 """
2727 Retrieve application data as set by :meth:`set_app_data`.
2729 :return: The application data
2730 """
2731 return self._app_data
2733 def set_app_data(self, data: Any) -> None:
2734 """
2735 Set application data
2737 :param data: The application data
2738 :return: None
2739 """
2740 self._app_data = data
2742 def get_shutdown(self) -> int:
2743 """
2744 Get the shutdown state of the Connection.
2746 :return: The shutdown state, a bitvector of SENT_SHUTDOWN,
2747 RECEIVED_SHUTDOWN.
2748 """
2749 return _lib.SSL_get_shutdown(self._ssl)
2751 def set_shutdown(self, state: int) -> None:
2752 """
2753 Set the shutdown state of the Connection.
2755 :param state: bitvector of SENT_SHUTDOWN, RECEIVED_SHUTDOWN.
2756 :return: None
2757 """
2758 if not isinstance(state, int):
2759 raise TypeError("state must be an integer")
2761 _lib.SSL_set_shutdown(self._ssl, state)
2763 def get_state_string(self) -> bytes:
2764 """
2765 Retrieve a verbose string detailing the state of the Connection.
2767 :return: A string representing the state
2768 """
2769 return _ffi.string(_lib.SSL_state_string_long(self._ssl))
2771 def server_random(self) -> bytes | None:
2772 """
2773 Retrieve the random value used with the server hello message.
2775 :return: A string representing the state
2776 """
2777 session = _lib.SSL_get_session(self._ssl)
2778 if session == _ffi.NULL:
2779 return None
2780 length = _lib.SSL_get_server_random(self._ssl, _ffi.NULL, 0)
2781 _openssl_assert(length > 0)
2782 outp = _no_zero_allocator("unsigned char[]", length)
2783 _lib.SSL_get_server_random(self._ssl, outp, length)
2784 return _ffi.buffer(outp, length)[:]
2786 def client_random(self) -> bytes | None:
2787 """
2788 Retrieve the random value used with the client hello message.
2790 :return: A string representing the state
2791 """
2792 session = _lib.SSL_get_session(self._ssl)
2793 if session == _ffi.NULL:
2794 return None
2796 length = _lib.SSL_get_client_random(self._ssl, _ffi.NULL, 0)
2797 _openssl_assert(length > 0)
2798 outp = _no_zero_allocator("unsigned char[]", length)
2799 _lib.SSL_get_client_random(self._ssl, outp, length)
2800 return _ffi.buffer(outp, length)[:]
2802 def master_key(self) -> bytes | None:
2803 """
2804 Retrieve the value of the master key for this session.
2806 :return: A string representing the state
2807 """
2808 session = _lib.SSL_get_session(self._ssl)
2809 if session == _ffi.NULL:
2810 return None
2812 length = _lib.SSL_SESSION_get_master_key(session, _ffi.NULL, 0)
2813 _openssl_assert(length > 0)
2814 outp = _no_zero_allocator("unsigned char[]", length)
2815 _lib.SSL_SESSION_get_master_key(session, outp, length)
2816 return _ffi.buffer(outp, length)[:]
2818 def export_keying_material(
2819 self, label: bytes, olen: int, context: bytes | None = None
2820 ) -> bytes:
2821 """
2822 Obtain keying material for application use.
2824 :param: label - a disambiguating label string as described in RFC 5705
2825 :param: olen - the length of the exported key material in bytes
2826 :param: context - a per-association context value
2827 :return: the exported key material bytes or None
2828 """
2829 outp = _no_zero_allocator("unsigned char[]", olen)
2830 context_buf = _ffi.NULL
2831 context_len = 0
2832 use_context = 0
2833 if context is not None:
2834 context_buf = context
2835 context_len = len(context)
2836 use_context = 1
2837 success = _lib.SSL_export_keying_material(
2838 self._ssl,
2839 outp,
2840 olen,
2841 label,
2842 len(label),
2843 context_buf,
2844 context_len,
2845 use_context,
2846 )
2847 _openssl_assert(success == 1)
2848 return _ffi.buffer(outp, olen)[:]
2850 def sock_shutdown(self, *args: Any, **kwargs: Any) -> None:
2851 """
2852 Call the :meth:`shutdown` method of the underlying socket.
2853 See :manpage:`shutdown(2)`.
2855 :return: What the socket's shutdown() method returns
2856 """
2857 return self._socket.shutdown(*args, **kwargs) # type: ignore[return-value, union-attr]
2859 @typing.overload
2860 def get_certificate(
2861 self, *, as_cryptography: typing.Literal[True]
2862 ) -> x509.Certificate | None:
2863 pass
2865 @typing.overload
2866 def get_certificate(
2867 self, *, as_cryptography: typing.Literal[False] = False
2868 ) -> X509 | None:
2869 pass
2871 def get_certificate(
2872 self,
2873 *,
2874 as_cryptography: typing.Literal[True] | typing.Literal[False] = False,
2875 ) -> X509 | x509.Certificate | None:
2876 """
2877 Retrieve the local certificate (if any)
2879 :param bool as_cryptography: Controls whether a
2880 ``cryptography.x509.Certificate`` or an ``OpenSSL.crypto.X509``
2881 object should be returned.
2883 :return: The local certificate
2884 """
2885 cert = _lib.SSL_get_certificate(self._ssl)
2886 if cert != _ffi.NULL:
2887 _lib.X509_up_ref(cert)
2888 pycert = X509._from_raw_x509_ptr(cert)
2889 if as_cryptography:
2890 return pycert.to_cryptography()
2891 return pycert
2892 return None
2894 @typing.overload
2895 def get_peer_certificate(
2896 self, *, as_cryptography: typing.Literal[True]
2897 ) -> x509.Certificate | None:
2898 pass
2900 @typing.overload
2901 def get_peer_certificate(
2902 self, *, as_cryptography: typing.Literal[False] = False
2903 ) -> X509 | None:
2904 pass
2906 def get_peer_certificate(
2907 self,
2908 *,
2909 as_cryptography: typing.Literal[True] | typing.Literal[False] = False,
2910 ) -> X509 | x509.Certificate | None:
2911 """
2912 Retrieve the other side's certificate (if any)
2914 :param bool as_cryptography: Controls whether a
2915 ``cryptography.x509.Certificate`` or an ``OpenSSL.crypto.X509``
2916 object should be returned.
2918 :return: The peer's certificate
2919 """
2920 cert = _lib.SSL_get_peer_certificate(self._ssl)
2921 if cert != _ffi.NULL:
2922 pycert = X509._from_raw_x509_ptr(cert)
2923 if as_cryptography:
2924 return pycert.to_cryptography()
2925 return pycert
2926 return None
2928 @staticmethod
2929 def _cert_stack_to_list(cert_stack: Any) -> list[X509]:
2930 """
2931 Internal helper to convert a STACK_OF(X509) to a list of X509
2932 instances.
2933 """
2934 result = []
2935 for i in range(_lib.sk_X509_num(cert_stack)):
2936 cert = _lib.sk_X509_value(cert_stack, i)
2937 _openssl_assert(cert != _ffi.NULL)
2938 res = _lib.X509_up_ref(cert)
2939 _openssl_assert(res >= 1)
2940 pycert = X509._from_raw_x509_ptr(cert)
2941 result.append(pycert)
2942 return result
2944 @staticmethod
2945 def _cert_stack_to_cryptography_list(
2946 cert_stack: Any,
2947 ) -> list[x509.Certificate]:
2948 """
2949 Internal helper to convert a STACK_OF(X509) to a list of X509
2950 instances.
2951 """
2952 result = []
2953 for i in range(_lib.sk_X509_num(cert_stack)):
2954 cert = _lib.sk_X509_value(cert_stack, i)
2955 _openssl_assert(cert != _ffi.NULL)
2956 res = _lib.X509_up_ref(cert)
2957 _openssl_assert(res >= 1)
2958 pycert = X509._from_raw_x509_ptr(cert)
2959 result.append(pycert.to_cryptography())
2960 return result
2962 @typing.overload
2963 def get_peer_cert_chain(
2964 self, *, as_cryptography: typing.Literal[True]
2965 ) -> list[x509.Certificate] | None:
2966 pass
2968 @typing.overload
2969 def get_peer_cert_chain(
2970 self, *, as_cryptography: typing.Literal[False] = False
2971 ) -> list[X509] | None:
2972 pass
2974 def get_peer_cert_chain(
2975 self,
2976 *,
2977 as_cryptography: typing.Literal[True] | typing.Literal[False] = False,
2978 ) -> list[X509] | list[x509.Certificate] | None:
2979 """
2980 Retrieve the other side's certificate (if any)
2982 :param bool as_cryptography: Controls whether a list of
2983 ``cryptography.x509.Certificate`` or ``OpenSSL.crypto.X509``
2984 object should be returned.
2986 :return: A list of X509 instances giving the peer's certificate chain,
2987 or None if it does not have one.
2988 """
2989 cert_stack = _lib.SSL_get_peer_cert_chain(self._ssl)
2990 if cert_stack == _ffi.NULL:
2991 return None
2993 if as_cryptography:
2994 return self._cert_stack_to_cryptography_list(cert_stack)
2995 return self._cert_stack_to_list(cert_stack)
2997 @typing.overload
2998 def get_verified_chain(
2999 self, *, as_cryptography: typing.Literal[True]
3000 ) -> list[x509.Certificate] | None:
3001 pass
3003 @typing.overload
3004 def get_verified_chain(
3005 self, *, as_cryptography: typing.Literal[False] = False
3006 ) -> list[X509] | None:
3007 pass
3009 def get_verified_chain(
3010 self,
3011 *,
3012 as_cryptography: typing.Literal[True] | typing.Literal[False] = False,
3013 ) -> list[X509] | list[x509.Certificate] | None:
3014 """
3015 Retrieve the verified certificate chain of the peer including the
3016 peer's end entity certificate. It must be called after a session has
3017 been successfully established. If peer verification was not successful
3018 the chain may be incomplete, invalid, or None.
3020 :param bool as_cryptography: Controls whether a list of
3021 ``cryptography.x509.Certificate`` or ``OpenSSL.crypto.X509``
3022 object should be returned.
3024 :return: A list of X509 instances giving the peer's verified
3025 certificate chain, or None if it does not have one.
3027 .. versionadded:: 20.0
3028 """
3029 # OpenSSL 1.1+
3030 cert_stack = _lib.SSL_get0_verified_chain(self._ssl)
3031 if cert_stack == _ffi.NULL:
3032 return None
3034 if as_cryptography:
3035 return self._cert_stack_to_cryptography_list(cert_stack)
3036 return self._cert_stack_to_list(cert_stack)
3038 def want_read(self) -> bool:
3039 """
3040 Checks if more data has to be read from the transport layer to complete
3041 an operation.
3043 :return: True iff more data has to be read
3044 """
3045 return _lib.SSL_want_read(self._ssl)
3047 def want_write(self) -> bool:
3048 """
3049 Checks if there is data to write to the transport layer to complete an
3050 operation.
3052 :return: True iff there is data to write
3053 """
3054 return _lib.SSL_want_write(self._ssl)
3056 def set_accept_state(self) -> None:
3057 """
3058 Set the connection to work in server mode. The handshake will be
3059 handled automatically by read/write.
3061 :return: None
3062 """
3063 _lib.SSL_set_accept_state(self._ssl)
3065 def set_connect_state(self) -> None:
3066 """
3067 Set the connection to work in client mode. The handshake will be
3068 handled automatically by read/write.
3070 :return: None
3071 """
3072 _lib.SSL_set_connect_state(self._ssl)
3074 def get_session(self) -> Session | None:
3075 """
3076 Returns the Session currently used.
3078 :return: An instance of :class:`OpenSSL.SSL.Session` or
3079 :obj:`None` if no session exists.
3081 .. versionadded:: 0.14
3082 """
3083 session = _lib.SSL_get1_session(self._ssl)
3084 if session == _ffi.NULL:
3085 return None
3087 pysession = Session.__new__(Session)
3088 pysession._session = _ffi.gc(session, _lib.SSL_SESSION_free)
3089 pysession._context = self._context
3090 return pysession
3092 def set_session(self, session: Session) -> None:
3093 """
3094 Set the session to be used when the TLS/SSL connection is established.
3096 The session must have been obtained, via :meth:`get_session`, from a
3097 :class:`Connection` that was using the same :class:`Context` as this
3098 one. OpenSSL requires (but does not verify) that sessions only be
3099 re-used with a compatible ``SSL_CTX``, so this is enforced here.
3101 :param session: A Session instance representing the session to use.
3102 :returns: None
3104 .. versionadded:: 0.14
3105 """
3106 if not isinstance(session, Session):
3107 raise TypeError("session must be a Session instance")
3109 if session._context is not self._context:
3110 raise ValueError(
3111 "session must have been created by a Connection using the "
3112 "same Context as this one"
3113 )
3115 result = _lib.SSL_set_session(self._ssl, session._session)
3116 _openssl_assert(result == 1)
3118 def _get_finished_message(
3119 self, function: Callable[[Any, Any, int], int]
3120 ) -> bytes | None:
3121 """
3122 Helper to implement :meth:`get_finished` and
3123 :meth:`get_peer_finished`.
3125 :param function: Either :data:`SSL_get_finished`: or
3126 :data:`SSL_get_peer_finished`.
3128 :return: :data:`None` if the desired message has not yet been
3129 received, otherwise the contents of the message.
3130 """
3131 # The OpenSSL documentation says nothing about what might happen if the
3132 # count argument given is zero. Specifically, it doesn't say whether
3133 # the output buffer may be NULL in that case or not. Inspection of the
3134 # implementation reveals that it calls memcpy() unconditionally.
3135 # Section 7.1.4, paragraph 1 of the C standard suggests that
3136 # memcpy(NULL, source, 0) is not guaranteed to produce defined (let
3137 # alone desirable) behavior (though it probably does on just about
3138 # every implementation...)
3139 #
3140 # Allocate a tiny buffer to pass in (instead of just passing NULL as
3141 # one might expect) for the initial call so as to be safe against this
3142 # potentially undefined behavior.
3143 empty = _ffi.new("char[]", 0)
3144 size = function(self._ssl, empty, 0)
3145 if size == 0:
3146 # No Finished message so far.
3147 return None
3149 buf = _no_zero_allocator("char[]", size)
3150 function(self._ssl, buf, size)
3151 return _ffi.buffer(buf, size)[:]
3153 def get_finished(self) -> bytes | None:
3154 """
3155 Obtain the latest TLS Finished message that we sent.
3157 :return: The contents of the message or :obj:`None` if the TLS
3158 handshake has not yet completed.
3160 .. versionadded:: 0.15
3161 """
3162 return self._get_finished_message(_lib.SSL_get_finished)
3164 def get_peer_finished(self) -> bytes | None:
3165 """
3166 Obtain the latest TLS Finished message that we received from the peer.
3168 :return: The contents of the message or :obj:`None` if the TLS
3169 handshake has not yet completed.
3171 .. versionadded:: 0.15
3172 """
3173 return self._get_finished_message(_lib.SSL_get_peer_finished)
3175 def get_cipher_name(self) -> str | None:
3176 """
3177 Obtain the name of the currently used cipher.
3179 :returns: The name of the currently used cipher or :obj:`None`
3180 if no connection has been established.
3182 .. versionadded:: 0.15
3183 """
3184 cipher = _lib.SSL_get_current_cipher(self._ssl)
3185 if cipher == _ffi.NULL:
3186 return None
3187 else:
3188 name = _ffi.string(_lib.SSL_CIPHER_get_name(cipher))
3189 return name.decode("utf-8")
3191 def get_cipher_bits(self) -> int | None:
3192 """
3193 Obtain the number of secret bits of the currently used cipher.
3195 :returns: The number of secret bits of the currently used cipher
3196 or :obj:`None` if no connection has been established.
3198 .. versionadded:: 0.15
3199 """
3200 cipher = _lib.SSL_get_current_cipher(self._ssl)
3201 if cipher == _ffi.NULL:
3202 return None
3203 else:
3204 return _lib.SSL_CIPHER_get_bits(cipher, _ffi.NULL)
3206 def get_cipher_version(self) -> str | None:
3207 """
3208 Obtain the protocol version of the currently used cipher.
3210 :returns: The protocol name of the currently used cipher
3211 or :obj:`None` if no connection has been established.
3213 .. versionadded:: 0.15
3214 """
3215 cipher = _lib.SSL_get_current_cipher(self._ssl)
3216 if cipher == _ffi.NULL:
3217 return None
3218 else:
3219 version = _ffi.string(_lib.SSL_CIPHER_get_version(cipher))
3220 return version.decode("utf-8")
3222 def get_protocol_version_name(self) -> str:
3223 """
3224 Retrieve the protocol version of the current connection.
3226 :returns: The TLS version of the current connection, for example
3227 the value for TLS 1.2 would be ``TLSv1.2``or ``Unknown``
3228 for connections that were not successfully established.
3229 """
3230 version = _ffi.string(_lib.SSL_get_version(self._ssl))
3231 return version.decode("utf-8")
3233 def get_protocol_version(self) -> int:
3234 """
3235 Retrieve the SSL or TLS protocol version of the current connection.
3237 :returns: The TLS version of the current connection. For example,
3238 it will return ``0x769`` for connections made over TLS version 1.
3239 """
3240 version = _lib.SSL_version(self._ssl)
3241 return version
3243 def set_alpn_protos(self, protos: list[bytes]) -> None:
3244 """
3245 Specify the client's ALPN protocol list.
3247 These protocols are offered to the server during protocol negotiation.
3249 :param protos: A list of the protocols to be offered to the server.
3250 This list should be a Python list of bytestrings representing the
3251 protocols to offer, e.g. ``[b'http/1.1', b'spdy/2']``.
3252 """
3253 # Different versions of OpenSSL are inconsistent about how they handle
3254 # empty proto lists (see #1043), so we avoid the problem entirely by
3255 # rejecting them ourselves.
3256 if not protos:
3257 raise ValueError("at least one protocol must be specified")
3259 # Take the list of protocols and join them together, prefixing them
3260 # with their lengths.
3261 protostr = b"".join(
3262 chain.from_iterable((bytes((len(p),)), p) for p in protos)
3263 )
3265 # Build a C string from the list. We don't need to save this off
3266 # because OpenSSL immediately copies the data out.
3267 input_str = _ffi.new("unsigned char[]", protostr)
3269 # https://www.openssl.org/docs/man1.1.0/man3/SSL_CTX_set_alpn_protos.html:
3270 # SSL_CTX_set_alpn_protos() and SSL_set_alpn_protos()
3271 # return 0 on success, and non-0 on failure.
3272 # WARNING: these functions reverse the return value convention.
3273 _openssl_assert(
3274 _lib.SSL_set_alpn_protos(self._ssl, input_str, len(protostr)) == 0
3275 )
3277 def get_alpn_proto_negotiated(self) -> bytes:
3278 """
3279 Get the protocol that was negotiated by ALPN.
3281 :returns: A bytestring of the protocol name. If no protocol has been
3282 negotiated yet, returns an empty bytestring.
3283 """
3284 data = _ffi.new("unsigned char **")
3285 data_len = _ffi.new("unsigned int *")
3287 _lib.SSL_get0_alpn_selected(self._ssl, data, data_len)
3289 if not data_len:
3290 return b""
3292 return _ffi.buffer(data[0], data_len[0])[:]
3294 def get_selected_srtp_profile(self) -> bytes:
3295 """
3296 Get the SRTP protocol which was negotiated.
3298 :returns: A bytestring of the SRTP profile name. If no profile has been
3299 negotiated yet, returns an empty bytestring.
3300 """
3301 profile = _lib.SSL_get_selected_srtp_profile(self._ssl)
3302 if not profile:
3303 return b""
3305 return _ffi.string(profile.name)
3307 @_requires_ssl_get0_group_name
3308 def get_group_name(self) -> str | None:
3309 """
3310 Get the name of the negotiated group for the key exchange.
3312 :return: A string giving the group name or :data:`None`.
3313 """
3314 # Do not remove this guard.
3315 # SSL_get0_group_name crashes with a segfault if called without
3316 # an established connection (should return NULL but doesn't).
3317 session = _lib.SSL_get_session(self._ssl)
3318 if session == _ffi.NULL:
3319 return None
3321 group_name = _lib.SSL_get0_group_name(self._ssl)
3322 if group_name == _ffi.NULL:
3323 return None
3325 return _ffi.string(group_name).decode("utf-8")
3327 def request_ocsp(self) -> None:
3328 """
3329 Called to request that the server sends stapled OCSP data, if
3330 available. If this is not called on the client side then the server
3331 will not send OCSP data. Should be used in conjunction with
3332 :meth:`Context.set_ocsp_client_callback`.
3333 """
3334 rc = _lib.SSL_set_tlsext_status_type(
3335 self._ssl, _lib.TLSEXT_STATUSTYPE_ocsp
3336 )
3337 _openssl_assert(rc == 1)
3339 def set_info_callback(
3340 self, callback: Callable[[Connection, int, int], None]
3341 ) -> None:
3342 """
3343 Set the information callback to *callback*. This function will be
3344 called from time to time during SSL handshakes.
3346 :param callback: The Python callback to use. This should take three
3347 arguments: a Connection object and two integers. The first integer
3348 specifies where in the SSL handshake the function was called, and
3349 the other the return code from a (possibly failed) internal
3350 function call.
3351 :return: None
3352 """
3354 @wraps(callback)
3355 def wrapper(ssl, where, return_code): # type: ignore[no-untyped-def]
3356 callback(Connection._reverse_mapping[ssl], where, return_code)
3358 self._info_callback = _ffi.callback(
3359 "void (*)(const SSL *, int, int)", wrapper
3360 )
3361 _lib.SSL_set_info_callback(self._ssl, self._info_callback)