Google Git
Sign in
chromium/external/github.com/pyca/pyopenssl/refs/heads/main/./src/OpenSSL/crypto.py
blob: 314b577fedf4be594e89081553254191b5659c20 [file] [edit]
from __future__ import annotations
import calendar
import datetime
import functools
import sys
import typing
from base64 import b16encode
from collections.abc import Sequence
from functools import partial
from typing import (
Any,
Callable,
Union,
)
if sys.version_info >= (3, 13):
from warnings import deprecated
else:
from typing_extensions import deprecated
from cryptography import utils, x509
from cryptography.hazmat.primitives.asymmetric import (
dsa,
ec,
ed448,
ed25519,
rsa,
)
from OpenSSL._util import StrOrBytesPath
from OpenSSL._util import (
byte_string as _byte_string,
)
from OpenSSL._util import (
exception_from_error_queue as _exception_from_error_queue,
)
from OpenSSL._util import (
ffi as _ffi,
)
from OpenSSL._util import (
lib as _lib,
)
from OpenSSL._util import (
make_assert as _make_assert,
)
from OpenSSL._util import (
path_bytes as _path_bytes,
)
__all__ = [
"FILETYPE_ASN1",
"FILETYPE_PEM",
"FILETYPE_TEXT",
"TYPE_DSA",
"TYPE_RSA",
"X509",
"Error",
"PKey",
"X509Name",
"X509Store",
"X509StoreContext",
"X509StoreContextError",
"X509StoreFlags",
"dump_certificate",
"dump_privatekey",
"dump_publickey",
"get_elliptic_curve",
"get_elliptic_curves",
"load_certificate",
"load_privatekey",
"load_publickey",
]
_PrivateKey = Union[
dsa.DSAPrivateKey,
ec.EllipticCurvePrivateKey,
ed25519.Ed25519PrivateKey,
ed448.Ed448PrivateKey,
rsa.RSAPrivateKey,
]
_PublicKey = Union[
dsa.DSAPublicKey,
ec.EllipticCurvePublicKey,
ed25519.Ed25519PublicKey,
ed448.Ed448PublicKey,
rsa.RSAPublicKey,
]
_Key = Union[_PrivateKey, _PublicKey]
PassphraseCallableT = Union[bytes, Callable[..., bytes]]
FILETYPE_PEM: int = _lib.SSL_FILETYPE_PEM
FILETYPE_ASN1: int = _lib.SSL_FILETYPE_ASN1
# TODO This was an API mistake. OpenSSL has no such constant.
FILETYPE_TEXT = 2**16 - 1
TYPE_RSA: int = _lib.EVP_PKEY_RSA
TYPE_DSA: int = _lib.EVP_PKEY_DSA
TYPE_DH: int = _lib.EVP_PKEY_DH
TYPE_EC: int = _lib.EVP_PKEY_EC
class Error(Exception):
"""
An error occurred in an `OpenSSL.crypto` API.
"""
_raise_current_error = partial(_exception_from_error_queue, Error)
_openssl_assert = _make_assert(Error)
def _new_mem_buf(buffer: bytes | None = None) -> Any:
"""
Allocate a new OpenSSL memory BIO.
Arrange for the garbage collector to clean it up automatically.
:param buffer: None or some bytes to use to put into the BIO so that they
can be read out.
"""
if buffer is None:
bio = _lib.BIO_new(_lib.BIO_s_mem())
free = _lib.BIO_free
else:
data = _ffi.new("char[]", buffer)
bio = _lib.BIO_new_mem_buf(data, len(buffer))
# Keep the memory alive as long as the bio is alive!
def free(bio: Any, ref: Any = data) -> Any:
return _lib.BIO_free(bio)
_openssl_assert(bio != _ffi.NULL)
bio = _ffi.gc(bio, free)
return bio
def _bio_to_string(bio: Any) -> bytes:
"""
Copy the contents of an OpenSSL BIO object into a Python byte string.
"""
result_buffer = _ffi.new("char**")
buffer_length = _lib.BIO_get_mem_data(bio, result_buffer)
return _ffi.buffer(result_buffer[0], buffer_length)[:]
def _set_asn1_time(boundary: Any, when: bytes) -> None:
"""
The the time value of an ASN1 time object.
@param boundary: An ASN1_TIME pointer (or an object safely
castable to that type) which will have its value set.
@param when: A string representation of the desired time value.
@raise TypeError: If C{when} is not a L{bytes} string.
@raise ValueError: If C{when} does not represent a time in the required
format.
@raise RuntimeError: If the time value cannot be set for some other
(unspecified) reason.
"""
if not isinstance(when, bytes):
raise TypeError("when must be a byte string")
# ASN1_TIME_set_string validates the string without writing anything
# when the destination is NULL.
_openssl_assert(boundary != _ffi.NULL)
set_result = _lib.ASN1_TIME_set_string(boundary, when)
if set_result == 0:
raise ValueError("Invalid string")
def _new_asn1_time(when: bytes) -> Any:
"""
Behaves like _set_asn1_time but returns a new ASN1_TIME object.
@param when: A string representation of the desired time value.
@raise TypeError: If C{when} is not a L{bytes} string.
@raise ValueError: If C{when} does not represent a time in the required
format.
@raise RuntimeError: If the time value cannot be set for some other
(unspecified) reason.
"""
ret = _lib.ASN1_TIME_new()
_openssl_assert(ret != _ffi.NULL)
ret = _ffi.gc(ret, _lib.ASN1_TIME_free)
_set_asn1_time(ret, when)
return ret
def _get_asn1_time(timestamp: Any) -> bytes | None:
"""
Retrieve the time value of an ASN1 time object.
@param timestamp: An ASN1_GENERALIZEDTIME* (or an object safely castable to
that type) from which the time value will be retrieved.
@return: The time value from C{timestamp} as a L{bytes} string in a certain
format. Or C{None} if the object contains no time value.
"""
string_timestamp = _ffi.cast("ASN1_STRING*", timestamp)
if _lib.ASN1_STRING_length(string_timestamp) == 0:
return None
elif (
_lib.ASN1_STRING_type(string_timestamp) == _lib.V_ASN1_GENERALIZEDTIME
):
return _ffi.string(_lib.ASN1_STRING_get0_data(string_timestamp))
else:
generalized_timestamp = _ffi.new("ASN1_GENERALIZEDTIME**")
_lib.ASN1_TIME_to_generalizedtime(timestamp, generalized_timestamp)
_openssl_assert(generalized_timestamp[0] != _ffi.NULL)
string_timestamp = _ffi.cast("ASN1_STRING*", generalized_timestamp[0])
string_data = _lib.ASN1_STRING_get0_data(string_timestamp)
string_result = _ffi.string(string_data)
_lib.ASN1_GENERALIZEDTIME_free(generalized_timestamp[0])
return string_result
class _X509NameInvalidator:
def __init__(self) -> None:
self._names: list[X509Name] = []
def add(self, name: X509Name) -> None:
self._names.append(name)
def clear(self) -> None:
for name in self._names:
# Breaks the object, but also prevents UAF!
del name._name
class PKey:
"""
A class representing an DSA or RSA public key or key pair.
"""
_only_public = False
_initialized = True
def __init__(self) -> None:
pkey = _lib.EVP_PKEY_new()
self._pkey = _ffi.gc(pkey, _lib.EVP_PKEY_free)
self._initialized = False
def to_cryptography_key(self) -> _Key:
"""
Export as a ``cryptography`` key.
:rtype: One of ``cryptography``'s `key interfaces`_.
.. _key interfaces: https://cryptography.io/en/latest/hazmat/\
primitives/asymmetric/rsa/#key-interfaces
.. versionadded:: 16.1.0
"""
from cryptography.hazmat.primitives.serialization import (
load_der_private_key,
load_der_public_key,
)
if self._only_public:
der = dump_publickey(FILETYPE_ASN1, self)
return typing.cast(_Key, load_der_public_key(der))
else:
der = _dump_privatekey_internal(FILETYPE_ASN1, self)
return typing.cast(_Key, load_der_private_key(der, password=None))
@classmethod
def from_cryptography_key(cls, crypto_key: _Key) -> PKey:
"""
Construct based on a ``cryptography`` *crypto_key*.
:param crypto_key: A ``cryptography`` key.
:type crypto_key: One of ``cryptography``'s `key interfaces`_.
:rtype: PKey
.. versionadded:: 16.1.0
"""
if not isinstance(
crypto_key,
(
dsa.DSAPrivateKey,
dsa.DSAPublicKey,
ec.EllipticCurvePrivateKey,
ec.EllipticCurvePublicKey,
ed25519.Ed25519PrivateKey,
ed25519.Ed25519PublicKey,
ed448.Ed448PrivateKey,
ed448.Ed448PublicKey,
rsa.RSAPrivateKey,
rsa.RSAPublicKey,
),
):
raise TypeError("Unsupported key type")
from cryptography.hazmat.primitives.serialization import (
Encoding,
NoEncryption,
PrivateFormat,
PublicFormat,
)
if isinstance(
crypto_key,
(
dsa.DSAPublicKey,
ec.EllipticCurvePublicKey,
ed25519.Ed25519PublicKey,
ed448.Ed448PublicKey,
rsa.RSAPublicKey,
),
):
return load_publickey(
FILETYPE_ASN1,
crypto_key.public_bytes(
Encoding.DER, PublicFormat.SubjectPublicKeyInfo
),
)
else:
der = crypto_key.private_bytes(
Encoding.DER, PrivateFormat.PKCS8, NoEncryption()
)
return load_privatekey(FILETYPE_ASN1, der)
@deprecated(
"PKey.generate_key is deprecated. You should use the key "
"generation APIs in cryptography instead."
)
def generate_key(self, type: int, bits: int) -> None:
"""
Generate a key pair of the given type, with the given number of bits.
This generates a key "into" the this object.
:param type: The key type.
:type type: :py:data:`TYPE_RSA` or :py:data:`TYPE_DSA`
:param bits: The number of bits.
:type bits: :py:data:`int` ``>= 0``
:raises TypeError: If :py:data:`type` or :py:data:`bits` isn't
of the appropriate type.
:raises ValueError: If the number of bits isn't an integer of
the appropriate size.
:return: ``None``
"""
if not isinstance(type, int):
raise TypeError("type must be an integer")
if not isinstance(bits, int):
raise TypeError("bits must be an integer")
if type == TYPE_RSA:
if bits <= 0:
raise ValueError("Invalid number of bits")
# TODO Check error return
exponent = _lib.BN_new()
exponent = _ffi.gc(exponent, _lib.BN_free)
_lib.BN_set_word(exponent, _lib.RSA_F4)
rsa = _lib.RSA_new()
result = _lib.RSA_generate_key_ex(rsa, bits, exponent, _ffi.NULL)
_openssl_assert(result == 1)
result = _lib.EVP_PKEY_assign_RSA(self._pkey, rsa)
_openssl_assert(result == 1)
elif type == TYPE_DSA:
dsa = _lib.DSA_new()
_openssl_assert(dsa != _ffi.NULL)
dsa = _ffi.gc(dsa, _lib.DSA_free)
res = _lib.DSA_generate_parameters_ex(
dsa, bits, _ffi.NULL, 0, _ffi.NULL, _ffi.NULL, _ffi.NULL
)
_openssl_assert(res == 1)
_openssl_assert(_lib.DSA_generate_key(dsa) == 1)
_openssl_assert(_lib.EVP_PKEY_set1_DSA(self._pkey, dsa) == 1)
else:
raise Error("No such key type")
self._initialized = True
@deprecated(
"PKey.check is deprecated. You should use the APIs in "
"cryptography instead."
)
def check(self) -> bool:
"""
Check the consistency of an RSA private key.
This is the Python equivalent of OpenSSL's ``RSA_check_key``.
:return: ``True`` if key is consistent.
:raise OpenSSL.crypto.Error: if the key is inconsistent.
:raise TypeError: if the key is of a type which cannot be checked.
Only RSA keys can currently be checked.
"""
if self._only_public:
raise TypeError("public key only")
if _lib.EVP_PKEY_type(self.type()) != _lib.EVP_PKEY_RSA:
raise TypeError("Only RSA keys can currently be checked.")
rsa = _lib.EVP_PKEY_get1_RSA(self._pkey)
rsa = _ffi.gc(rsa, _lib.RSA_free)
result = _lib.RSA_check_key(rsa)
if result == 1:
return True
_raise_current_error()
def type(self) -> int:
"""
Returns the type of the key
:return: The type of the key.
"""
return _lib.EVP_PKEY_id(self._pkey)
def bits(self) -> int:
"""
Returns the number of bits of the key
:return: The number of bits of the key.
"""
return _lib.EVP_PKEY_bits(self._pkey)
class _EllipticCurve:
"""
A representation of a supported elliptic curve.
@cvar _curves: :py:obj:`None` until an attempt is made to load the curves.
Thereafter, a :py:type:`set` containing :py:type:`_EllipticCurve`
instances each of which represents one curve supported by the system.
@type _curves: :py:type:`NoneType` or :py:type:`set`
"""
_curves = None
def __ne__(self, other: Any) -> bool:
"""
Implement cooperation with the right-hand side argument of ``!=``.
Python 3 seems to have dropped this cooperation in this very narrow
circumstance.
"""
if isinstance(other, _EllipticCurve):
return super().__ne__(other)
return NotImplemented
@classmethod
def _load_elliptic_curves(cls, lib: Any) -> set[_EllipticCurve]:
"""
Get the curves supported by OpenSSL.
:param lib: The OpenSSL library binding object.
:return: A :py:type:`set` of ``cls`` instances giving the names of the
elliptic curves the underlying library supports.
"""
num_curves = lib.EC_get_builtin_curves(_ffi.NULL, 0)
builtin_curves = _ffi.new("EC_builtin_curve[]", num_curves)
# The return value on this call should be num_curves again. We
# could check it to make sure but if it *isn't* then.. what could
# we do? Abort the whole process, I suppose...? -exarkun
lib.EC_get_builtin_curves(builtin_curves, num_curves)
return set(cls.from_nid(lib, c.nid) for c in builtin_curves)
@classmethod
def _get_elliptic_curves(cls, lib: Any) -> set[_EllipticCurve]:
"""
Get, cache, and return the curves supported by OpenSSL.
:param lib: The OpenSSL library binding object.
:return: A :py:type:`set` of ``cls`` instances giving the names of the
elliptic curves the underlying library supports.
"""
if cls._curves is None:
cls._curves = cls._load_elliptic_curves(lib)
return cls._curves
@classmethod
def from_nid(cls, lib: Any, nid: int) -> _EllipticCurve:
"""
Instantiate a new :py:class:`_EllipticCurve` associated with the given
OpenSSL NID.
:param lib: The OpenSSL library binding object.
:param nid: The OpenSSL NID the resulting curve object will represent.
This must be a curve NID (and not, for example, a hash NID) or
subsequent operations will fail in unpredictable ways.
:type nid: :py:class:`int`
:return: The curve object.
"""
return cls(lib, nid, _ffi.string(lib.OBJ_nid2sn(nid)).decode("ascii"))
def __init__(self, lib: Any, nid: int, name: str) -> None:
"""
:param _lib: The :py:mod:`cryptography` binding instance used to
interface with OpenSSL.
:param _nid: The OpenSSL NID identifying the curve this object
represents.
:type _nid: :py:class:`int`
:param name: The OpenSSL short name identifying the curve this object
represents.
:type name: :py:class:`unicode`
"""
self._lib = lib
self._nid = nid
self.name = name
def __repr__(self) -> str:
return f"<Curve {self.name!r}>"
def _to_EC_KEY(self) -> Any:
"""
Create a new OpenSSL EC_KEY structure initialized to use this curve.
The structure is automatically garbage collected when the Python object
is garbage collected.
"""
key = self._lib.EC_KEY_new_by_curve_name(self._nid)
return _ffi.gc(key, _lib.EC_KEY_free)
@deprecated(
"get_elliptic_curves is deprecated. You should use the APIs in "
"cryptography instead."
)
def get_elliptic_curves() -> set[_EllipticCurve]:
"""
Return a set of objects representing the elliptic curves supported in the
OpenSSL build in use.
The curve objects have a :py:class:`unicode` ``name`` attribute by which
they identify themselves.
The curve objects are useful as values for the argument accepted by
:py:meth:`Context.set_tmp_ecdh` to specify which elliptical curve should be
used for ECDHE key exchange.
"""
return _EllipticCurve._get_elliptic_curves(_lib)
@deprecated(
"get_elliptic_curve is deprecated. You should use the APIs in "
"cryptography instead."
)
def get_elliptic_curve(name: str) -> _EllipticCurve:
"""
Return a single curve object selected by name.
See :py:func:`get_elliptic_curves` for information about curve objects.
:param name: The OpenSSL short name identifying the curve object to
retrieve.
:type name: :py:class:`unicode`
If the named curve is not supported then :py:class:`ValueError` is raised.
"""
for curve in get_elliptic_curves():
if curve.name == name:
return curve
raise ValueError("unknown curve name", name)
@deprecated(
"X509Name support in pyOpenSSL is deprecated. You should use the "
"APIs in cryptography."
)
@functools.total_ordering
class X509Name:
"""
An X.509 Distinguished Name.
:ivar countryName: The country of the entity.
:ivar C: Alias for :py:attr:`countryName`.
:ivar stateOrProvinceName: The state or province of the entity.
:ivar ST: Alias for :py:attr:`stateOrProvinceName`.
:ivar localityName: The locality of the entity.
:ivar L: Alias for :py:attr:`localityName`.
:ivar organizationName: The organization name of the entity.
:ivar O: Alias for :py:attr:`organizationName`.
:ivar organizationalUnitName: The organizational unit of the entity.
:ivar OU: Alias for :py:attr:`organizationalUnitName`
:ivar commonName: The common name of the entity.
:ivar CN: Alias for :py:attr:`commonName`.
:ivar emailAddress: The e-mail address of the entity.
"""
def __init__(self, name: X509Name) -> None:
"""
Create a new X509Name, copying the given X509Name instance.
:param name: The name to copy.
:type name: :py:class:`X509Name`
"""
name = _lib.X509_NAME_dup(name._name)
self._name: Any = _ffi.gc(name, _lib.X509_NAME_free)
def __setattr__(self, name: str, value: Any) -> None:
if name.startswith("_"):
return super().__setattr__(name, value)
# Note: we really do not want str subclasses here, so we do not use
# isinstance.
if type(name) is not str:
raise TypeError(
f"attribute name must be string, not "
f"'{type(value).__name__:.200}'"
)
nid = _lib.OBJ_txt2nid(_byte_string(name))
if nid == _lib.NID_undef:
try:
_raise_current_error()
except Error:
pass
raise AttributeError("No such attribute")
# If there's an old entry for this NID, remove it
for i in range(_lib.X509_NAME_entry_count(self._name)):
ent = _lib.X509_NAME_get_entry(self._name, i)
ent_obj = _lib.X509_NAME_ENTRY_get_object(ent)
ent_nid = _lib.OBJ_obj2nid(ent_obj)
if nid == ent_nid:
ent = _lib.X509_NAME_delete_entry(self._name, i)
_lib.X509_NAME_ENTRY_free(ent)
break
if isinstance(value, str):
value = value.encode("utf-8")
add_result = _lib.X509_NAME_add_entry_by_NID(
self._name, nid, _lib.MBSTRING_UTF8, value, len(value), -1, 0
)
if not add_result:
_raise_current_error()
def __getattr__(self, name: str) -> str | None:
"""
Find attribute. An X509Name object has the following attributes:
countryName (alias C), stateOrProvince (alias ST), locality (alias L),
organization (alias O), organizationalUnit (alias OU), commonName
(alias CN) and more...
"""
nid = _lib.OBJ_txt2nid(_byte_string(name))
if nid == _lib.NID_undef:
# This is a bit weird. OBJ_txt2nid indicated failure, but it seems
# a lower level function, a2d_ASN1_OBJECT, also feels the need to
# push something onto the error queue. If we don't clean that up
# now, someone else will bump into it later and be quite confused.
# See lp#314814.
try:
_raise_current_error()
except Error:
pass
raise AttributeError("No such attribute")
entry_index = _lib.X509_NAME_get_index_by_NID(self._name, nid, -1)
if entry_index == -1:
return None
entry = _lib.X509_NAME_get_entry(self._name, entry_index)
data = _lib.X509_NAME_ENTRY_get_data(entry)
result_buffer = _ffi.new("unsigned char**")
data_length = _lib.ASN1_STRING_to_UTF8(result_buffer, data)
_openssl_assert(data_length >= 0)
try:
result = _ffi.buffer(result_buffer[0], data_length)[:].decode(
"utf-8"
)
finally:
# XXX untested
_lib.OPENSSL_free(result_buffer[0])
return result
def __eq__(self, other: Any) -> bool:
if not isinstance(other, X509Name):
return NotImplemented
return _lib.X509_NAME_cmp(self._name, other._name) == 0
def __lt__(self, other: Any) -> bool:
if not isinstance(other, X509Name):
return NotImplemented
return _lib.X509_NAME_cmp(self._name, other._name) < 0
def __repr__(self) -> str:
"""
String representation of an X509Name
"""
result_buffer = _ffi.new("char[]", 512)
format_result = _lib.X509_NAME_oneline(
self._name, result_buffer, len(result_buffer)
)
_openssl_assert(format_result != _ffi.NULL)
return "<X509Name object '{}'>".format(
_ffi.string(result_buffer).decode("utf-8"),
)
def hash(self) -> int:
"""
Return an integer representation of the first four bytes of the
MD5 digest of the DER representation of the name.
This is the Python equivalent of OpenSSL's ``X509_NAME_hash``.
:return: The (integer) hash of this name.
:rtype: :py:class:`int`
"""
return _lib.X509_NAME_hash(self._name)
def der(self) -> bytes:
"""
Return the DER encoding of this name.
:return: The DER encoded form of this name.
:rtype: :py:class:`bytes`
"""
result_buffer = _ffi.new("unsigned char**")
encode_result = _lib.i2d_X509_NAME(self._name, result_buffer)
_openssl_assert(encode_result >= 0)
string_result = _ffi.buffer(result_buffer[0], encode_result)[:]
_lib.OPENSSL_free(result_buffer[0])
return string_result
def get_components(self) -> list[tuple[bytes, bytes]]:
"""
Returns the components of this name, as a sequence of 2-tuples.
:return: The components of this name.
:rtype: :py:class:`list` of ``name, value`` tuples.
"""
result = []
for i in range(_lib.X509_NAME_entry_count(self._name)):
ent = _lib.X509_NAME_get_entry(self._name, i)
fname = _lib.X509_NAME_ENTRY_get_object(ent)
fval = _lib.X509_NAME_ENTRY_get_data(ent)
nid = _lib.OBJ_obj2nid(fname)
name = _lib.OBJ_nid2sn(nid)
# ffi.string does not handle strings containing NULL bytes
# (which may have been generated by old, broken software)
value = _ffi.buffer(
_lib.ASN1_STRING_get0_data(fval), _lib.ASN1_STRING_length(fval)
)[:]
result.append((_ffi.string(name), value))
return result
class X509:
"""
An X.509 certificate.
"""
def __init__(self) -> None:
x509 = _lib.X509_new()
_openssl_assert(x509 != _ffi.NULL)
self._x509 = _ffi.gc(x509, _lib.X509_free)
self._issuer_invalidator = _X509NameInvalidator()
self._subject_invalidator = _X509NameInvalidator()
@classmethod
def _from_raw_x509_ptr(cls, x509: Any) -> X509:
cert = cls.__new__(cls)
cert._x509 = _ffi.gc(x509, _lib.X509_free)
cert._issuer_invalidator = _X509NameInvalidator()
cert._subject_invalidator = _X509NameInvalidator()
return cert
def to_cryptography(self) -> x509.Certificate:
"""
Export as a ``cryptography`` certificate.
:rtype: ``cryptography.x509.Certificate``
.. versionadded:: 17.1.0
"""
from cryptography.x509 import load_der_x509_certificate
der = dump_certificate(FILETYPE_ASN1, self)
return load_der_x509_certificate(der)
@classmethod
def from_cryptography(cls, crypto_cert: x509.Certificate) -> X509:
"""
Construct based on a ``cryptography`` *crypto_cert*.
:param crypto_key: A ``cryptography`` X.509 certificate.
:type crypto_key: ``cryptography.x509.Certificate``
:rtype: X509
.. versionadded:: 17.1.0
"""
if not isinstance(crypto_cert, x509.Certificate):
raise TypeError("Must be a certificate")
from cryptography.hazmat.primitives.serialization import Encoding
der = crypto_cert.public_bytes(Encoding.DER)
return load_certificate(FILETYPE_ASN1, der)
@deprecated(
"X509.set_version is deprecated. You should use "
"cryptography's CertificateBuilder instead."
)
def set_version(self, version: int) -> None:
"""
Set the version number of the certificate. Note that the
version value is zero-based, eg. a value of 0 is V1.
:param version: The version number of the certificate.
:type version: :py:class:`int`
:return: ``None``
"""
if not isinstance(version, int):
raise TypeError("version must be an integer")
_openssl_assert(_lib.X509_set_version(self._x509, version) == 1)
def get_version(self) -> int:
"""
Return the version number of the certificate.
:return: The version number of the certificate.
:rtype: :py:class:`int`
"""
return _lib.X509_get_version(self._x509)
def get_pubkey(self) -> PKey:
"""
Get the public key of the certificate.
:return: The public key.
:rtype: :py:class:`PKey`
"""
pkey = PKey.__new__(PKey)
pkey._pkey = _lib.X509_get_pubkey(self._x509)
if pkey._pkey == _ffi.NULL:
_raise_current_error()
pkey._pkey = _ffi.gc(pkey._pkey, _lib.EVP_PKEY_free)
pkey._only_public = True
return pkey
@deprecated(
"X509.set_pubkey is deprecated. You should use "
"cryptography's CertificateBuilder instead."
)
def set_pubkey(self, pkey: PKey) -> None:
"""
Set the public key of the certificate.
:param pkey: The public key.
:type pkey: :py:class:`PKey`
:return: :py:data:`None`
"""
if not isinstance(pkey, PKey):
raise TypeError("pkey must be a PKey instance")
set_result = _lib.X509_set_pubkey(self._x509, pkey._pkey)
_openssl_assert(set_result == 1)
@deprecated(
"X509.sign is deprecated. You should use "
"cryptography's CertificateBuilder instead."
)
def sign(self, pkey: PKey, digest: str) -> None:
"""
Sign the certificate with this key and digest type.
:param pkey: The key to sign with.
:type pkey: :py:class:`PKey`
:param digest: The name of the message digest to use.
:type digest: :py:class:`str`
:return: :py:data:`None`
"""
if not isinstance(pkey, PKey):
raise TypeError("pkey must be a PKey instance")
if pkey._only_public:
raise ValueError("Key only has public part")
if not pkey._initialized:
raise ValueError("Key is uninitialized")
evp_md = _lib.EVP_get_digestbyname(_byte_string(digest))
if evp_md == _ffi.NULL:
raise ValueError("No such digest method")
sign_result = _lib.X509_sign(self._x509, pkey._pkey, evp_md)
_openssl_assert(sign_result > 0)
def get_signature_algorithm(self) -> bytes:
"""
Return the signature algorithm used in the certificate.
:return: The name of the algorithm.
:rtype: :py:class:`bytes`
:raises ValueError: If the signature algorithm is undefined.
.. versionadded:: 0.13
"""
sig_alg = _lib.X509_get0_tbs_sigalg(self._x509)
alg = _ffi.new("ASN1_OBJECT **")
_lib.X509_ALGOR_get0(alg, _ffi.NULL, _ffi.NULL, sig_alg)
nid = _lib.OBJ_obj2nid(alg[0])
if nid == _lib.NID_undef:
raise ValueError("Undefined signature algorithm")
return _ffi.string(_lib.OBJ_nid2ln(nid))
def digest(self, digest_name: str) -> bytes:
"""
Return the digest of the X509 object.
:param digest_name: The name of the digest algorithm to use.
:type digest_name: :py:class:`str`
:return: The digest of the object, formatted as
:py:const:`b":"`-delimited hex pairs.
:rtype: :py:class:`bytes`
"""
digest = _lib.EVP_get_digestbyname(_byte_string(digest_name))
if digest == _ffi.NULL:
raise ValueError("No such digest method")
result_buffer = _ffi.new("unsigned char[]", _lib.EVP_MAX_MD_SIZE)
result_length = _ffi.new("unsigned int[]", 1)
result_length[0] = len(result_buffer)
digest_result = _lib.X509_digest(
self._x509, digest, result_buffer, result_length
)
_openssl_assert(digest_result == 1)
return b":".join(
[
b16encode(ch).upper()
for ch in _ffi.buffer(result_buffer, result_length[0])
]
)
def subject_name_hash(self) -> int:
"""
Return the hash of the X509 subject.
:return: The hash of the subject.
:rtype: :py:class:`int`
"""
return _lib.X509_subject_name_hash(self._x509)
@deprecated(
"X509.set_serial_number is deprecated. You should use "
"cryptography's CertificateBuilder instead."
)
def set_serial_number(self, serial: int) -> None:
"""
Set the serial number of the certificate.
:param serial: The new serial number.
:type serial: :py:class:`int`
:return: :py:data`None`
"""
if not isinstance(serial, int):
raise TypeError("serial must be an integer")
hex_serial = hex(serial)[2:]
hex_serial_bytes = hex_serial.encode("ascii")
bignum_serial = _ffi.new("BIGNUM**")
# BN_hex2bn stores the result in &bignum.
result = _lib.BN_hex2bn(bignum_serial, hex_serial_bytes)
_openssl_assert(result != _ffi.NULL)
asn1_serial = _lib.BN_to_ASN1_INTEGER(bignum_serial[0], _ffi.NULL)
_lib.BN_free(bignum_serial[0])
_openssl_assert(asn1_serial != _ffi.NULL)
asn1_serial = _ffi.gc(asn1_serial, _lib.ASN1_INTEGER_free)
set_result = _lib.X509_set_serialNumber(self._x509, asn1_serial)
_openssl_assert(set_result == 1)
def get_serial_number(self) -> int:
"""
Return the serial number of this certificate.
:return: The serial number.
:rtype: int
"""
asn1_serial = _lib.X509_get_serialNumber(self._x509)
bignum_serial = _lib.ASN1_INTEGER_to_BN(asn1_serial, _ffi.NULL)
try:
hex_serial = _lib.BN_bn2hex(bignum_serial)
try:
hexstring_serial = _ffi.string(hex_serial)
serial = int(hexstring_serial, 16)
return serial
finally:
_lib.OPENSSL_free(hex_serial)
finally:
_lib.BN_free(bignum_serial)
@deprecated(
"X509.gmtime_adj_notAfter is deprecated. You should use "
"cryptography's CertificateBuilder instead."
)
def gmtime_adj_notAfter(self, amount: int) -> None:
"""
Adjust the time stamp on which the certificate stops being valid.
:param int amount: The number of seconds by which to adjust the
timestamp.
:return: ``None``
"""
if not isinstance(amount, int):
raise TypeError("amount must be an integer")
notAfter = _lib.X509_getm_notAfter(self._x509)
_lib.X509_gmtime_adj(notAfter, amount)
@deprecated(
"X509.gmtime_adj_notBefore is deprecated. You should use "
"cryptography's CertificateBuilder instead."
)
def gmtime_adj_notBefore(self, amount: int) -> None:
"""
Adjust the timestamp on which the certificate starts being valid.
:param amount: The number of seconds by which to adjust the timestamp.
:return: ``None``
"""
if not isinstance(amount, int):
raise TypeError("amount must be an integer")
notBefore = _lib.X509_getm_notBefore(self._x509)
_lib.X509_gmtime_adj(notBefore, amount)
def has_expired(self) -> bool:
"""
Check whether the certificate has expired.
:return: ``True`` if the certificate has expired, ``False`` otherwise.
:rtype: bool
"""
time_bytes = self.get_notAfter()
if time_bytes is None:
raise ValueError("Unable to determine notAfter")
time_string = time_bytes.decode("utf-8")
not_after = datetime.datetime.strptime(time_string, "%Y%m%d%H%M%SZ")
UTC = datetime.timezone.utc
utcnow = datetime.datetime.now(UTC).replace(tzinfo=None)
return not_after < utcnow
def _get_boundary_time(self, which: Any) -> bytes | None:
return _get_asn1_time(which(self._x509))
def get_notBefore(self) -> bytes | None:
"""
Get the timestamp at which the certificate starts being valid.
The timestamp is formatted as an ASN.1 TIME::
YYYYMMDDhhmmssZ
:return: A timestamp string, or ``None`` if there is none.
:rtype: bytes or NoneType
"""
return self._get_boundary_time(_lib.X509_getm_notBefore)
def _set_boundary_time(
self, which: Callable[..., Any], when: bytes
) -> None:
return _set_asn1_time(which(self._x509), when)
@deprecated(
"X509.set_notBefore is deprecated. You should use "
"cryptography's CertificateBuilder instead."
)
def set_notBefore(self, when: bytes) -> None:
"""
Set the timestamp at which the certificate starts being valid.
The timestamp is formatted as an ASN.1 TIME::
YYYYMMDDhhmmssZ
:param bytes when: A timestamp string.
:return: ``None``
"""
return self._set_boundary_time(_lib.X509_getm_notBefore, when)
def get_notAfter(self) -> bytes | None:
"""
Get the timestamp at which the certificate stops being valid.
The timestamp is formatted as an ASN.1 TIME::
YYYYMMDDhhmmssZ
:return: A timestamp string, or ``None`` if there is none.
:rtype: bytes or NoneType
"""
return self._get_boundary_time(_lib.X509_getm_notAfter)
@deprecated(
"X509.set_notAfter is deprecated. You should use "
"cryptography's CertificateBuilder instead."
)
def set_notAfter(self, when: bytes) -> None:
"""
Set the timestamp at which the certificate stops being valid.
The timestamp is formatted as an ASN.1 TIME::
YYYYMMDDhhmmssZ
:param bytes when: A timestamp string.
:return: ``None``
"""
return self._set_boundary_time(_lib.X509_getm_notAfter, when)
def _get_name(self, which: Any) -> X509Name:
# Bypass X509Name.__new__, which warns that X509Name is deprecated;
# callers that should warn are decorated individually.
name = object.__new__(X509Name)
name._name = which(self._x509)
_openssl_assert(name._name != _ffi.NULL)
# The name is owned by the X509 structure. As long as the X509Name
# Python object is alive, keep the X509 Python object alive.
name._owner = self
return name
def _set_name(self, which: Any, name: X509Name) -> None:
if not isinstance(name, X509Name):
raise TypeError("name must be an X509Name")
set_result = which(self._x509, name._name)
_openssl_assert(set_result == 1)
@deprecated(
"X509.get_issuer is deprecated. You should use "
"cryptography's X.509 APIs instead."
)
def get_issuer(self) -> X509Name:
"""
Return the issuer of this certificate.
This creates a new :class:`X509Name` that wraps the underlying issuer
name field on the certificate. Modifying it will modify the underlying
certificate, and will have the effect of modifying any other
:class:`X509Name` that refers to this issuer.
:return: The issuer of this certificate.
:rtype: :class:`X509Name`
"""
name = self._get_name(_lib.X509_get_issuer_name)
self._issuer_invalidator.add(name)
return name
@deprecated(
"X509.set_issuer is deprecated. You should use "
"cryptography's CertificateBuilder instead."
)
def set_issuer(self, issuer: X509Name) -> None:
"""
Set the issuer of this certificate.
:param issuer: The issuer.
:type issuer: :py:class:`X509Name`
:return: ``None``
"""
self._set_name(_lib.X509_set_issuer_name, issuer)
self._issuer_invalidator.clear()
@deprecated(
"X509.get_subject is deprecated. You should use "
"cryptography's X.509 APIs instead."
)
def get_subject(self) -> X509Name:
"""
Return the subject of this certificate.
This creates a new :class:`X509Name` that wraps the underlying subject
name field on the certificate. Modifying it will modify the underlying
certificate, and will have the effect of modifying any other
:class:`X509Name` that refers to this subject.
:return: The subject of this certificate.
:rtype: :class:`X509Name`
"""
name = self._get_name(_lib.X509_get_subject_name)
self._subject_invalidator.add(name)
return name
@deprecated(
"X509.set_subject is deprecated. You should use "
"cryptography's CertificateBuilder instead."
)
def set_subject(self, subject: X509Name) -> None:
"""
Set the subject of this certificate.
:param subject: The subject.
:type subject: :py:class:`X509Name`
:return: ``None``
"""
self._set_name(_lib.X509_set_subject_name, subject)
self._subject_invalidator.clear()
def get_extension_count(self) -> int:
"""
Get the number of extensions on this certificate.
:return: The number of extensions.
:rtype: :py:class:`int`
.. versionadded:: 0.12
"""
return _lib.X509_get_ext_count(self._x509)
class X509StoreFlags:
"""
Flags for X509 verification, used to change the behavior of
:class:`X509Store`.
See `OpenSSL Verification Flags`_ for details.
.. _OpenSSL Verification Flags:
https://www.openssl.org/docs/manmaster/man3/X509_VERIFY_PARAM_set_flags.html
"""
CRL_CHECK: int = _lib.X509_V_FLAG_CRL_CHECK
CRL_CHECK_ALL: int = _lib.X509_V_FLAG_CRL_CHECK_ALL
IGNORE_CRITICAL: int = _lib.X509_V_FLAG_IGNORE_CRITICAL
X509_STRICT: int = _lib.X509_V_FLAG_X509_STRICT
ALLOW_PROXY_CERTS: int = _lib.X509_V_FLAG_ALLOW_PROXY_CERTS
POLICY_CHECK: int = _lib.X509_V_FLAG_POLICY_CHECK
EXPLICIT_POLICY: int = _lib.X509_V_FLAG_EXPLICIT_POLICY
INHIBIT_MAP: int = _lib.X509_V_FLAG_INHIBIT_MAP
CHECK_SS_SIGNATURE: int = _lib.X509_V_FLAG_CHECK_SS_SIGNATURE
PARTIAL_CHAIN: int = _lib.X509_V_FLAG_PARTIAL_CHAIN
class X509Store:
"""
An X.509 store.
An X.509 store is used to describe a context in which to verify a
certificate. A description of a context may include a set of certificates
to trust, a set of certificate revocation lists, verification flags and
more.
An X.509 store, being only a description, cannot be used by itself to
verify a certificate. To carry out the actual verification process, see
:class:`X509StoreContext`.
"""
def __init__(self) -> None:
store = _lib.X509_STORE_new()
self._store = _ffi.gc(store, _lib.X509_STORE_free)
def add_cert(self, cert: X509) -> None:
"""
Adds a trusted certificate to this store.
Adding a certificate with this method adds this certificate as a
*trusted* certificate.
:param X509 cert: The certificate to add to this store.
:raises TypeError: If the certificate is not an :class:`X509`.
:raises OpenSSL.crypto.Error: If OpenSSL was unhappy with your
certificate.
:return: ``None`` if the certificate was added successfully.
"""
if not isinstance(cert, X509):
raise TypeError()
res = _lib.X509_STORE_add_cert(self._store, cert._x509)
_openssl_assert(res == 1)
def add_crl(self, crl: x509.CertificateRevocationList) -> None:
"""
Add a certificate revocation list to this store.
The certificate revocation lists added to a store will only be used if
the associated flags are configured to check certificate revocation
lists.
.. versionadded:: 16.1.0
:param crl: The certificate revocation list to add to this store.
:type crl: ``cryptography.x509.CertificateRevocationList``
:return: ``None`` if the certificate revocation list was added
successfully.
"""
if isinstance(crl, x509.CertificateRevocationList):
from cryptography.hazmat.primitives.serialization import Encoding
bio = _new_mem_buf(crl.public_bytes(Encoding.DER))
openssl_crl = _lib.d2i_X509_CRL_bio(bio, _ffi.NULL)
_openssl_assert(openssl_crl != _ffi.NULL)
crl = _ffi.gc(openssl_crl, _lib.X509_CRL_free)
else:
raise TypeError(
"CRL must be of type "
"cryptography.x509.CertificateRevocationList"
)
_openssl_assert(_lib.X509_STORE_add_crl(self._store, crl) != 0)
def set_flags(self, flags: int) -> None:
"""
Set verification flags to this store.
Verification flags can be combined by oring them together.
.. note::
Setting a verification flag sometimes requires clients to add
additional information to the store, otherwise a suitable error will
be raised.
For example, in setting flags to enable CRL checking a
suitable CRL must be added to the store otherwise an error will be
raised.
.. versionadded:: 16.1.0
:param int flags: The verification flags to set on this store.
See :class:`X509StoreFlags` for available constants.
:return: ``None`` if the verification flags were successfully set.
"""
_openssl_assert(_lib.X509_STORE_set_flags(self._store, flags) != 0)
def set_time(self, vfy_time: datetime.datetime) -> None:
"""
Set the time against which the certificates are verified.
Normally the current time is used.
.. note::
For example, you can determine if a certificate was valid at a given
time.
.. versionadded:: 17.0.0
:param datetime vfy_time: The verification time to set on this store.
:return: ``None`` if the verification time was successfully set.
"""
param = _lib.X509_VERIFY_PARAM_new()
param = _ffi.gc(param, _lib.X509_VERIFY_PARAM_free)
_lib.X509_VERIFY_PARAM_set_time(
param, calendar.timegm(vfy_time.timetuple())
)
_openssl_assert(_lib.X509_STORE_set1_param(self._store, param) != 0)
def load_locations(
self,
cafile: StrOrBytesPath | None,
capath: StrOrBytesPath | None = None,
) -> None:
"""
Let X509Store know where we can find trusted certificates for the
certificate chain. Note that the certificates have to be in PEM
format.
If *capath* is passed, it must be a directory prepared using the
``c_rehash`` tool included with OpenSSL. Either, but not both, of
*cafile* or *capath* may be ``None``.
.. note::
Both *cafile* and *capath* may be set simultaneously.
Call this method multiple times to add more than one location.
For example, CA certificates, and certificate revocation list bundles
may be passed in *cafile* in subsequent calls to this method.
.. versionadded:: 20.0
:param cafile: In which file we can find the certificates (``bytes`` or
``unicode``).
:param capath: In which directory we can find the certificates
(``bytes`` or ``unicode``).
:return: ``None`` if the locations were set successfully.
:raises OpenSSL.crypto.Error: If both *cafile* and *capath* is ``None``
or the locations could not be set for any reason.
"""
if cafile is None:
cafile = _ffi.NULL
else:
cafile = _path_bytes(cafile)
if capath is None:
capath = _ffi.NULL
else:
capath = _path_bytes(capath)
load_result = _lib.X509_STORE_load_locations(
self._store, cafile, capath
)
if not load_result:
_raise_current_error()
class X509StoreContextError(Exception):
"""
An exception raised when an error occurred while verifying a certificate
using `OpenSSL.X509StoreContext.verify_certificate`.
:ivar certificate: The certificate which caused verificate failure.
:type certificate: :class:`X509`
"""
def __init__(
self, message: str, errors: list[Any], certificate: X509
) -> None:
super().__init__(message)
self.errors = errors
self.certificate = certificate
class X509StoreContext:
"""
An X.509 store context.
An X.509 store context is used to carry out the actual verification process
of a certificate in a described context. For describing such a context, see
:class:`X509Store`.
:param X509Store store: The certificates which will be trusted for the
purposes of any verifications.
:param X509 certificate: The certificate to be verified.
:param chain: List of untrusted certificates that may be used for building
the certificate chain. May be ``None``.
:type chain: :class:`list` of :class:`X509`
"""
def __init__(
self,
store: X509Store,
certificate: X509,
chain: Sequence[X509] | None = None,
) -> None:
self._store = store
self._cert = certificate
self._chain = self._build_certificate_stack(chain)
@staticmethod
def _build_certificate_stack(
certificates: Sequence[X509] | None,
) -> None:
def cleanup(s: Any) -> None:
# Equivalent to sk_X509_pop_free, but we don't
# currently have a CFFI binding for that available
for i in range(_lib.sk_X509_num(s)):
x = _lib.sk_X509_value(s, i)
_lib.X509_free(x)
_lib.sk_X509_free(s)
if certificates is None or len(certificates) == 0:
return _ffi.NULL
stack = _lib.sk_X509_new_null()
_openssl_assert(stack != _ffi.NULL)
stack = _ffi.gc(stack, cleanup)
for cert in certificates:
if not isinstance(cert, X509):
raise TypeError("One of the elements is not an X509 instance")
_openssl_assert(_lib.X509_up_ref(cert._x509) > 0)
if _lib.sk_X509_push(stack, cert._x509) <= 0:
_lib.X509_free(cert._x509)
_raise_current_error()
return stack
@staticmethod
def _exception_from_context(store_ctx: Any) -> X509StoreContextError:
"""
Convert an OpenSSL native context error failure into a Python
exception.
When a call to native OpenSSL X509_verify_cert fails, additional
information about the failure can be obtained from the store context.
"""
message = _ffi.string(
_lib.X509_verify_cert_error_string(
_lib.X509_STORE_CTX_get_error(store_ctx)
)
).decode("utf-8")
errors = [
_lib.X509_STORE_CTX_get_error(store_ctx),
_lib.X509_STORE_CTX_get_error_depth(store_ctx),
message,
]
# A context error should always be associated with a certificate, so we
# expect this call to never return :class:`None`.
_x509 = _lib.X509_STORE_CTX_get_current_cert(store_ctx)
_cert = _lib.X509_dup(_x509)
pycert = X509._from_raw_x509_ptr(_cert)
return X509StoreContextError(message, errors, pycert)
def _verify_certificate(self) -> Any:
"""
Verifies the certificate and runs an X509_STORE_CTX containing the
results.
:raises X509StoreContextError: If an error occurred when validating a
certificate in the context. Sets ``certificate`` attribute to
indicate which certificate caused the error.
"""
store_ctx = _lib.X509_STORE_CTX_new()
_openssl_assert(store_ctx != _ffi.NULL)
store_ctx = _ffi.gc(store_ctx, _lib.X509_STORE_CTX_free)
ret = _lib.X509_STORE_CTX_init(
store_ctx, self._store._store, self._cert._x509, self._chain
)
_openssl_assert(ret == 1)
ret = _lib.X509_verify_cert(store_ctx)
if ret <= 0:
raise self._exception_from_context(store_ctx)
return store_ctx
def set_store(self, store: X509Store) -> None:
"""
Set the context's X.509 store.
.. versionadded:: 0.15
:param X509Store store: The store description which will be used for
the purposes of any *future* verifications.
"""
self._store = store
def verify_certificate(self) -> None:
"""
Verify a certificate in a context.
.. versionadded:: 0.15
:raises X509StoreContextError: If an error occurred when validating a
certificate in the context. Sets ``certificate`` attribute to
indicate which certificate caused the error.
"""
self._verify_certificate()
def get_verified_chain(self) -> list[X509]:
"""
Verify a certificate in a context and return the complete validated
chain.
:raises X509StoreContextError: If an error occurred when validating a
certificate in the context. Sets ``certificate`` attribute to
indicate which certificate caused the error.
.. versionadded:: 20.0
"""
store_ctx = self._verify_certificate()
# Note: X509_STORE_CTX_get1_chain returns a deep copy of the chain.
cert_stack = _lib.X509_STORE_CTX_get1_chain(store_ctx)
_openssl_assert(cert_stack != _ffi.NULL)
result = []
for i in range(_lib.sk_X509_num(cert_stack)):
cert = _lib.sk_X509_value(cert_stack, i)
_openssl_assert(cert != _ffi.NULL)
pycert = X509._from_raw_x509_ptr(cert)
result.append(pycert)
# Free the stack but not the members which are freed by the X509 class.
_lib.sk_X509_free(cert_stack)
return result
def load_certificate(type: int, buffer: bytes) -> X509:
"""
Load a certificate (X509) from the string *buffer* encoded with the
type *type*.
:param type: The file type (one of FILETYPE_PEM, FILETYPE_ASN1)
:param bytes buffer: The buffer the certificate is stored in
:return: The X509 object
"""
if isinstance(buffer, str):
buffer = buffer.encode("ascii")
bio = _new_mem_buf(buffer)
if type == FILETYPE_PEM:
x509 = _lib.PEM_read_bio_X509(bio, _ffi.NULL, _ffi.NULL, _ffi.NULL)
elif type == FILETYPE_ASN1:
x509 = _lib.d2i_X509_bio(bio, _ffi.NULL)
else:
raise ValueError("type argument must be FILETYPE_PEM or FILETYPE_ASN1")
if x509 == _ffi.NULL:
_raise_current_error()
return X509._from_raw_x509_ptr(x509)
def dump_certificate(type: int, cert: X509) -> bytes:
"""
Dump the certificate *cert* into a buffer string encoded with the type
*type*.
:param type: The file type (one of FILETYPE_PEM, FILETYPE_ASN1, or
FILETYPE_TEXT)
:param cert: The certificate to dump
:return: The buffer with the dumped certificate in
"""
bio = _new_mem_buf()
if type == FILETYPE_PEM:
result_code = _lib.PEM_write_bio_X509(bio, cert._x509)
elif type == FILETYPE_ASN1:
result_code = _lib.i2d_X509_bio(bio, cert._x509)
elif type == FILETYPE_TEXT:
result_code = _lib.X509_print_ex(bio, cert._x509, 0, 0)
else:
raise ValueError(
"type argument must be FILETYPE_PEM, FILETYPE_ASN1, or "
"FILETYPE_TEXT"
)
_openssl_assert(result_code == 1)
return _bio_to_string(bio)
def dump_publickey(type: int, pkey: PKey) -> bytes:
"""
Dump a public key to a buffer.
:param type: The file type (one of :data:`FILETYPE_PEM` or
:data:`FILETYPE_ASN1`).
:param PKey pkey: The public key to dump
:return: The buffer with the dumped key in it.
:rtype: bytes
"""
bio = _new_mem_buf()
if type == FILETYPE_PEM:
write_bio = _lib.PEM_write_bio_PUBKEY
elif type == FILETYPE_ASN1:
write_bio = _lib.i2d_PUBKEY_bio
else:
raise ValueError("type argument must be FILETYPE_PEM or FILETYPE_ASN1")
result_code = write_bio(bio, pkey._pkey)
if result_code != 1: # pragma: no cover
_raise_current_error()
return _bio_to_string(bio)
def dump_privatekey(
type: int,
pkey: PKey,
cipher: str | None = None,
passphrase: PassphraseCallableT | None = None,
) -> bytes:
"""
Dump the private key *pkey* into a buffer string encoded with the type
*type*. Optionally (if *type* is :const:`FILETYPE_PEM`) encrypting it
using *cipher* and *passphrase*.
:param type: The file type (one of :const:`FILETYPE_PEM`,
:const:`FILETYPE_ASN1`, or :const:`FILETYPE_TEXT`)
:param PKey pkey: The PKey to dump
:param cipher: (optional) if encrypted PEM format, the cipher to use
:param passphrase: (optional) if encrypted PEM format, this can be either
the passphrase to use, or a callback for providing the passphrase.
:return: The buffer with the dumped key in
:rtype: bytes
.. deprecated:: 26.3.0
Use the serialization APIs on ``cryptography`` private key types
instead.
"""
bio = _new_mem_buf()
if not isinstance(pkey, PKey):
raise TypeError("pkey must be a PKey")
if cipher is not None:
if passphrase is None:
raise TypeError(
"if a value is given for cipher "
"one must also be given for passphrase"
)
cipher_obj = _lib.EVP_get_cipherbyname(_byte_string(cipher))
if cipher_obj == _ffi.NULL:
raise ValueError("Invalid cipher name")
else:
cipher_obj = _ffi.NULL
helper = _PassphraseHelper(type, passphrase)
if type == FILETYPE_PEM:
result_code = _lib.PEM_write_bio_PrivateKey(
bio,
pkey._pkey,
cipher_obj,
_ffi.NULL,
0,
helper.callback,
helper.callback_args,
)
helper.raise_if_problem()
elif type == FILETYPE_ASN1:
result_code = _lib.i2d_PrivateKey_bio(bio, pkey._pkey)
elif type == FILETYPE_TEXT:
if _lib.EVP_PKEY_id(pkey._pkey) != _lib.EVP_PKEY_RSA:
raise TypeError("Only RSA keys are supported for FILETYPE_TEXT")
rsa = _ffi.gc(_lib.EVP_PKEY_get1_RSA(pkey._pkey), _lib.RSA_free)
result_code = _lib.RSA_print(bio, rsa, 0)
else:
raise ValueError(
"type argument must be FILETYPE_PEM, FILETYPE_ASN1, or "
"FILETYPE_TEXT"
)
_openssl_assert(result_code != 0)
return _bio_to_string(bio)
_dump_privatekey_internal = dump_privatekey
utils.deprecated(
dump_privatekey,
__name__,
(
"dump_privatekey is deprecated. You should use the APIs in "
"cryptography."
),
DeprecationWarning,
name="dump_privatekey",
)
class _PassphraseHelper:
def __init__(
self,
type: int,
passphrase: PassphraseCallableT | None,
more_args: bool = False,
truncate: bool = False,
) -> None:
if type != FILETYPE_PEM and passphrase is not None:
raise ValueError(
"only FILETYPE_PEM key format supports encryption"
)
self._passphrase = passphrase
self._more_args = more_args
self._truncate = truncate
self._problems: list[Exception] = []
@property
def callback(self) -> Any:
if self._passphrase is None:
return _ffi.NULL
elif isinstance(self._passphrase, bytes) or callable(self._passphrase):
return _ffi.callback("pem_password_cb", self._read_passphrase)
else:
raise TypeError(
"Last argument must be a byte string or a callable."
)
@property
def callback_args(self) -> Any:
if self._passphrase is None:
return _ffi.NULL
elif isinstance(self._passphrase, bytes) or callable(self._passphrase):
return _ffi.NULL
else:
raise TypeError(
"Last argument must be a byte string or a callable."
)
def raise_if_problem(self, exceptionType: type[Exception] = Error) -> None:
if self._problems:
# Flush the OpenSSL error queue
try:
_exception_from_error_queue(exceptionType)
except exceptionType:
pass
raise self._problems.pop(0)
def _read_passphrase(
self, buf: Any, size: int, rwflag: Any, userdata: Any
) -> int:
try:
if callable(self._passphrase):
if self._more_args:
result = self._passphrase(size, rwflag, userdata)
else:
result = self._passphrase(rwflag)
else:
assert self._passphrase is not None
result = self._passphrase
if not isinstance(result, bytes):
raise ValueError("Bytes expected")
if len(result) > size:
if self._truncate:
result = result[:size]
else:
raise ValueError(
"passphrase returned by callback is too long"
)
for i in range(len(result)):
buf[i] = result[i : i + 1]
return len(result)
except Exception as e:
self._problems.append(e)
return 0
def load_publickey(type: int, buffer: str | bytes) -> PKey:
"""
Load a public key from a buffer.
:param type: The file type (one of :data:`FILETYPE_PEM`,
:data:`FILETYPE_ASN1`).
:param buffer: The buffer the key is stored in.
:type buffer: A Python string object, either unicode or bytestring.
:return: The PKey object.
:rtype: :class:`PKey`
"""
if isinstance(buffer, str):
buffer = buffer.encode("ascii")
bio = _new_mem_buf(buffer)
if type == FILETYPE_PEM:
evp_pkey = _lib.PEM_read_bio_PUBKEY(
bio, _ffi.NULL, _ffi.NULL, _ffi.NULL
)
elif type == FILETYPE_ASN1:
evp_pkey = _lib.d2i_PUBKEY_bio(bio, _ffi.NULL)
else:
raise ValueError("type argument must be FILETYPE_PEM or FILETYPE_ASN1")
if evp_pkey == _ffi.NULL:
_raise_current_error()
pkey = PKey.__new__(PKey)
pkey._pkey = _ffi.gc(evp_pkey, _lib.EVP_PKEY_free)
pkey._only_public = True
return pkey
def load_privatekey(
type: int,
buffer: str | bytes,
passphrase: PassphraseCallableT | None = None,
) -> PKey:
"""
Load a private key (PKey) from the string *buffer* encoded with the type
*type*.
:param type: The file type (one of FILETYPE_PEM, FILETYPE_ASN1)
:param buffer: The buffer the key is stored in
:param passphrase: (optional) if encrypted PEM format, this can be
either the passphrase to use, or a callback for
providing the passphrase.
:return: The PKey object
"""
if isinstance(buffer, str):
buffer = buffer.encode("ascii")
bio = _new_mem_buf(buffer)
helper = _PassphraseHelper(type, passphrase)
if type == FILETYPE_PEM:
evp_pkey = _lib.PEM_read_bio_PrivateKey(
bio, _ffi.NULL, helper.callback, helper.callback_args
)
helper.raise_if_problem()
elif type == FILETYPE_ASN1:
evp_pkey = _lib.d2i_PrivateKey_bio(bio, _ffi.NULL)
else:
raise ValueError("type argument must be FILETYPE_PEM or FILETYPE_ASN1")
if evp_pkey == _ffi.NULL:
_raise_current_error()
pkey = PKey.__new__(PKey)
pkey._pkey = _ffi.gc(evp_pkey, _lib.EVP_PKEY_free)
return pkey