Edit File - File Manager

Edit File: decode_asn1.py

# 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 ipaddress

import six

from cryptography import x509
from cryptography.hazmat._der import DERReader, INTEGER, NULL, SEQUENCE
from cryptography.x509.extensions import _TLS_FEATURE_TYPE_TO_ENUM
from cryptography.x509.name import _ASN1_TYPE_TO_ENUM
from cryptography.x509.oid import (
    CRLEntryExtensionOID,
    CertificatePoliciesOID,
    ExtensionOID,
    OCSPExtensionOID,
)

def _obj2txt(backend, obj):
    # Set to 80 on the recommendation of
    # https://www.openssl.org/docs/crypto/OBJ_nid2ln.html#return_values
    #
    # But OIDs longer than this occur in real life (e.g. Active
    # Directory makes some very long OIDs).  So we need to detect
    # and properly handle the case where the default buffer is not
    # big enough.
    #
    buf_len = 80
    buf = backend._ffi.new("char[]", buf_len)

# 'res' is the number of bytes that *would* be written if the
    # buffer is large enough.  If 'res' > buf_len - 1, we need to
    # alloc a big-enough buffer and go again.
    res = backend._lib.OBJ_obj2txt(buf, buf_len, obj, 1)
    if res > buf_len - 1:  # account for terminating null byte
        buf_len = res + 1
        buf = backend._ffi.new("char[]", buf_len)
        res = backend._lib.OBJ_obj2txt(buf, buf_len, obj, 1)
    backend.openssl_assert(res > 0)
    return backend._ffi.buffer(buf, res)[:].decode()

def _decode_x509_name_entry(backend, x509_name_entry):
    obj = backend._lib.X509_NAME_ENTRY_get_object(x509_name_entry)
    backend.openssl_assert(obj != backend._ffi.NULL)
    data = backend._lib.X509_NAME_ENTRY_get_data(x509_name_entry)
    backend.openssl_assert(data != backend._ffi.NULL)
    value = _asn1_string_to_utf8(backend, data)
    oid = _obj2txt(backend, obj)
    type = _ASN1_TYPE_TO_ENUM[data.type]

return x509.NameAttribute(x509.ObjectIdentifier(oid), value, type)

def _decode_x509_name(backend, x509_name):
    count = backend._lib.X509_NAME_entry_count(x509_name)
    attributes = []
    prev_set_id = -1
    for x in range(count):
        entry = backend._lib.X509_NAME_get_entry(x509_name, x)
        attribute = _decode_x509_name_entry(backend, entry)
        set_id = backend._lib.Cryptography_X509_NAME_ENTRY_set(entry)
        if set_id != prev_set_id:
            attributes.append({attribute})
        else:
            # is in the same RDN a previous entry
            attributes[-1].add(attribute)
        prev_set_id = set_id

return x509.Name(x509.RelativeDistinguishedName(rdn) for rdn in attributes)

def _decode_general_names(backend, gns):
    num = backend._lib.sk_GENERAL_NAME_num(gns)
    names = []
    for i in range(num):
        gn = backend._lib.sk_GENERAL_NAME_value(gns, i)
        backend.openssl_assert(gn != backend._ffi.NULL)
        names.append(_decode_general_name(backend, gn))

return names

