FM / openssl

/usr/lib/python3/dist-packages/cryptography/hazmat/backends/openssl/

/usr/lib/python3/dist-packages/cryptography/hazmat/backends/openssl

Name	Size	Mode	Actions
__pycache__/	-	0755	rm
aead.py	5700	0644	edit dl rm
backend.py	106341	0644	edit dl rm
ciphers.py	9231	0644	edit dl rm
cmac.py	2846	0644	edit dl rm
decode_asn1.py	32254	0644	edit dl rm
dh.py	11209	0644	edit dl rm
dsa.py	10684	0644	edit dl rm
ec.py	14431	0644	edit dl rm
ed448.py	5743	0644	edit dl rm
ed25519.py	5789	0644	edit dl rm
encode_asn1.py	23988	0644	edit dl rm
hashes.py	3090	0644	edit dl rm
hmac.py	2933	0644	edit dl rm
ocsp.py	14617	0644	edit dl rm
poly1305.py	2347	0644	edit dl rm
rsa.py	20919	0644	edit dl rm
utils.py	2283	0644	edit dl rm
x448.py	4141	0644	edit dl rm
x509.py	22831	0644	edit dl rm
x25519.py	4622	0644	edit dl rm
__init__.py	271	0644	edit dl rm

Edit: /usr/lib/python3/dist-packages/cryptography/hazmat/backends/openssl/ocsp.py (14617B)

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

import datetime
import typing

from cryptography import utils, x509
from cryptography.exceptions import UnsupportedAlgorithm
from cryptography.hazmat.backends.openssl.decode_asn1 import (
    _CRL_ENTRY_REASON_CODE_TO_ENUM,
    _asn1_integer_to_int,
    _asn1_string_to_bytes,
    _decode_x509_name,
    _obj2txt,
    _parse_asn1_generalized_time,
)
from cryptography.hazmat.backends.openssl.x509 import _Certificate
from cryptography.hazmat.primitives import hashes, serialization
from cryptography.x509.ocsp import (
    OCSPCertStatus,
    OCSPRequest,
    OCSPResponse,
    OCSPResponseStatus,
    _CERT_STATUS_TO_ENUM,
    _OIDS_TO_HASH,
    _RESPONSE_STATUS_TO_ENUM,
)

def _issuer_key_hash(backend, cert_id):
    key_hash = backend._ffi.new("ASN1_OCTET_STRING **")
    res = backend._lib.OCSP_id_get0_info(
        backend._ffi.NULL,
        backend._ffi.NULL,
        key_hash,
        backend._ffi.NULL,
        cert_id,
    )
    backend.openssl_assert(res == 1)
    backend.openssl_assert(key_hash[0] != backend._ffi.NULL)
    return _asn1_string_to_bytes(backend, key_hash[0])

def _issuer_name_hash(backend, cert_id):
    name_hash = backend._ffi.new("ASN1_OCTET_STRING **")
    res = backend._lib.OCSP_id_get0_info(
        name_hash,
        backend._ffi.NULL,
        backend._ffi.NULL,
        backend._ffi.NULL,
        cert_id,
    )
    backend.openssl_assert(res == 1)
    backend.openssl_assert(name_hash[0] != backend._ffi.NULL)
    return _asn1_string_to_bytes(backend, name_hash[0])

def _serial_number(backend, cert_id):
    num = backend._ffi.new("ASN1_INTEGER **")
    res = backend._lib.OCSP_id_get0_info(
        backend._ffi.NULL, backend._ffi.NULL, backend._ffi.NULL, num, cert_id
    )
    backend.openssl_assert(res == 1)
    backend.openssl_assert(num[0] != backend._ffi.NULL)
    return _asn1_integer_to_int(backend, num[0])

