diff options
author | Alex Gaynor <alex.gaynor@gmail.com> | 2016-01-10 12:10:05 -0500 |
---|---|---|
committer | Alex Gaynor <alex.gaynor@gmail.com> | 2016-01-10 12:10:05 -0500 |
commit | 3f6f8f593e0bd9910d406ec56bfeceba5a2badbf (patch) | |
tree | 894a48662e6bc88ce5f0fa3ffc7130905a5b22dc | |
parent | cba511154b1ecc4dbea19a2f2ffbb908c66a3aed (diff) | |
parent | 3917fab05b8ce62cf1ed11ea66d4a67c27fc50b7 (diff) | |
download | cryptography-3f6f8f593e0bd9910d406ec56bfeceba5a2badbf.tar.gz cryptography-3f6f8f593e0bd9910d406ec56bfeceba5a2badbf.tar.bz2 cryptography-3f6f8f593e0bd9910d406ec56bfeceba5a2badbf.zip |
Merge pull request #2659 from reaperhulk/move-encode-functions
move openssl asn1 encode functions to a new module
-rw-r--r-- | src/cryptography/hazmat/backends/openssl/backend.py | 595 | ||||
-rw-r--r-- | src/cryptography/hazmat/backends/openssl/encode_asn1.py | 599 |
2 files changed, 607 insertions, 587 deletions
diff --git a/src/cryptography/hazmat/backends/openssl/backend.py b/src/cryptography/hazmat/backends/openssl/backend.py index c21d5427..57d36acd 100644 --- a/src/cryptography/hazmat/backends/openssl/backend.py +++ b/src/cryptography/hazmat/backends/openssl/backend.py @@ -10,8 +10,6 @@ import datetime import itertools from contextlib import contextmanager -import idna - import six from cryptography import utils, x509 @@ -31,15 +29,20 @@ from cryptography.hazmat.backends.openssl.dsa import ( from cryptography.hazmat.backends.openssl.ec import ( _EllipticCurvePrivateKey, _EllipticCurvePublicKey ) +from cryptography.hazmat.backends.openssl.encode_asn1 import ( + _CRL_ENTRY_EXTENSION_ENCODE_HANDLERS, + _CRL_EXTENSION_ENCODE_HANDLERS, _EXTENSION_ENCODE_HANDLERS, + _encode_asn1_int_gc, _encode_asn1_str_gc, _encode_name_gc, + _txt2obj_gc, +) from cryptography.hazmat.backends.openssl.hashes import _HashContext from cryptography.hazmat.backends.openssl.hmac import _HMACContext from cryptography.hazmat.backends.openssl.rsa import ( _RSAPrivateKey, _RSAPublicKey ) from cryptography.hazmat.backends.openssl.x509 import ( - _CRL_ENTRY_REASON_ENUM_TO_CODE, _Certificate, _CertificateRevocationList, - _CertificateSigningRequest, _DISTPOINT_TYPE_FULLNAME, - _DISTPOINT_TYPE_RELATIVENAME, _RevokedCertificate + _Certificate, _CertificateRevocationList, + _CertificateSigningRequest, _RevokedCertificate ) from cryptography.hazmat.bindings._openssl import ffi as _ffi from cryptography.hazmat.bindings.openssl import binding @@ -54,593 +57,11 @@ from cryptography.hazmat.primitives.ciphers.algorithms import ( from cryptography.hazmat.primitives.ciphers.modes import ( CBC, CFB, CFB8, CTR, ECB, GCM, OFB ) -from cryptography.x509.oid import CRLEntryExtensionOID, ExtensionOID, NameOID _MemoryBIO = collections.namedtuple("_MemoryBIO", ["bio", "char_ptr"]) -def _encode_asn1_int(backend, x): - """ - Converts a python integer to an ASN1_INTEGER. The returned ASN1_INTEGER - will not be garbage collected (to support adding them to structs that take - ownership of the object). Be sure to register it for GC if it will be - discarded after use. - - """ - # Convert Python integer to OpenSSL "bignum" in case value exceeds - # machine's native integer limits (note: `int_to_bn` doesn't automatically - # GC). - i = backend._int_to_bn(x) - i = backend._ffi.gc(i, backend._lib.BN_free) - - # Wrap in an ASN.1 integer. Don't GC -- as documented. - i = backend._lib.BN_to_ASN1_INTEGER(i, backend._ffi.NULL) - backend.openssl_assert(i != backend._ffi.NULL) - return i - - -def _encode_asn1_int_gc(backend, x): - i = _encode_asn1_int(backend, x) - i = backend._ffi.gc(i, backend._lib.ASN1_INTEGER_free) - return i - - -def _encode_asn1_str(backend, data, length): - """ - Create an ASN1_OCTET_STRING from a Python byte string. - """ - s = backend._lib.ASN1_OCTET_STRING_new() - res = backend._lib.ASN1_OCTET_STRING_set(s, data, length) - backend.openssl_assert(res == 1) - return s - - -def _encode_asn1_utf8_str(backend, string): - """ - Create an ASN1_UTF8STRING from a Python unicode string. - This object will be an ASN1_STRING with UTF8 type in OpenSSL and - can be decoded with ASN1_STRING_to_UTF8. - """ - s = backend._lib.ASN1_UTF8STRING_new() - res = backend._lib.ASN1_STRING_set( - s, string.encode("utf8"), len(string.encode("utf8")) - ) - backend.openssl_assert(res == 1) - return s - - -def _encode_asn1_str_gc(backend, data, length): - s = _encode_asn1_str(backend, data, length) - s = backend._ffi.gc(s, backend._lib.ASN1_OCTET_STRING_free) - return s - - -def _encode_extension_to_der(backend, i2d_func, value): - pp = backend._ffi.new("unsigned char **") - r = i2d_func(value, pp) - backend.openssl_assert(r > 0) - pp = backend._ffi.gc( - pp, lambda pointer: backend._lib.OPENSSL_free(pointer[0]) - ) - return pp, r - - -def _encode_inhibit_any_policy(backend, inhibit_any_policy): - asn1int = _encode_asn1_int_gc(backend, inhibit_any_policy.skip_certs) - return _encode_extension_to_der( - backend, backend._lib.i2d_ASN1_INTEGER, asn1int - ) - - -def _encode_name(backend, attributes): - """ - The X509_NAME created will not be gc'd. Use _encode_name_gc if needed. - """ - subject = backend._lib.X509_NAME_new() - for attribute in attributes: - value = attribute.value.encode('utf8') - obj = _txt2obj_gc(backend, attribute.oid.dotted_string) - if attribute.oid == NameOID.COUNTRY_NAME: - # Per RFC5280 Appendix A.1 countryName should be encoded as - # PrintableString, not UTF8String - type = backend._lib.MBSTRING_ASC - else: - type = backend._lib.MBSTRING_UTF8 - res = backend._lib.X509_NAME_add_entry_by_OBJ( - subject, obj, type, value, -1, -1, 0, - ) - backend.openssl_assert(res == 1) - return subject - - -def _encode_name_gc(backend, attributes): - subject = _encode_name(backend, attributes) - subject = backend._ffi.gc(subject, backend._lib.X509_NAME_free) - return subject - - -def _encode_crl_number(backend, crl_number): - asn1int = _encode_asn1_int_gc(backend, crl_number.crl_number) - return _encode_extension_to_der( - backend, backend._lib.i2d_ASN1_INTEGER, asn1int - ) - - -def _encode_crl_reason(backend, crl_reason): - asn1enum = backend._lib.ASN1_ENUMERATED_new() - backend.openssl_assert(asn1enum != backend._ffi.NULL) - asn1enum = backend._ffi.gc(asn1enum, backend._lib.ASN1_ENUMERATED_free) - res = backend._lib.ASN1_ENUMERATED_set( - asn1enum, _CRL_ENTRY_REASON_ENUM_TO_CODE[crl_reason.reason] - ) - backend.openssl_assert(res == 1) - - return _encode_extension_to_der( - backend, backend._lib.i2d_ASN1_ENUMERATED, asn1enum - ) - - -def _encode_invalidity_date(backend, invalidity_date): - time = backend._lib.ASN1_GENERALIZEDTIME_set( - backend._ffi.NULL, calendar.timegm( - invalidity_date.invalidity_date.timetuple() - ) - ) - backend.openssl_assert(time != backend._ffi.NULL) - time = backend._ffi.gc(time, backend._lib.ASN1_GENERALIZEDTIME_free) - - return _encode_extension_to_der( - backend, backend._lib.i2d_ASN1_GENERALIZEDTIME, time - ) - - -def _encode_certificate_policies(backend, certificate_policies): - cp = backend._lib.sk_POLICYINFO_new_null() - backend.openssl_assert(cp != backend._ffi.NULL) - cp = backend._ffi.gc(cp, backend._lib.sk_POLICYINFO_free) - for policy_info in certificate_policies: - pi = backend._lib.POLICYINFO_new() - backend.openssl_assert(pi != backend._ffi.NULL) - res = backend._lib.sk_POLICYINFO_push(cp, pi) - backend.openssl_assert(res >= 1) - oid = _txt2obj(backend, policy_info.policy_identifier.dotted_string) - pi.policyid = oid - if policy_info.policy_qualifiers: - pqis = backend._lib.sk_POLICYQUALINFO_new_null() - backend.openssl_assert(pqis != backend._ffi.NULL) - for qualifier in policy_info.policy_qualifiers: - pqi = backend._lib.POLICYQUALINFO_new() - backend.openssl_assert(pqi != backend._ffi.NULL) - res = backend._lib.sk_POLICYQUALINFO_push(pqis, pqi) - backend.openssl_assert(res >= 1) - if isinstance(qualifier, six.text_type): - pqi.pqualid = _txt2obj( - backend, x509.OID_CPS_QUALIFIER.dotted_string - ) - pqi.d.cpsuri = _encode_asn1_str( - backend, - qualifier.encode("ascii"), - len(qualifier.encode("ascii")) - ) - else: - assert isinstance(qualifier, x509.UserNotice) - pqi.pqualid = _txt2obj( - backend, x509.OID_CPS_USER_NOTICE.dotted_string - ) - un = backend._lib.USERNOTICE_new() - backend.openssl_assert(un != backend._ffi.NULL) - pqi.d.usernotice = un - if qualifier.explicit_text: - un.exptext = _encode_asn1_utf8_str( - backend, qualifier.explicit_text - ) - - un.noticeref = _encode_notice_reference( - backend, qualifier.notice_reference - ) - - pi.qualifiers = pqis - - return _encode_extension_to_der( - backend, backend._lib.i2d_CERTIFICATEPOLICIES, cp - ) - - -def _encode_notice_reference(backend, notice): - if notice is None: - return backend._ffi.NULL - else: - nr = backend._lib.NOTICEREF_new() - backend.openssl_assert(nr != backend._ffi.NULL) - # organization is a required field - nr.organization = _encode_asn1_utf8_str(backend, notice.organization) - - notice_stack = backend._lib.sk_ASN1_INTEGER_new_null() - nr.noticenos = notice_stack - for number in notice.notice_numbers: - num = _encode_asn1_int(backend, number) - res = backend._lib.sk_ASN1_INTEGER_push(notice_stack, num) - backend.openssl_assert(res >= 1) - - return nr - - -def _txt2obj(backend, name): - """ - Converts a Python string with an ASN.1 object ID in dotted form to a - ASN1_OBJECT. - """ - name = name.encode('ascii') - obj = backend._lib.OBJ_txt2obj(name, 1) - backend.openssl_assert(obj != backend._ffi.NULL) - return obj - - -def _txt2obj_gc(backend, name): - obj = _txt2obj(backend, name) - obj = backend._ffi.gc(obj, backend._lib.ASN1_OBJECT_free) - return obj - - -def _encode_ocsp_nocheck(backend, ext): - """ - The OCSP No Check extension is defined as a null ASN.1 value. Rather than - calling OpenSSL we can return a Python bytestring value in a list. - """ - return [b"\x05\x00"], 2 - - -def _encode_key_usage(backend, key_usage): - set_bit = backend._lib.ASN1_BIT_STRING_set_bit - ku = backend._lib.ASN1_BIT_STRING_new() - ku = backend._ffi.gc(ku, backend._lib.ASN1_BIT_STRING_free) - res = set_bit(ku, 0, key_usage.digital_signature) - backend.openssl_assert(res == 1) - res = set_bit(ku, 1, key_usage.content_commitment) - backend.openssl_assert(res == 1) - res = set_bit(ku, 2, key_usage.key_encipherment) - backend.openssl_assert(res == 1) - res = set_bit(ku, 3, key_usage.data_encipherment) - backend.openssl_assert(res == 1) - res = set_bit(ku, 4, key_usage.key_agreement) - backend.openssl_assert(res == 1) - res = set_bit(ku, 5, key_usage.key_cert_sign) - backend.openssl_assert(res == 1) - res = set_bit(ku, 6, key_usage.crl_sign) - backend.openssl_assert(res == 1) - if key_usage.key_agreement: - res = set_bit(ku, 7, key_usage.encipher_only) - backend.openssl_assert(res == 1) - res = set_bit(ku, 8, key_usage.decipher_only) - backend.openssl_assert(res == 1) - else: - res = set_bit(ku, 7, 0) - backend.openssl_assert(res == 1) - res = set_bit(ku, 8, 0) - backend.openssl_assert(res == 1) - - return _encode_extension_to_der( - backend, backend._lib.i2d_ASN1_BIT_STRING, ku - ) - - -def _encode_authority_key_identifier(backend, authority_keyid): - akid = backend._lib.AUTHORITY_KEYID_new() - backend.openssl_assert(akid != backend._ffi.NULL) - akid = backend._ffi.gc(akid, backend._lib.AUTHORITY_KEYID_free) - if authority_keyid.key_identifier is not None: - akid.keyid = _encode_asn1_str( - backend, - authority_keyid.key_identifier, - len(authority_keyid.key_identifier) - ) - - if authority_keyid.authority_cert_issuer is not None: - akid.issuer = _encode_general_names( - backend, authority_keyid.authority_cert_issuer - ) - - if authority_keyid.authority_cert_serial_number is not None: - akid.serial = _encode_asn1_int( - backend, authority_keyid.authority_cert_serial_number - ) - - return _encode_extension_to_der( - backend, backend._lib.i2d_AUTHORITY_KEYID, akid - ) - - -def _encode_basic_constraints(backend, basic_constraints): - constraints = backend._lib.BASIC_CONSTRAINTS_new() - constraints = backend._ffi.gc( - constraints, backend._lib.BASIC_CONSTRAINTS_free - ) - constraints.ca = 255 if basic_constraints.ca else 0 - if basic_constraints.ca and basic_constraints.path_length is not None: - constraints.pathlen = _encode_asn1_int( - backend, basic_constraints.path_length - ) - - return _encode_extension_to_der( - backend, backend._lib.i2d_BASIC_CONSTRAINTS, constraints - ) - - -def _encode_authority_information_access(backend, authority_info_access): - aia = backend._lib.sk_ACCESS_DESCRIPTION_new_null() - backend.openssl_assert(aia != backend._ffi.NULL) - aia = backend._ffi.gc( - aia, backend._lib.sk_ACCESS_DESCRIPTION_free - ) - for access_description in authority_info_access: - ad = backend._lib.ACCESS_DESCRIPTION_new() - method = _txt2obj( - backend, access_description.access_method.dotted_string - ) - gn = _encode_general_name(backend, access_description.access_location) - ad.method = method - ad.location = gn - res = backend._lib.sk_ACCESS_DESCRIPTION_push(aia, ad) - backend.openssl_assert(res >= 1) - - return _encode_extension_to_der( - backend, backend._lib.i2d_AUTHORITY_INFO_ACCESS, aia - ) - - -def _encode_general_names(backend, names): - general_names = backend._lib.GENERAL_NAMES_new() - backend.openssl_assert(general_names != backend._ffi.NULL) - for name in names: - gn = _encode_general_name(backend, name) - res = backend._lib.sk_GENERAL_NAME_push(general_names, gn) - backend.openssl_assert(res != 0) - - return general_names - - -def _encode_alt_name(backend, san): - general_names = _encode_general_names(backend, san) - general_names = backend._ffi.gc( - general_names, backend._lib.GENERAL_NAMES_free - ) - return _encode_extension_to_der( - backend, backend._lib.i2d_GENERAL_NAMES, general_names - ) - - -def _encode_subject_key_identifier(backend, ski): - asn1_str = _encode_asn1_str_gc(backend, ski.digest, len(ski.digest)) - return _encode_extension_to_der( - backend, backend._lib.i2d_ASN1_OCTET_STRING, asn1_str - ) - - -def _encode_general_name(backend, name): - if isinstance(name, x509.DNSName): - gn = backend._lib.GENERAL_NAME_new() - backend.openssl_assert(gn != backend._ffi.NULL) - gn.type = backend._lib.GEN_DNS - - ia5 = backend._lib.ASN1_IA5STRING_new() - backend.openssl_assert(ia5 != backend._ffi.NULL) - - if name.value.startswith(u"*."): - value = b"*." + idna.encode(name.value[2:]) - else: - value = idna.encode(name.value) - - res = backend._lib.ASN1_STRING_set(ia5, value, len(value)) - backend.openssl_assert(res == 1) - gn.d.dNSName = ia5 - elif isinstance(name, x509.RegisteredID): - gn = backend._lib.GENERAL_NAME_new() - backend.openssl_assert(gn != backend._ffi.NULL) - gn.type = backend._lib.GEN_RID - obj = backend._lib.OBJ_txt2obj( - name.value.dotted_string.encode('ascii'), 1 - ) - backend.openssl_assert(obj != backend._ffi.NULL) - gn.d.registeredID = obj - elif isinstance(name, x509.DirectoryName): - gn = backend._lib.GENERAL_NAME_new() - backend.openssl_assert(gn != backend._ffi.NULL) - dir_name = _encode_name(backend, name.value) - gn.type = backend._lib.GEN_DIRNAME - gn.d.directoryName = dir_name - elif isinstance(name, x509.IPAddress): - gn = backend._lib.GENERAL_NAME_new() - backend.openssl_assert(gn != backend._ffi.NULL) - ipaddr = _encode_asn1_str( - backend, name.value.packed, len(name.value.packed) - ) - gn.type = backend._lib.GEN_IPADD - gn.d.iPAddress = ipaddr - elif isinstance(name, x509.OtherName): - gn = backend._lib.GENERAL_NAME_new() - backend.openssl_assert(gn != backend._ffi.NULL) - other_name = backend._lib.OTHERNAME_new() - backend.openssl_assert(other_name != backend._ffi.NULL) - - type_id = backend._lib.OBJ_txt2obj( - name.type_id.dotted_string.encode('ascii'), 1 - ) - backend.openssl_assert(type_id != backend._ffi.NULL) - data = backend._ffi.new("unsigned char[]", name.value) - data_ptr_ptr = backend._ffi.new("unsigned char **") - data_ptr_ptr[0] = data - value = backend._lib.d2i_ASN1_TYPE( - backend._ffi.NULL, data_ptr_ptr, len(name.value) - ) - if value == backend._ffi.NULL: - backend._consume_errors() - raise ValueError("Invalid ASN.1 data") - other_name.type_id = type_id - other_name.value = value - gn.type = backend._lib.GEN_OTHERNAME - gn.d.otherName = other_name - elif isinstance(name, x509.RFC822Name): - gn = backend._lib.GENERAL_NAME_new() - backend.openssl_assert(gn != backend._ffi.NULL) - asn1_str = _encode_asn1_str( - backend, name._encoded, len(name._encoded) - ) - gn.type = backend._lib.GEN_EMAIL - gn.d.rfc822Name = asn1_str - elif isinstance(name, x509.UniformResourceIdentifier): - gn = backend._lib.GENERAL_NAME_new() - backend.openssl_assert(gn != backend._ffi.NULL) - asn1_str = _encode_asn1_str( - backend, name._encoded, len(name._encoded) - ) - gn.type = backend._lib.GEN_URI - gn.d.uniformResourceIdentifier = asn1_str - else: - raise ValueError( - "{0} is an unknown GeneralName type".format(name) - ) - - return gn - - -def _encode_extended_key_usage(backend, extended_key_usage): - eku = backend._lib.sk_ASN1_OBJECT_new_null() - eku = backend._ffi.gc(eku, backend._lib.sk_ASN1_OBJECT_free) - for oid in extended_key_usage: - obj = _txt2obj(backend, oid.dotted_string) - res = backend._lib.sk_ASN1_OBJECT_push(eku, obj) - backend.openssl_assert(res >= 1) - - eku_ptr = backend._ffi.cast("EXTENDED_KEY_USAGE *", eku) - return _encode_extension_to_der( - backend, backend._lib.i2d_EXTENDED_KEY_USAGE, eku_ptr - ) - - -_CRLREASONFLAGS = { - x509.ReasonFlags.key_compromise: 1, - x509.ReasonFlags.ca_compromise: 2, - x509.ReasonFlags.affiliation_changed: 3, - x509.ReasonFlags.superseded: 4, - x509.ReasonFlags.cessation_of_operation: 5, - x509.ReasonFlags.certificate_hold: 6, - x509.ReasonFlags.privilege_withdrawn: 7, - x509.ReasonFlags.aa_compromise: 8, -} - - -def _encode_crl_distribution_points(backend, crl_distribution_points): - cdp = backend._lib.sk_DIST_POINT_new_null() - cdp = backend._ffi.gc(cdp, backend._lib.sk_DIST_POINT_free) - for point in crl_distribution_points: - dp = backend._lib.DIST_POINT_new() - backend.openssl_assert(dp != backend._ffi.NULL) - - if point.reasons: - bitmask = backend._lib.ASN1_BIT_STRING_new() - backend.openssl_assert(bitmask != backend._ffi.NULL) - dp.reasons = bitmask - for reason in point.reasons: - res = backend._lib.ASN1_BIT_STRING_set_bit( - bitmask, _CRLREASONFLAGS[reason], 1 - ) - backend.openssl_assert(res == 1) - - if point.full_name: - dpn = backend._lib.DIST_POINT_NAME_new() - backend.openssl_assert(dpn != backend._ffi.NULL) - dpn.type = _DISTPOINT_TYPE_FULLNAME - dpn.name.fullname = _encode_general_names(backend, point.full_name) - dp.distpoint = dpn - - if point.relative_name: - dpn = backend._lib.DIST_POINT_NAME_new() - backend.openssl_assert(dpn != backend._ffi.NULL) - dpn.type = _DISTPOINT_TYPE_RELATIVENAME - name = _encode_name_gc(backend, point.relative_name) - relativename = backend._lib.sk_X509_NAME_ENTRY_dup(name.entries) - backend.openssl_assert(relativename != backend._ffi.NULL) - dpn.name.relativename = relativename - dp.distpoint = dpn - - if point.crl_issuer: - dp.CRLissuer = _encode_general_names(backend, point.crl_issuer) - - res = backend._lib.sk_DIST_POINT_push(cdp, dp) - backend.openssl_assert(res >= 1) - - return _encode_extension_to_der( - backend, backend._lib.i2d_CRL_DIST_POINTS, cdp - ) - - -def _encode_name_constraints(backend, name_constraints): - nc = backend._lib.NAME_CONSTRAINTS_new() - assert nc != backend._ffi.NULL - nc = backend._ffi.gc(nc, backend._lib.NAME_CONSTRAINTS_free) - permitted = _encode_general_subtree( - backend, name_constraints.permitted_subtrees - ) - nc.permittedSubtrees = permitted - excluded = _encode_general_subtree( - backend, name_constraints.excluded_subtrees - ) - nc.excludedSubtrees = excluded - - return _encode_extension_to_der( - backend, backend._lib.Cryptography_i2d_NAME_CONSTRAINTS, nc - ) - - -def _encode_general_subtree(backend, subtrees): - if subtrees is None: - return backend._ffi.NULL - else: - general_subtrees = backend._lib.sk_GENERAL_SUBTREE_new_null() - for name in subtrees: - gs = backend._lib.GENERAL_SUBTREE_new() - gs.base = _encode_general_name(backend, name) - res = backend._lib.sk_GENERAL_SUBTREE_push(general_subtrees, gs) - assert res >= 1 - - return general_subtrees - - -_EXTENSION_ENCODE_HANDLERS = { - ExtensionOID.BASIC_CONSTRAINTS: _encode_basic_constraints, - ExtensionOID.SUBJECT_KEY_IDENTIFIER: _encode_subject_key_identifier, - ExtensionOID.KEY_USAGE: _encode_key_usage, - ExtensionOID.SUBJECT_ALTERNATIVE_NAME: _encode_alt_name, - ExtensionOID.ISSUER_ALTERNATIVE_NAME: _encode_alt_name, - ExtensionOID.EXTENDED_KEY_USAGE: _encode_extended_key_usage, - ExtensionOID.AUTHORITY_KEY_IDENTIFIER: _encode_authority_key_identifier, - ExtensionOID.CERTIFICATE_POLICIES: _encode_certificate_policies, - ExtensionOID.AUTHORITY_INFORMATION_ACCESS: ( - _encode_authority_information_access - ), - ExtensionOID.CRL_DISTRIBUTION_POINTS: _encode_crl_distribution_points, - ExtensionOID.INHIBIT_ANY_POLICY: _encode_inhibit_any_policy, - ExtensionOID.OCSP_NO_CHECK: _encode_ocsp_nocheck, - ExtensionOID.NAME_CONSTRAINTS: _encode_name_constraints, -} - -_CRL_EXTENSION_ENCODE_HANDLERS = { - ExtensionOID.ISSUER_ALTERNATIVE_NAME: _encode_alt_name, - ExtensionOID.AUTHORITY_KEY_IDENTIFIER: _encode_authority_key_identifier, - ExtensionOID.AUTHORITY_INFORMATION_ACCESS: ( - _encode_authority_information_access - ), - ExtensionOID.CRL_NUMBER: _encode_crl_number, -} - -_CRL_ENTRY_EXTENSION_ENCODE_HANDLERS = { - CRLEntryExtensionOID.CERTIFICATE_ISSUER: _encode_alt_name, - CRLEntryExtensionOID.CRL_REASON: _encode_crl_reason, - CRLEntryExtensionOID.INVALIDITY_DATE: _encode_invalidity_date, -} - - class _PasswordUserdata(object): def __init__(self, password): self.password = password diff --git a/src/cryptography/hazmat/backends/openssl/encode_asn1.py b/src/cryptography/hazmat/backends/openssl/encode_asn1.py new file mode 100644 index 00000000..dc41670d --- /dev/null +++ b/src/cryptography/hazmat/backends/openssl/encode_asn1.py @@ -0,0 +1,599 @@ +# 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 calendar + +import idna + +import six + +from cryptography import x509 +from cryptography.hazmat.backends.openssl.x509 import ( + _CRL_ENTRY_REASON_ENUM_TO_CODE, _DISTPOINT_TYPE_FULLNAME, + _DISTPOINT_TYPE_RELATIVENAME +) +from cryptography.x509.oid import CRLEntryExtensionOID, ExtensionOID, NameOID + + +def _encode_asn1_int(backend, x): + """ + Converts a python integer to an ASN1_INTEGER. The returned ASN1_INTEGER + will not be garbage collected (to support adding them to structs that take + ownership of the object). Be sure to register it for GC if it will be + discarded after use. + + """ + # Convert Python integer to OpenSSL "bignum" in case value exceeds + # machine's native integer limits (note: `int_to_bn` doesn't automatically + # GC). + i = backend._int_to_bn(x) + i = backend._ffi.gc(i, backend._lib.BN_free) + + # Wrap in an ASN.1 integer. Don't GC -- as documented. + i = backend._lib.BN_to_ASN1_INTEGER(i, backend._ffi.NULL) + backend.openssl_assert(i != backend._ffi.NULL) + return i + + +def _encode_asn1_int_gc(backend, x): + i = _encode_asn1_int(backend, x) + i = backend._ffi.gc(i, backend._lib.ASN1_INTEGER_free) + return i + + +def _encode_asn1_str(backend, data, length): + """ + Create an ASN1_OCTET_STRING from a Python byte string. + """ + s = backend._lib.ASN1_OCTET_STRING_new() + res = backend._lib.ASN1_OCTET_STRING_set(s, data, length) + backend.openssl_assert(res == 1) + return s + + +def _encode_asn1_utf8_str(backend, string): + """ + Create an ASN1_UTF8STRING from a Python unicode string. + This object will be an ASN1_STRING with UTF8 type in OpenSSL and + can be decoded with ASN1_STRING_to_UTF8. + """ + s = backend._lib.ASN1_UTF8STRING_new() + res = backend._lib.ASN1_STRING_set( + s, string.encode("utf8"), len(string.encode("utf8")) + ) + backend.openssl_assert(res == 1) + return s + + +def _encode_asn1_str_gc(backend, data, length): + s = _encode_asn1_str(backend, data, length) + s = backend._ffi.gc(s, backend._lib.ASN1_OCTET_STRING_free) + return s + + +def _encode_extension_to_der(backend, i2d_func, value): + pp = backend._ffi.new("unsigned char **") + r = i2d_func(value, pp) + backend.openssl_assert(r > 0) + pp = backend._ffi.gc( + pp, lambda pointer: backend._lib.OPENSSL_free(pointer[0]) + ) + return pp, r + + +def _encode_inhibit_any_policy(backend, inhibit_any_policy): + asn1int = _encode_asn1_int_gc(backend, inhibit_any_policy.skip_certs) + return _encode_extension_to_der( + backend, backend._lib.i2d_ASN1_INTEGER, asn1int + ) + + +def _encode_name(backend, attributes): + """ + The X509_NAME created will not be gc'd. Use _encode_name_gc if needed. + """ + subject = backend._lib.X509_NAME_new() + for attribute in attributes: + value = attribute.value.encode('utf8') + obj = _txt2obj_gc(backend, attribute.oid.dotted_string) + if attribute.oid == NameOID.COUNTRY_NAME: + # Per RFC5280 Appendix A.1 countryName should be encoded as + # PrintableString, not UTF8String + type = backend._lib.MBSTRING_ASC + else: + type = backend._lib.MBSTRING_UTF8 + res = backend._lib.X509_NAME_add_entry_by_OBJ( + subject, obj, type, value, -1, -1, 0, + ) + backend.openssl_assert(res == 1) + return subject + + +def _encode_name_gc(backend, attributes): + subject = _encode_name(backend, attributes) + subject = backend._ffi.gc(subject, backend._lib.X509_NAME_free) + return subject + + +def _encode_crl_number(backend, crl_number): + asn1int = _encode_asn1_int_gc(backend, crl_number.crl_number) + return _encode_extension_to_der( + backend, backend._lib.i2d_ASN1_INTEGER, asn1int + ) + + +def _encode_crl_reason(backend, crl_reason): + asn1enum = backend._lib.ASN1_ENUMERATED_new() + backend.openssl_assert(asn1enum != backend._ffi.NULL) + asn1enum = backend._ffi.gc(asn1enum, backend._lib.ASN1_ENUMERATED_free) + res = backend._lib.ASN1_ENUMERATED_set( + asn1enum, _CRL_ENTRY_REASON_ENUM_TO_CODE[crl_reason.reason] + ) + backend.openssl_assert(res == 1) + + return _encode_extension_to_der( + backend, backend._lib.i2d_ASN1_ENUMERATED, asn1enum + ) + + +def _encode_invalidity_date(backend, invalidity_date): + time = backend._lib.ASN1_GENERALIZEDTIME_set( + backend._ffi.NULL, calendar.timegm( + invalidity_date.invalidity_date.timetuple() + ) + ) + backend.openssl_assert(time != backend._ffi.NULL) + time = backend._ffi.gc(time, backend._lib.ASN1_GENERALIZEDTIME_free) + + return _encode_extension_to_der( + backend, backend._lib.i2d_ASN1_GENERALIZEDTIME, time + ) + + +def _encode_certificate_policies(backend, certificate_policies): + cp = backend._lib.sk_POLICYINFO_new_null() + backend.openssl_assert(cp != backend._ffi.NULL) + cp = backend._ffi.gc(cp, backend._lib.sk_POLICYINFO_free) + for policy_info in certificate_policies: + pi = backend._lib.POLICYINFO_new() + backend.openssl_assert(pi != backend._ffi.NULL) + res = backend._lib.sk_POLICYINFO_push(cp, pi) + backend.openssl_assert(res >= 1) + oid = _txt2obj(backend, policy_info.policy_identifier.dotted_string) + pi.policyid = oid + if policy_info.policy_qualifiers: + pqis = backend._lib.sk_POLICYQUALINFO_new_null() + backend.openssl_assert(pqis != backend._ffi.NULL) + for qualifier in policy_info.policy_qualifiers: + pqi = backend._lib.POLICYQUALINFO_new() + backend.openssl_assert(pqi != backend._ffi.NULL) + res = backend._lib.sk_POLICYQUALINFO_push(pqis, pqi) + backend.openssl_assert(res >= 1) + if isinstance(qualifier, six.text_type): + pqi.pqualid = _txt2obj( + backend, x509.OID_CPS_QUALIFIER.dotted_string + ) + pqi.d.cpsuri = _encode_asn1_str( + backend, + qualifier.encode("ascii"), + len(qualifier.encode("ascii")) + ) + else: + assert isinstance(qualifier, x509.UserNotice) + pqi.pqualid = _txt2obj( + backend, x509.OID_CPS_USER_NOTICE.dotted_string + ) + un = backend._lib.USERNOTICE_new() + backend.openssl_assert(un != backend._ffi.NULL) + pqi.d.usernotice = un + if qualifier.explicit_text: + un.exptext = _encode_asn1_utf8_str( + backend, qualifier.explicit_text + ) + + un.noticeref = _encode_notice_reference( + backend, qualifier.notice_reference + ) + + pi.qualifiers = pqis + + return _encode_extension_to_der( + backend, backend._lib.i2d_CERTIFICATEPOLICIES, cp + ) + + +def _encode_notice_reference(backend, notice): + if notice is None: + return backend._ffi.NULL + else: + nr = backend._lib.NOTICEREF_new() + backend.openssl_assert(nr != backend._ffi.NULL) + # organization is a required field + nr.organization = _encode_asn1_utf8_str(backend, notice.organization) + + notice_stack = backend._lib.sk_ASN1_INTEGER_new_null() + nr.noticenos = notice_stack + for number in notice.notice_numbers: + num = _encode_asn1_int(backend, number) + res = backend._lib.sk_ASN1_INTEGER_push(notice_stack, num) + backend.openssl_assert(res >= 1) + + return nr + + +def _txt2obj(backend, name): + """ + Converts a Python string with an ASN.1 object ID in dotted form to a + ASN1_OBJECT. + """ + name = name.encode('ascii') + obj = backend._lib.OBJ_txt2obj(name, 1) + backend.openssl_assert(obj != backend._ffi.NULL) + return obj + + +def _txt2obj_gc(backend, name): + obj = _txt2obj(backend, name) + obj = backend._ffi.gc(obj, backend._lib.ASN1_OBJECT_free) + return obj + + +def _encode_ocsp_nocheck(backend, ext): + """ + The OCSP No Check extension is defined as a null ASN.1 value. Rather than + calling OpenSSL we can return a Python bytestring value in a list. + """ + return [b"\x05\x00"], 2 + + +def _encode_key_usage(backend, key_usage): + set_bit = backend._lib.ASN1_BIT_STRING_set_bit + ku = backend._lib.ASN1_BIT_STRING_new() + ku = backend._ffi.gc(ku, backend._lib.ASN1_BIT_STRING_free) + res = set_bit(ku, 0, key_usage.digital_signature) + backend.openssl_assert(res == 1) + res = set_bit(ku, 1, key_usage.content_commitment) + backend.openssl_assert(res == 1) + res = set_bit(ku, 2, key_usage.key_encipherment) + backend.openssl_assert(res == 1) + res = set_bit(ku, 3, key_usage.data_encipherment) + backend.openssl_assert(res == 1) + res = set_bit(ku, 4, key_usage.key_agreement) + backend.openssl_assert(res == 1) + res = set_bit(ku, 5, key_usage.key_cert_sign) + backend.openssl_assert(res == 1) + res = set_bit(ku, 6, key_usage.crl_sign) + backend.openssl_assert(res == 1) + if key_usage.key_agreement: + res = set_bit(ku, 7, key_usage.encipher_only) + backend.openssl_assert(res == 1) + res = set_bit(ku, 8, key_usage.decipher_only) + backend.openssl_assert(res == 1) + else: + res = set_bit(ku, 7, 0) + backend.openssl_assert(res == 1) + res = set_bit(ku, 8, 0) + backend.openssl_assert(res == 1) + + return _encode_extension_to_der( + backend, backend._lib.i2d_ASN1_BIT_STRING, ku + ) + + +def _encode_authority_key_identifier(backend, authority_keyid): + akid = backend._lib.AUTHORITY_KEYID_new() + backend.openssl_assert(akid != backend._ffi.NULL) + akid = backend._ffi.gc(akid, backend._lib.AUTHORITY_KEYID_free) + if authority_keyid.key_identifier is not None: + akid.keyid = _encode_asn1_str( + backend, + authority_keyid.key_identifier, + len(authority_keyid.key_identifier) + ) + + if authority_keyid.authority_cert_issuer is not None: + akid.issuer = _encode_general_names( + backend, authority_keyid.authority_cert_issuer + ) + + if authority_keyid.authority_cert_serial_number is not None: + akid.serial = _encode_asn1_int( + backend, authority_keyid.authority_cert_serial_number + ) + + return _encode_extension_to_der( + backend, backend._lib.i2d_AUTHORITY_KEYID, akid + ) + + +def _encode_basic_constraints(backend, basic_constraints): + constraints = backend._lib.BASIC_CONSTRAINTS_new() + constraints = backend._ffi.gc( + constraints, backend._lib.BASIC_CONSTRAINTS_free + ) + constraints.ca = 255 if basic_constraints.ca else 0 + if basic_constraints.ca and basic_constraints.path_length is not None: + constraints.pathlen = _encode_asn1_int( + backend, basic_constraints.path_length + ) + + return _encode_extension_to_der( + backend, backend._lib.i2d_BASIC_CONSTRAINTS, constraints + ) + + +def _encode_authority_information_access(backend, authority_info_access): + aia = backend._lib.sk_ACCESS_DESCRIPTION_new_null() + backend.openssl_assert(aia != backend._ffi.NULL) + aia = backend._ffi.gc( + aia, backend._lib.sk_ACCESS_DESCRIPTION_free + ) + for access_description in authority_info_access: + ad = backend._lib.ACCESS_DESCRIPTION_new() + method = _txt2obj( + backend, access_description.access_method.dotted_string + ) + gn = _encode_general_name(backend, access_description.access_location) + ad.method = method + ad.location = gn + res = backend._lib.sk_ACCESS_DESCRIPTION_push(aia, ad) + backend.openssl_assert(res >= 1) + + return _encode_extension_to_der( + backend, backend._lib.i2d_AUTHORITY_INFO_ACCESS, aia + ) + + +def _encode_general_names(backend, names): + general_names = backend._lib.GENERAL_NAMES_new() + backend.openssl_assert(general_names != backend._ffi.NULL) + for name in names: + gn = _encode_general_name(backend, name) + res = backend._lib.sk_GENERAL_NAME_push(general_names, gn) + backend.openssl_assert(res != 0) + + return general_names + + +def _encode_alt_name(backend, san): + general_names = _encode_general_names(backend, san) + general_names = backend._ffi.gc( + general_names, backend._lib.GENERAL_NAMES_free + ) + return _encode_extension_to_der( + backend, backend._lib.i2d_GENERAL_NAMES, general_names + ) + + +def _encode_subject_key_identifier(backend, ski): + asn1_str = _encode_asn1_str_gc(backend, ski.digest, len(ski.digest)) + return _encode_extension_to_der( + backend, backend._lib.i2d_ASN1_OCTET_STRING, asn1_str + ) + + +def _encode_general_name(backend, name): + if isinstance(name, x509.DNSName): + gn = backend._lib.GENERAL_NAME_new() + backend.openssl_assert(gn != backend._ffi.NULL) + gn.type = backend._lib.GEN_DNS + + ia5 = backend._lib.ASN1_IA5STRING_new() + backend.openssl_assert(ia5 != backend._ffi.NULL) + + if name.value.startswith(u"*."): + value = b"*." + idna.encode(name.value[2:]) + else: + value = idna.encode(name.value) + + res = backend._lib.ASN1_STRING_set(ia5, value, len(value)) + backend.openssl_assert(res == 1) + gn.d.dNSName = ia5 + elif isinstance(name, x509.RegisteredID): + gn = backend._lib.GENERAL_NAME_new() + backend.openssl_assert(gn != backend._ffi.NULL) + gn.type = backend._lib.GEN_RID + obj = backend._lib.OBJ_txt2obj( + name.value.dotted_string.encode('ascii'), 1 + ) + backend.openssl_assert(obj != backend._ffi.NULL) + gn.d.registeredID = obj + elif isinstance(name, x509.DirectoryName): + gn = backend._lib.GENERAL_NAME_new() + backend.openssl_assert(gn != backend._ffi.NULL) + dir_name = _encode_name(backend, name.value) + gn.type = backend._lib.GEN_DIRNAME + gn.d.directoryName = dir_name + elif isinstance(name, x509.IPAddress): + gn = backend._lib.GENERAL_NAME_new() + backend.openssl_assert(gn != backend._ffi.NULL) + ipaddr = _encode_asn1_str( + backend, name.value.packed, len(name.value.packed) + ) + gn.type = backend._lib.GEN_IPADD + gn.d.iPAddress = ipaddr + elif isinstance(name, x509.OtherName): + gn = backend._lib.GENERAL_NAME_new() + backend.openssl_assert(gn != backend._ffi.NULL) + other_name = backend._lib.OTHERNAME_new() + backend.openssl_assert(other_name != backend._ffi.NULL) + + type_id = backend._lib.OBJ_txt2obj( + name.type_id.dotted_string.encode('ascii'), 1 + ) + backend.openssl_assert(type_id != backend._ffi.NULL) + data = backend._ffi.new("unsigned char[]", name.value) + data_ptr_ptr = backend._ffi.new("unsigned char **") + data_ptr_ptr[0] = data + value = backend._lib.d2i_ASN1_TYPE( + backend._ffi.NULL, data_ptr_ptr, len(name.value) + ) + if value == backend._ffi.NULL: + backend._consume_errors() + raise ValueError("Invalid ASN.1 data") + other_name.type_id = type_id + other_name.value = value + gn.type = backend._lib.GEN_OTHERNAME + gn.d.otherName = other_name + elif isinstance(name, x509.RFC822Name): + gn = backend._lib.GENERAL_NAME_new() + backend.openssl_assert(gn != backend._ffi.NULL) + asn1_str = _encode_asn1_str( + backend, name._encoded, len(name._encoded) + ) + gn.type = backend._lib.GEN_EMAIL + gn.d.rfc822Name = asn1_str + elif isinstance(name, x509.UniformResourceIdentifier): + gn = backend._lib.GENERAL_NAME_new() + backend.openssl_assert(gn != backend._ffi.NULL) + asn1_str = _encode_asn1_str( + backend, name._encoded, len(name._encoded) + ) + gn.type = backend._lib.GEN_URI + gn.d.uniformResourceIdentifier = asn1_str + else: + raise ValueError( + "{0} is an unknown GeneralName type".format(name) + ) + + return gn + + +def _encode_extended_key_usage(backend, extended_key_usage): + eku = backend._lib.sk_ASN1_OBJECT_new_null() + eku = backend._ffi.gc(eku, backend._lib.sk_ASN1_OBJECT_free) + for oid in extended_key_usage: + obj = _txt2obj(backend, oid.dotted_string) + res = backend._lib.sk_ASN1_OBJECT_push(eku, obj) + backend.openssl_assert(res >= 1) + + eku_ptr = backend._ffi.cast("EXTENDED_KEY_USAGE *", eku) + return _encode_extension_to_der( + backend, backend._lib.i2d_EXTENDED_KEY_USAGE, eku_ptr + ) + + +_CRLREASONFLAGS = { + x509.ReasonFlags.key_compromise: 1, + x509.ReasonFlags.ca_compromise: 2, + x509.ReasonFlags.affiliation_changed: 3, + x509.ReasonFlags.superseded: 4, + x509.ReasonFlags.cessation_of_operation: 5, + x509.ReasonFlags.certificate_hold: 6, + x509.ReasonFlags.privilege_withdrawn: 7, + x509.ReasonFlags.aa_compromise: 8, +} + + +def _encode_crl_distribution_points(backend, crl_distribution_points): + cdp = backend._lib.sk_DIST_POINT_new_null() + cdp = backend._ffi.gc(cdp, backend._lib.sk_DIST_POINT_free) + for point in crl_distribution_points: + dp = backend._lib.DIST_POINT_new() + backend.openssl_assert(dp != backend._ffi.NULL) + + if point.reasons: + bitmask = backend._lib.ASN1_BIT_STRING_new() + backend.openssl_assert(bitmask != backend._ffi.NULL) + dp.reasons = bitmask + for reason in point.reasons: + res = backend._lib.ASN1_BIT_STRING_set_bit( + bitmask, _CRLREASONFLAGS[reason], 1 + ) + backend.openssl_assert(res == 1) + + if point.full_name: + dpn = backend._lib.DIST_POINT_NAME_new() + backend.openssl_assert(dpn != backend._ffi.NULL) + dpn.type = _DISTPOINT_TYPE_FULLNAME + dpn.name.fullname = _encode_general_names(backend, point.full_name) + dp.distpoint = dpn + + if point.relative_name: + dpn = backend._lib.DIST_POINT_NAME_new() + backend.openssl_assert(dpn != backend._ffi.NULL) + dpn.type = _DISTPOINT_TYPE_RELATIVENAME + name = _encode_name_gc(backend, point.relative_name) + relativename = backend._lib.sk_X509_NAME_ENTRY_dup(name.entries) + backend.openssl_assert(relativename != backend._ffi.NULL) + dpn.name.relativename = relativename + dp.distpoint = dpn + + if point.crl_issuer: + dp.CRLissuer = _encode_general_names(backend, point.crl_issuer) + + res = backend._lib.sk_DIST_POINT_push(cdp, dp) + backend.openssl_assert(res >= 1) + + return _encode_extension_to_der( + backend, backend._lib.i2d_CRL_DIST_POINTS, cdp + ) + + +def _encode_name_constraints(backend, name_constraints): + nc = backend._lib.NAME_CONSTRAINTS_new() + assert nc != backend._ffi.NULL + nc = backend._ffi.gc(nc, backend._lib.NAME_CONSTRAINTS_free) + permitted = _encode_general_subtree( + backend, name_constraints.permitted_subtrees + ) + nc.permittedSubtrees = permitted + excluded = _encode_general_subtree( + backend, name_constraints.excluded_subtrees + ) + nc.excludedSubtrees = excluded + + return _encode_extension_to_der( + backend, backend._lib.Cryptography_i2d_NAME_CONSTRAINTS, nc + ) + + +def _encode_general_subtree(backend, subtrees): + if subtrees is None: + return backend._ffi.NULL + else: + general_subtrees = backend._lib.sk_GENERAL_SUBTREE_new_null() + for name in subtrees: + gs = backend._lib.GENERAL_SUBTREE_new() + gs.base = _encode_general_name(backend, name) + res = backend._lib.sk_GENERAL_SUBTREE_push(general_subtrees, gs) + assert res >= 1 + + return general_subtrees + + +_EXTENSION_ENCODE_HANDLERS = { + ExtensionOID.BASIC_CONSTRAINTS: _encode_basic_constraints, + ExtensionOID.SUBJECT_KEY_IDENTIFIER: _encode_subject_key_identifier, + ExtensionOID.KEY_USAGE: _encode_key_usage, + ExtensionOID.SUBJECT_ALTERNATIVE_NAME: _encode_alt_name, + ExtensionOID.ISSUER_ALTERNATIVE_NAME: _encode_alt_name, + ExtensionOID.EXTENDED_KEY_USAGE: _encode_extended_key_usage, + ExtensionOID.AUTHORITY_KEY_IDENTIFIER: _encode_authority_key_identifier, + ExtensionOID.CERTIFICATE_POLICIES: _encode_certificate_policies, + ExtensionOID.AUTHORITY_INFORMATION_ACCESS: ( + _encode_authority_information_access + ), + ExtensionOID.CRL_DISTRIBUTION_POINTS: _encode_crl_distribution_points, + ExtensionOID.INHIBIT_ANY_POLICY: _encode_inhibit_any_policy, + ExtensionOID.OCSP_NO_CHECK: _encode_ocsp_nocheck, + ExtensionOID.NAME_CONSTRAINTS: _encode_name_constraints, +} + +_CRL_EXTENSION_ENCODE_HANDLERS = { + ExtensionOID.ISSUER_ALTERNATIVE_NAME: _encode_alt_name, + ExtensionOID.AUTHORITY_KEY_IDENTIFIER: _encode_authority_key_identifier, + ExtensionOID.AUTHORITY_INFORMATION_ACCESS: ( + _encode_authority_information_access + ), + ExtensionOID.CRL_NUMBER: _encode_crl_number, +} + +_CRL_ENTRY_EXTENSION_ENCODE_HANDLERS = { + CRLEntryExtensionOID.CERTIFICATE_ISSUER: _encode_alt_name, + CRLEntryExtensionOID.CRL_REASON: _encode_crl_reason, + CRLEntryExtensionOID.INVALIDITY_DATE: _encode_invalidity_date, +} |