def _decode_general_name(backend, gn):
    if gn.type == backend._lib.GEN_DNS:
        # Convert to bytes and then decode to utf8. We don't use
        # asn1_string_to_utf8 here because it doesn't properly convert
        # utf8 from ia5strings.
        data = _asn1_string_to_bytes(backend, gn.d.dNSName).decode("utf8")
        # We don't use the constructor for DNSName so we can bypass validation
        # This allows us to create DNSName objects that have unicode chars
        # when a certificate (against the RFC) contains them.
        return x509.DNSName._init_without_validation(data)
    elif gn.type == backend._lib.GEN_URI:
        # Convert to bytes and then decode to utf8. We don't use
        # asn1_string_to_utf8 here because it doesn't properly convert
        # utf8 from ia5strings.
        data = _asn1_string_to_bytes(
            backend, gn.d.uniformResourceIdentifier
        ).decode("utf8")
        # We don't use the constructor for URI so we can bypass validation
        # This allows us to create URI objects that have unicode chars
        # when a certificate (against the RFC) contains them.
        return x509.UniformResourceIdentifier._init_without_validation(data)
    elif gn.type == backend._lib.GEN_RID:
        oid = _obj2txt(backend, gn.d.registeredID)
        return x509.RegisteredID(x509.ObjectIdentifier(oid))
    elif gn.type == backend._lib.GEN_IPADD:
        data = _asn1_string_to_bytes(backend, gn.d.iPAddress)
        data_len = len(data)
        if data_len == 8 or data_len == 32:
            # This is an IPv4 or IPv6 Network and not a single IP. This
            # type of data appears in Name Constraints. Unfortunately,
            # ipaddress doesn't support packed bytes + netmask. Additionally,
            # IPv6Network can only handle CIDR rather than the full 16 byte
            # netmask. To handle this we convert the netmask to integer, then
            # find the first 0 bit, which will be the prefix. If another 1
            # bit is present after that the netmask is invalid.
            base = ipaddress.ip_address(data[: data_len // 2])
            netmask = ipaddress.ip_address(data[data_len // 2 :])
            bits = bin(int(netmask))[2:]
            prefix = bits.find("0")
            # If no 0 bits are found it is a /32 or /128
            if prefix == -1:
                prefix = len(bits)

if "1" in bits[prefix:]:
                raise ValueError("Invalid netmask")

ip = ipaddress.ip_network(base.exploded + u"/{}".format(prefix))
        else:
            ip = ipaddress.ip_address(data)

return x509.IPAddress(ip)
    elif gn.type == backend._lib.GEN_DIRNAME:
        return x509.DirectoryName(
            _decode_x509_name(backend, gn.d.directoryName)
        )
    elif gn.type == backend._lib.GEN_EMAIL:
        # Convert to bytes and then decode to utf8. We don't use
        # asn1_string_to_utf8 here because it doesn't properly convert
        # utf8 from ia5strings.
        data = _asn1_string_to_bytes(backend, gn.d.rfc822Name).decode("utf8")
        # We don't use the constructor for RFC822Name so we can bypass
        # validation. This allows us to create RFC822Name objects that have
        # unicode chars when a certificate (against the RFC) contains them.
        return x509.RFC822Name._init_without_validation(data)
    elif gn.type == backend._lib.GEN_OTHERNAME:
        type_id = _obj2txt(backend, gn.d.otherName.type_id)
        value = _asn1_to_der(backend, gn.d.otherName.value)
        return x509.OtherName(x509.ObjectIdentifier(type_id), value)
    else:
        # x400Address or ediPartyName
        raise x509.UnsupportedGeneralNameType(
            "{} is not a supported type".format(
                x509._GENERAL_NAMES.get(gn.type, gn.type)
            ),
            gn.type,
        )

def _decode_ocsp_no_check(backend, ext):
    return x509.OCSPNoCheck()

def _decode_crl_number(backend, ext):
    asn1_int = backend._ffi.cast("ASN1_INTEGER *", ext)
    asn1_int = backend._ffi.gc(asn1_int, backend._lib.ASN1_INTEGER_free)
    return x509.CRLNumber(_asn1_integer_to_int(backend, asn1_int))

def _decode_delta_crl_indicator(backend, ext):
    asn1_int = backend._ffi.cast("ASN1_INTEGER *", ext)
    asn1_int = backend._ffi.gc(asn1_int, backend._lib.ASN1_INTEGER_free)
    return x509.DeltaCRLIndicator(_asn1_integer_to_int(backend, asn1_int))

class _X509ExtensionParser(object):
    def __init__(self, backend, ext_count, get_ext, handlers):
        self.ext_count = ext_count
        self.get_ext = get_ext
        self.handlers = handlers
        self._backend = backend

def parse(self, x509_obj):
        extensions = []
        seen_oids = set()
        for i in range(self.ext_count(x509_obj)):
            ext = self.get_ext(x509_obj, i)
            self._backend.openssl_assert(ext != self._backend._ffi.NULL)
            crit = self._backend._lib.X509_EXTENSION_get_critical(ext)
            critical = crit == 1
            oid = x509.ObjectIdentifier(
                _obj2txt(
                    self._backend,
                    self._backend._lib.X509_EXTENSION_get_object(ext),
                )
            )
            if oid in seen_oids:
                raise x509.DuplicateExtension(
                    "Duplicate {} extension found".format(oid), oid
                )

# These OIDs are only supported in OpenSSL 1.1.0+ but we want
            # to support them in all versions of OpenSSL so we decode them
            # ourselves.
            if oid == ExtensionOID.TLS_FEATURE:
                # The extension contents are a SEQUENCE OF INTEGERs.
                data = self._backend._lib.X509_EXTENSION_get_data(ext)
                data_bytes = _asn1_string_to_bytes(self._backend, data)
                features = DERReader(data_bytes).read_single_element(SEQUENCE)
                parsed = []
                while not features.is_empty():
                    parsed.append(features.read_element(INTEGER).as_integer())
                # Map the features to their enum value.
                value = x509.TLSFeature(
                    [_TLS_FEATURE_TYPE_TO_ENUM[x] for x in parsed]
                )
                extensions.append(x509.Extension(oid, critical, value))
                seen_oids.add(oid)
                continue
            elif oid == ExtensionOID.PRECERT_POISON:
                data = self._backend._lib.X509_EXTENSION_get_data(ext)
                # The contents of the extension must be an ASN.1 NULL.
                reader = DERReader(_asn1_string_to_bytes(self._backend, data))
                reader.read_single_element(NULL).check_empty()
                extensions.append(
                    x509.Extension(oid, critical, x509.PrecertPoison())
                )
                seen_oids.add(oid)
                continue

try:
                handler = self.handlers[oid]
            except KeyError:
                # Dump the DER payload into an UnrecognizedExtension object
                data = self._backend._lib.X509_EXTENSION_get_data(ext)
                self._backend.openssl_assert(data != self._backend._ffi.NULL)
                der = self._backend._ffi.buffer(data.data, data.length)[:]
                unrecognized = x509.UnrecognizedExtension(oid, der)
                extensions.append(x509.Extension(oid, critical, unrecognized))
            else:
                ext_data = self._backend._lib.X509V3_EXT_d2i(ext)
                if ext_data == self._backend._ffi.NULL:
                    self._backend._consume_errors()
                    raise ValueError(
                        "The {} extension is invalid and can't be "
                        "parsed".format(oid)
                    )

value = handler(self._backend, ext_data)
                extensions.append(x509.Extension(oid, critical, value))

seen_oids.add(oid)

return x509.Extensions(extensions)

def _decode_certificate_policies(backend, cp):
    cp = backend._ffi.cast("Cryptography_STACK_OF_POLICYINFO *", cp)
    cp = backend._ffi.gc(cp, backend._lib.CERTIFICATEPOLICIES_free)

num = backend._lib.sk_POLICYINFO_num(cp)
    certificate_policies = []
    for i in range(num):
        qualifiers = None
        pi = backend._lib.sk_POLICYINFO_value(cp, i)
        oid = x509.ObjectIdentifier(_obj2txt(backend, pi.policyid))
        if pi.qualifiers != backend._ffi.NULL:
            qnum = backend._lib.sk_POLICYQUALINFO_num(pi.qualifiers)
            qualifiers = []
            for j in range(qnum):
                pqi = backend._lib.sk_POLICYQUALINFO_value(pi.qualifiers, j)
                pqualid = x509.ObjectIdentifier(_obj2txt(backend, pqi.pqualid))
                if pqualid == CertificatePoliciesOID.CPS_QUALIFIER:
                    cpsuri = backend._ffi.buffer(
                        pqi.d.cpsuri.data, pqi.d.cpsuri.length
                    )[:].decode("ascii")
                    qualifiers.append(cpsuri)
                else:
                    assert pqualid == CertificatePoliciesOID.CPS_USER_NOTICE
                    user_notice = _decode_user_notice(
                        backend, pqi.d.usernotice
                    )
                    qualifiers.append(user_notice)

certificate_policies.append(x509.PolicyInformation(oid, qualifiers))

return x509.CertificatePolicies(certificate_policies)

def _decode_user_notice(backend, un):
    explicit_text = None
    notice_reference = None

if un.exptext != backend._ffi.NULL:
        explicit_text = _asn1_string_to_utf8(backend, un.exptext)

if un.noticeref != backend._ffi.NULL:
        organization = _asn1_string_to_utf8(backend, un.noticeref.organization)

num = backend._lib.sk_ASN1_INTEGER_num(un.noticeref.noticenos)
        notice_numbers = []
        for i in range(num):
            asn1_int = backend._lib.sk_ASN1_INTEGER_value(
                un.noticeref.noticenos, i
            )
            notice_num = _asn1_integer_to_int(backend, asn1_int)
            notice_numbers.append(notice_num)

notice_reference = x509.NoticeReference(organization, notice_numbers)

return x509.UserNotice(notice_reference, explicit_text)

def _decode_basic_constraints(backend, bc_st):
    basic_constraints = backend._ffi.cast("BASIC_CONSTRAINTS *", bc_st)
    basic_constraints = backend._ffi.gc(
        basic_constraints, backend._lib.BASIC_CONSTRAINTS_free
    )
    # The byte representation of an ASN.1 boolean true is \xff. OpenSSL
    # chooses to just map this to its ordinal value, so true is 255 and
    # false is 0.
    ca = basic_constraints.ca == 255
    path_length = _asn1_integer_to_int_or_none(
        backend, basic_constraints.pathlen
    )

return x509.BasicConstraints(ca, path_length)

def _decode_subject_key_identifier(backend, asn1_string):
    asn1_string = backend._ffi.cast("ASN1_OCTET_STRING *", asn1_string)
    asn1_string = backend._ffi.gc(
        asn1_string, backend._lib.ASN1_OCTET_STRING_free
    )
    return x509.SubjectKeyIdentifier(
        backend._ffi.buffer(asn1_string.data, asn1_string.length)[:]
    )

def _decode_authority_key_identifier(backend, akid):
    akid = backend._ffi.cast("AUTHORITY_KEYID *", akid)
    akid = backend._ffi.gc(akid, backend._lib.AUTHORITY_KEYID_free)
    key_identifier = None
    authority_cert_issuer = None

if akid.keyid != backend._ffi.NULL:
        key_identifier = backend._ffi.buffer(
            akid.keyid.data, akid.keyid.length
        )[:]

if akid.issuer != backend._ffi.NULL:
        authority_cert_issuer = _decode_general_names(backend, akid.issuer)

authority_cert_serial_number = _asn1_integer_to_int_or_none(
        backend, akid.serial
    )

return x509.AuthorityKeyIdentifier(
        key_identifier, authority_cert_issuer, authority_cert_serial_number
    )

def _decode_information_access(backend, ia):
    ia = backend._ffi.cast("Cryptography_STACK_OF_ACCESS_DESCRIPTION *", ia)
    ia = backend._ffi.gc(
        ia,
        lambda x: backend._lib.sk_ACCESS_DESCRIPTION_pop_free(
            x,
            backend._ffi.addressof(
                backend._lib._original_lib, "ACCESS_DESCRIPTION_free"
            ),
        ),
    )
    num = backend._lib.sk_ACCESS_DESCRIPTION_num(ia)
    access_descriptions = []
    for i in range(num):
        ad = backend._lib.sk_ACCESS_DESCRIPTION_value(ia, i)
        backend.openssl_assert(ad.method != backend._ffi.NULL)
        oid = x509.ObjectIdentifier(_obj2txt(backend, ad.method))
        backend.openssl_assert(ad.location != backend._ffi.NULL)
        gn = _decode_general_name(backend, ad.location)
        access_descriptions.append(x509.AccessDescription(oid, gn))

return access_descriptions

def _decode_authority_information_access(backend, aia):
    access_descriptions = _decode_information_access(backend, aia)
    return x509.AuthorityInformationAccess(access_descriptions)

def _decode_subject_information_access(backend, aia):
    access_descriptions = _decode_information_access(backend, aia)
    return x509.SubjectInformationAccess(access_descriptions)

def _decode_key_usage(backend, bit_string):
    bit_string = backend._ffi.cast("ASN1_BIT_STRING *", bit_string)
    bit_string = backend._ffi.gc(bit_string, backend._lib.ASN1_BIT_STRING_free)
    get_bit = backend._lib.ASN1_BIT_STRING_get_bit
    digital_signature = get_bit(bit_string, 0) == 1
    content_commitment = get_bit(bit_string, 1) == 1
    key_encipherment = get_bit(bit_string, 2) == 1
    data_encipherment = get_bit(bit_string, 3) == 1
    key_agreement = get_bit(bit_string, 4) == 1
    key_cert_sign = get_bit(bit_string, 5) == 1
    crl_sign = get_bit(bit_string, 6) == 1
    encipher_only = get_bit(bit_string, 7) == 1
    decipher_only = get_bit(bit_string, 8) == 1
    return x509.KeyUsage(
        digital_signature,
        content_commitment,
        key_encipherment,
        data_encipherment,
        key_agreement,
        key_cert_sign,
        crl_sign,
        encipher_only,
        decipher_only,
    )

def _decode_general_names_extension(backend, gns):
    gns = backend._ffi.cast("GENERAL_NAMES *", gns)
    gns = backend._ffi.gc(gns, backend._lib.GENERAL_NAMES_free)
    general_names = _decode_general_names(backend, gns)
    return general_names

def _decode_subject_alt_name(backend, ext):
    return x509.SubjectAlternativeName(
        _decode_general_names_extension(backend, ext)
    )

def _decode_issuer_alt_name(backend, ext):
    return x509.IssuerAlternativeName(
        _decode_general_names_extension(backend, ext)
    )

def _decode_name_constraints(backend, nc):
    nc = backend._ffi.cast("NAME_CONSTRAINTS *", nc)
    nc = backend._ffi.gc(nc, backend._lib.NAME_CONSTRAINTS_free)
    permitted = _decode_general_subtrees(backend, nc.permittedSubtrees)
    excluded = _decode_general_subtrees(backend, nc.excludedSubtrees)
    return x509.NameConstraints(
        permitted_subtrees=permitted, excluded_subtrees=excluded
    )

def _decode_general_subtrees(backend, stack_subtrees):
    if stack_subtrees == backend._ffi.NULL:
        return None

num = backend._lib.sk_GENERAL_SUBTREE_num(stack_subtrees)
    subtrees = []

for i in range(num):
        obj = backend._lib.sk_GENERAL_SUBTREE_value(stack_subtrees, i)
        backend.openssl_assert(obj != backend._ffi.NULL)
        name = _decode_general_name(backend, obj.base)
        subtrees.append(name)

return subtrees

def _decode_issuing_dist_point(backend, idp):
    idp = backend._ffi.cast("ISSUING_DIST_POINT *", idp)
    idp = backend._ffi.gc(idp, backend._lib.ISSUING_DIST_POINT_free)
    if idp.distpoint != backend._ffi.NULL:
        full_name, relative_name = _decode_distpoint(backend, idp.distpoint)
    else:
        full_name = None
        relative_name = None

only_user = idp.onlyuser == 255
    only_ca = idp.onlyCA == 255
    indirect_crl = idp.indirectCRL == 255
    only_attr = idp.onlyattr == 255
    if idp.onlysomereasons != backend._ffi.NULL:
        only_some_reasons = _decode_reasons(backend, idp.onlysomereasons)
    else:
        only_some_reasons = None

return x509.IssuingDistributionPoint(
        full_name,
        relative_name,
        only_user,
        only_ca,
        only_some_reasons,
        indirect_crl,
        only_attr,
    )

def _decode_policy_constraints(backend, pc):
    pc = backend._ffi.cast("POLICY_CONSTRAINTS *", pc)
    pc = backend._ffi.gc(pc, backend._lib.POLICY_CONSTRAINTS_free)

require_explicit_policy = _asn1_integer_to_int_or_none(
        backend, pc.requireExplicitPolicy
    )
    inhibit_policy_mapping = _asn1_integer_to_int_or_none(
        backend, pc.inhibitPolicyMapping
    )

return x509.PolicyConstraints(
        require_explicit_policy, inhibit_policy_mapping
    )

def _decode_extended_key_usage(backend, sk):
    sk = backend._ffi.cast("Cryptography_STACK_OF_ASN1_OBJECT *", sk)
    sk = backend._ffi.gc(sk, backend._lib.sk_ASN1_OBJECT_free)
    num = backend._lib.sk_ASN1_OBJECT_num(sk)
    ekus = []

for i in range(num):
        obj = backend._lib.sk_ASN1_OBJECT_value(sk, i)
        backend.openssl_assert(obj != backend._ffi.NULL)
        oid = x509.ObjectIdentifier(_obj2txt(backend, obj))
        ekus.append(oid)

return x509.ExtendedKeyUsage(ekus)

_DISTPOINT_TYPE_FULLNAME = 0
_DISTPOINT_TYPE_RELATIVENAME = 1

def _decode_dist_points(backend, cdps):
    cdps = backend._ffi.cast("Cryptography_STACK_OF_DIST_POINT *", cdps)
    cdps = backend._ffi.gc(cdps, backend._lib.CRL_DIST_POINTS_free)

num = backend._lib.sk_DIST_POINT_num(cdps)
    dist_points = []
    for i in range(num):
        full_name = None
        relative_name = None
        crl_issuer = None
        reasons = None
        cdp = backend._lib.sk_DIST_POINT_value(cdps, i)
        if cdp.reasons != backend._ffi.NULL:
            reasons = _decode_reasons(backend, cdp.reasons)

if cdp.CRLissuer != backend._ffi.NULL:
            crl_issuer = _decode_general_names(backend, cdp.CRLissuer)

# Certificates may have a crl_issuer/reasons and no distribution
        # point so make sure it's not null.
        if cdp.distpoint != backend._ffi.NULL:
            full_name, relative_name = _decode_distpoint(
                backend, cdp.distpoint
            )

dist_points.append(
            x509.DistributionPoint(
                full_name, relative_name, reasons, crl_issuer
            )
        )

return dist_points

# ReasonFlags ::= BIT STRING {
#      unused                  (0),
#      keyCompromise           (1),
#      cACompromise            (2),
#      affiliationChanged      (3),
#      superseded              (4),
#      cessationOfOperation    (5),
#      certificateHold         (6),
#      privilegeWithdrawn      (7),
#      aACompromise            (8) }
_REASON_BIT_MAPPING = {
    1: x509.ReasonFlags.key_compromise,
    2: x509.ReasonFlags.ca_compromise,
    3: x509.ReasonFlags.affiliation_changed,
    4: x509.ReasonFlags.superseded,
    5: x509.ReasonFlags.cessation_of_operation,
    6: x509.ReasonFlags.certificate_hold,
    7: x509.ReasonFlags.privilege_withdrawn,
    8: x509.ReasonFlags.aa_compromise,
}

def _decode_reasons(backend, reasons):
    # We will check each bit from RFC 5280
    enum_reasons = []
    for bit_position, reason in six.iteritems(_REASON_BIT_MAPPING):
        if backend._lib.ASN1_BIT_STRING_get_bit(reasons, bit_position):
            enum_reasons.append(reason)

return frozenset(enum_reasons)

def _decode_distpoint(backend, distpoint):
    if distpoint.type == _DISTPOINT_TYPE_FULLNAME:
        full_name = _decode_general_names(backend, distpoint.name.fullname)
        return full_name, None

# OpenSSL code doesn't test for a specific type for
    # relativename, everything that isn't fullname is considered
    # relativename.  Per RFC 5280:
    #
    # DistributionPointName ::= CHOICE {
    #      fullName                [0]      GeneralNames,
    #      nameRelativeToCRLIssuer [1]      RelativeDistinguishedName }
    rns = distpoint.name.relativename
    rnum = backend._lib.sk_X509_NAME_ENTRY_num(rns)
    attributes = set()
    for i in range(rnum):
        rn = backend._lib.sk_X509_NAME_ENTRY_value(rns, i)
        backend.openssl_assert(rn != backend._ffi.NULL)
        attributes.add(_decode_x509_name_entry(backend, rn))

relative_name = x509.RelativeDistinguishedName(attributes)

return None, relative_name

def _decode_crl_distribution_points(backend, cdps):
    dist_points = _decode_dist_points(backend, cdps)
    return x509.CRLDistributionPoints(dist_points)

def _decode_freshest_crl(backend, cdps):
    dist_points = _decode_dist_points(backend, cdps)
    return x509.FreshestCRL(dist_points)

def _decode_inhibit_any_policy(backend, asn1_int):
    asn1_int = backend._ffi.cast("ASN1_INTEGER *", asn1_int)
    asn1_int = backend._ffi.gc(asn1_int, backend._lib.ASN1_INTEGER_free)
    skip_certs = _asn1_integer_to_int(backend, asn1_int)
    return x509.InhibitAnyPolicy(skip_certs)

def _decode_scts(backend, asn1_scts):
    from cryptography.hazmat.backends.openssl.x509 import (
        _SignedCertificateTimestamp,
    )

asn1_scts = backend._ffi.cast("Cryptography_STACK_OF_SCT *", asn1_scts)
    asn1_scts = backend._ffi.gc(asn1_scts, backend._lib.SCT_LIST_free)

scts = []
    for i in range(backend._lib.sk_SCT_num(asn1_scts)):
        sct = backend._lib.sk_SCT_value(asn1_scts, i)

scts.append(_SignedCertificateTimestamp(backend, asn1_scts, sct))
    return scts

def _decode_precert_signed_certificate_timestamps(backend, asn1_scts):
    return x509.PrecertificateSignedCertificateTimestamps(
        _decode_scts(backend, asn1_scts)
    )

def _decode_signed_certificate_timestamps(backend, asn1_scts):
    return x509.SignedCertificateTimestamps(_decode_scts(backend, asn1_scts))

#    CRLReason ::= ENUMERATED {
#        unspecified             (0),
#        keyCompromise           (1),
#        cACompromise            (2),
#        affiliationChanged      (3),
#        superseded              (4),
#        cessationOfOperation    (5),
#        certificateHold         (6),
#             -- value 7 is not used
#        removeFromCRL           (8),
#        privilegeWithdrawn      (9),
#        aACompromise           (10) }
_CRL_ENTRY_REASON_CODE_TO_ENUM = {
    0: x509.ReasonFlags.unspecified,
    1: x509.ReasonFlags.key_compromise,
    2: x509.ReasonFlags.ca_compromise,
    3: x509.ReasonFlags.affiliation_changed,
    4: x509.ReasonFlags.superseded,
    5: x509.ReasonFlags.cessation_of_operation,
    6: x509.ReasonFlags.certificate_hold,
    8: x509.ReasonFlags.remove_from_crl,
    9: x509.ReasonFlags.privilege_withdrawn,
    10: x509.ReasonFlags.aa_compromise,
}

_CRL_ENTRY_REASON_ENUM_TO_CODE = {
    x509.ReasonFlags.unspecified: 0,
    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.remove_from_crl: 8,
    x509.ReasonFlags.privilege_withdrawn: 9,
    x509.ReasonFlags.aa_compromise: 10,
}

def _decode_crl_reason(backend, enum):
    enum = backend._ffi.cast("ASN1_ENUMERATED *", enum)
    enum = backend._ffi.gc(enum, backend._lib.ASN1_ENUMERATED_free)
    code = backend._lib.ASN1_ENUMERATED_get(enum)

try:
        return x509.CRLReason(_CRL_ENTRY_REASON_CODE_TO_ENUM[code])
    except KeyError:
        raise ValueError("Unsupported reason code: {}".format(code))

def _decode_invalidity_date(backend, inv_date):
    generalized_time = backend._ffi.cast("ASN1_GENERALIZEDTIME *", inv_date)
    generalized_time = backend._ffi.gc(
        generalized_time, backend._lib.ASN1_GENERALIZEDTIME_free
    )
    return x509.InvalidityDate(
        _parse_asn1_generalized_time(backend, generalized_time)
    )

def _decode_cert_issuer(backend, gns):
    gns = backend._ffi.cast("GENERAL_NAMES *", gns)
    gns = backend._ffi.gc(gns, backend._lib.GENERAL_NAMES_free)
    general_names = _decode_general_names(backend, gns)
    return x509.CertificateIssuer(general_names)

def _asn1_to_der(backend, asn1_type):
    buf = backend._ffi.new("unsigned char **")
    res = backend._lib.i2d_ASN1_TYPE(asn1_type, buf)
    backend.openssl_assert(res >= 0)
    backend.openssl_assert(buf[0] != backend._ffi.NULL)
    buf = backend._ffi.gc(
        buf, lambda buffer: backend._lib.OPENSSL_free(buffer[0])
    )
    return backend._ffi.buffer(buf[0], res)[:]

def _asn1_integer_to_int(backend, asn1_int):
    bn = backend._lib.ASN1_INTEGER_to_BN(asn1_int, backend._ffi.NULL)
    backend.openssl_assert(bn != backend._ffi.NULL)
    bn = backend._ffi.gc(bn, backend._lib.BN_free)
    return backend._bn_to_int(bn)

def _asn1_integer_to_int_or_none(backend, asn1_int):
    if asn1_int == backend._ffi.NULL:
        return None
    else:
        return _asn1_integer_to_int(backend, asn1_int)

def _asn1_string_to_bytes(backend, asn1_string):
    return backend._ffi.buffer(asn1_string.data, asn1_string.length)[:]

def _asn1_string_to_ascii(backend, asn1_string):
    return _asn1_string_to_bytes(backend, asn1_string).decode("ascii")

def _asn1_string_to_utf8(backend, asn1_string):
    buf = backend._ffi.new("unsigned char **")
    res = backend._lib.ASN1_STRING_to_UTF8(buf, asn1_string)
    if res == -1:
        raise ValueError(
            "Unsupported ASN1 string type. Type: {}".format(asn1_string.type)
        )

backend.openssl_assert(buf[0] != backend._ffi.NULL)
    buf = backend._ffi.gc(
        buf, lambda buffer: backend._lib.OPENSSL_free(buffer[0])
    )
    return backend._ffi.buffer(buf[0], res)[:].decode("utf8")

def _parse_asn1_time(backend, asn1_time):
    backend.openssl_assert(asn1_time != backend._ffi.NULL)
    generalized_time = backend._lib.ASN1_TIME_to_generalizedtime(
        asn1_time, backend._ffi.NULL
    )
    if generalized_time == backend._ffi.NULL:
        raise ValueError(
            "Couldn't parse ASN.1 time as generalizedtime {!r}".format(
                _asn1_string_to_bytes(backend, asn1_time)
            )
        )

generalized_time = backend._ffi.gc(
        generalized_time, backend._lib.ASN1_GENERALIZEDTIME_free
    )
    return _parse_asn1_generalized_time(backend, generalized_time)

def _parse_asn1_generalized_time(backend, generalized_time):
    time = _asn1_string_to_ascii(
        backend, backend._ffi.cast("ASN1_STRING *", generalized_time)
    )
    return datetime.datetime.strptime(time, "%Y%m%d%H%M%SZ")

def _decode_nonce(backend, nonce):
    nonce = backend._ffi.cast("ASN1_OCTET_STRING *", nonce)
    nonce = backend._ffi.gc(nonce, backend._lib.ASN1_OCTET_STRING_free)
    return x509.OCSPNonce(_asn1_string_to_bytes(backend, nonce))

_EXTENSION_HANDLERS_BASE = {
    ExtensionOID.BASIC_CONSTRAINTS: _decode_basic_constraints,
    ExtensionOID.SUBJECT_KEY_IDENTIFIER: _decode_subject_key_identifier,
    ExtensionOID.KEY_USAGE: _decode_key_usage,
    ExtensionOID.SUBJECT_ALTERNATIVE_NAME: _decode_subject_alt_name,
    ExtensionOID.EXTENDED_KEY_USAGE: _decode_extended_key_usage,
    ExtensionOID.AUTHORITY_KEY_IDENTIFIER: _decode_authority_key_identifier,
    ExtensionOID.AUTHORITY_INFORMATION_ACCESS: (
        _decode_authority_information_access
    ),
    ExtensionOID.SUBJECT_INFORMATION_ACCESS: (
        _decode_subject_information_access
    ),
    ExtensionOID.CERTIFICATE_POLICIES: _decode_certificate_policies,
    ExtensionOID.CRL_DISTRIBUTION_POINTS: _decode_crl_distribution_points,
    ExtensionOID.FRESHEST_CRL: _decode_freshest_crl,
    ExtensionOID.OCSP_NO_CHECK: _decode_ocsp_no_check,
    ExtensionOID.INHIBIT_ANY_POLICY: _decode_inhibit_any_policy,
    ExtensionOID.ISSUER_ALTERNATIVE_NAME: _decode_issuer_alt_name,
    ExtensionOID.NAME_CONSTRAINTS: _decode_name_constraints,
    ExtensionOID.POLICY_CONSTRAINTS: _decode_policy_constraints,
}
_EXTENSION_HANDLERS_SCT = {
    ExtensionOID.PRECERT_SIGNED_CERTIFICATE_TIMESTAMPS: (
        _decode_precert_signed_certificate_timestamps
    )
}

_REVOKED_EXTENSION_HANDLERS = {
    CRLEntryExtensionOID.CRL_REASON: _decode_crl_reason,
    CRLEntryExtensionOID.INVALIDITY_DATE: _decode_invalidity_date,
    CRLEntryExtensionOID.CERTIFICATE_ISSUER: _decode_cert_issuer,
}

_CRL_EXTENSION_HANDLERS = {
    ExtensionOID.CRL_NUMBER: _decode_crl_number,
    ExtensionOID.DELTA_CRL_INDICATOR: _decode_delta_crl_indicator,
    ExtensionOID.AUTHORITY_KEY_IDENTIFIER: _decode_authority_key_identifier,
    ExtensionOID.ISSUER_ALTERNATIVE_NAME: _decode_issuer_alt_name,
    ExtensionOID.AUTHORITY_INFORMATION_ACCESS: (
        _decode_authority_information_access
    ),
    ExtensionOID.ISSUING_DISTRIBUTION_POINT: _decode_issuing_dist_point,
    ExtensionOID.FRESHEST_CRL: _decode_freshest_crl,
}

_OCSP_REQ_EXTENSION_HANDLERS = {
    OCSPExtensionOID.NONCE: _decode_nonce,
}

_OCSP_BASICRESP_EXTENSION_HANDLERS = {
    OCSPExtensionOID.NONCE: _decode_nonce,
}

_OCSP_SINGLERESP_EXTENSION_HANDLERS_SCT = {
    ExtensionOID.SIGNED_CERTIFICATE_TIMESTAMPS: (
        _decode_signed_certificate_timestamps
    )
}

Back to File Manager