def _hash_algorithm(backend, cert_id):
    asn1obj = backend._ffi.new("ASN1_OBJECT **")
    res = backend._lib.OCSP_id_get0_info(
        backend._ffi.NULL,
        asn1obj,
        backend._ffi.NULL,
        backend._ffi.NULL,
        cert_id,
    )
    backend.openssl_assert(res == 1)
    backend.openssl_assert(asn1obj[0] != backend._ffi.NULL)
    oid = _obj2txt(backend, asn1obj[0])
    try:
        return _OIDS_TO_HASH[oid]
    except KeyError:
        raise UnsupportedAlgorithm(
            "Signature algorithm OID: {} not recognized".format(oid)
        )

class _OCSPResponse(OCSPResponse):
    def __init__(self, backend, ocsp_response):
        self._backend = backend
        self._ocsp_response = ocsp_response
        status = self._backend._lib.OCSP_response_status(self._ocsp_response)
        self._backend.openssl_assert(status in _RESPONSE_STATUS_TO_ENUM)
        self._status = _RESPONSE_STATUS_TO_ENUM[status]
        if self._status is OCSPResponseStatus.SUCCESSFUL:
            basic = self._backend._lib.OCSP_response_get1_basic(
                self._ocsp_response
            )
            self._backend.openssl_assert(basic != self._backend._ffi.NULL)
            self._basic = self._backend._ffi.gc(
                basic, self._backend._lib.OCSP_BASICRESP_free
            )
            num_resp = self._backend._lib.OCSP_resp_count(self._basic)
            if num_resp != 1:
                raise ValueError(
                    "OCSP response contains more than one SINGLERESP structure"
                    ", which this library does not support. "
                    "{} found".format(num_resp)
                )
            self._single = self._backend._lib.OCSP_resp_get0(self._basic, 0)
            self._backend.openssl_assert(
                self._single != self._backend._ffi.NULL
            )
            self._cert_id = self._backend._lib.OCSP_SINGLERESP_get0_id(
                self._single
            )
            self._backend.openssl_assert(
                self._cert_id != self._backend._ffi.NULL
            )

response_status = utils.read_only_property("_status")

def _requires_successful_response(self) -> None:
        if self.response_status != OCSPResponseStatus.SUCCESSFUL:
            raise ValueError(
                "OCSP response status is not successful so the property "
                "has no value"
            )

@property
    def signature_algorithm_oid(self) -> x509.ObjectIdentifier:
        self._requires_successful_response()
        alg = self._backend._lib.OCSP_resp_get0_tbs_sigalg(self._basic)
        self._backend.openssl_assert(alg != self._backend._ffi.NULL)
        oid = _obj2txt(self._backend, alg.algorithm)
        return x509.ObjectIdentifier(oid)

@property
    def signature_hash_algorithm(
        self,
    ) -> typing.Optional[hashes.HashAlgorithm]:
        self._requires_successful_response()
        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(self) -> bytes:
        self._requires_successful_response()
        sig = self._backend._lib.OCSP_resp_get0_signature(self._basic)
        self._backend.openssl_assert(sig != self._backend._ffi.NULL)
        return _asn1_string_to_bytes(self._backend, sig)

@property
    def tbs_response_bytes(self) -> bytes:
        self._requires_successful_response()
        respdata = self._backend._lib.OCSP_resp_get0_respdata(self._basic)
        self._backend.openssl_assert(respdata != self._backend._ffi.NULL)
        pp = self._backend._ffi.new("unsigned char **")
        res = self._backend._lib.i2d_OCSP_RESPDATA(respdata, pp)
        self._backend.openssl_assert(pp[0] != self._backend._ffi.NULL)
        pp = self._backend._ffi.gc(
            pp, lambda pointer: self._backend._lib.OPENSSL_free(pointer[0])
        )
        self._backend.openssl_assert(res > 0)
        return self._backend._ffi.buffer(pp[0], res)[:]

