Coverage for /pythoncovmergedfiles/medio/medio/usr/local/lib/python3.11/site-packages/OpenSSL/crypto.py: 34%
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
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 calendar
4import datetime
5import functools
6import sys
7import typing
8from base64 import b16encode
9from collections.abc import Sequence
10from functools import partial
11from typing import (
12 Any,
13 Callable,
14 Union,
15)
17if sys.version_info >= (3, 13):
18 from warnings import deprecated
19else:
20 from typing_extensions import deprecated
22from cryptography import utils, x509
23from cryptography.hazmat.primitives.asymmetric import (
24 dsa,
25 ec,
26 ed448,
27 ed25519,
28 rsa,
29)
31from OpenSSL._util import StrOrBytesPath
32from OpenSSL._util import (
33 byte_string as _byte_string,
34)
35from OpenSSL._util import (
36 exception_from_error_queue as _exception_from_error_queue,
37)
38from OpenSSL._util import (
39 ffi as _ffi,
40)
41from OpenSSL._util import (
42 lib as _lib,
43)
44from OpenSSL._util import (
45 make_assert as _make_assert,
46)
47from OpenSSL._util import (
48 path_bytes as _path_bytes,
49)
51__all__ = [
52 "FILETYPE_ASN1",
53 "FILETYPE_PEM",
54 "FILETYPE_TEXT",
55 "TYPE_DSA",
56 "TYPE_RSA",
57 "X509",
58 "Error",
59 "PKey",
60 "X509Name",
61 "X509Store",
62 "X509StoreContext",
63 "X509StoreContextError",
64 "X509StoreFlags",
65 "dump_certificate",
66 "dump_privatekey",
67 "dump_publickey",
68 "get_elliptic_curve",
69 "get_elliptic_curves",
70 "load_certificate",
71 "load_privatekey",
72 "load_publickey",
73]
76_PrivateKey = Union[
77 dsa.DSAPrivateKey,
78 ec.EllipticCurvePrivateKey,
79 ed25519.Ed25519PrivateKey,
80 ed448.Ed448PrivateKey,
81 rsa.RSAPrivateKey,
82]
83_PublicKey = Union[
84 dsa.DSAPublicKey,
85 ec.EllipticCurvePublicKey,
86 ed25519.Ed25519PublicKey,
87 ed448.Ed448PublicKey,
88 rsa.RSAPublicKey,
89]
90_Key = Union[_PrivateKey, _PublicKey]
91PassphraseCallableT = Union[bytes, Callable[..., bytes]]
94FILETYPE_PEM: int = _lib.SSL_FILETYPE_PEM
95FILETYPE_ASN1: int = _lib.SSL_FILETYPE_ASN1
97# TODO This was an API mistake. OpenSSL has no such constant.
98FILETYPE_TEXT = 2**16 - 1
100TYPE_RSA: int = _lib.EVP_PKEY_RSA
101TYPE_DSA: int = _lib.EVP_PKEY_DSA
102TYPE_DH: int = _lib.EVP_PKEY_DH
103TYPE_EC: int = _lib.EVP_PKEY_EC
106class Error(Exception):
107 """
108 An error occurred in an `OpenSSL.crypto` API.
109 """
112_raise_current_error = partial(_exception_from_error_queue, Error)
113_openssl_assert = _make_assert(Error)
116def _new_mem_buf(buffer: bytes | None = None) -> Any:
117 """
118 Allocate a new OpenSSL memory BIO.
120 Arrange for the garbage collector to clean it up automatically.
122 :param buffer: None or some bytes to use to put into the BIO so that they
123 can be read out.
124 """
125 if buffer is None:
126 bio = _lib.BIO_new(_lib.BIO_s_mem())
127 free = _lib.BIO_free
128 else:
129 data = _ffi.new("char[]", buffer)
130 bio = _lib.BIO_new_mem_buf(data, len(buffer))
132 # Keep the memory alive as long as the bio is alive!
133 def free(bio: Any, ref: Any = data) -> Any:
134 return _lib.BIO_free(bio)
136 _openssl_assert(bio != _ffi.NULL)
138 bio = _ffi.gc(bio, free)
139 return bio
142def _bio_to_string(bio: Any) -> bytes:
143 """
144 Copy the contents of an OpenSSL BIO object into a Python byte string.
145 """
146 result_buffer = _ffi.new("char**")
147 buffer_length = _lib.BIO_get_mem_data(bio, result_buffer)
148 return _ffi.buffer(result_buffer[0], buffer_length)[:]
151def _set_asn1_time(boundary: Any, when: bytes) -> None:
152 """
153 The the time value of an ASN1 time object.
155 @param boundary: An ASN1_TIME pointer (or an object safely
156 castable to that type) which will have its value set.
157 @param when: A string representation of the desired time value.
159 @raise TypeError: If C{when} is not a L{bytes} string.
160 @raise ValueError: If C{when} does not represent a time in the required
161 format.
162 @raise RuntimeError: If the time value cannot be set for some other
163 (unspecified) reason.
164 """
165 if not isinstance(when, bytes):
166 raise TypeError("when must be a byte string")
167 # ASN1_TIME_set_string validates the string without writing anything
168 # when the destination is NULL.
169 _openssl_assert(boundary != _ffi.NULL)
171 set_result = _lib.ASN1_TIME_set_string(boundary, when)
172 if set_result == 0:
173 raise ValueError("Invalid string")
176def _new_asn1_time(when: bytes) -> Any:
177 """
178 Behaves like _set_asn1_time but returns a new ASN1_TIME object.
180 @param when: A string representation of the desired time value.
182 @raise TypeError: If C{when} is not a L{bytes} string.
183 @raise ValueError: If C{when} does not represent a time in the required
184 format.
185 @raise RuntimeError: If the time value cannot be set for some other
186 (unspecified) reason.
187 """
188 ret = _lib.ASN1_TIME_new()
189 _openssl_assert(ret != _ffi.NULL)
190 ret = _ffi.gc(ret, _lib.ASN1_TIME_free)
191 _set_asn1_time(ret, when)
192 return ret
195def _get_asn1_time(timestamp: Any) -> bytes | None:
196 """
197 Retrieve the time value of an ASN1 time object.
199 @param timestamp: An ASN1_GENERALIZEDTIME* (or an object safely castable to
200 that type) from which the time value will be retrieved.
202 @return: The time value from C{timestamp} as a L{bytes} string in a certain
203 format. Or C{None} if the object contains no time value.
204 """
205 string_timestamp = _ffi.cast("ASN1_STRING*", timestamp)
206 if _lib.ASN1_STRING_length(string_timestamp) == 0:
207 return None
208 elif (
209 _lib.ASN1_STRING_type(string_timestamp) == _lib.V_ASN1_GENERALIZEDTIME
210 ):
211 return _ffi.string(_lib.ASN1_STRING_get0_data(string_timestamp))
212 else:
213 generalized_timestamp = _ffi.new("ASN1_GENERALIZEDTIME**")
214 _lib.ASN1_TIME_to_generalizedtime(timestamp, generalized_timestamp)
215 _openssl_assert(generalized_timestamp[0] != _ffi.NULL)
217 string_timestamp = _ffi.cast("ASN1_STRING*", generalized_timestamp[0])
218 string_data = _lib.ASN1_STRING_get0_data(string_timestamp)
219 string_result = _ffi.string(string_data)
220 _lib.ASN1_GENERALIZEDTIME_free(generalized_timestamp[0])
221 return string_result
224class _X509NameInvalidator:
225 def __init__(self) -> None:
226 self._names: list[X509Name] = []
228 def add(self, name: X509Name) -> None:
229 self._names.append(name)
231 def clear(self) -> None:
232 for name in self._names:
233 # Breaks the object, but also prevents UAF!
234 del name._name
237class PKey:
238 """
239 A class representing an DSA or RSA public key or key pair.
240 """
242 _only_public = False
243 _initialized = True
245 def __init__(self) -> None:
246 pkey = _lib.EVP_PKEY_new()
247 self._pkey = _ffi.gc(pkey, _lib.EVP_PKEY_free)
248 self._initialized = False
250 def to_cryptography_key(self) -> _Key:
251 """
252 Export as a ``cryptography`` key.
254 :rtype: One of ``cryptography``'s `key interfaces`_.
256 .. _key interfaces: https://cryptography.io/en/latest/hazmat/\
257 primitives/asymmetric/rsa/#key-interfaces
259 .. versionadded:: 16.1.0
260 """
261 from cryptography.hazmat.primitives.serialization import (
262 load_der_private_key,
263 load_der_public_key,
264 )
266 if self._only_public:
267 der = dump_publickey(FILETYPE_ASN1, self)
268 return typing.cast(_Key, load_der_public_key(der))
269 else:
270 der = _dump_privatekey_internal(FILETYPE_ASN1, self)
271 return typing.cast(_Key, load_der_private_key(der, password=None))
273 @classmethod
274 def from_cryptography_key(cls, crypto_key: _Key) -> PKey:
275 """
276 Construct based on a ``cryptography`` *crypto_key*.
278 :param crypto_key: A ``cryptography`` key.
279 :type crypto_key: One of ``cryptography``'s `key interfaces`_.
281 :rtype: PKey
283 .. versionadded:: 16.1.0
284 """
285 if not isinstance(
286 crypto_key,
287 (
288 dsa.DSAPrivateKey,
289 dsa.DSAPublicKey,
290 ec.EllipticCurvePrivateKey,
291 ec.EllipticCurvePublicKey,
292 ed25519.Ed25519PrivateKey,
293 ed25519.Ed25519PublicKey,
294 ed448.Ed448PrivateKey,
295 ed448.Ed448PublicKey,
296 rsa.RSAPrivateKey,
297 rsa.RSAPublicKey,
298 ),
299 ):
300 raise TypeError("Unsupported key type")
302 from cryptography.hazmat.primitives.serialization import (
303 Encoding,
304 NoEncryption,
305 PrivateFormat,
306 PublicFormat,
307 )
309 if isinstance(
310 crypto_key,
311 (
312 dsa.DSAPublicKey,
313 ec.EllipticCurvePublicKey,
314 ed25519.Ed25519PublicKey,
315 ed448.Ed448PublicKey,
316 rsa.RSAPublicKey,
317 ),
318 ):
319 return load_publickey(
320 FILETYPE_ASN1,
321 crypto_key.public_bytes(
322 Encoding.DER, PublicFormat.SubjectPublicKeyInfo
323 ),
324 )
325 else:
326 der = crypto_key.private_bytes(
327 Encoding.DER, PrivateFormat.PKCS8, NoEncryption()
328 )
329 return load_privatekey(FILETYPE_ASN1, der)
331 @deprecated(
332 "PKey.generate_key is deprecated. You should use the key "
333 "generation APIs in cryptography instead."
334 )
335 def generate_key(self, type: int, bits: int) -> None:
336 """
337 Generate a key pair of the given type, with the given number of bits.
339 This generates a key "into" the this object.
341 :param type: The key type.
342 :type type: :py:data:`TYPE_RSA` or :py:data:`TYPE_DSA`
343 :param bits: The number of bits.
344 :type bits: :py:data:`int` ``>= 0``
345 :raises TypeError: If :py:data:`type` or :py:data:`bits` isn't
346 of the appropriate type.
347 :raises ValueError: If the number of bits isn't an integer of
348 the appropriate size.
349 :return: ``None``
350 """
351 if not isinstance(type, int):
352 raise TypeError("type must be an integer")
354 if not isinstance(bits, int):
355 raise TypeError("bits must be an integer")
357 if type == TYPE_RSA:
358 if bits <= 0:
359 raise ValueError("Invalid number of bits")
361 # TODO Check error return
362 exponent = _lib.BN_new()
363 exponent = _ffi.gc(exponent, _lib.BN_free)
364 _lib.BN_set_word(exponent, _lib.RSA_F4)
366 rsa = _lib.RSA_new()
368 result = _lib.RSA_generate_key_ex(rsa, bits, exponent, _ffi.NULL)
369 _openssl_assert(result == 1)
371 result = _lib.EVP_PKEY_assign_RSA(self._pkey, rsa)
372 _openssl_assert(result == 1)
374 elif type == TYPE_DSA:
375 dsa = _lib.DSA_new()
376 _openssl_assert(dsa != _ffi.NULL)
378 dsa = _ffi.gc(dsa, _lib.DSA_free)
379 res = _lib.DSA_generate_parameters_ex(
380 dsa, bits, _ffi.NULL, 0, _ffi.NULL, _ffi.NULL, _ffi.NULL
381 )
382 _openssl_assert(res == 1)
384 _openssl_assert(_lib.DSA_generate_key(dsa) == 1)
385 _openssl_assert(_lib.EVP_PKEY_set1_DSA(self._pkey, dsa) == 1)
386 else:
387 raise Error("No such key type")
389 self._initialized = True
391 @deprecated(
392 "PKey.check is deprecated. You should use the APIs in "
393 "cryptography instead."
394 )
395 def check(self) -> bool:
396 """
397 Check the consistency of an RSA private key.
399 This is the Python equivalent of OpenSSL's ``RSA_check_key``.
401 :return: ``True`` if key is consistent.
403 :raise OpenSSL.crypto.Error: if the key is inconsistent.
405 :raise TypeError: if the key is of a type which cannot be checked.
406 Only RSA keys can currently be checked.
407 """
408 if self._only_public:
409 raise TypeError("public key only")
411 if _lib.EVP_PKEY_type(self.type()) != _lib.EVP_PKEY_RSA:
412 raise TypeError("Only RSA keys can currently be checked.")
414 rsa = _lib.EVP_PKEY_get1_RSA(self._pkey)
415 rsa = _ffi.gc(rsa, _lib.RSA_free)
416 result = _lib.RSA_check_key(rsa)
417 if result == 1:
418 return True
419 _raise_current_error()
421 def type(self) -> int:
422 """
423 Returns the type of the key
425 :return: The type of the key.
426 """
427 return _lib.EVP_PKEY_id(self._pkey)
429 def bits(self) -> int:
430 """
431 Returns the number of bits of the key
433 :return: The number of bits of the key.
434 """
435 return _lib.EVP_PKEY_bits(self._pkey)
438class _EllipticCurve:
439 """
440 A representation of a supported elliptic curve.
442 @cvar _curves: :py:obj:`None` until an attempt is made to load the curves.
443 Thereafter, a :py:type:`set` containing :py:type:`_EllipticCurve`
444 instances each of which represents one curve supported by the system.
445 @type _curves: :py:type:`NoneType` or :py:type:`set`
446 """
448 _curves = None
450 def __ne__(self, other: Any) -> bool:
451 """
452 Implement cooperation with the right-hand side argument of ``!=``.
454 Python 3 seems to have dropped this cooperation in this very narrow
455 circumstance.
456 """
457 if isinstance(other, _EllipticCurve):
458 return super().__ne__(other)
459 return NotImplemented
461 @classmethod
462 def _load_elliptic_curves(cls, lib: Any) -> set[_EllipticCurve]:
463 """
464 Get the curves supported by OpenSSL.
466 :param lib: The OpenSSL library binding object.
468 :return: A :py:type:`set` of ``cls`` instances giving the names of the
469 elliptic curves the underlying library supports.
470 """
471 num_curves = lib.EC_get_builtin_curves(_ffi.NULL, 0)
472 builtin_curves = _ffi.new("EC_builtin_curve[]", num_curves)
473 # The return value on this call should be num_curves again. We
474 # could check it to make sure but if it *isn't* then.. what could
475 # we do? Abort the whole process, I suppose...? -exarkun
476 lib.EC_get_builtin_curves(builtin_curves, num_curves)
477 return set(cls.from_nid(lib, c.nid) for c in builtin_curves)
479 @classmethod
480 def _get_elliptic_curves(cls, lib: Any) -> set[_EllipticCurve]:
481 """
482 Get, cache, and return the curves supported by OpenSSL.
484 :param lib: The OpenSSL library binding object.
486 :return: A :py:type:`set` of ``cls`` instances giving the names of the
487 elliptic curves the underlying library supports.
488 """
489 if cls._curves is None:
490 cls._curves = cls._load_elliptic_curves(lib)
491 return cls._curves
493 @classmethod
494 def from_nid(cls, lib: Any, nid: int) -> _EllipticCurve:
495 """
496 Instantiate a new :py:class:`_EllipticCurve` associated with the given
497 OpenSSL NID.
499 :param lib: The OpenSSL library binding object.
501 :param nid: The OpenSSL NID the resulting curve object will represent.
502 This must be a curve NID (and not, for example, a hash NID) or
503 subsequent operations will fail in unpredictable ways.
504 :type nid: :py:class:`int`
506 :return: The curve object.
507 """
508 return cls(lib, nid, _ffi.string(lib.OBJ_nid2sn(nid)).decode("ascii"))
510 def __init__(self, lib: Any, nid: int, name: str) -> None:
511 """
512 :param _lib: The :py:mod:`cryptography` binding instance used to
513 interface with OpenSSL.
515 :param _nid: The OpenSSL NID identifying the curve this object
516 represents.
517 :type _nid: :py:class:`int`
519 :param name: The OpenSSL short name identifying the curve this object
520 represents.
521 :type name: :py:class:`unicode`
522 """
523 self._lib = lib
524 self._nid = nid
525 self.name = name
527 def __repr__(self) -> str:
528 return f"<Curve {self.name!r}>"
530 def _to_EC_KEY(self) -> Any:
531 """
532 Create a new OpenSSL EC_KEY structure initialized to use this curve.
534 The structure is automatically garbage collected when the Python object
535 is garbage collected.
536 """
537 key = self._lib.EC_KEY_new_by_curve_name(self._nid)
538 return _ffi.gc(key, _lib.EC_KEY_free)
541@deprecated(
542 "get_elliptic_curves is deprecated. You should use the APIs in "
543 "cryptography instead."
544)
545def get_elliptic_curves() -> set[_EllipticCurve]:
546 """
547 Return a set of objects representing the elliptic curves supported in the
548 OpenSSL build in use.
550 The curve objects have a :py:class:`unicode` ``name`` attribute by which
551 they identify themselves.
553 The curve objects are useful as values for the argument accepted by
554 :py:meth:`Context.set_tmp_ecdh` to specify which elliptical curve should be
555 used for ECDHE key exchange.
556 """
557 return _EllipticCurve._get_elliptic_curves(_lib)
560@deprecated(
561 "get_elliptic_curve is deprecated. You should use the APIs in "
562 "cryptography instead."
563)
564def get_elliptic_curve(name: str) -> _EllipticCurve:
565 """
566 Return a single curve object selected by name.
568 See :py:func:`get_elliptic_curves` for information about curve objects.
570 :param name: The OpenSSL short name identifying the curve object to
571 retrieve.
572 :type name: :py:class:`unicode`
574 If the named curve is not supported then :py:class:`ValueError` is raised.
575 """
576 for curve in get_elliptic_curves():
577 if curve.name == name:
578 return curve
579 raise ValueError("unknown curve name", name)
582@deprecated(
583 "X509Name support in pyOpenSSL is deprecated. You should use the "
584 "APIs in cryptography."
585)
586@functools.total_ordering
587class X509Name:
588 """
589 An X.509 Distinguished Name.
591 :ivar countryName: The country of the entity.
592 :ivar C: Alias for :py:attr:`countryName`.
594 :ivar stateOrProvinceName: The state or province of the entity.
595 :ivar ST: Alias for :py:attr:`stateOrProvinceName`.
597 :ivar localityName: The locality of the entity.
598 :ivar L: Alias for :py:attr:`localityName`.
600 :ivar organizationName: The organization name of the entity.
601 :ivar O: Alias for :py:attr:`organizationName`.
603 :ivar organizationalUnitName: The organizational unit of the entity.
604 :ivar OU: Alias for :py:attr:`organizationalUnitName`
606 :ivar commonName: The common name of the entity.
607 :ivar CN: Alias for :py:attr:`commonName`.
609 :ivar emailAddress: The e-mail address of the entity.
610 """
612 def __init__(self, name: X509Name) -> None:
613 """
614 Create a new X509Name, copying the given X509Name instance.
616 :param name: The name to copy.
617 :type name: :py:class:`X509Name`
618 """
619 name = _lib.X509_NAME_dup(name._name)
620 self._name: Any = _ffi.gc(name, _lib.X509_NAME_free)
622 def __setattr__(self, name: str, value: Any) -> None:
623 if name.startswith("_"):
624 return super().__setattr__(name, value)
626 # Note: we really do not want str subclasses here, so we do not use
627 # isinstance.
628 if type(name) is not str:
629 raise TypeError(
630 f"attribute name must be string, not "
631 f"'{type(value).__name__:.200}'"
632 )
634 nid = _lib.OBJ_txt2nid(_byte_string(name))
635 if nid == _lib.NID_undef:
636 try:
637 _raise_current_error()
638 except Error:
639 pass
640 raise AttributeError("No such attribute")
642 # If there's an old entry for this NID, remove it
643 for i in range(_lib.X509_NAME_entry_count(self._name)):
644 ent = _lib.X509_NAME_get_entry(self._name, i)
645 ent_obj = _lib.X509_NAME_ENTRY_get_object(ent)
646 ent_nid = _lib.OBJ_obj2nid(ent_obj)
647 if nid == ent_nid:
648 ent = _lib.X509_NAME_delete_entry(self._name, i)
649 _lib.X509_NAME_ENTRY_free(ent)
650 break
652 if isinstance(value, str):
653 value = value.encode("utf-8")
655 add_result = _lib.X509_NAME_add_entry_by_NID(
656 self._name, nid, _lib.MBSTRING_UTF8, value, len(value), -1, 0
657 )
658 if not add_result:
659 _raise_current_error()
661 def __getattr__(self, name: str) -> str | None:
662 """
663 Find attribute. An X509Name object has the following attributes:
664 countryName (alias C), stateOrProvince (alias ST), locality (alias L),
665 organization (alias O), organizationalUnit (alias OU), commonName
666 (alias CN) and more...
667 """
668 nid = _lib.OBJ_txt2nid(_byte_string(name))
669 if nid == _lib.NID_undef:
670 # This is a bit weird. OBJ_txt2nid indicated failure, but it seems
671 # a lower level function, a2d_ASN1_OBJECT, also feels the need to
672 # push something onto the error queue. If we don't clean that up
673 # now, someone else will bump into it later and be quite confused.
674 # See lp#314814.
675 try:
676 _raise_current_error()
677 except Error:
678 pass
679 raise AttributeError("No such attribute")
681 entry_index = _lib.X509_NAME_get_index_by_NID(self._name, nid, -1)
682 if entry_index == -1:
683 return None
685 entry = _lib.X509_NAME_get_entry(self._name, entry_index)
686 data = _lib.X509_NAME_ENTRY_get_data(entry)
688 result_buffer = _ffi.new("unsigned char**")
689 data_length = _lib.ASN1_STRING_to_UTF8(result_buffer, data)
690 _openssl_assert(data_length >= 0)
692 try:
693 result = _ffi.buffer(result_buffer[0], data_length)[:].decode(
694 "utf-8"
695 )
696 finally:
697 # XXX untested
698 _lib.OPENSSL_free(result_buffer[0])
699 return result
701 def __eq__(self, other: Any) -> bool:
702 if not isinstance(other, X509Name):
703 return NotImplemented
705 return _lib.X509_NAME_cmp(self._name, other._name) == 0
707 def __lt__(self, other: Any) -> bool:
708 if not isinstance(other, X509Name):
709 return NotImplemented
711 return _lib.X509_NAME_cmp(self._name, other._name) < 0
713 def __repr__(self) -> str:
714 """
715 String representation of an X509Name
716 """
717 result_buffer = _ffi.new("char[]", 512)
718 format_result = _lib.X509_NAME_oneline(
719 self._name, result_buffer, len(result_buffer)
720 )
721 _openssl_assert(format_result != _ffi.NULL)
723 return "<X509Name object '{}'>".format(
724 _ffi.string(result_buffer).decode("utf-8"),
725 )
727 def hash(self) -> int:
728 """
729 Return an integer representation of the first four bytes of the
730 MD5 digest of the DER representation of the name.
732 This is the Python equivalent of OpenSSL's ``X509_NAME_hash``.
734 :return: The (integer) hash of this name.
735 :rtype: :py:class:`int`
736 """
737 return _lib.X509_NAME_hash(self._name)
739 def der(self) -> bytes:
740 """
741 Return the DER encoding of this name.
743 :return: The DER encoded form of this name.
744 :rtype: :py:class:`bytes`
745 """
746 result_buffer = _ffi.new("unsigned char**")
747 encode_result = _lib.i2d_X509_NAME(self._name, result_buffer)
748 _openssl_assert(encode_result >= 0)
750 string_result = _ffi.buffer(result_buffer[0], encode_result)[:]
751 _lib.OPENSSL_free(result_buffer[0])
752 return string_result
754 def get_components(self) -> list[tuple[bytes, bytes]]:
755 """
756 Returns the components of this name, as a sequence of 2-tuples.
758 :return: The components of this name.
759 :rtype: :py:class:`list` of ``name, value`` tuples.
760 """
761 result = []
762 for i in range(_lib.X509_NAME_entry_count(self._name)):
763 ent = _lib.X509_NAME_get_entry(self._name, i)
765 fname = _lib.X509_NAME_ENTRY_get_object(ent)
766 fval = _lib.X509_NAME_ENTRY_get_data(ent)
768 nid = _lib.OBJ_obj2nid(fname)
769 name = _lib.OBJ_nid2sn(nid)
771 # ffi.string does not handle strings containing NULL bytes
772 # (which may have been generated by old, broken software)
773 value = _ffi.buffer(
774 _lib.ASN1_STRING_get0_data(fval), _lib.ASN1_STRING_length(fval)
775 )[:]
776 result.append((_ffi.string(name), value))
778 return result
781class X509:
782 """
783 An X.509 certificate.
784 """
786 def __init__(self) -> None:
787 x509 = _lib.X509_new()
788 _openssl_assert(x509 != _ffi.NULL)
789 self._x509 = _ffi.gc(x509, _lib.X509_free)
791 self._issuer_invalidator = _X509NameInvalidator()
792 self._subject_invalidator = _X509NameInvalidator()
794 @classmethod
795 def _from_raw_x509_ptr(cls, x509: Any) -> X509:
796 cert = cls.__new__(cls)
797 cert._x509 = _ffi.gc(x509, _lib.X509_free)
798 cert._issuer_invalidator = _X509NameInvalidator()
799 cert._subject_invalidator = _X509NameInvalidator()
800 return cert
802 def to_cryptography(self) -> x509.Certificate:
803 """
804 Export as a ``cryptography`` certificate.
806 :rtype: ``cryptography.x509.Certificate``
808 .. versionadded:: 17.1.0
809 """
810 from cryptography.x509 import load_der_x509_certificate
812 der = dump_certificate(FILETYPE_ASN1, self)
813 return load_der_x509_certificate(der)
815 @classmethod
816 def from_cryptography(cls, crypto_cert: x509.Certificate) -> X509:
817 """
818 Construct based on a ``cryptography`` *crypto_cert*.
820 :param crypto_key: A ``cryptography`` X.509 certificate.
821 :type crypto_key: ``cryptography.x509.Certificate``
823 :rtype: X509
825 .. versionadded:: 17.1.0
826 """
827 if not isinstance(crypto_cert, x509.Certificate):
828 raise TypeError("Must be a certificate")
830 from cryptography.hazmat.primitives.serialization import Encoding
832 der = crypto_cert.public_bytes(Encoding.DER)
833 return load_certificate(FILETYPE_ASN1, der)
835 @deprecated(
836 "X509.set_version is deprecated. You should use "
837 "cryptography's CertificateBuilder instead."
838 )
839 def set_version(self, version: int) -> None:
840 """
841 Set the version number of the certificate. Note that the
842 version value is zero-based, eg. a value of 0 is V1.
844 :param version: The version number of the certificate.
845 :type version: :py:class:`int`
847 :return: ``None``
848 """
849 if not isinstance(version, int):
850 raise TypeError("version must be an integer")
852 _openssl_assert(_lib.X509_set_version(self._x509, version) == 1)
854 def get_version(self) -> int:
855 """
856 Return the version number of the certificate.
858 :return: The version number of the certificate.
859 :rtype: :py:class:`int`
860 """
861 return _lib.X509_get_version(self._x509)
863 def get_pubkey(self) -> PKey:
864 """
865 Get the public key of the certificate.
867 :return: The public key.
868 :rtype: :py:class:`PKey`
869 """
870 pkey = PKey.__new__(PKey)
871 pkey._pkey = _lib.X509_get_pubkey(self._x509)
872 if pkey._pkey == _ffi.NULL:
873 _raise_current_error()
874 pkey._pkey = _ffi.gc(pkey._pkey, _lib.EVP_PKEY_free)
875 pkey._only_public = True
876 return pkey
878 @deprecated(
879 "X509.set_pubkey is deprecated. You should use "
880 "cryptography's CertificateBuilder instead."
881 )
882 def set_pubkey(self, pkey: PKey) -> None:
883 """
884 Set the public key of the certificate.
886 :param pkey: The public key.
887 :type pkey: :py:class:`PKey`
889 :return: :py:data:`None`
890 """
891 if not isinstance(pkey, PKey):
892 raise TypeError("pkey must be a PKey instance")
894 set_result = _lib.X509_set_pubkey(self._x509, pkey._pkey)
895 _openssl_assert(set_result == 1)
897 @deprecated(
898 "X509.sign is deprecated. You should use "
899 "cryptography's CertificateBuilder instead."
900 )
901 def sign(self, pkey: PKey, digest: str) -> None:
902 """
903 Sign the certificate with this key and digest type.
905 :param pkey: The key to sign with.
906 :type pkey: :py:class:`PKey`
908 :param digest: The name of the message digest to use.
909 :type digest: :py:class:`str`
911 :return: :py:data:`None`
912 """
913 if not isinstance(pkey, PKey):
914 raise TypeError("pkey must be a PKey instance")
916 if pkey._only_public:
917 raise ValueError("Key only has public part")
919 if not pkey._initialized:
920 raise ValueError("Key is uninitialized")
922 evp_md = _lib.EVP_get_digestbyname(_byte_string(digest))
923 if evp_md == _ffi.NULL:
924 raise ValueError("No such digest method")
926 sign_result = _lib.X509_sign(self._x509, pkey._pkey, evp_md)
927 _openssl_assert(sign_result > 0)
929 def get_signature_algorithm(self) -> bytes:
930 """
931 Return the signature algorithm used in the certificate.
933 :return: The name of the algorithm.
934 :rtype: :py:class:`bytes`
936 :raises ValueError: If the signature algorithm is undefined.
938 .. versionadded:: 0.13
939 """
940 sig_alg = _lib.X509_get0_tbs_sigalg(self._x509)
941 alg = _ffi.new("ASN1_OBJECT **")
942 _lib.X509_ALGOR_get0(alg, _ffi.NULL, _ffi.NULL, sig_alg)
943 nid = _lib.OBJ_obj2nid(alg[0])
944 if nid == _lib.NID_undef:
945 raise ValueError("Undefined signature algorithm")
946 return _ffi.string(_lib.OBJ_nid2ln(nid))
948 def digest(self, digest_name: str) -> bytes:
949 """
950 Return the digest of the X509 object.
952 :param digest_name: The name of the digest algorithm to use.
953 :type digest_name: :py:class:`str`
955 :return: The digest of the object, formatted as
956 :py:const:`b":"`-delimited hex pairs.
957 :rtype: :py:class:`bytes`
958 """
959 digest = _lib.EVP_get_digestbyname(_byte_string(digest_name))
960 if digest == _ffi.NULL:
961 raise ValueError("No such digest method")
963 result_buffer = _ffi.new("unsigned char[]", _lib.EVP_MAX_MD_SIZE)
964 result_length = _ffi.new("unsigned int[]", 1)
965 result_length[0] = len(result_buffer)
967 digest_result = _lib.X509_digest(
968 self._x509, digest, result_buffer, result_length
969 )
970 _openssl_assert(digest_result == 1)
972 return b":".join(
973 [
974 b16encode(ch).upper()
975 for ch in _ffi.buffer(result_buffer, result_length[0])
976 ]
977 )
979 def subject_name_hash(self) -> int:
980 """
981 Return the hash of the X509 subject.
983 :return: The hash of the subject.
984 :rtype: :py:class:`int`
985 """
986 return _lib.X509_subject_name_hash(self._x509)
988 @deprecated(
989 "X509.set_serial_number is deprecated. You should use "
990 "cryptography's CertificateBuilder instead."
991 )
992 def set_serial_number(self, serial: int) -> None:
993 """
994 Set the serial number of the certificate.
996 :param serial: The new serial number.
997 :type serial: :py:class:`int`
999 :return: :py:data`None`
1000 """
1001 if not isinstance(serial, int):
1002 raise TypeError("serial must be an integer")
1004 hex_serial = hex(serial)[2:]
1005 hex_serial_bytes = hex_serial.encode("ascii")
1007 bignum_serial = _ffi.new("BIGNUM**")
1009 # BN_hex2bn stores the result in &bignum.
1010 result = _lib.BN_hex2bn(bignum_serial, hex_serial_bytes)
1011 _openssl_assert(result != _ffi.NULL)
1013 asn1_serial = _lib.BN_to_ASN1_INTEGER(bignum_serial[0], _ffi.NULL)
1014 _lib.BN_free(bignum_serial[0])
1015 _openssl_assert(asn1_serial != _ffi.NULL)
1016 asn1_serial = _ffi.gc(asn1_serial, _lib.ASN1_INTEGER_free)
1017 set_result = _lib.X509_set_serialNumber(self._x509, asn1_serial)
1018 _openssl_assert(set_result == 1)
1020 def get_serial_number(self) -> int:
1021 """
1022 Return the serial number of this certificate.
1024 :return: The serial number.
1025 :rtype: int
1026 """
1027 asn1_serial = _lib.X509_get_serialNumber(self._x509)
1028 bignum_serial = _lib.ASN1_INTEGER_to_BN(asn1_serial, _ffi.NULL)
1029 try:
1030 hex_serial = _lib.BN_bn2hex(bignum_serial)
1031 try:
1032 hexstring_serial = _ffi.string(hex_serial)
1033 serial = int(hexstring_serial, 16)
1034 return serial
1035 finally:
1036 _lib.OPENSSL_free(hex_serial)
1037 finally:
1038 _lib.BN_free(bignum_serial)
1040 @deprecated(
1041 "X509.gmtime_adj_notAfter is deprecated. You should use "
1042 "cryptography's CertificateBuilder instead."
1043 )
1044 def gmtime_adj_notAfter(self, amount: int) -> None:
1045 """
1046 Adjust the time stamp on which the certificate stops being valid.
1048 :param int amount: The number of seconds by which to adjust the
1049 timestamp.
1050 :return: ``None``
1051 """
1052 if not isinstance(amount, int):
1053 raise TypeError("amount must be an integer")
1055 notAfter = _lib.X509_getm_notAfter(self._x509)
1056 _lib.X509_gmtime_adj(notAfter, amount)
1058 @deprecated(
1059 "X509.gmtime_adj_notBefore is deprecated. You should use "
1060 "cryptography's CertificateBuilder instead."
1061 )
1062 def gmtime_adj_notBefore(self, amount: int) -> None:
1063 """
1064 Adjust the timestamp on which the certificate starts being valid.
1066 :param amount: The number of seconds by which to adjust the timestamp.
1067 :return: ``None``
1068 """
1069 if not isinstance(amount, int):
1070 raise TypeError("amount must be an integer")
1072 notBefore = _lib.X509_getm_notBefore(self._x509)
1073 _lib.X509_gmtime_adj(notBefore, amount)
1075 def has_expired(self) -> bool:
1076 """
1077 Check whether the certificate has expired.
1079 :return: ``True`` if the certificate has expired, ``False`` otherwise.
1080 :rtype: bool
1081 """
1082 time_bytes = self.get_notAfter()
1083 if time_bytes is None:
1084 raise ValueError("Unable to determine notAfter")
1085 time_string = time_bytes.decode("utf-8")
1086 not_after = datetime.datetime.strptime(time_string, "%Y%m%d%H%M%SZ")
1088 UTC = datetime.timezone.utc
1089 utcnow = datetime.datetime.now(UTC).replace(tzinfo=None)
1090 return not_after < utcnow
1092 def _get_boundary_time(self, which: Any) -> bytes | None:
1093 return _get_asn1_time(which(self._x509))
1095 def get_notBefore(self) -> bytes | None:
1096 """
1097 Get the timestamp at which the certificate starts being valid.
1099 The timestamp is formatted as an ASN.1 TIME::
1101 YYYYMMDDhhmmssZ
1103 :return: A timestamp string, or ``None`` if there is none.
1104 :rtype: bytes or NoneType
1105 """
1106 return self._get_boundary_time(_lib.X509_getm_notBefore)
1108 def _set_boundary_time(
1109 self, which: Callable[..., Any], when: bytes
1110 ) -> None:
1111 return _set_asn1_time(which(self._x509), when)
1113 @deprecated(
1114 "X509.set_notBefore is deprecated. You should use "
1115 "cryptography's CertificateBuilder instead."
1116 )
1117 def set_notBefore(self, when: bytes) -> None:
1118 """
1119 Set the timestamp at which the certificate starts being valid.
1121 The timestamp is formatted as an ASN.1 TIME::
1123 YYYYMMDDhhmmssZ
1125 :param bytes when: A timestamp string.
1126 :return: ``None``
1127 """
1128 return self._set_boundary_time(_lib.X509_getm_notBefore, when)
1130 def get_notAfter(self) -> bytes | None:
1131 """
1132 Get the timestamp at which the certificate stops being valid.
1134 The timestamp is formatted as an ASN.1 TIME::
1136 YYYYMMDDhhmmssZ
1138 :return: A timestamp string, or ``None`` if there is none.
1139 :rtype: bytes or NoneType
1140 """
1141 return self._get_boundary_time(_lib.X509_getm_notAfter)
1143 @deprecated(
1144 "X509.set_notAfter is deprecated. You should use "
1145 "cryptography's CertificateBuilder instead."
1146 )
1147 def set_notAfter(self, when: bytes) -> None:
1148 """
1149 Set the timestamp at which the certificate stops being valid.
1151 The timestamp is formatted as an ASN.1 TIME::
1153 YYYYMMDDhhmmssZ
1155 :param bytes when: A timestamp string.
1156 :return: ``None``
1157 """
1158 return self._set_boundary_time(_lib.X509_getm_notAfter, when)
1160 def _get_name(self, which: Any) -> X509Name:
1161 # Bypass X509Name.__new__, which warns that X509Name is deprecated;
1162 # callers that should warn are decorated individually.
1163 name = object.__new__(X509Name)
1164 name._name = which(self._x509)
1165 _openssl_assert(name._name != _ffi.NULL)
1167 # The name is owned by the X509 structure. As long as the X509Name
1168 # Python object is alive, keep the X509 Python object alive.
1169 name._owner = self
1171 return name
1173 def _set_name(self, which: Any, name: X509Name) -> None:
1174 if not isinstance(name, X509Name):
1175 raise TypeError("name must be an X509Name")
1176 set_result = which(self._x509, name._name)
1177 _openssl_assert(set_result == 1)
1179 @deprecated(
1180 "X509.get_issuer is deprecated. You should use "
1181 "cryptography's X.509 APIs instead."
1182 )
1183 def get_issuer(self) -> X509Name:
1184 """
1185 Return the issuer of this certificate.
1187 This creates a new :class:`X509Name` that wraps the underlying issuer
1188 name field on the certificate. Modifying it will modify the underlying
1189 certificate, and will have the effect of modifying any other
1190 :class:`X509Name` that refers to this issuer.
1192 :return: The issuer of this certificate.
1193 :rtype: :class:`X509Name`
1194 """
1195 name = self._get_name(_lib.X509_get_issuer_name)
1196 self._issuer_invalidator.add(name)
1197 return name
1199 @deprecated(
1200 "X509.set_issuer is deprecated. You should use "
1201 "cryptography's CertificateBuilder instead."
1202 )
1203 def set_issuer(self, issuer: X509Name) -> None:
1204 """
1205 Set the issuer of this certificate.
1207 :param issuer: The issuer.
1208 :type issuer: :py:class:`X509Name`
1210 :return: ``None``
1211 """
1212 self._set_name(_lib.X509_set_issuer_name, issuer)
1213 self._issuer_invalidator.clear()
1215 @deprecated(
1216 "X509.get_subject is deprecated. You should use "
1217 "cryptography's X.509 APIs instead."
1218 )
1219 def get_subject(self) -> X509Name:
1220 """
1221 Return the subject of this certificate.
1223 This creates a new :class:`X509Name` that wraps the underlying subject
1224 name field on the certificate. Modifying it will modify the underlying
1225 certificate, and will have the effect of modifying any other
1226 :class:`X509Name` that refers to this subject.
1228 :return: The subject of this certificate.
1229 :rtype: :class:`X509Name`
1230 """
1231 name = self._get_name(_lib.X509_get_subject_name)
1232 self._subject_invalidator.add(name)
1233 return name
1235 @deprecated(
1236 "X509.set_subject is deprecated. You should use "
1237 "cryptography's CertificateBuilder instead."
1238 )
1239 def set_subject(self, subject: X509Name) -> None:
1240 """
1241 Set the subject of this certificate.
1243 :param subject: The subject.
1244 :type subject: :py:class:`X509Name`
1246 :return: ``None``
1247 """
1248 self._set_name(_lib.X509_set_subject_name, subject)
1249 self._subject_invalidator.clear()
1251 def get_extension_count(self) -> int:
1252 """
1253 Get the number of extensions on this certificate.
1255 :return: The number of extensions.
1256 :rtype: :py:class:`int`
1258 .. versionadded:: 0.12
1259 """
1260 return _lib.X509_get_ext_count(self._x509)
1263class X509StoreFlags:
1264 """
1265 Flags for X509 verification, used to change the behavior of
1266 :class:`X509Store`.
1268 See `OpenSSL Verification Flags`_ for details.
1270 .. _OpenSSL Verification Flags:
1271 https://www.openssl.org/docs/manmaster/man3/X509_VERIFY_PARAM_set_flags.html
1272 """
1274 CRL_CHECK: int = _lib.X509_V_FLAG_CRL_CHECK
1275 CRL_CHECK_ALL: int = _lib.X509_V_FLAG_CRL_CHECK_ALL
1276 IGNORE_CRITICAL: int = _lib.X509_V_FLAG_IGNORE_CRITICAL
1277 X509_STRICT: int = _lib.X509_V_FLAG_X509_STRICT
1278 ALLOW_PROXY_CERTS: int = _lib.X509_V_FLAG_ALLOW_PROXY_CERTS
1279 POLICY_CHECK: int = _lib.X509_V_FLAG_POLICY_CHECK
1280 EXPLICIT_POLICY: int = _lib.X509_V_FLAG_EXPLICIT_POLICY
1281 INHIBIT_MAP: int = _lib.X509_V_FLAG_INHIBIT_MAP
1282 CHECK_SS_SIGNATURE: int = _lib.X509_V_FLAG_CHECK_SS_SIGNATURE
1283 PARTIAL_CHAIN: int = _lib.X509_V_FLAG_PARTIAL_CHAIN
1286class X509Store:
1287 """
1288 An X.509 store.
1290 An X.509 store is used to describe a context in which to verify a
1291 certificate. A description of a context may include a set of certificates
1292 to trust, a set of certificate revocation lists, verification flags and
1293 more.
1295 An X.509 store, being only a description, cannot be used by itself to
1296 verify a certificate. To carry out the actual verification process, see
1297 :class:`X509StoreContext`.
1298 """
1300 def __init__(self) -> None:
1301 store = _lib.X509_STORE_new()
1302 self._store = _ffi.gc(store, _lib.X509_STORE_free)
1304 def add_cert(self, cert: X509) -> None:
1305 """
1306 Adds a trusted certificate to this store.
1308 Adding a certificate with this method adds this certificate as a
1309 *trusted* certificate.
1311 :param X509 cert: The certificate to add to this store.
1313 :raises TypeError: If the certificate is not an :class:`X509`.
1315 :raises OpenSSL.crypto.Error: If OpenSSL was unhappy with your
1316 certificate.
1318 :return: ``None`` if the certificate was added successfully.
1319 """
1320 if not isinstance(cert, X509):
1321 raise TypeError()
1323 res = _lib.X509_STORE_add_cert(self._store, cert._x509)
1324 _openssl_assert(res == 1)
1326 def add_crl(self, crl: x509.CertificateRevocationList) -> None:
1327 """
1328 Add a certificate revocation list to this store.
1330 The certificate revocation lists added to a store will only be used if
1331 the associated flags are configured to check certificate revocation
1332 lists.
1334 .. versionadded:: 16.1.0
1336 :param crl: The certificate revocation list to add to this store.
1337 :type crl: ``cryptography.x509.CertificateRevocationList``
1338 :return: ``None`` if the certificate revocation list was added
1339 successfully.
1340 """
1341 if isinstance(crl, x509.CertificateRevocationList):
1342 from cryptography.hazmat.primitives.serialization import Encoding
1344 bio = _new_mem_buf(crl.public_bytes(Encoding.DER))
1345 openssl_crl = _lib.d2i_X509_CRL_bio(bio, _ffi.NULL)
1346 _openssl_assert(openssl_crl != _ffi.NULL)
1347 crl = _ffi.gc(openssl_crl, _lib.X509_CRL_free)
1348 else:
1349 raise TypeError(
1350 "CRL must be of type "
1351 "cryptography.x509.CertificateRevocationList"
1352 )
1354 _openssl_assert(_lib.X509_STORE_add_crl(self._store, crl) != 0)
1356 def set_flags(self, flags: int) -> None:
1357 """
1358 Set verification flags to this store.
1360 Verification flags can be combined by oring them together.
1362 .. note::
1364 Setting a verification flag sometimes requires clients to add
1365 additional information to the store, otherwise a suitable error will
1366 be raised.
1368 For example, in setting flags to enable CRL checking a
1369 suitable CRL must be added to the store otherwise an error will be
1370 raised.
1372 .. versionadded:: 16.1.0
1374 :param int flags: The verification flags to set on this store.
1375 See :class:`X509StoreFlags` for available constants.
1376 :return: ``None`` if the verification flags were successfully set.
1377 """
1378 _openssl_assert(_lib.X509_STORE_set_flags(self._store, flags) != 0)
1380 def set_time(self, vfy_time: datetime.datetime) -> None:
1381 """
1382 Set the time against which the certificates are verified.
1384 Normally the current time is used.
1386 .. note::
1388 For example, you can determine if a certificate was valid at a given
1389 time.
1391 .. versionadded:: 17.0.0
1393 :param datetime vfy_time: The verification time to set on this store.
1394 :return: ``None`` if the verification time was successfully set.
1395 """
1396 param = _lib.X509_VERIFY_PARAM_new()
1397 param = _ffi.gc(param, _lib.X509_VERIFY_PARAM_free)
1399 _lib.X509_VERIFY_PARAM_set_time(
1400 param, calendar.timegm(vfy_time.timetuple())
1401 )
1402 _openssl_assert(_lib.X509_STORE_set1_param(self._store, param) != 0)
1404 def load_locations(
1405 self,
1406 cafile: StrOrBytesPath | None,
1407 capath: StrOrBytesPath | None = None,
1408 ) -> None:
1409 """
1410 Let X509Store know where we can find trusted certificates for the
1411 certificate chain. Note that the certificates have to be in PEM
1412 format.
1414 If *capath* is passed, it must be a directory prepared using the
1415 ``c_rehash`` tool included with OpenSSL. Either, but not both, of
1416 *cafile* or *capath* may be ``None``.
1418 .. note::
1420 Both *cafile* and *capath* may be set simultaneously.
1422 Call this method multiple times to add more than one location.
1423 For example, CA certificates, and certificate revocation list bundles
1424 may be passed in *cafile* in subsequent calls to this method.
1426 .. versionadded:: 20.0
1428 :param cafile: In which file we can find the certificates (``bytes`` or
1429 ``unicode``).
1430 :param capath: In which directory we can find the certificates
1431 (``bytes`` or ``unicode``).
1433 :return: ``None`` if the locations were set successfully.
1435 :raises OpenSSL.crypto.Error: If both *cafile* and *capath* is ``None``
1436 or the locations could not be set for any reason.
1438 """
1439 if cafile is None:
1440 cafile = _ffi.NULL
1441 else:
1442 cafile = _path_bytes(cafile)
1444 if capath is None:
1445 capath = _ffi.NULL
1446 else:
1447 capath = _path_bytes(capath)
1449 load_result = _lib.X509_STORE_load_locations(
1450 self._store, cafile, capath
1451 )
1452 if not load_result:
1453 _raise_current_error()
1456class X509StoreContextError(Exception):
1457 """
1458 An exception raised when an error occurred while verifying a certificate
1459 using `OpenSSL.X509StoreContext.verify_certificate`.
1461 :ivar certificate: The certificate which caused verificate failure.
1462 :type certificate: :class:`X509`
1463 """
1465 def __init__(
1466 self, message: str, errors: list[Any], certificate: X509
1467 ) -> None:
1468 super().__init__(message)
1469 self.errors = errors
1470 self.certificate = certificate
1473class X509StoreContext:
1474 """
1475 An X.509 store context.
1477 An X.509 store context is used to carry out the actual verification process
1478 of a certificate in a described context. For describing such a context, see
1479 :class:`X509Store`.
1481 :param X509Store store: The certificates which will be trusted for the
1482 purposes of any verifications.
1483 :param X509 certificate: The certificate to be verified.
1484 :param chain: List of untrusted certificates that may be used for building
1485 the certificate chain. May be ``None``.
1486 :type chain: :class:`list` of :class:`X509`
1487 """
1489 def __init__(
1490 self,
1491 store: X509Store,
1492 certificate: X509,
1493 chain: Sequence[X509] | None = None,
1494 ) -> None:
1495 self._store = store
1496 self._cert = certificate
1497 self._chain = self._build_certificate_stack(chain)
1499 @staticmethod
1500 def _build_certificate_stack(
1501 certificates: Sequence[X509] | None,
1502 ) -> None:
1503 def cleanup(s: Any) -> None:
1504 # Equivalent to sk_X509_pop_free, but we don't
1505 # currently have a CFFI binding for that available
1506 for i in range(_lib.sk_X509_num(s)):
1507 x = _lib.sk_X509_value(s, i)
1508 _lib.X509_free(x)
1509 _lib.sk_X509_free(s)
1511 if certificates is None or len(certificates) == 0:
1512 return _ffi.NULL
1514 stack = _lib.sk_X509_new_null()
1515 _openssl_assert(stack != _ffi.NULL)
1516 stack = _ffi.gc(stack, cleanup)
1518 for cert in certificates:
1519 if not isinstance(cert, X509):
1520 raise TypeError("One of the elements is not an X509 instance")
1522 _openssl_assert(_lib.X509_up_ref(cert._x509) > 0)
1523 if _lib.sk_X509_push(stack, cert._x509) <= 0:
1524 _lib.X509_free(cert._x509)
1525 _raise_current_error()
1527 return stack
1529 @staticmethod
1530 def _exception_from_context(store_ctx: Any) -> X509StoreContextError:
1531 """
1532 Convert an OpenSSL native context error failure into a Python
1533 exception.
1535 When a call to native OpenSSL X509_verify_cert fails, additional
1536 information about the failure can be obtained from the store context.
1537 """
1538 message = _ffi.string(
1539 _lib.X509_verify_cert_error_string(
1540 _lib.X509_STORE_CTX_get_error(store_ctx)
1541 )
1542 ).decode("utf-8")
1543 errors = [
1544 _lib.X509_STORE_CTX_get_error(store_ctx),
1545 _lib.X509_STORE_CTX_get_error_depth(store_ctx),
1546 message,
1547 ]
1548 # A context error should always be associated with a certificate, so we
1549 # expect this call to never return :class:`None`.
1550 _x509 = _lib.X509_STORE_CTX_get_current_cert(store_ctx)
1551 _cert = _lib.X509_dup(_x509)
1552 pycert = X509._from_raw_x509_ptr(_cert)
1553 return X509StoreContextError(message, errors, pycert)
1555 def _verify_certificate(self) -> Any:
1556 """
1557 Verifies the certificate and runs an X509_STORE_CTX containing the
1558 results.
1560 :raises X509StoreContextError: If an error occurred when validating a
1561 certificate in the context. Sets ``certificate`` attribute to
1562 indicate which certificate caused the error.
1563 """
1564 store_ctx = _lib.X509_STORE_CTX_new()
1565 _openssl_assert(store_ctx != _ffi.NULL)
1566 store_ctx = _ffi.gc(store_ctx, _lib.X509_STORE_CTX_free)
1568 ret = _lib.X509_STORE_CTX_init(
1569 store_ctx, self._store._store, self._cert._x509, self._chain
1570 )
1571 _openssl_assert(ret == 1)
1573 ret = _lib.X509_verify_cert(store_ctx)
1574 if ret <= 0:
1575 raise self._exception_from_context(store_ctx)
1577 return store_ctx
1579 def set_store(self, store: X509Store) -> None:
1580 """
1581 Set the context's X.509 store.
1583 .. versionadded:: 0.15
1585 :param X509Store store: The store description which will be used for
1586 the purposes of any *future* verifications.
1587 """
1588 self._store = store
1590 def verify_certificate(self) -> None:
1591 """
1592 Verify a certificate in a context.
1594 .. versionadded:: 0.15
1596 :raises X509StoreContextError: If an error occurred when validating a
1597 certificate in the context. Sets ``certificate`` attribute to
1598 indicate which certificate caused the error.
1599 """
1600 self._verify_certificate()
1602 def get_verified_chain(self) -> list[X509]:
1603 """
1604 Verify a certificate in a context and return the complete validated
1605 chain.
1607 :raises X509StoreContextError: If an error occurred when validating a
1608 certificate in the context. Sets ``certificate`` attribute to
1609 indicate which certificate caused the error.
1611 .. versionadded:: 20.0
1612 """
1613 store_ctx = self._verify_certificate()
1615 # Note: X509_STORE_CTX_get1_chain returns a deep copy of the chain.
1616 cert_stack = _lib.X509_STORE_CTX_get1_chain(store_ctx)
1617 _openssl_assert(cert_stack != _ffi.NULL)
1619 result = []
1620 for i in range(_lib.sk_X509_num(cert_stack)):
1621 cert = _lib.sk_X509_value(cert_stack, i)
1622 _openssl_assert(cert != _ffi.NULL)
1623 pycert = X509._from_raw_x509_ptr(cert)
1624 result.append(pycert)
1626 # Free the stack but not the members which are freed by the X509 class.
1627 _lib.sk_X509_free(cert_stack)
1628 return result
1631def load_certificate(type: int, buffer: bytes) -> X509:
1632 """
1633 Load a certificate (X509) from the string *buffer* encoded with the
1634 type *type*.
1636 :param type: The file type (one of FILETYPE_PEM, FILETYPE_ASN1)
1638 :param bytes buffer: The buffer the certificate is stored in
1640 :return: The X509 object
1641 """
1642 if isinstance(buffer, str):
1643 buffer = buffer.encode("ascii")
1645 bio = _new_mem_buf(buffer)
1647 if type == FILETYPE_PEM:
1648 x509 = _lib.PEM_read_bio_X509(bio, _ffi.NULL, _ffi.NULL, _ffi.NULL)
1649 elif type == FILETYPE_ASN1:
1650 x509 = _lib.d2i_X509_bio(bio, _ffi.NULL)
1651 else:
1652 raise ValueError("type argument must be FILETYPE_PEM or FILETYPE_ASN1")
1654 if x509 == _ffi.NULL:
1655 _raise_current_error()
1657 return X509._from_raw_x509_ptr(x509)
1660def dump_certificate(type: int, cert: X509) -> bytes:
1661 """
1662 Dump the certificate *cert* into a buffer string encoded with the type
1663 *type*.
1665 :param type: The file type (one of FILETYPE_PEM, FILETYPE_ASN1, or
1666 FILETYPE_TEXT)
1667 :param cert: The certificate to dump
1668 :return: The buffer with the dumped certificate in
1669 """
1670 bio = _new_mem_buf()
1672 if type == FILETYPE_PEM:
1673 result_code = _lib.PEM_write_bio_X509(bio, cert._x509)
1674 elif type == FILETYPE_ASN1:
1675 result_code = _lib.i2d_X509_bio(bio, cert._x509)
1676 elif type == FILETYPE_TEXT:
1677 result_code = _lib.X509_print_ex(bio, cert._x509, 0, 0)
1678 else:
1679 raise ValueError(
1680 "type argument must be FILETYPE_PEM, FILETYPE_ASN1, or "
1681 "FILETYPE_TEXT"
1682 )
1684 _openssl_assert(result_code == 1)
1685 return _bio_to_string(bio)
1688def dump_publickey(type: int, pkey: PKey) -> bytes:
1689 """
1690 Dump a public key to a buffer.
1692 :param type: The file type (one of :data:`FILETYPE_PEM` or
1693 :data:`FILETYPE_ASN1`).
1694 :param PKey pkey: The public key to dump
1695 :return: The buffer with the dumped key in it.
1696 :rtype: bytes
1697 """
1698 bio = _new_mem_buf()
1699 if type == FILETYPE_PEM:
1700 write_bio = _lib.PEM_write_bio_PUBKEY
1701 elif type == FILETYPE_ASN1:
1702 write_bio = _lib.i2d_PUBKEY_bio
1703 else:
1704 raise ValueError("type argument must be FILETYPE_PEM or FILETYPE_ASN1")
1706 result_code = write_bio(bio, pkey._pkey)
1707 if result_code != 1: # pragma: no cover
1708 _raise_current_error()
1710 return _bio_to_string(bio)
1713def dump_privatekey(
1714 type: int,
1715 pkey: PKey,
1716 cipher: str | None = None,
1717 passphrase: PassphraseCallableT | None = None,
1718) -> bytes:
1719 """
1720 Dump the private key *pkey* into a buffer string encoded with the type
1721 *type*. Optionally (if *type* is :const:`FILETYPE_PEM`) encrypting it
1722 using *cipher* and *passphrase*.
1724 :param type: The file type (one of :const:`FILETYPE_PEM`,
1725 :const:`FILETYPE_ASN1`, or :const:`FILETYPE_TEXT`)
1726 :param PKey pkey: The PKey to dump
1727 :param cipher: (optional) if encrypted PEM format, the cipher to use
1728 :param passphrase: (optional) if encrypted PEM format, this can be either
1729 the passphrase to use, or a callback for providing the passphrase.
1731 :return: The buffer with the dumped key in
1732 :rtype: bytes
1734 .. deprecated:: 26.3.0
1735 Use the serialization APIs on ``cryptography`` private key types
1736 instead.
1737 """
1738 bio = _new_mem_buf()
1740 if not isinstance(pkey, PKey):
1741 raise TypeError("pkey must be a PKey")
1743 if cipher is not None:
1744 if passphrase is None:
1745 raise TypeError(
1746 "if a value is given for cipher "
1747 "one must also be given for passphrase"
1748 )
1749 cipher_obj = _lib.EVP_get_cipherbyname(_byte_string(cipher))
1750 if cipher_obj == _ffi.NULL:
1751 raise ValueError("Invalid cipher name")
1752 else:
1753 cipher_obj = _ffi.NULL
1755 helper = _PassphraseHelper(type, passphrase)
1756 if type == FILETYPE_PEM:
1757 result_code = _lib.PEM_write_bio_PrivateKey(
1758 bio,
1759 pkey._pkey,
1760 cipher_obj,
1761 _ffi.NULL,
1762 0,
1763 helper.callback,
1764 helper.callback_args,
1765 )
1766 helper.raise_if_problem()
1767 elif type == FILETYPE_ASN1:
1768 result_code = _lib.i2d_PrivateKey_bio(bio, pkey._pkey)
1769 elif type == FILETYPE_TEXT:
1770 if _lib.EVP_PKEY_id(pkey._pkey) != _lib.EVP_PKEY_RSA:
1771 raise TypeError("Only RSA keys are supported for FILETYPE_TEXT")
1773 rsa = _ffi.gc(_lib.EVP_PKEY_get1_RSA(pkey._pkey), _lib.RSA_free)
1774 result_code = _lib.RSA_print(bio, rsa, 0)
1775 else:
1776 raise ValueError(
1777 "type argument must be FILETYPE_PEM, FILETYPE_ASN1, or "
1778 "FILETYPE_TEXT"
1779 )
1781 _openssl_assert(result_code != 0)
1783 return _bio_to_string(bio)
1786_dump_privatekey_internal = dump_privatekey
1788utils.deprecated(
1789 dump_privatekey,
1790 __name__,
1791 (
1792 "dump_privatekey is deprecated. You should use the APIs in "
1793 "cryptography."
1794 ),
1795 DeprecationWarning,
1796 name="dump_privatekey",
1797)
1800class _PassphraseHelper:
1801 def __init__(
1802 self,
1803 type: int,
1804 passphrase: PassphraseCallableT | None,
1805 more_args: bool = False,
1806 truncate: bool = False,
1807 ) -> None:
1808 if type != FILETYPE_PEM and passphrase is not None:
1809 raise ValueError(
1810 "only FILETYPE_PEM key format supports encryption"
1811 )
1812 self._passphrase = passphrase
1813 self._more_args = more_args
1814 self._truncate = truncate
1815 self._problems: list[Exception] = []
1817 @property
1818 def callback(self) -> Any:
1819 if self._passphrase is None:
1820 return _ffi.NULL
1821 elif isinstance(self._passphrase, bytes) or callable(self._passphrase):
1822 return _ffi.callback("pem_password_cb", self._read_passphrase)
1823 else:
1824 raise TypeError(
1825 "Last argument must be a byte string or a callable."
1826 )
1828 @property
1829 def callback_args(self) -> Any:
1830 if self._passphrase is None:
1831 return _ffi.NULL
1832 elif isinstance(self._passphrase, bytes) or callable(self._passphrase):
1833 return _ffi.NULL
1834 else:
1835 raise TypeError(
1836 "Last argument must be a byte string or a callable."
1837 )
1839 def raise_if_problem(self, exceptionType: type[Exception] = Error) -> None:
1840 if self._problems:
1841 # Flush the OpenSSL error queue
1842 try:
1843 _exception_from_error_queue(exceptionType)
1844 except exceptionType:
1845 pass
1847 raise self._problems.pop(0)
1849 def _read_passphrase(
1850 self, buf: Any, size: int, rwflag: Any, userdata: Any
1851 ) -> int:
1852 try:
1853 if callable(self._passphrase):
1854 if self._more_args:
1855 result = self._passphrase(size, rwflag, userdata)
1856 else:
1857 result = self._passphrase(rwflag)
1858 else:
1859 assert self._passphrase is not None
1860 result = self._passphrase
1861 if not isinstance(result, bytes):
1862 raise ValueError("Bytes expected")
1863 if len(result) > size:
1864 if self._truncate:
1865 result = result[:size]
1866 else:
1867 raise ValueError(
1868 "passphrase returned by callback is too long"
1869 )
1870 for i in range(len(result)):
1871 buf[i] = result[i : i + 1]
1872 return len(result)
1873 except Exception as e:
1874 self._problems.append(e)
1875 return 0
1878def load_publickey(type: int, buffer: str | bytes) -> PKey:
1879 """
1880 Load a public key from a buffer.
1882 :param type: The file type (one of :data:`FILETYPE_PEM`,
1883 :data:`FILETYPE_ASN1`).
1884 :param buffer: The buffer the key is stored in.
1885 :type buffer: A Python string object, either unicode or bytestring.
1886 :return: The PKey object.
1887 :rtype: :class:`PKey`
1888 """
1889 if isinstance(buffer, str):
1890 buffer = buffer.encode("ascii")
1892 bio = _new_mem_buf(buffer)
1894 if type == FILETYPE_PEM:
1895 evp_pkey = _lib.PEM_read_bio_PUBKEY(
1896 bio, _ffi.NULL, _ffi.NULL, _ffi.NULL
1897 )
1898 elif type == FILETYPE_ASN1:
1899 evp_pkey = _lib.d2i_PUBKEY_bio(bio, _ffi.NULL)
1900 else:
1901 raise ValueError("type argument must be FILETYPE_PEM or FILETYPE_ASN1")
1903 if evp_pkey == _ffi.NULL:
1904 _raise_current_error()
1906 pkey = PKey.__new__(PKey)
1907 pkey._pkey = _ffi.gc(evp_pkey, _lib.EVP_PKEY_free)
1908 pkey._only_public = True
1909 return pkey
1912def load_privatekey(
1913 type: int,
1914 buffer: str | bytes,
1915 passphrase: PassphraseCallableT | None = None,
1916) -> PKey:
1917 """
1918 Load a private key (PKey) from the string *buffer* encoded with the type
1919 *type*.
1921 :param type: The file type (one of FILETYPE_PEM, FILETYPE_ASN1)
1922 :param buffer: The buffer the key is stored in
1923 :param passphrase: (optional) if encrypted PEM format, this can be
1924 either the passphrase to use, or a callback for
1925 providing the passphrase.
1927 :return: The PKey object
1928 """
1929 if isinstance(buffer, str):
1930 buffer = buffer.encode("ascii")
1932 bio = _new_mem_buf(buffer)
1934 helper = _PassphraseHelper(type, passphrase)
1935 if type == FILETYPE_PEM:
1936 evp_pkey = _lib.PEM_read_bio_PrivateKey(
1937 bio, _ffi.NULL, helper.callback, helper.callback_args
1938 )
1939 helper.raise_if_problem()
1940 elif type == FILETYPE_ASN1:
1941 evp_pkey = _lib.d2i_PrivateKey_bio(bio, _ffi.NULL)
1942 else:
1943 raise ValueError("type argument must be FILETYPE_PEM or FILETYPE_ASN1")
1945 if evp_pkey == _ffi.NULL:
1946 _raise_current_error()
1948 pkey = PKey.__new__(PKey)
1949 pkey._pkey = _ffi.gc(evp_pkey, _lib.EVP_PKEY_free)
1950 return pkey