diff options
| author | Paul Kehrer <paul.l.kehrer@gmail.com> | 2021-09-28 20:09:08 +0800 |
|---|---|---|
| committer | GitHub <noreply@github.com> | 2021-09-28 08:09:08 -0400 |
| commit | 792fd2242d4b58811e33c43218fba80c3f93233d (patch) | |
| tree | 08e7df97e9838a7ce5f63b132a510dd2309c064d /src/cryptography | |
| parent | e415c65206bf6e0eb2569865dd681a10ae995fe8 (diff) | |
| download | cryptography-792fd2242d4b58811e33c43218fba80c3f93233d.tar.gz | |
Parse CSRs in pure rust (#6312)
* Parse CSRs in pure rust
* cargo fmt
* various improvements
* remove more unneeded things
* remove more things
* fix most coverage issues
* refactor and a new test using a fresh vector
* update TODOs
* address review feedback
* simplify
* implicit required now supported
* review feedback
* try to satisfy the coverage gods
* simplify more
* add a new test
Diffstat (limited to 'src/cryptography')
| -rw-r--r-- | src/cryptography/hazmat/backends/interfaces.py | 12 | ||||
| -rw-r--r-- | src/cryptography/hazmat/backends/openssl/backend.py | 64 | ||||
| -rw-r--r-- | src/cryptography/hazmat/backends/openssl/decode_asn1.py | 117 | ||||
| -rw-r--r-- | src/cryptography/hazmat/backends/openssl/x509.py | 178 | ||||
| -rw-r--r-- | src/cryptography/hazmat/bindings/_rust/x509.pyi | 3 | ||||
| -rw-r--r-- | src/cryptography/x509/base.py | 14 |
6 files changed, 53 insertions, 335 deletions
diff --git a/src/cryptography/hazmat/backends/interfaces.py b/src/cryptography/hazmat/backends/interfaces.py index db708d4cd..d3d7396e9 100644 --- a/src/cryptography/hazmat/backends/interfaces.py +++ b/src/cryptography/hazmat/backends/interfaces.py @@ -281,18 +281,6 @@ class DERSerializationBackend(metaclass=abc.ABCMeta): class X509Backend(metaclass=abc.ABCMeta): @abc.abstractmethod - def load_der_x509_csr(self, data: bytes) -> "CertificateSigningRequest": - """ - Load an X.509 CSR from DER encoded data. - """ - - @abc.abstractmethod - def load_pem_x509_csr(self, data: bytes) -> "CertificateSigningRequest": - """ - Load an X.509 CSR from PEM encoded data. - """ - - @abc.abstractmethod def create_x509_csr( self, builder: "CertificateSigningRequestBuilder", diff --git a/src/cryptography/hazmat/backends/openssl/backend.py b/src/cryptography/hazmat/backends/openssl/backend.py index 53d6cf555..8901ed76f 100644 --- a/src/cryptography/hazmat/backends/openssl/backend.py +++ b/src/cryptography/hazmat/backends/openssl/backend.py @@ -18,7 +18,6 @@ from cryptography.hazmat.backends.openssl.ciphers import _CipherContext from cryptography.hazmat.backends.openssl.cmac import _CMACContext from cryptography.hazmat.backends.openssl.decode_asn1 import ( _CRL_ENTRY_REASON_ENUM_TO_CODE, - _X509ExtensionParser, ) from cryptography.hazmat.backends.openssl.dh import ( _DHParameters, @@ -74,7 +73,6 @@ from cryptography.hazmat.backends.openssl.x448 import ( _X448PublicKey, ) from cryptography.hazmat.backends.openssl.x509 import ( - _CertificateSigningRequest, _RawRevokedCertificate, ) from cryptography.hazmat.bindings._rust import ( @@ -192,7 +190,6 @@ class Backend(BackendInterface): self._cipher_registry = {} self._register_default_ciphers() - self._register_x509_ext_parsers() self._register_x509_encoders() if self._fips_enabled and self._lib.CRYPTOGRAPHY_NEEDS_OSRANDOM_ENGINE: warnings.warn( @@ -418,14 +415,6 @@ class Backend(BackendInterface): SM4, mode_cls, GetCipherByName("sm4-{mode.name}") ) - def _register_x509_ext_parsers(self): - self._csr_extension_parser = _X509ExtensionParser( - self, - ext_count=self._lib.sk_X509_EXTENSION_num, - get_ext=self._lib.sk_X509_EXTENSION_value, - rust_callback=rust_x509.parse_csr_extension, - ) - def _register_x509_encoders(self): self._extension_encode_handlers = _EXTENSION_ENCODE_HANDLERS.copy() self._crl_extension_encode_handlers = ( @@ -893,7 +882,7 @@ class Backend(BackendInterface): builder: x509.CertificateSigningRequestBuilder, private_key: PRIVATE_KEY_TYPES, algorithm: typing.Optional[hashes.HashAlgorithm], - ) -> _CertificateSigningRequest: + ) -> x509.CertificateSigningRequest: if not isinstance(builder, x509.CertificateSigningRequestBuilder): raise TypeError("Builder type mismatch.") self._x509_check_signature_params(private_key, algorithm) @@ -967,7 +956,7 @@ class Backend(BackendInterface): errors = self._consume_errors_with_text() raise ValueError("Signing failed", errors) - return _CertificateSigningRequest(self, x509_req) + return self._ossl2csr(x509_req) def create_x509_certificate( self, @@ -1372,6 +1361,22 @@ class Backend(BackendInterface): self.openssl_assert(res == 1) return rust_x509.load_der_x509_certificate(self._read_mem_bio(bio)) + def _csr2ossl(self, csr: x509.CertificateSigningRequest) -> typing.Any: + data = csr.public_bytes(serialization.Encoding.DER) + mem_bio = self._bytes_to_bio(data) + x509_req = self._lib.d2i_X509_REQ_bio(mem_bio.bio, self._ffi.NULL) + self.openssl_assert(x509_req != self._ffi.NULL) + x509_req = self._ffi.gc(x509_req, self._lib.X509_REQ_free) + return x509_req + + def _ossl2csr( + self, x509_req: typing.Any + ) -> x509.CertificateSigningRequest: + bio = self._create_mem_bio_gc() + res = self._lib.i2d_X509_REQ_bio(bio, x509_req) + self.openssl_assert(res == 1) + return rust_x509.load_der_x509_csr(self._read_mem_bio(bio)) + def _crl_is_signature_valid( self, crl: x509.CertificateRevocationList, public_key: PUBLIC_KEY_TYPES ) -> bool: @@ -1401,31 +1406,20 @@ class Backend(BackendInterface): return True - def load_pem_x509_csr(self, data: bytes) -> _CertificateSigningRequest: - mem_bio = self._bytes_to_bio(data) - x509_req = self._lib.PEM_read_bio_X509_REQ( - mem_bio.bio, self._ffi.NULL, self._ffi.NULL, self._ffi.NULL - ) - if x509_req == self._ffi.NULL: - self._consume_errors() - raise ValueError( - "Unable to load request. See https://cryptography.io/en/" - "latest/faq.html#why-can-t-i-import-my-pem-file for more" - " details." - ) - - x509_req = self._ffi.gc(x509_req, self._lib.X509_REQ_free) - return _CertificateSigningRequest(self, x509_req) + def _csr_is_signature_valid( + self, csr: x509.CertificateSigningRequest + ) -> bool: + x509_req = self._csr2ossl(csr) + pkey = self._lib.X509_REQ_get_pubkey(x509_req) + self.openssl_assert(pkey != self._ffi.NULL) + pkey = self._ffi.gc(pkey, self._lib.EVP_PKEY_free) + res = self._lib.X509_REQ_verify(x509_req, pkey) - def load_der_x509_csr(self, data: bytes) -> _CertificateSigningRequest: - mem_bio = self._bytes_to_bio(data) - x509_req = self._lib.d2i_X509_REQ_bio(mem_bio.bio, self._ffi.NULL) - if x509_req == self._ffi.NULL: + if res != 1: self._consume_errors() - raise ValueError("Unable to load request") + return False - x509_req = self._ffi.gc(x509_req, self._lib.X509_REQ_free) - return _CertificateSigningRequest(self, x509_req) + return True def _load_key(self, openssl_read_func, convert_func, data, password): mem_bio = self._bytes_to_bio(data) diff --git a/src/cryptography/hazmat/backends/openssl/decode_asn1.py b/src/cryptography/hazmat/backends/openssl/decode_asn1.py index 8ce49a039..8d97a13bf 100644 --- a/src/cryptography/hazmat/backends/openssl/decode_asn1.py +++ b/src/cryptography/hazmat/backends/openssl/decode_asn1.py @@ -3,109 +3,7 @@ # for complete details. -import typing - from cryptography import x509 -from cryptography.x509.name import _ASN1_TYPE_TO_ENUM - - -def _obj2txt(backend, obj): - # Set to 80 on the recommendation of - # https://www.openssl.org/docs/crypto/OBJ_nid2ln.html#return_values - # - # But OIDs longer than this occur in real life (e.g. Active - # Directory makes some very long OIDs). So we need to detect - # and properly handle the case where the default buffer is not - # big enough. - # - buf_len = 80 - buf = backend._ffi.new("char[]", buf_len) - - # 'res' is the number of bytes that *would* be written if the - # buffer is large enough. If 'res' > buf_len - 1, we need to - # alloc a big-enough buffer and go again. - res = backend._lib.OBJ_obj2txt(buf, buf_len, obj, 1) - if res > buf_len - 1: # account for terminating null byte - buf_len = res + 1 - buf = backend._ffi.new("char[]", buf_len) - res = backend._lib.OBJ_obj2txt(buf, buf_len, obj, 1) - backend.openssl_assert(res > 0) - return backend._ffi.buffer(buf, res)[:].decode() - - -def _decode_x509_name_entry(backend, x509_name_entry): - obj = backend._lib.X509_NAME_ENTRY_get_object(x509_name_entry) - backend.openssl_assert(obj != backend._ffi.NULL) - data = backend._lib.X509_NAME_ENTRY_get_data(x509_name_entry) - backend.openssl_assert(data != backend._ffi.NULL) - value = _asn1_string_to_utf8(backend, data) - oid = _obj2txt(backend, obj) - type = _ASN1_TYPE_TO_ENUM[data.type] - - return x509.NameAttribute(x509.ObjectIdentifier(oid), value, type) - - -def _decode_x509_name(backend, x509_name): - count = backend._lib.X509_NAME_entry_count(x509_name) - attributes = [] - prev_set_id = -1 - for x in range(count): - entry = backend._lib.X509_NAME_get_entry(x509_name, x) - attribute = _decode_x509_name_entry(backend, entry) - set_id = backend._lib.X509_NAME_ENTRY_set(entry) - if set_id != prev_set_id: - attributes.append({attribute}) - else: - # is in the same RDN a previous entry - attributes[-1].add(attribute) - prev_set_id = set_id - - return x509.Name(x509.RelativeDistinguishedName(rdn) for rdn in attributes) - - -class _X509ExtensionParser(object): - def __init__(self, backend, ext_count, get_ext, rust_callback): - self.ext_count = ext_count - self.get_ext = get_ext - self.rust_callback = rust_callback - self._backend = backend - - def parse(self, x509_obj): - extensions: typing.List[x509.Extension[x509.ExtensionType]] = [] - seen_oids = set() - for i in range(self.ext_count(x509_obj)): - ext = self.get_ext(x509_obj, i) - self._backend.openssl_assert(ext != self._backend._ffi.NULL) - crit = self._backend._lib.X509_EXTENSION_get_critical(ext) - critical = crit == 1 - oid = x509.ObjectIdentifier( - _obj2txt( - self._backend, - self._backend._lib.X509_EXTENSION_get_object(ext), - ) - ) - if oid in seen_oids: - raise x509.DuplicateExtension( - "Duplicate {} extension found".format(oid), oid - ) - - # Try to parse this with the rust callback first - oid_ptr = self._backend._lib.X509_EXTENSION_get_object(ext) - oid_der_bytes = self._backend._ffi.buffer( - self._backend._lib.Cryptography_OBJ_get0_data(oid_ptr), - self._backend._lib.Cryptography_OBJ_length(oid_ptr), - )[:] - data = self._backend._lib.X509_EXTENSION_get_data(ext) - data_bytes = _asn1_string_to_bytes(self._backend, data) - ext_obj = self.rust_callback(oid_der_bytes, data_bytes) - if ext_obj is None: - ext_obj = x509.UnrecognizedExtension(oid, data_bytes) - - extensions.append(x509.Extension(oid, critical, ext_obj)) - seen_oids.add(oid) - - return x509.Extensions(extensions) - _DISTPOINT_TYPE_FULLNAME = 0 _DISTPOINT_TYPE_RELATIVENAME = 1 @@ -134,18 +32,3 @@ _CRL_ENTRY_REASON_ENUM_TO_CODE = { x509.ReasonFlags.privilege_withdrawn: 9, x509.ReasonFlags.aa_compromise: 10, } - - -def _asn1_string_to_bytes(backend, asn1_string): - return backend._ffi.buffer(asn1_string.data, asn1_string.length)[:] - - -def _asn1_string_to_utf8(backend, asn1_string) -> str: - buf = backend._ffi.new("unsigned char **") - res = backend._lib.ASN1_STRING_to_UTF8(buf, asn1_string) - backend.openssl_assert(res >= 0) - backend.openssl_assert(buf[0] != backend._ffi.NULL) - buf = backend._ffi.gc( - buf, lambda buffer: backend._lib.OPENSSL_free(buffer[0]) - ) - return backend._ffi.buffer(buf[0], res)[:].decode("utf8") diff --git a/src/cryptography/hazmat/backends/openssl/x509.py b/src/cryptography/hazmat/backends/openssl/x509.py index 52d08a024..874a922c9 100644 --- a/src/cryptography/hazmat/backends/openssl/x509.py +++ b/src/cryptography/hazmat/backends/openssl/x509.py @@ -4,23 +4,9 @@ import datetime -import typing import warnings from cryptography import utils, x509 -from cryptography.exceptions import UnsupportedAlgorithm -from cryptography.hazmat._oid import _SIG_OIDS_TO_HASH -from cryptography.hazmat.backends.openssl.decode_asn1 import ( - _asn1_string_to_bytes, - _decode_x509_name, - _obj2txt, -) -from cryptography.hazmat.backends.openssl.encode_asn1 import ( - _txt2obj_gc, -) -from cryptography.hazmat.primitives import hashes, serialization -from cryptography.x509.base import PUBLIC_KEY_TYPES -from cryptography.x509.name import _ASN1Type # This exists for pyOpenSSL compatibility and SHOULD NOT BE USED @@ -34,159 +20,17 @@ def _Certificate(backend, x509) -> x509.Certificate: # noqa: N802 return backend._ossl2cert(x509) -class _CertificateSigningRequest(x509.CertificateSigningRequest): - def __init__(self, backend, x509_req): - self._backend = backend - self._x509_req = x509_req - - def __eq__(self, other: object) -> bool: - if not isinstance(other, _CertificateSigningRequest): - return NotImplemented - - self_bytes = self.public_bytes(serialization.Encoding.DER) - other_bytes = other.public_bytes(serialization.Encoding.DER) - return self_bytes == other_bytes - - def __ne__(self, other: object) -> bool: - return not self == other - - def __hash__(self) -> int: - return hash(self.public_bytes(serialization.Encoding.DER)) - - def public_key(self) -> PUBLIC_KEY_TYPES: - pkey = self._backend._lib.X509_REQ_get_pubkey(self._x509_req) - self._backend.openssl_assert(pkey != self._backend._ffi.NULL) - pkey = self._backend._ffi.gc(pkey, self._backend._lib.EVP_PKEY_free) - return self._backend._evp_pkey_to_public_key(pkey) - - @property - def subject(self) -> x509.Name: - subject = self._backend._lib.X509_REQ_get_subject_name(self._x509_req) - self._backend.openssl_assert(subject != self._backend._ffi.NULL) - return _decode_x509_name(self._backend, subject) - - @property - def signature_hash_algorithm( - self, - ) -> typing.Optional[hashes.HashAlgorithm]: - oid = self.signature_algorithm_oid - try: - return _SIG_OIDS_TO_HASH[oid] - except KeyError: - raise UnsupportedAlgorithm( - "Signature algorithm OID:{} not recognized".format(oid) - ) - - @property - def signature_algorithm_oid(self) -> x509.ObjectIdentifier: - alg = self._backend._ffi.new("X509_ALGOR **") - self._backend._lib.X509_REQ_get0_signature( - self._x509_req, self._backend._ffi.NULL, alg - ) - self._backend.openssl_assert(alg[0] != self._backend._ffi.NULL) - oid = _obj2txt(self._backend, alg[0].algorithm) - return x509.ObjectIdentifier(oid) - - @utils.cached_property - def extensions(self) -> x509.Extensions: - x509_exts = self._backend._lib.X509_REQ_get_extensions(self._x509_req) - x509_exts = self._backend._ffi.gc( - x509_exts, - lambda x: self._backend._lib.sk_X509_EXTENSION_pop_free( - x, - self._backend._ffi.addressof( - self._backend._lib._original_lib, "X509_EXTENSION_free" - ), - ), - ) - return self._backend._csr_extension_parser.parse(x509_exts) - - def public_bytes(self, encoding: serialization.Encoding) -> bytes: - bio = self._backend._create_mem_bio_gc() - if encoding is serialization.Encoding.PEM: - res = self._backend._lib.PEM_write_bio_X509_REQ( - bio, self._x509_req - ) - elif encoding is serialization.Encoding.DER: - res = self._backend._lib.i2d_X509_REQ_bio(bio, self._x509_req) - else: - raise TypeError("encoding must be an item from the Encoding enum") - - self._backend.openssl_assert(res == 1) - return self._backend._read_mem_bio(bio) - - @property - def tbs_certrequest_bytes(self) -> bytes: - pp = self._backend._ffi.new("unsigned char **") - res = self._backend._lib.i2d_re_X509_REQ_tbs(self._x509_req, pp) - self._backend.openssl_assert(res > 0) - pp = self._backend._ffi.gc( - pp, lambda pointer: self._backend._lib.OPENSSL_free(pointer[0]) - ) - return self._backend._ffi.buffer(pp[0], res)[:] - - @property - def signature(self) -> bytes: - sig = self._backend._ffi.new("ASN1_BIT_STRING **") - self._backend._lib.X509_REQ_get0_signature( - self._x509_req, sig, self._backend._ffi.NULL - ) - self._backend.openssl_assert(sig[0] != self._backend._ffi.NULL) - return _asn1_string_to_bytes(self._backend, sig[0]) - - @property - def is_signature_valid(self) -> bool: - pkey = self._backend._lib.X509_REQ_get_pubkey(self._x509_req) - self._backend.openssl_assert(pkey != self._backend._ffi.NULL) - pkey = self._backend._ffi.gc(pkey, self._backend._lib.EVP_PKEY_free) - res = self._backend._lib.X509_REQ_verify(self._x509_req, pkey) - - if res != 1: - self._backend._consume_errors() - return False - - return True - - def get_attribute_for_oid(self, oid: x509.ObjectIdentifier) -> bytes: - obj = _txt2obj_gc(self._backend, oid.dotted_string) - pos = self._backend._lib.X509_REQ_get_attr_by_OBJ( - self._x509_req, obj, -1 - ) - if pos == -1: - raise x509.AttributeNotFound( - "No {} attribute was found".format(oid), oid - ) - - attr = self._backend._lib.X509_REQ_get_attr(self._x509_req, pos) - self._backend.openssl_assert(attr != self._backend._ffi.NULL) - # We don't support multiple valued attributes for now. - self._backend.openssl_assert( - self._backend._lib.X509_ATTRIBUTE_count(attr) == 1 - ) - asn1_type = self._backend._lib.X509_ATTRIBUTE_get0_type(attr, 0) - self._backend.openssl_assert(asn1_type != self._backend._ffi.NULL) - # We need this to ensure that our C type cast is safe. - # Also this should always be a sane string type, but we'll see if - # that is true in the real world... - if asn1_type.type not in ( - _ASN1Type.UTF8String.value, - _ASN1Type.PrintableString.value, - _ASN1Type.IA5String.value, - ): - raise ValueError( - "OID {} has a disallowed ASN.1 type: {}".format( - oid, asn1_type.type - ) - ) - - data = self._backend._lib.X509_ATTRIBUTE_get0_data( - attr, 0, asn1_type.type, self._backend._ffi.NULL - ) - self._backend.openssl_assert(data != self._backend._ffi.NULL) - # This cast is safe iff we assert on the type above to ensure - # that it is always a type of ASN1_STRING - data = self._backend._ffi.cast("ASN1_STRING *", data) - return _asn1_string_to_bytes(self._backend, data) +# This exists for pyOpenSSL compatibility and SHOULD NOT BE USED +# WE WILL REMOVE THIS VERY SOON. +def _CertificateSigningRequest( # noqa: N802 + backend, x509_req +) -> x509.CertificateSigningRequest: + warnings.warn( + "This version of cryptography contains a temporary pyOpenSSL " + "fallback path. Upgrade pyOpenSSL now.", + utils.DeprecatedIn35, + ) + return backend._ossl2csr(x509_req) class _RawRevokedCertificate(x509.RevokedCertificate): diff --git a/src/cryptography/hazmat/bindings/_rust/x509.pyi b/src/cryptography/hazmat/bindings/_rust/x509.pyi index 08b4e6c2e..84ef0044b 100644 --- a/src/cryptography/hazmat/bindings/_rust/x509.pyi +++ b/src/cryptography/hazmat/bindings/_rust/x509.pyi @@ -10,6 +10,8 @@ def load_pem_x509_certificate(data: bytes) -> x509.Certificate: ... def load_der_x509_certificate(data: bytes) -> x509.Certificate: ... def load_pem_x509_crl(data: bytes) -> x509.CertificateRevocationList: ... def load_der_x509_crl(data: bytes) -> x509.CertificateRevocationList: ... +def load_pem_x509_csr(data: bytes) -> x509.CertificateSigningRequest: ... +def load_der_x509_csr(data: bytes) -> x509.CertificateSigningRequest: ... def encode_precertificate_signed_certificate_timestamps( extension: x509.PrecertificateSignedCertificateTimestamps, ) -> bytes: ... @@ -18,3 +20,4 @@ class Sct: ... class Certificate: ... class RevokedCertificate: ... class CertificateRevocationList: ... +class CertificateSigningRequest: ... diff --git a/src/cryptography/x509/base.py b/src/cryptography/x509/base.py index 17ab61fe7..bd28d9a56 100644 --- a/src/cryptography/x509/base.py +++ b/src/cryptography/x509/base.py @@ -425,6 +425,10 @@ class CertificateSigningRequest(metaclass=abc.ABCMeta): """ +# Runtime isinstance checks need this since the rust class is not a subclass. +CertificateSigningRequest.register(rust_x509.CertificateSigningRequest) + + # Backend argument preserved for API compatibility, but ignored. def load_pem_x509_certificate( data: bytes, backend: typing.Any = None @@ -439,26 +443,28 @@ def load_der_x509_certificate( return rust_x509.load_der_x509_certificate(data) +# Backend argument preserved for API compatibility, but ignored. def load_pem_x509_csr( data: bytes, backend: typing.Optional[Backend] = None ) -> CertificateSigningRequest: - backend = _get_backend(backend) - return backend.load_pem_x509_csr(data) + return rust_x509.load_pem_x509_csr(data) +# Backend argument preserved for API compatibility, but ignored. def load_der_x509_csr( data: bytes, backend: typing.Optional[Backend] = None ) -> CertificateSigningRequest: - backend = _get_backend(backend) - return backend.load_der_x509_csr(data) + return rust_x509.load_der_x509_csr(data) +# Backend argument preserved for API compatibility, but ignored. def load_pem_x509_crl( data: bytes, backend: typing.Optional[Backend] = None ) -> CertificateRevocationList: return rust_x509.load_pem_x509_crl(data) +# Backend argument preserved for API compatibility, but ignored. def load_der_x509_crl( data: bytes, backend: typing.Optional[Backend] = None ) -> CertificateRevocationList: |