@property
    def certificates(self) -> typing.List[x509.Certificate]:
        self._requires_successful_response()
        sk_x509 = self._backend._lib.OCSP_resp_get0_certs(self._basic)
        num = self._backend._lib.sk_X509_num(sk_x509)
        certs: typing.List[x509.Certificate] = []
        for i in range(num):
            x509_ptr = self._backend._lib.sk_X509_value(sk_x509, i)
            self._backend.openssl_assert(x509_ptr != self._backend._ffi.NULL)
            cert = _Certificate(self._backend, x509_ptr)
            # We need to keep the OCSP response that the certificate came from
            # alive until the Certificate object itself goes out of scope, so
            # we give it a private reference.
            cert._ocsp_resp_ref = self
            certs.append(cert)

return certs

@property
    def responder_key_hash(self) -> typing.Optional[bytes]:
        self._requires_successful_response()
        _, asn1_string = self._responder_key_name()
        if asn1_string == self._backend._ffi.NULL:
            return None
        else:
            return _asn1_string_to_bytes(self._backend, asn1_string)

@property
    def responder_name(self) -> typing.Optional[x509.Name]:
        self._requires_successful_response()
        x509_name, _ = self._responder_key_name()
        if x509_name == self._backend._ffi.NULL:
            return None
        else:
            return _decode_x509_name(self._backend, x509_name)

def _responder_key_name(self):
        asn1_string = self._backend._ffi.new("ASN1_OCTET_STRING **")
        x509_name = self._backend._ffi.new("X509_NAME **")
        res = self._backend._lib.OCSP_resp_get0_id(
            self._basic, asn1_string, x509_name
        )
        self._backend.openssl_assert(res == 1)
        return x509_name[0], asn1_string[0]

@property
    def produced_at(self) -> datetime.datetime:
        self._requires_successful_response()
        produced_at = self._backend._lib.OCSP_resp_get0_produced_at(
            self._basic
        )
        return _parse_asn1_generalized_time(self._backend, produced_at)

@property
    def certificate_status(self) -> OCSPCertStatus:
        self._requires_successful_response()
        status = self._backend._lib.OCSP_single_get0_status(
            self._single,
            self._backend._ffi.NULL,
            self._backend._ffi.NULL,
            self._backend._ffi.NULL,
            self._backend._ffi.NULL,
        )
        self._backend.openssl_assert(status in _CERT_STATUS_TO_ENUM)
        return _CERT_STATUS_TO_ENUM[status]

@property
    def revocation_time(self) -> typing.Optional[datetime.datetime]:
        self._requires_successful_response()
        if self.certificate_status is not OCSPCertStatus.REVOKED:
            return None

asn1_time = self._backend._ffi.new("ASN1_GENERALIZEDTIME **")
        self._backend._lib.OCSP_single_get0_status(
            self._single,
            self._backend._ffi.NULL,
            asn1_time,
            self._backend._ffi.NULL,
            self._backend._ffi.NULL,
        )
        self._backend.openssl_assert(asn1_time[0] != self._backend._ffi.NULL)
        return _parse_asn1_generalized_time(self._backend, asn1_time[0])

@property
    def revocation_reason(self) -> typing.Optional[x509.ReasonFlags]:
        self._requires_successful_response()
        if self.certificate_status is not OCSPCertStatus.REVOKED:
            return None

reason_ptr = self._backend._ffi.new("int *")
        self._backend._lib.OCSP_single_get0_status(
            self._single,
            reason_ptr,
            self._backend._ffi.NULL,
            self._backend._ffi.NULL,
            self._backend._ffi.NULL,
        )
        # If no reason is encoded OpenSSL returns -1
        if reason_ptr[0] == -1:
            return None
        else:
            self._backend.openssl_assert(
                reason_ptr[0] in _CRL_ENTRY_REASON_CODE_TO_ENUM
            )
            return _CRL_ENTRY_REASON_CODE_TO_ENUM[reason_ptr[0]]

