support OCSP response parsing (#4452)

* support OCSP response parsing

* move the decorator to make pep8 happy

* add some missing docs

* review feedback

* more review feedback

2 files changed, 254 insertions, 4 deletions

diff --git a/src/cryptography/hazmat/backends/openssl/backend.py b/src/cryptography/hazmat/backends/openssl/backend.py
index 8b4792b4..f374a8e3 100644
--- a/src/cryptography/hazmat/backends/openssl/backend.py
+++ b/src/cryptography/hazmat/backends/openssl/backend.py
@@ -42,7 +42,9 @@ from cryptography.hazmat.backends.openssl.encode_asn1 import (
 )
 from cryptography.hazmat.backends.openssl.hashes import _HashContext
 from cryptography.hazmat.backends.openssl.hmac import _HMACContext
-from cryptography.hazmat.backends.openssl.ocsp import _OCSPRequest
+from cryptography.hazmat.backends.openssl.ocsp import (
+    _OCSPRequest, _OCSPResponse
+)
 from cryptography.hazmat.backends.openssl.rsa import (
     _RSAPrivateKey, _RSAPublicKey
 )
@@ -1441,6 +1443,16 @@ class Backend(object):
         request = self._ffi.gc(request, self._lib.OCSP_REQUEST_free)
         return _OCSPRequest(self, request)
 
+    def load_der_ocsp_response(self, data):
+        mem_bio = self._bytes_to_bio(data)
+        response = self._lib.d2i_OCSP_RESPONSE_bio(mem_bio.bio, self._ffi.NULL)
+        if response == self._ffi.NULL:
+            self._consume_errors()
+            raise ValueError("Unable to load OCSP response")
+
+        response = self._ffi.gc(response, self._lib.OCSP_RESPONSE_free)
+        return _OCSPResponse(self, response)
+
     def create_ocsp_request(self, builder):
         ocsp_req = self._lib.OCSP_REQUEST_new()
         self.openssl_assert(ocsp_req != self._ffi.NULL)
diff --git a/src/cryptography/hazmat/backends/openssl/ocsp.py b/src/cryptography/hazmat/backends/openssl/ocsp.py
index 420d7eb6..f3f18cb0 100644
--- a/src/cryptography/hazmat/backends/openssl/ocsp.py
+++ b/src/cryptography/hazmat/backends/openssl/ocsp.py
@@ -4,13 +4,35 @@
 
 from __future__ import absolute_import, division, print_function
 
-from cryptography import utils
+import functools
+
+from cryptography import utils, x509
 from cryptography.exceptions import UnsupportedAlgorithm
 from cryptography.hazmat.backends.openssl.decode_asn1 import (
-    _OCSP_REQ_EXT_PARSER, _asn1_integer_to_int, _asn1_string_to_bytes, _obj2txt
+    _CRL_ENTRY_REASON_CODE_TO_ENUM, _OCSP_REQ_EXT_PARSER, _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 serialization
-from cryptography.x509.ocsp import OCSPRequest, _OIDS_TO_HASH
+from cryptography.x509.ocsp import (
+    OCSPCertStatus, OCSPRequest, OCSPResponse, OCSPResponseStatus,
+    _CERT_STATUS_TO_ENUM, _OIDS_TO_HASH, _RESPONSE_STATUS_TO_ENUM,
+)
+
+
+def _requires_successful_response(func):
+    @functools.wraps(func)
+    def wrapper(self, *args):
+        if self.response_status != OCSPResponseStatus.SUCCESSFUL:
+            raise ValueError(
+                "OCSP response status is not successful so the property "
+                "has no value"
+            )
+        else:
+            return func(self, *args)
+
+    return wrapper
 
 
 def _issuer_key_hash(backend, cert_id):
@@ -63,6 +85,222 @@ def _hash_algorithm(backend, cert_id):
         )
 
 
+@utils.register_interface(OCSPResponse)
+class _OCSPResponse(object):
+    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
+            )
+            self._backend.openssl_assert(
+                self._backend._lib.OCSP_resp_count(self._basic) == 1
+            )
+            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")
+
+    @property
+    @_requires_successful_response
+    def signature_algorithm_oid(self):
+        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
+    @_requires_successful_response
+    def signature(self):
+        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
+    @_requires_successful_response
+    def tbs_response_bytes(self):
+        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
+    @_requires_successful_response
+    def certificates(self):
+        sk_x509 = self._backend._lib.OCSP_resp_get0_certs(self._basic)
+        num = self._backend._lib.sk_X509_num(sk_x509)
+        certs = []
+        for i in range(num):
+            x509 = self._backend._lib.sk_X509_value(sk_x509, i)
+            self._backend.openssl_assert(x509 != self._backend._ffi.NULL)
+            cert = _Certificate(self._backend, x509)
+            # 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 = self
+            certs.append(cert)
+
+        return certs
+
+    @property
+    @_requires_successful_response
+    def responder_key_hash(self):
+        _, 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
+    @_requires_successful_response
+    def responder_name(self):
+        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
+    @_requires_successful_response
+    def produced_at(self):
+        produced_at = self._backend._lib.OCSP_resp_get0_produced_at(
+            self._basic
+        )
+        return _parse_asn1_generalized_time(self._backend, produced_at)
+
+    @property
+    @_requires_successful_response
+    def certificate_status(self):
+        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
+    @_requires_successful_response
+    def revocation_time(self):
+        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
+    @_requires_successful_response
+    def revocation_reason(self):
+        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
+    @_requires_successful_response
+    def this_update(self):
+        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
+    @_requires_successful_response
+    def next_update(self):
+        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,
+        )
+        self._backend.openssl_assert(asn1_time[0] != self._backend._ffi.NULL)
+        return _parse_asn1_generalized_time(self._backend, asn1_time[0])
+
+    @property
+    @_requires_successful_response
+    def issuer_key_hash(self):
+        return _issuer_key_hash(self._backend, self._cert_id)
+
+    @property
+    @_requires_successful_response
+    def issuer_name_hash(self):
+        return _issuer_name_hash(self._backend, self._cert_id)
+
+    @property
+    @_requires_successful_response
+    def hash_algorithm(self):
+        return _hash_algorithm(self._backend, self._cert_id)
+
+    @property
+    @_requires_successful_response
+    def serial_number(self):
+        return _serial_number(self._backend, self._cert_id)
+
+
 @utils.register_interface(OCSPRequest)
 class _OCSPRequest(object):
     def __init__(self, backend, ocsp_request):