From fe8e94980e0f1daccfed8cd19a1a5751e02965c1 Mon Sep 17 00:00:00 2001 From: Erik Trauschke Date: Thu, 7 May 2015 08:47:26 -0700 Subject: [PATCH] adding CRL interfaces --- .../hazmat/backends/openssl/backend.py | 64 +++- src/cryptography/hazmat/backends/openssl/x509.py | 308 +++++++++++++++++--- src/cryptography/hazmat/bindings/openssl/asn1.py | 1 + src/cryptography/hazmat/bindings/openssl/x509.py | 3 + .../hazmat/bindings/openssl/x509name.py | 5 + src/cryptography/hazmat/bindings/openssl/x509v3.py | 24 ++ src/cryptography/x509.py | 160 ++++++++++ tests/test_x509.py | 152 ++++++++++ tests/test_x509_ext.py | 60 ++++ .../x509/custom/revocation_ca.pem | 14 + .../x509/custom/revocation_cert_revoked.pem | 13 + .../x509/custom/revocation_cert_valid.pem | 13 + .../x509/custom/revocation_crl.pem | 8 + 13 files changed, 787 insertions(+), 38 deletions(-) create mode 100644 vectors/cryptography_vectors/x509/custom/revocation_ca.pem create mode 100644 vectors/cryptography_vectors/x509/custom/revocation_cert_revoked.pem create mode 100644 vectors/cryptography_vectors/x509/custom/revocation_cert_valid.pem create mode 100644 vectors/cryptography_vectors/x509/custom/revocation_crl.pem diff --git a/src/cryptography/hazmat/backends/openssl/backend.py b/src/cryptography/hazmat/backends/openssl/backend.py index 665771a..4020551 100644 --- a/src/cryptography/hazmat/backends/openssl/backend.py +++ b/src/cryptography/hazmat/backends/openssl/backend.py @@ -5,6 +5,7 @@ from __future__ import absolute_import, division, print_function import collections +import datetime import itertools from contextlib import contextmanager @@ -35,7 +36,7 @@ from cryptography.hazmat.backends.openssl.rsa import ( _RSAPrivateKey, _RSAPublicKey ) from cryptography.hazmat.backends.openssl.x509 import ( - _Certificate, _CertificateSigningRequest + _Certificate, _CertificateSigningRequest, _CertificateRevocationList ) from cryptography.hazmat.bindings.openssl.binding import Binding from cryptography.hazmat.primitives import hashes, serialization @@ -525,6 +526,29 @@ class Backend(object): else: raise UnsupportedAlgorithm("Unsupported key type.") + def _key_to_evp_key(self, pkey): + + evp_key = self._lib.EVP_PKEY_new() + assert evp_key != self._ffi.NULL + evp_key = self._ffi.gc(evp_key, self._lib.EVP_PKEY_free) + + if isinstance(pkey, _RSAPrivateKey) or isinstance(pkey, _RSAPublicKey): + res = self._lib.EVP_PKEY_set1_RSA(evp_key, pkey._rsa_cdata); + assert res == 1 + return evp_key + elif isinstance(pkey, _DSAPrivateKey) or isinstance(pkey, _DSAPublicKey): + res = self._lib.EVP_PKEY_set1_DSA(evp_key, pkey._dsa_cdata); + assert res == 1 + return evp_key + elif (isinstance(pkey, _EllipticCurvePrivateKey) or + isinstance(pkey, _EllipticCurvePublicKey)): + res = self._lib.EVP_PKEY_set1_EC_KEY(evp_key, pkey._ec_key); + assert res == 1 + return evp_key + else: + raise TypeError( + "{0} is not a supported key type.".format(type(pkey))) + def _pem_password_cb(self, password): """ Generate a pem_password_cb function pointer that copied the password to @@ -800,6 +824,28 @@ class Backend(object): else: self._handle_key_loading_error() + def load_pem_x509_crl(self, data): + mem_bio = self._bytes_to_bio(data) + x509_crl = self._lib.PEM_read_bio_X509_CRL( + mem_bio.bio, self._ffi.NULL, self._ffi.NULL, self._ffi.NULL + ) + if x509_crl == self._ffi.NULL: + self._consume_errors() + raise ValueError("Unable to load certificate") + + x509_crl = self._ffi.gc(x509_crl, self._lib.X509_CRL_free) + return _CertificateRevocationList(self, x509_crl) + + def load_der_x509_crl(self, data): + mem_bio = self._bytes_to_bio(data) + x509_crl = self._lib.d2i_X509_CRL_bio(mem_bio.bio, self._ffi.NULL) + if x509_crl == self._ffi.NULL: + self._consume_errors() + raise ValueError("Unable to load request") + + x509_crl = self._ffi.gc(x509_crl, self._lib.X509_CRL_free) + return _CertificateRevocationList(self, x509_crl) + def load_pem_x509_certificate(self, data): mem_bio = self._bytes_to_bio(data) x509 = self._lib.PEM_read_bio_X509( @@ -1274,6 +1320,22 @@ class Backend(object): assert res == 1 return self._read_mem_bio(bio) + def _parse_asn1_time(self, asn1_time): + assert asn1_time != self._ffi.NULL + generalized_time = self._lib.ASN1_TIME_to_generalizedtime( + asn1_time, self._ffi.NULL + ) + assert generalized_time != self._ffi.NULL + generalized_time = self._ffi.gc( + generalized_time, self._lib.ASN1_GENERALIZEDTIME_free + ) + time = self._ffi.string( + self._lib.ASN1_STRING_data( + self._ffi.cast("ASN1_STRING *", generalized_time) + ) + ).decode("ascii") + return datetime.datetime.strptime(time, "%Y%m%d%H%M%SZ") + class GetCipherByName(object): def __init__(self, fmt): diff --git a/src/cryptography/hazmat/backends/openssl/x509.py b/src/cryptography/hazmat/backends/openssl/x509.py index 42ca138..ff22e6c 100644 --- a/src/cryptography/hazmat/backends/openssl/x509.py +++ b/src/cryptography/hazmat/backends/openssl/x509.py @@ -38,33 +38,33 @@ def _obj2txt(backend, obj): return backend._ffi.buffer(buf, res)[:].decode() +def _build_x509_name_entry(backend, x509_name_entry): + + obj = backend._lib.X509_NAME_ENTRY_get_object(x509_name_entry) + assert obj != backend._ffi.NULL + data = backend._lib.X509_NAME_ENTRY_get_data(x509_name_entry) + assert data != backend._ffi.NULL + buf = backend._ffi.new("unsigned char **") + res = backend._lib.ASN1_STRING_to_UTF8(buf, data) + assert res >= 0 + assert buf[0] != backend._ffi.NULL + buf = backend._ffi.gc( + buf, lambda buffer: backend._lib.OPENSSL_free(buffer[0]) + ) + value = backend._ffi.buffer(buf[0], res)[:].decode('utf8') + oid = _obj2txt(backend, obj) + + return x509.NameAttribute(x509.ObjectIdentifier(oid), value) + def _build_x509_name(backend, x509_name): count = backend._lib.X509_NAME_entry_count(x509_name) attributes = [] for x in range(count): entry = backend._lib.X509_NAME_get_entry(x509_name, x) - obj = backend._lib.X509_NAME_ENTRY_get_object(entry) - assert obj != backend._ffi.NULL - data = backend._lib.X509_NAME_ENTRY_get_data(entry) - assert data != backend._ffi.NULL - buf = backend._ffi.new("unsigned char **") - res = backend._lib.ASN1_STRING_to_UTF8(buf, data) - assert res >= 0 - assert buf[0] != backend._ffi.NULL - buf = backend._ffi.gc( - buf, lambda buffer: backend._lib.OPENSSL_free(buffer[0]) - ) - value = backend._ffi.buffer(buf[0], res)[:].decode('utf8') - oid = _obj2txt(backend, obj) - attributes.append( - x509.NameAttribute( - x509.ObjectIdentifier(oid), value - ) - ) + attributes.append(_build_x509_name_entry(backend, entry)) return x509.Name(attributes) - def _build_general_name(backend, gn): if gn.type == backend._lib.GEN_DNS: data = backend._ffi.buffer(gn.d.dNSName.data, gn.d.dNSName.length)[:] @@ -199,28 +199,12 @@ class _Certificate(object): @property def not_valid_before(self): asn1_time = self._backend._lib.X509_get_notBefore(self._x509) - return self._parse_asn1_time(asn1_time) + return self._backend._parse_asn1_time(asn1_time) @property def not_valid_after(self): asn1_time = self._backend._lib.X509_get_notAfter(self._x509) - return self._parse_asn1_time(asn1_time) - - def _parse_asn1_time(self, asn1_time): - assert asn1_time != self._backend._ffi.NULL - generalized_time = self._backend._lib.ASN1_TIME_to_generalizedtime( - asn1_time, self._backend._ffi.NULL - ) - assert generalized_time != self._backend._ffi.NULL - generalized_time = self._backend._ffi.gc( - generalized_time, self._backend._lib.ASN1_GENERALIZEDTIME_free - ) - time = self._backend._ffi.string( - self._backend._lib.ASN1_STRING_data( - self._backend._ffi.cast("ASN1_STRING *", generalized_time) - ) - ).decode("ascii") - return datetime.datetime.strptime(time, "%Y%m%d%H%M%SZ") + return self._backend._parse_asn1_time(asn1_time) @property def issuer(self): @@ -273,6 +257,8 @@ class _Certificate(object): value = self._build_authority_key_identifier(ext) elif oid == x509.OID_AUTHORITY_INFORMATION_ACCESS: value = self._build_authority_information_access(ext) + elif oid == x509.OID_CRL_DISTRIBUTION_POINTS: + value = self._build_crl_distribution_points(ext) elif critical: raise x509.UnsupportedExtension( "{0} is not currently supported".format(oid), oid @@ -450,6 +436,48 @@ class _Certificate(object): return x509.ExtendedKeyUsage(ekus) + def _build_crl_distribution_points(self, ext): + cdps = self._backend._ffi.cast( + "Cryptography_STACK_OF_DIST_POINT *", + self._backend._lib.X509V3_EXT_d2i(ext) + ) + assert cdps != self._backend._ffi.NULL + cdps = self._backend._ffi.gc( + cdps, self._backend._lib.sk_DIST_POINT_free) + num = self._backend._lib.sk_DIST_POINT_num(cdps) + + dist_points = [] + for i in range(num): + # TODO: so far this doesn't support reasons and cRLIssuer. + # Most use cases can make do with just the distributionPoint. + cdp = self._backend._lib.sk_DIST_POINT_value(cdps, i) + # Type 0 is fullName, there is no #define for it in the code. + if cdp.distpoint.type == 0: + gns = cdp.distpoint.name.fullname + gnum = self._backend._lib.sk_GENERAL_NAME_num(gns) + names = [] + for i in range(gnum): + gn = self._backend._lib.sk_GENERAL_NAME_value(gns, i) + assert gn != self._backend._ffi.NULL + names.append(_build_general_name(self._backend, gn)) + dist_points.append(x509.CRLDistributionPoint(fullname=names)) + # OpenSSL code doesn't test for a specific type for relativename, + # everything that isn't fullname is considered relativename. + else: + xns = cdp.distpoint.name.relativename + xnum = self._backend._lib.sk_X509_NAME_ENTRY_num(xns) + name_entries = [] + for i in range(xnum): + xn = self._backend._lib.sk_X509_NAME_ENTRY_value(xns, i) + assert xn != self._backend._ffi.NULL + name_entries.append(_build_x509_name_entry( + self._backend, xn)) + + dist_points.append(x509.CRLDistributionPoint( + relativename=x509.Name(name_entries))) + + return dist_points + @utils.register_interface(x509.CertificateSigningRequest) class _CertificateSigningRequest(object): @@ -478,3 +506,209 @@ class _CertificateSigningRequest(object): raise UnsupportedAlgorithm( "Signature algorithm OID:{0} not recognized".format(oid) ) + + +@utils.register_interface(x509.RevokedCertificate) +class _RevokedCertificate(object): + def __init__(self, backend, x509_revoked): + self._backend = backend + self._x509_revoked = x509_revoked + + self.__serialNumber = None + self.__revocationDate = None + self.__extensions = None + + @property + def serialNumber(self): + if self.__serialNumber: + return self.__serialNumber + + bn = self._backend._lib.ASN1_INTEGER_to_BN( + self._x509_revoked.serialNumber, self._backend._ffi.NULL + ) + assert bn != self._backend._ffi.NULL + bn = self._backend._ffi.gc(bn, self._backend._lib.BN_free) + self.__serialNumber = self._backend._bn_to_int(bn) + return self.__serialNumber + + @property + def revocationDate(self): + if self.__revocationDate: + return self.__revocationDate + + self.__revocationDate = self._backend._parse_asn1_time( + self._x509_revoked.revocationDate) + return self.__revocationDate + + @property + def extensions(self): + if self.__extensions: + return self.__extensions + + extensions = [] + seen_oids = set() + extcount = self._backend._lib.X509_REVOKED_get_ext_count( + self._x509_revoked) + for i in range(0, extcount): + ext = self._backend._lib.X509_REVOKED_get_ext(self._x509_revoked, i) + assert ext != self._backend._ffi.NULL + crit = self._backend._lib.X509_EXTENSION_get_critical(ext) + critical = crit == 1 + oid = x509.ObjectIdentifier(_obj2txt(self._backend, ext.object)) + if oid in seen_oids: + raise x509.DuplicateExtension( + "Duplicate {0} extension found".format(oid), oid + ) + + # TODO: so far we just support CRLReason which should be enough for + # most use cases. + if oid == x509.OID_CRLREASON: + value = self._build_crl_reason(ext) + elif critical: + raise x509.UnsupportedExtension( + "{0} is not currently supported".format(oid), oid + ) + else: + # Unsupported non-critical extension, silently skipping for now + seen_oids.add(oid) + continue + + seen_oids.add(oid) + extensions.append(x509.Extension(oid, critical, value)) + + self.__extensions = x509.Extensions(extensions) + return self.__extensions + + def get_reason(self): + try: + re = self.extensions.get_extension_for_oid(x509.OID_CRLREASON) + except x509.ExtensionNotFound: + return None + return re.value + + def _build_crl_reason(self, ext): + enum = self._backend._lib.X509V3_EXT_d2i(ext) + assert enum != self._backend._ffi.NULL + enum = self._backend._ffi.cast("ASN1_ENUMERATED *", enum) + enum = self._backend._ffi.gc( + enum, self._backend._lib.ASN1_ENUMERATED_free + ) + code = self._backend._lib.ASN1_ENUMERATED_get(enum) + return x509.CrlReason(code) + + +@utils.register_interface(x509.CertificateRevocationList) +class _CertificateRevocationList(object): + def __init__(self, backend, x509_crl): + self._backend = backend + self._x509_crl = x509_crl + + self.__revoked = None + self.__issuer = None + self.__nextUpdate = None + self.__lastUpdate = None + + def fingerprint(self, algorithm): + h = hashes.Hash(algorithm, self._backend) + bio = self._backend._create_mem_bio() + res = self._backend._lib.i2d_X509_CRL_bio( + bio, self._x509_crl + ) + assert res == 1 + der = self._backend._read_mem_bio(bio) + h.update(der) + return h.finalize() + + @property + def signature_hash_algorithm(self): + oid = _obj2txt(self._backend, self._x509_crl.sig_alg.algorithm) + try: + return x509._SIG_OIDS_TO_HASH[oid] + except KeyError: + raise UnsupportedAlgorithm( + "Signature algorithm OID:{0} not recognized".format(oid) + ) + + @property + def issuer(self): + if self.__issuer: + return self.__issuer + + issuer = self._backend._lib.X509_CRL_get_issuer(self._x509_crl) + assert issuer != self._backend._ffi.NULL + self.__issuer = _build_x509_name(self._backend, issuer) + return self.__issuer + + @property + def nextUpdate(self): + if self.__nextUpdate: + return self.__nextUpdate + + nextUpdate = self._backend._lib.X509_CRL_get_nextUpdate(self._x509_crl) + assert nextUpdate != self._backend._ffi.NULL + self.__nextUpdate = self._backend._parse_asn1_time(nextUpdate) + return self.__nextUpdate + + @property + def lastUpdate(self): + if self.__lastUpdate: + return self.__lastUpdate + + lastUpdate = self._backend._lib.X509_CRL_get_lastUpdate(self._x509_crl) + assert lastUpdate != self._backend._ffi.NULL + self.__lastUpdate = self._backend._parse_asn1_time(lastUpdate) + return self.__lastUpdate + + @property + def revoked(self): + if self.__revoked: + return self.__revoked + + revoked = self._backend._lib.X509_CRL_get_REVOKED(self._x509_crl) + assert revoked != self._backend._ffi.NULL + + num = self._backend._lib.sk_X509_REVOKED_num(revoked) + revoked_list = [] + for i in range(num): + r = self._backend._lib.sk_X509_REVOKED_value(revoked, i) + assert r != self._backend._ffi.NULL + revoked_list.append(_RevokedCertificate(self._backend, r)) + + self.__revoked = revoked_list + return self.__revoked + + def verify(self, pubkey): + evp_key = self._backend._key_to_evp_key(pubkey) + + res = self._backend._lib.X509_CRL_verify(self._x509_crl, evp_key) + return res == 1 + + # Additional convenience methods/properties. + + def is_revoked(self, cert): + """ + Check if certificate 'cert' is marked as revoked in this CRL. + """ + + if cert.issuer != self.issuer: + raise ValueError("Certificate issuer does not match!") + for r in self.revoked: + if cert.serial == r.serialNumber: + return True + return False + + def get_revoked(self, cert): + """ + Return RevokedCertificate object for the given 'cert' if it is part + of this CRL. + """ + + if cert.issuer != self.issuer: + raise ValueError("Certificate issuer does not match!") + for r in self.revoked: + if cert.serial == r.serialNumber: + return r + return None + + + diff --git a/src/cryptography/hazmat/bindings/openssl/asn1.py b/src/cryptography/hazmat/bindings/openssl/asn1.py index 475bd05..b13445d 100644 --- a/src/cryptography/hazmat/bindings/openssl/asn1.py +++ b/src/cryptography/hazmat/bindings/openssl/asn1.py @@ -113,6 +113,7 @@ void ASN1_GENERALIZEDTIME_free(ASN1_GENERALIZEDTIME *); ASN1_ENUMERATED *ASN1_ENUMERATED_new(void); void ASN1_ENUMERATED_free(ASN1_ENUMERATED *); int ASN1_ENUMERATED_set(ASN1_ENUMERATED *, long); +long ASN1_ENUMERATED_get(ASN1_ENUMERATED *); ASN1_VALUE *ASN1_item_d2i(ASN1_VALUE **, const unsigned char **, long, const ASN1_ITEM *); diff --git a/src/cryptography/hazmat/bindings/openssl/x509.py b/src/cryptography/hazmat/bindings/openssl/x509.py index fa6a16b..3125016 100644 --- a/src/cryptography/hazmat/bindings/openssl/x509.py +++ b/src/cryptography/hazmat/bindings/openssl/x509.py @@ -64,6 +64,7 @@ typedef struct { typedef struct { X509_CRL_INFO *crl; + X509_ALGOR *sig_alg; ...; } X509_CRL; @@ -183,6 +184,7 @@ int i2d_X509_CRL_bio(BIO *, X509_CRL *); int X509_CRL_print(BIO *, X509_CRL *); int X509_CRL_set_issuer_name(X509_CRL *, X509_NAME *); int X509_CRL_sign(X509_CRL *, EVP_PKEY *, const EVP_MD *); +int X509_CRL_verify(X509_CRL *, EVP_PKEY *); int X509_CRL_get_ext_count(X509_CRL *); X509_EXTENSION *X509_CRL_get_ext(X509_CRL *, int); int X509_CRL_add_ext(X509_CRL *, X509_EXTENSION *, int); @@ -285,6 +287,7 @@ int X509_CRL_get_version(X509_CRL *); ASN1_TIME *X509_CRL_get_lastUpdate(X509_CRL *); ASN1_TIME *X509_CRL_get_nextUpdate(X509_CRL *); X509_NAME *X509_CRL_get_issuer(X509_CRL *); +Cryptography_STACK_OF_X509_REVOKED *X509_CRL_get_REVOKED(X509_CRL *); /* These aren't macros these arguments are all const X on openssl > 1.0.x */ int X509_CRL_set_lastUpdate(X509_CRL *, ASN1_TIME *); diff --git a/src/cryptography/hazmat/bindings/openssl/x509name.py b/src/cryptography/hazmat/bindings/openssl/x509name.py index bda92eb..a7dde87 100644 --- a/src/cryptography/hazmat/bindings/openssl/x509name.py +++ b/src/cryptography/hazmat/bindings/openssl/x509name.py @@ -11,12 +11,14 @@ INCLUDES = """ * See the comment above Cryptography_STACK_OF_X509 in x509.py */ typedef STACK_OF(X509_NAME) Cryptography_STACK_OF_X509_NAME; +typedef STACK_OF(X509_NAME_ENTRY) Cryptography_STACK_OF_X509_NAME_ENTRY; """ TYPES = """ typedef ... X509_NAME; typedef ... X509_NAME_ENTRY; typedef ... Cryptography_STACK_OF_X509_NAME; +typedef ... Cryptography_STACK_OF_X509_NAME_ENTRY; """ FUNCTIONS = """ @@ -48,6 +50,9 @@ int sk_X509_NAME_num(Cryptography_STACK_OF_X509_NAME *); int sk_X509_NAME_push(Cryptography_STACK_OF_X509_NAME *, X509_NAME *); X509_NAME *sk_X509_NAME_value(Cryptography_STACK_OF_X509_NAME *, int); void sk_X509_NAME_free(Cryptography_STACK_OF_X509_NAME *); +int sk_X509_NAME_ENTRY_num(Cryptography_STACK_OF_X509_NAME_ENTRY *); +X509_NAME_ENTRY *sk_X509_NAME_ENTRY_value( + Cryptography_STACK_OF_X509_NAME_ENTRY *, int); """ CUSTOMIZATIONS = """ diff --git a/src/cryptography/hazmat/bindings/openssl/x509v3.py b/src/cryptography/hazmat/bindings/openssl/x509v3.py index c2b6860..4a77f0e 100644 --- a/src/cryptography/hazmat/bindings/openssl/x509v3.py +++ b/src/cryptography/hazmat/bindings/openssl/x509v3.py @@ -20,6 +20,7 @@ typedef LHASH_OF(CONF_VALUE) Cryptography_LHASH_OF_CONF_VALUE; typedef LHASH Cryptography_LHASH_OF_CONF_VALUE; #endif typedef STACK_OF(ACCESS_DESCRIPTION) Cryptography_STACK_OF_ACCESS_DESCRIPTION; +typedef STACK_OF(DIST_POINT) Cryptography_STACK_OF_DIST_POINT; """ TYPES = """ @@ -101,6 +102,26 @@ typedef struct { } ACCESS_DESCRIPTION; typedef ... Cryptography_LHASH_OF_CONF_VALUE; + + +typedef ... Cryptography_STACK_OF_DIST_POINT; + +typedef struct { + int type; + union { + GENERAL_NAMES *fullname; + Cryptography_STACK_OF_X509_NAME_ENTRY *relativename; + } name; + X509_NAME *dpname; +} DIST_POINT_NAME; + +typedef struct { + DIST_POINT_NAME *distpoint; + ASN1_BIT_STRING *reasons; + GENERAL_NAMES *CRLissuer; + int dp_reasons; +} DIST_POINT; + """ @@ -138,6 +159,9 @@ X509_EXTENSION *X509V3_EXT_conf_nid(Cryptography_LHASH_OF_CONF_VALUE *, const X509V3_EXT_METHOD *X509V3_EXT_get(X509_EXTENSION *); const X509V3_EXT_METHOD *X509V3_EXT_get_nid(int); +void sk_DIST_POINT_free(Cryptography_STACK_OF_DIST_POINT *); +int sk_DIST_POINT_num(Cryptography_STACK_OF_DIST_POINT *); +DIST_POINT *sk_DIST_POINT_value(Cryptography_STACK_OF_DIST_POINT *, int); """ CUSTOMIZATIONS = """ diff --git a/src/cryptography/x509.py b/src/cryptography/x509.py index 0d87cd5..fe9a8d9 100644 --- a/src/cryptography/x509.py +++ b/src/cryptography/x509.py @@ -55,6 +55,7 @@ _OID_NAMES = { "2.5.29.17": "subjectAltName", "2.5.29.18": "issuerAltName", "2.5.29.19": "basicConstraints", + "2.5.29.21": "CRLReason", "2.5.29.30": "nameConstraints", "2.5.29.31": "cRLDistributionPoints", "2.5.29.32": "certificatePolicies", @@ -106,6 +107,14 @@ def load_der_x509_csr(data, backend): return backend.load_der_x509_csr(data) +def load_pem_x509_crl(data, backend): + return backend.load_pem_x509_crl(data) + + +def load_der_x509_crl(data, backend): + return backend.load_der_x509_crl(data) + + class InvalidVersion(Exception): def __init__(self, msg, parsed_version): super(InvalidVersion, self).__init__(msg) @@ -222,6 +231,7 @@ OID_KEY_USAGE = ObjectIdentifier("2.5.29.15") OID_SUBJECT_ALTERNATIVE_NAME = ObjectIdentifier("2.5.29.17") OID_ISSUER_ALTERNATIVE_NAME = ObjectIdentifier("2.5.29.18") OID_BASIC_CONSTRAINTS = ObjectIdentifier("2.5.29.19") +OID_CRLREASON = ObjectIdentifier("2.5.29.21") OID_NAME_CONSTRAINTS = ObjectIdentifier("2.5.29.30") OID_CRL_DISTRIBUTION_POINTS = ObjectIdentifier("2.5.29.31") OID_CERTIFICATE_POLICIES = ObjectIdentifier("2.5.29.32") @@ -277,6 +287,86 @@ class Extension(object): "value={0.value})>").format(self) +class CrlReason(object): + + __CRL_REASON = { + 0: "unspecified", + 1: "keyCompromise", + 2: "cACompromise", + 3: "affiliationChanged", + 4: "superseded", + 5: "cessationOfOperation", + 6: "certificateHold", + 8: "removeFromCRL", + } + + @staticmethod + def get_reason(code): + return CrlReason.__CRL_REASON[code] + + def __init__(self, reason_code): + if reason_code not in self.__CRL_REASON: + raise KeyError("No such CRL reason: {0}".format(reason_code)) + + self._code = reason_code + + code = utils.read_only_property("_code") + + @property + def reason_str(self): + return self.__CRL_REASON[self._code] + + def __repr__(self): + return "".format( + self.get_reason(), self.code() + ) + + +class CRLDistributionPoint(object): + + TYPE_FULL = "full" + TYPE_RELATIVE = "relative" + + def __init__(self, fullname=None, relativename=None, reasons=None, + crl_issuer=None): + if fullname and relativename: + raise ValueError( + "fullname and relativename are mutually exclusive.") + if not (fullname or relativename): + raise ValueError( + "Either fullname or relativename must be given.") + + + if fullname and not isinstance(fullname, list): + self._fullname = [fullname] + else: + self._fullname = fullname + self._relativename = relativename + self._reasons = reasons + self._crl_issuer = crl_issuer + + crl_issuer = utils.read_only_property("_crl_issuer") + fullname = utils.read_only_property("_fullname") + reasons = utils.read_only_property("_reasons") + relativename = utils.read_only_property("_relativename") + + @property + def name_type(self): + if self._fullname: + return self.TYPE_FULL + elif self._relativename: + return self.TYPE_RELATIVE + else: + raise AssertionError("Invalid CRLDistributionPoint object.") + + def __repr__(self): + if self._fullname: + out = "fullname: {0}".format(", ".join(self._fullname)) + else: + out = "relativename: {0}".format(relativename) + + return "".format(out) + class ExtendedKeyUsage(object): def __init__(self, usages): if not all(isinstance(x, ObjectIdentifier) for x in usages): @@ -845,3 +935,73 @@ class CertificateSigningRequest(object): Returns a HashAlgorithm corresponding to the type of the digest signed in the certificate. """ + + +@six.add_metaclass(abc.ABCMeta) +class RevokedCertificate(object): + + @abc.abstractproperty + def serialNumber(self): + """ + Returns the serial number of the revoked certificate. + """ + + @abc.abstractproperty + def revocationDate(self): + """ + Returns the date of when this certificate was revoked. + """ + + @abc.abstractproperty + def extensions(self): + + """ + Returns the extensions for the revoked certificate object. + """ + + +@six.add_metaclass(abc.ABCMeta) +class CertificateRevocationList(object): + + @abc.abstractmethod + def fingerprint(self, algorithm): + """ + Returns bytes using digest passed. + """ + + @abc.abstractproperty + def signature_hash_algorithm(self): + """ + Returns a HashAlgorithm corresponding to the type of the digest signed + in the certificate. + """ + + @abc.abstractproperty + def issuer(self): + """ + Returns the X509Name with the issuer of this CRL. + """ + + @abc.abstractproperty + def nextUpdate(self): + """ + Returns the date of next update for this CRL. + """ + + @abc.abstractproperty + def lastUpdate(self): + """ + Returns the date of last update for this CRL. + """ + + @abc.abstractproperty + def revoked(self): + """ + Returns a list of RevokedCertificate objects for this CRL. + """ + + @abc.abstractmethod + def verify(self, pubkey): + """ + Verify this CRL against a given 'pubkey'. + """ diff --git a/tests/test_x509.py b/tests/test_x509.py index 8561f1f..9c40239 100644 --- a/tests/test_x509.py +++ b/tests/test_x509.py @@ -30,6 +30,158 @@ def _load_cert(filename, loader, backend): ) return cert +@pytest.mark.requires_backend_interface(interface=X509Backend) +class TestCertificateRevocationList(object): + def test_load_pem_crl(self, backend): + crl = _load_cert( + os.path.join("x509", "custom", "revocation_crl.pem"), + x509.load_pem_x509_crl, + backend + ) + + assert isinstance(crl, x509.CertificateRevocationList) + fingerprint = binascii.hexlify(crl.fingerprint(hashes.SHA1())) + assert fingerprint == b"1d4e9a6453271c0e8ac6552091bfdc0b6036bf5e" + assert isinstance(crl.signature_hash_algorithm, hashes.SHA256) + + def test_load_der_crl(self, backend): + crl = _load_cert( + os.path.join("x509", "PKITS_data", "crls", "GoodCACRL.crl"), + x509.load_der_x509_crl, + backend + ) + + assert isinstance(crl, x509.CertificateRevocationList) + fingerprint = binascii.hexlify(crl.fingerprint(hashes.SHA1())) + assert fingerprint == b"dd3db63c50f4c4a13e090f14053227cb1011a5ad" + assert isinstance(crl.signature_hash_algorithm, hashes.SHA256) + + def test_issuer(self, backend): + crl = _load_cert( + os.path.join("x509", "PKITS_data", "crls", "GoodCACRL.crl"), + x509.load_der_x509_crl, + backend + ) + + issuer = crl.issuer + assert isinstance(issuer, x509.Name) + assert list(issuer) == [ + x509.NameAttribute(x509.OID_COUNTRY_NAME, 'US'), + x509.NameAttribute( + x509.OID_ORGANIZATION_NAME, 'Test Certificates 2011' + ), + x509.NameAttribute(x509.OID_COMMON_NAME, 'Good CA') + ] + assert issuer.get_attributes_for_oid(x509.OID_COMMON_NAME) == [ + x509.NameAttribute(x509.OID_COMMON_NAME, 'Good CA') + ] + + def test_update_dates(self, backend): + crl = _load_cert( + os.path.join("x509", "custom", "revocation_crl.pem"), + x509.load_pem_x509_crl, + backend + ) + + + assert isinstance(crl.nextUpdate, datetime.datetime) + assert isinstance(crl.lastUpdate, datetime.datetime) + + assert crl.nextUpdate.isoformat() == "2018-01-30T18:02:39" + assert crl.lastUpdate.isoformat() == "2015-05-06T18:02:39" + + + def test_revoked_basics(self, backend): + crl = _load_cert( + os.path.join("x509", "custom", "revocation_crl.pem"), + x509.load_pem_x509_crl, + backend + ) + + assert isinstance(crl.revoked, list) + for i in crl.revoked: + assert isinstance(i, x509.RevokedCertificate) + assert isinstance(i.serialNumber, int) + assert isinstance(i.revocationDate, datetime.datetime) + + assert crl.revoked[0].serialNumber == 0x01 + + # test verify() with key from issuing CA + ca = _load_cert( + os.path.join("x509", "custom", "revocation_ca.pem"), + x509.load_pem_x509_certificate, + backend + ) + assert crl.verify(ca.public_key()) + + # now try if it fails with random key + pubkey = backend.generate_rsa_private_key(113, 2048).public_key() + assert not crl.verify(pubkey) + + # test convenience functions + + # revoked cert + rev_cert = _load_cert( + os.path.join("x509", "custom", "revocation_cert_revoked.pem"), + x509.load_pem_x509_certificate, + backend + ) + # valid (unrevoked) cert from correct issuer + valid_cert = _load_cert( + os.path.join("x509", "custom", "revocation_cert_valid.pem"), + x509.load_pem_x509_certificate, + backend + ) + # unrelated cert from different issuer + unrel_cert = _load_cert( + os.path.join("x509", "custom", "post2000utctime.pem"), + x509.load_pem_x509_certificate, + backend + ) + + with pytest.raises(ValueError): + crl.is_revoked(unrel_cert) + with pytest.raises(ValueError): + crl.get_revoked(unrel_cert) + + assert crl.is_revoked(rev_cert) + assert crl.get_revoked(rev_cert).serialNumber == 0x01 + assert (crl.get_revoked(rev_cert).revocationDate.isoformat() == + "2015-05-06T18:02:39") + + assert not crl.is_revoked(valid_cert) + assert crl.get_revoked(valid_cert) == None + + def test_revoked_reason(self, backend): + crl = _load_cert( + os.path.join("x509", "PKITS_data", "crls", "GoodCACRL.crl"), + x509.load_der_x509_crl, + backend + ) + + assert isinstance(crl.revoked, list) + for i in crl.revoked: + assert isinstance(i, x509.RevokedCertificate) + assert isinstance(i.serialNumber, int) + assert isinstance(i.revocationDate, datetime.datetime) + + assert crl.revoked[0].serialNumber == 0x0e + assert crl.revoked[1].serialNumber == 0x0f + + rev0 = crl.revoked[0] + assert isinstance(rev0.extensions, x509.Extensions) + reason = rev0.extensions.get_extension_for_oid(x509.OID_CRLREASON).value + assert reason + assert isinstance(reason, x509.CrlReason) + assert reason.reason_str == "keyCompromise" + assert reason.code == 1 + + # test convenience function get_reason() + reason = crl.revoked[1].get_reason() + assert isinstance(reason, x509.CrlReason) + assert reason.reason_str == "keyCompromise" + assert reason.code == 1 + @pytest.mark.requires_backend_interface(interface=RSABackend) @pytest.mark.requires_backend_interface(interface=X509Backend) diff --git a/tests/test_x509_ext.py b/tests/test_x509_ext.py index 8a22795..ff42b77 100644 --- a/tests/test_x509_ext.py +++ b/tests/test_x509_ext.py @@ -38,6 +38,21 @@ class TestExtension(object): "_length=None)>)>" ) +class TestCrlReason(object): + def test_valid_codes(self): + for i in range(9): + # code 7 is undefined + if i == 7: + with pytest.raises(KeyError): + x509.CrlReason(7) + continue + + cr = x509.CrlReason(i) + assert isinstance(cr.reason_str, str) + assert cr.code == i + # test static helper function + x509.CrlReason.get_reason(i) + class TestKeyUsage(object): def test_key_agreement_false_encipher_decipher_true(self): @@ -304,6 +319,51 @@ class TestBasicConstraints(object): ) +@pytest.mark.requires_backend_interface(interface=X509Backend) +class TestCrlDistributionPoints(object): + + def test_crldp(self, backend): + cert = _load_cert( + os.path.join( + "x509", "PKITS_data", "certs", "ValidcRLIssuerTest28EE.crt" + ), + x509.load_der_x509_certificate, + backend + ) + + crldps = cert.extensions.get_extension_for_oid( + x509.OID_CRL_DISTRIBUTION_POINTS).value + assert isinstance(crldps, list) + + assert crldps[0].name_type == x509.CRLDistributionPoint.TYPE_FULL + assert isinstance(crldps[0].fullname[0], x509.DirectoryName) + + + dirname = crldps[0].fullname[0].value + assert (dirname.get_attributes_for_oid( + x509.ObjectIdentifier("2.5.4.3"))[0].value == + 'indirect CRL for indirectCRL CA3') + + def test_crldp_relname(self, backend): + cert = _load_cert( + os.path.join( + "x509", "custom", "revocation_cert_valid.pem" + ), + x509.load_pem_x509_certificate, + backend + ) + + crldps = cert.extensions.get_extension_for_oid( + x509.OID_CRL_DISTRIBUTION_POINTS).value + + assert crldps[0].name_type == x509.CRLDistributionPoint.TYPE_RELATIVE + + relname = crldps[0].relativename + assert (relname.get_attributes_for_oid( + x509.ObjectIdentifier("2.5.4.3"))[0].value == + 'test') + + class TestExtendedKeyUsage(object): def test_not_all_oids(self): with pytest.raises(TypeError): diff --git a/vectors/cryptography_vectors/x509/custom/revocation_ca.pem b/vectors/cryptography_vectors/x509/custom/revocation_ca.pem new file mode 100644 index 0000000..6173a6d --- /dev/null +++ b/vectors/cryptography_vectors/x509/custom/revocation_ca.pem @@ -0,0 +1,14 @@ +-----BEGIN CERTIFICATE----- +MIICIDCCAYmgAwIBAgIJAKb/cLoZmUlMMA0GCSqGSIb3DQEBCwUAMCkxDTALBgNV +BAoMBHB5Q0ExGDAWBgNVBAMMD2NyeXB0b2dyYXBoeS5pbzAeFw0xNTA1MDYxODAy +MzlaFw0xODAxMzAxODAyMzlaMCkxDTALBgNVBAoMBHB5Q0ExGDAWBgNVBAMMD2Ny +eXB0b2dyYXBoeS5pbzCBnzANBgkqhkiG9w0BAQEFAAOBjQAwgYkCgYEA06Hq7eX3 +QKbfiET4OQkyqqUHuWWzQdu2tpxHYIcZNQl3DIM2uuMZ/ngMAJ6oW1xxAGfpcKnh +D24agKo6KpUBaiXNpzJe4+/3p5sk4r4fwnBsURjDjsUvJEwytOxlGzkNXuif/fBH +Sp6psQdQk+YTCZSxypJeF9RsDUluilQZ5MsCAwEAAaNQME4wHQYDVR0OBBYEFHd6 +hJYCOyp98rfBNzU7JerNhpc4MB8GA1UdIwQYMBaAFHd6hJYCOyp98rfBNzU7JerN +hpc4MAwGA1UdEwQFMAMBAf8wDQYJKoZIhvcNAQELBQADgYEAhDh/kEKfsqXi47cr +1L3Bc+S1pwxVUq3achJzJvxceL2vo+N3dyYozY4XZVk4e8phm6jMnO0WVf58ZR+f +ZO/ZU+cXYmnvo0fqBrmp3KtjzBJsrlfWuzI3ou3y5SJ3toOl887REeuuhm0fnSe2 +eKtPg9n8whDh5Lg7HOsZ58urdoA= +-----END CERTIFICATE----- diff --git a/vectors/cryptography_vectors/x509/custom/revocation_cert_revoked.pem b/vectors/cryptography_vectors/x509/custom/revocation_cert_revoked.pem new file mode 100644 index 0000000..48df679 --- /dev/null +++ b/vectors/cryptography_vectors/x509/custom/revocation_cert_revoked.pem @@ -0,0 +1,13 @@ +-----BEGIN CERTIFICATE----- +MIIB4DCCAUmgAwIBAgIBATANBgkqhkiG9w0BAQsFADApMQ0wCwYDVQQKDARweUNB +MRgwFgYDVQQDDA9jcnlwdG9ncmFwaHkuaW8wHhcNMTUwNTA2MTgwMjM5WhcNMTgw +MTMwMTgwMjM5WjAhMQ0wCwYDVQQKDARweUNBMRAwDgYDVQQDDAdyZXZva2VkMIGf +MA0GCSqGSIb3DQEBAQUAA4GNADCBiQKBgQDV+UZGnoJ6vAkBMJ47oTeCvuCsvvzI +2rPUFIWg5UsHnB5Rx50jdelGY9jDASwrk+dzPI6XbXejn2Pc5SUKpe0s8XtbufgE +v5nAqHX1j/dnBO51Q5d4dBKXjxs2wFtyDOUPzW4wOPjRLR8KDm4RptTh3nhfMvTS +gD0zV6A+UuuueQIDAQABoyAwHjAcBgNVHR8EFTATMBGgD6ENMAsGA1UEAwwEdGVz +dDANBgkqhkiG9w0BAQsFAAOBgQB2Dx/7epBmvX+ZPLj8srCj2gwtGA8s2lFoh+Rn +z4krSvENpNCKLqJUYuBRSWeSuQ80+3T2seNR0TQSKMahorSHrl5PRL5OJgv8bP7w +xvQ6BohYwsgdeAaD/39M/K/ZLWRaXwPxynrtHNcez+jdkV7bPXJtJjrIg+RcOIAU +I0hIgQ== +-----END CERTIFICATE----- diff --git a/vectors/cryptography_vectors/x509/custom/revocation_cert_valid.pem b/vectors/cryptography_vectors/x509/custom/revocation_cert_valid.pem new file mode 100644 index 0000000..51f4672 --- /dev/null +++ b/vectors/cryptography_vectors/x509/custom/revocation_cert_valid.pem @@ -0,0 +1,13 @@ +-----BEGIN CERTIFICATE----- +MIIB4zCCAUygAwIBAgIBAjANBgkqhkiG9w0BAQsFADApMQ0wCwYDVQQKDARweUNB +MRgwFgYDVQQDDA9jcnlwdG9ncmFwaHkuaW8wHhcNMTUwNTA2MTgwMjM5WhcNMTgw +MTMwMTgwMjM5WjAkMQ0wCwYDVQQKDARweUNBMRMwEQYDVQQDDApub25yZXZva2Vk +MIGfMA0GCSqGSIb3DQEBAQUAA4GNADCBiQKBgQCrF2X5TslhunqSa+sk6b8iEPhS +oWL7YOuFyyNE0+XYlXNjCy05BHPBLZtDAscJoqSubHEHL3f3sBmqwzDp9r/AZFhG +jCZpWG4oTk0jOEvxoVlXR/Lob/MNTYwdgIwkKGGO0N832Xa9sLPmD/QJj+7FG863 +PxzXWcvvfraHz9SpZwIDAQABoyAwHjAcBgNVHR8EFTATMBGgD6ENMAsGA1UEAwwE +dGVzdDANBgkqhkiG9w0BAQsFAAOBgQCY34MbBO5Jf9beJtODpdBw45nG2O8ia2do +y/ZP+wwi3f7iqojR7FNReFgR3/Un9iMWrGjA/8Yo+wr+3pI5hQLSCZ5ES2HbWvmF +3lhRjmPd2FTURUOJ0Su+MxNXwq6t2c6QnvwLSZ73cIFN2XycL+RLyhVpDCUtOww6 +EFUPNOwJvw== +-----END CERTIFICATE----- diff --git a/vectors/cryptography_vectors/x509/custom/revocation_crl.pem b/vectors/cryptography_vectors/x509/custom/revocation_crl.pem new file mode 100644 index 0000000..23ef26d --- /dev/null +++ b/vectors/cryptography_vectors/x509/custom/revocation_crl.pem @@ -0,0 +1,8 @@ +-----BEGIN X509 CRL----- +MIIBAzBuMA0GCSqGSIb3DQEBCwUAMCkxDTALBgNVBAoMBHB5Q0ExGDAWBgNVBAMM +D2NyeXB0b2dyYXBoeS5pbxcNMTUwNTA2MTgwMjM5WhcNMTgwMTMwMTgwMjM5WjAU +MBICAQEXDTE1MDUwNjE4MDIzOVowDQYJKoZIhvcNAQELBQADgYEAmhFwgH4Oj3vx +D3MC0zNbMwl1MpTSyhVqCllquubIO/18rWchI9GhA+96mkkP6vkfaLTN4siVPerC +xjC9Br4PxkQSu7vMg42cjmMPG/a1zO2NwdPvlNWZM6FMr+WpiIS80Ih1VrWZ6mos +oQ23azbF9IiP2Id0tCuZRc3fVn3RcJY= +-----END X509 CRL----- -- 1.7.9.2