D7net
Home
Console
Upload
information
Create File
Create Folder
About
Tools
:
/
usr
/
lib64
/
python3.6
/
site-packages
/
cryptography
/
hazmat
/
backends
/
openssl
/
Filename :
x509.py
back
Copy
# This file is dual licensed under the terms of the Apache License, Version # 2.0, and the BSD License. See the LICENSE file in the root of this repository # for complete details. from __future__ import absolute_import, division, print_function import datetime import operator from cryptography import utils, x509 from cryptography.exceptions import UnsupportedAlgorithm from cryptography.hazmat.backends.openssl.decode_asn1 import ( _asn1_integer_to_int, _asn1_string_to_bytes, _decode_x509_name, _obj2txt, _parse_asn1_time, ) from cryptography.hazmat.backends.openssl.encode_asn1 import ( _encode_asn1_int_gc, _txt2obj_gc, ) from cryptography.hazmat.primitives import hashes, serialization from cryptography.hazmat.primitives.asymmetric import dsa, ec, rsa from cryptography.x509.name import _ASN1Type @utils.register_interface(x509.Certificate) class _Certificate(object): def __init__(self, backend, x509_cert): self._backend = backend self._x509 = x509_cert version = self._backend._lib.X509_get_version(self._x509) if version == 0: self._version = x509.Version.v1 elif version == 2: self._version = x509.Version.v3 else: raise x509.InvalidVersion( "{} is not a valid X509 version".format(version), version ) def __repr__(self): return "<Certificate(subject={}, ...)>".format(self.subject) def __eq__(self, other): if not isinstance(other, x509.Certificate): return NotImplemented res = self._backend._lib.X509_cmp(self._x509, other._x509) return res == 0 def __ne__(self, other): return not self == other def __hash__(self): return hash(self.public_bytes(serialization.Encoding.DER)) def __deepcopy__(self, memo): return self def fingerprint(self, algorithm): h = hashes.Hash(algorithm, self._backend) h.update(self.public_bytes(serialization.Encoding.DER)) return h.finalize() version = utils.read_only_property("_version") @property def serial_number(self): asn1_int = self._backend._lib.X509_get_serialNumber(self._x509) self._backend.openssl_assert(asn1_int != self._backend._ffi.NULL) return _asn1_integer_to_int(self._backend, asn1_int) @property def serial(self): return self.serial_number def public_key(self): pkey = self._backend._lib.X509_get_pubkey(self._x509) if pkey == self._backend._ffi.NULL: # Remove errors from the stack. self._backend._consume_errors() raise ValueError("Certificate public key is of an unknown type") pkey = self._backend._ffi.gc(pkey, self._backend._lib.EVP_PKEY_free) return self._backend._evp_pkey_to_public_key(pkey) @property def not_valid_before(self): asn1_time = self._backend._lib.X509_getm_notBefore(self._x509) return _parse_asn1_time(self._backend, asn1_time) @property def not_valid_after(self): asn1_time = self._backend._lib.X509_getm_notAfter(self._x509) return _parse_asn1_time(self._backend, asn1_time) @property def issuer(self): issuer = self._backend._lib.X509_get_issuer_name(self._x509) self._backend.openssl_assert(issuer != self._backend._ffi.NULL) return _decode_x509_name(self._backend, issuer) @property def subject(self): subject = self._backend._lib.X509_get_subject_name(self._x509) self._backend.openssl_assert(subject != self._backend._ffi.NULL) return _decode_x509_name(self._backend, subject) @property def signature_hash_algorithm(self): oid = self.signature_algorithm_oid try: return x509._SIG_OIDS_TO_HASH[oid] except KeyError: raise UnsupportedAlgorithm( "Signature algorithm OID:{} not recognized".format(oid) ) @property def signature_algorithm_oid(self): alg = self._backend._ffi.new("X509_ALGOR **") self._backend._lib.X509_get0_signature( self._backend._ffi.NULL, alg, self._x509 ) 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): return self._backend._certificate_extension_parser.parse(self._x509) @property def signature(self): sig = self._backend._ffi.new("ASN1_BIT_STRING **") self._backend._lib.X509_get0_signature( sig, self._backend._ffi.NULL, self._x509 ) self._backend.openssl_assert(sig[0] != self._backend._ffi.NULL) return _asn1_string_to_bytes(self._backend, sig[0]) @property def tbs_certificate_bytes(self): pp = self._backend._ffi.new("unsigned char **") res = self._backend._lib.i2d_re_X509_tbs(self._x509, 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)[:] def public_bytes(self, encoding): bio = self._backend._create_mem_bio_gc() if encoding is serialization.Encoding.PEM: res = self._backend._lib.PEM_write_bio_X509(bio, self._x509) elif encoding is serialization.Encoding.DER: res = self._backend._lib.i2d_X509_bio(bio, self._x509) 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) @utils.register_interface(x509.RevokedCertificate) class _RevokedCertificate(object): def __init__(self, backend, crl, x509_revoked): self._backend = backend # The X509_REVOKED_value is a X509_REVOKED * that has # no reference counting. This means when X509_CRL_free is # called then the CRL and all X509_REVOKED * are freed. Since # you can retain a reference to a single revoked certificate # and let the CRL fall out of scope we need to retain a # private reference to the CRL inside the RevokedCertificate # object to prevent the gc from being called inappropriately. self._crl = crl self._x509_revoked = x509_revoked @property def serial_number(self): asn1_int = self._backend._lib.X509_REVOKED_get0_serialNumber( self._x509_revoked ) self._backend.openssl_assert(asn1_int != self._backend._ffi.NULL) return _asn1_integer_to_int(self._backend, asn1_int) @property def revocation_date(self): return _parse_asn1_time( self._backend, self._backend._lib.X509_REVOKED_get0_revocationDate( self._x509_revoked ), ) @utils.cached_property def extensions(self): return self._backend._revoked_cert_extension_parser.parse( self._x509_revoked ) @utils.register_interface(x509.CertificateRevocationList) class _CertificateRevocationList(object): def __init__(self, backend, x509_crl): self._backend = backend self._x509_crl = x509_crl def __eq__(self, other): if not isinstance(other, x509.CertificateRevocationList): return NotImplemented res = self._backend._lib.X509_CRL_cmp(self._x509_crl, other._x509_crl) return res == 0 def __ne__(self, other): return not self == other def fingerprint(self, algorithm): h = hashes.Hash(algorithm, self._backend) bio = self._backend._create_mem_bio_gc() res = self._backend._lib.i2d_X509_CRL_bio(bio, self._x509_crl) self._backend.openssl_assert(res == 1) der = self._backend._read_mem_bio(bio) h.update(der) return h.finalize() @utils.cached_property def _sorted_crl(self): # X509_CRL_get0_by_serial sorts in place, which breaks a variety of # things we don't want to break (like iteration and the signature). # Let's dupe it and sort that instead. dup = self._backend._lib.X509_CRL_dup(self._x509_crl) self._backend.openssl_assert(dup != self._backend._ffi.NULL) dup = self._backend._ffi.gc(dup, self._backend._lib.X509_CRL_free) return dup def get_revoked_certificate_by_serial_number(self, serial_number): revoked = self._backend._ffi.new("X509_REVOKED **") asn1_int = _encode_asn1_int_gc(self._backend, serial_number) res = self._backend._lib.X509_CRL_get0_by_serial( self._sorted_crl, revoked, asn1_int ) if res == 0: return None else: self._backend.openssl_assert(revoked[0] != self._backend._ffi.NULL) return _RevokedCertificate( self._backend, self._sorted_crl, revoked[0] ) @property def signature_hash_algorithm(self): oid = self.signature_algorithm_oid try: return x509._SIG_OIDS_TO_HASH[oid] except KeyError: raise UnsupportedAlgorithm( "Signature algorithm OID:{} not recognized".format(oid) ) @property def signature_algorithm_oid(self): alg = self._backend._ffi.new("X509_ALGOR **") self._backend._lib.X509_CRL_get0_signature( self._x509_crl, 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) @property def issuer(self): issuer = self._backend._lib.X509_CRL_get_issuer(self._x509_crl) self._backend.openssl_assert(issuer != self._backend._ffi.NULL) return _decode_x509_name(self._backend, issuer) @property def next_update(self): nu = self._backend._lib.X509_CRL_get_nextUpdate(self._x509_crl) self._backend.openssl_assert(nu != self._backend._ffi.NULL) return _parse_asn1_time(self._backend, nu) @property def last_update(self): lu = self._backend._lib.X509_CRL_get_lastUpdate(self._x509_crl) self._backend.openssl_assert(lu != self._backend._ffi.NULL) return _parse_asn1_time(self._backend, lu) @property def signature(self): sig = self._backend._ffi.new("ASN1_BIT_STRING **") self._backend._lib.X509_CRL_get0_signature( self._x509_crl, 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 tbs_certlist_bytes(self): pp = self._backend._ffi.new("unsigned char **") res = self._backend._lib.i2d_re_X509_CRL_tbs(self._x509_crl, 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)[:] def public_bytes(self, encoding): bio = self._backend._create_mem_bio_gc() if encoding is serialization.Encoding.PEM: res = self._backend._lib.PEM_write_bio_X509_CRL( bio, self._x509_crl ) elif encoding is serialization.Encoding.DER: res = self._backend._lib.i2d_X509_CRL_bio(bio, self._x509_crl) 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) def _revoked_cert(self, idx): revoked = self._backend._lib.X509_CRL_get_REVOKED(self._x509_crl) r = self._backend._lib.sk_X509_REVOKED_value(revoked, idx) self._backend.openssl_assert(r != self._backend._ffi.NULL) return _RevokedCertificate(self._backend, self, r) def __iter__(self): for i in range(len(self)): yield self._revoked_cert(i) def __getitem__(self, idx): if isinstance(idx, slice): start, stop, step = idx.indices(len(self)) return [self._revoked_cert(i) for i in range(start, stop, step)] else: idx = operator.index(idx) if idx < 0: idx += len(self) if not 0 <= idx < len(self): raise IndexError return self._revoked_cert(idx) def __len__(self): revoked = self._backend._lib.X509_CRL_get_REVOKED(self._x509_crl) if revoked == self._backend._ffi.NULL: return 0 else: return self._backend._lib.sk_X509_REVOKED_num(revoked) @utils.cached_property def extensions(self): return self._backend._crl_extension_parser.parse(self._x509_crl) def is_signature_valid(self, public_key): if not isinstance( public_key, (dsa.DSAPublicKey, rsa.RSAPublicKey, ec.EllipticCurvePublicKey), ): raise TypeError( "Expecting one of DSAPublicKey, RSAPublicKey," " or EllipticCurvePublicKey." ) res = self._backend._lib.X509_CRL_verify( self._x509_crl, public_key._evp_pkey ) if res != 1: self._backend._consume_errors() return False return True @utils.register_interface(x509.CertificateSigningRequest) class _CertificateSigningRequest(object): def __init__(self, backend, x509_req): self._backend = backend self._x509_req = x509_req def __eq__(self, other): 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): return not self == other def __hash__(self): return hash(self.public_bytes(serialization.Encoding.DER)) def public_key(self): 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): 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): oid = self.signature_algorithm_oid try: return x509._SIG_OIDS_TO_HASH[oid] except KeyError: raise UnsupportedAlgorithm( "Signature algorithm OID:{} not recognized".format(oid) ) @property def signature_algorithm_oid(self): 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_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): 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): 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): 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): 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): 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) @utils.register_interface( x509.certificate_transparency.SignedCertificateTimestamp ) class _SignedCertificateTimestamp(object): def __init__(self, backend, sct_list, sct): self._backend = backend # Keep the SCT_LIST that this SCT came from alive. self._sct_list = sct_list self._sct = sct @property def version(self): version = self._backend._lib.SCT_get_version(self._sct) assert version == self._backend._lib.SCT_VERSION_V1 return x509.certificate_transparency.Version.v1 @property def log_id(self): out = self._backend._ffi.new("unsigned char **") log_id_length = self._backend._lib.SCT_get0_log_id(self._sct, out) assert log_id_length >= 0 return self._backend._ffi.buffer(out[0], log_id_length)[:] @property def timestamp(self): timestamp = self._backend._lib.SCT_get_timestamp(self._sct) milliseconds = timestamp % 1000 return datetime.datetime.utcfromtimestamp(timestamp // 1000).replace( microsecond=milliseconds * 1000 ) @property def entry_type(self): entry_type = self._backend._lib.SCT_get_log_entry_type(self._sct) # We currently only support loading SCTs from the X.509 extension, so # we only have precerts. assert entry_type == self._backend._lib.CT_LOG_ENTRY_TYPE_PRECERT return x509.certificate_transparency.LogEntryType.PRE_CERTIFICATE @property def _signature(self): ptrptr = self._backend._ffi.new("unsigned char **") res = self._backend._lib.SCT_get0_signature(self._sct, ptrptr) self._backend.openssl_assert(res > 0) self._backend.openssl_assert(ptrptr[0] != self._backend._ffi.NULL) return self._backend._ffi.buffer(ptrptr[0], res)[:] def __hash__(self): return hash(self._signature) def __eq__(self, other): if not isinstance(other, _SignedCertificateTimestamp): return NotImplemented return self._signature == other._signature def __ne__(self, other): return not self == other