@property
    def this_update(self) -> datetime.datetime:
        self._requires_successful_response()
        asn1_time = self._backend._ffi.new("ASN1_GENERALIZEDTIME **")
        self._backend._lib.OCSP_single_get0_status(
            self._single,
            self._backend._ffi.NULL,
            self._backend._ffi.NULL,
            asn1_time,
            self._backend._ffi.NULL,
        )
        self._backend.openssl_assert(asn1_time[0] != self._backend._ffi.NULL)
        return _parse_asn1_generalized_time(self._backend, asn1_time[0])

@property
    def next_update(self) -> typing.Optional[datetime.datetime]:
        self._requires_successful_response()
        asn1_time = self._backend._ffi.new("ASN1_GENERALIZEDTIME **")
        self._backend._lib.OCSP_single_get0_status(
            self._single,
            self._backend._ffi.NULL,
            self._backend._ffi.NULL,
            self._backend._ffi.NULL,
            asn1_time,
        )
        if asn1_time[0] != self._backend._ffi.NULL:
            return _parse_asn1_generalized_time(self._backend, asn1_time[0])
        else:
            return None

@property
    def issuer_key_hash(self) -> bytes:
        self._requires_successful_response()
        return _issuer_key_hash(self._backend, self._cert_id)

@property
    def issuer_name_hash(self) -> bytes:
        self._requires_successful_response()
        return _issuer_name_hash(self._backend, self._cert_id)

@property
    def hash_algorithm(self) -> hashes.HashAlgorithm:
        self._requires_successful_response()
        return _hash_algorithm(self._backend, self._cert_id)

@property
    def serial_number(self) -> int:
        self._requires_successful_response()
        return _serial_number(self._backend, self._cert_id)

@utils.cached_property
    def extensions(self) -> x509.Extensions:
        self._requires_successful_response()
        return self._backend._ocsp_basicresp_ext_parser.parse(self._basic)

@utils.cached_property
    def single_extensions(self) -> x509.Extensions:
        self._requires_successful_response()
        return self._backend._ocsp_singleresp_ext_parser.parse(self._single)

def public_bytes(self, encoding: serialization.Encoding) -> bytes:
        if encoding is not serialization.Encoding.DER:
            raise ValueError("The only allowed encoding value is Encoding.DER")

bio = self._backend._create_mem_bio_gc()
        res = self._backend._lib.i2d_OCSP_RESPONSE_bio(
            bio, self._ocsp_response
        )
        self._backend.openssl_assert(res > 0)
        return self._backend._read_mem_bio(bio)

class _OCSPRequest(OCSPRequest):
    def __init__(self, backend, ocsp_request):
        if backend._lib.OCSP_request_onereq_count(ocsp_request) > 1:
            raise NotImplementedError(
                "OCSP request contains more than one request"
            )
        self._backend = backend
        self._ocsp_request = ocsp_request
        self._request = self._backend._lib.OCSP_request_onereq_get0(
            self._ocsp_request, 0
        )
        self._backend.openssl_assert(self._request != self._backend._ffi.NULL)
        self._cert_id = self._backend._lib.OCSP_onereq_get0_id(self._request)
        self._backend.openssl_assert(self._cert_id != self._backend._ffi.NULL)

@property
    def issuer_key_hash(self) -> bytes:
        return _issuer_key_hash(self._backend, self._cert_id)

@property
    def issuer_name_hash(self) -> bytes:
        return _issuer_name_hash(self._backend, self._cert_id)

@property
    def serial_number(self) -> int:
        return _serial_number(self._backend, self._cert_id)

@property
    def hash_algorithm(self) -> hashes.HashAlgorithm:
        return _hash_algorithm(self._backend, self._cert_id)

@utils.cached_property
    def extensions(self) -> x509.Extensions:
        return self._backend._ocsp_req_ext_parser.parse(self._ocsp_request)

bio = self._backend._create_mem_bio_gc()
        res = self._backend._lib.i2d_OCSP_REQUEST_bio(bio, self._ocsp_request)
        self._backend.openssl_assert(res > 0)
        return self._backend._read_mem_bio(bio)

cmd: