✏️ Editing: ocsp.py Read Only

"""Tools for checking certificate revocation."""
from datetime import datetime
from datetime import timedelta
import logging
import re
import subprocess
from subprocess import PIPE
from typing import Optional
from typing import Tuple
import warnings

from cryptography import x509
from cryptography.exceptions import InvalidSignature
from cryptography.exceptions import UnsupportedAlgorithm
from cryptography.hazmat.backends import default_backend
from cryptography.hazmat.primitives import hashes
from cryptography.hazmat.primitives import serialization
from cryptography.x509 import ocsp
import pytz
import requests

from certbot import crypto_util
from certbot import errors
from certbot import util
from certbot.compat.os import getenv
from certbot.interfaces import RenewableCert

logger = logging.getLogger(__name__)

class RevocationChecker:
    """This class figures out OCSP checking on this system, and performs it."""

def __init__(self, enforce_openssl_binary_usage: bool = False) -> None:
        self.broken = False
        self.use_openssl_binary = enforce_openssl_binary_usage

if self.use_openssl_binary:
            if not util.exe_exists("openssl"):
                logger.info("openssl not installed, can't check revocation")
                self.broken = True
                return

# New versions of openssl want -header var=val, old ones want -header var val
            test_host_format = subprocess.run(["openssl", "ocsp", "-header", "var", "val"],
                                     stdout=PIPE, stderr=PIPE, universal_newlines=True,
                                     check=False, env=util.env_no_snap_for_external_calls())
            if "Missing =" in test_host_format.stderr:
                self.host_args = lambda host: ["Host=" + host]
            else:
                self.host_args = lambda host: ["Host", host]

def ocsp_revoked(self, cert: RenewableCert) -> bool:
        """Get revoked status for a particular cert version.

.. todo:: Make this a non-blocking call

:param `.interfaces.RenewableCert` cert: Certificate object
        :returns: True if revoked; False if valid or the check failed or cert is expired.
        :rtype: bool

"""
        return self.ocsp_revoked_by_paths(cert.cert_path, cert.chain_path)

def ocsp_revoked_by_paths(self, cert_path: str, chain_path: str, timeout: int = 10) -> bool:
        """Performs the OCSP revocation check

:param str cert_path: Certificate filepath
        :param str chain_path: Certificate chain
        :param int timeout: Timeout (in seconds) for the OCSP query

:returns: True if revoked; False if valid or the check failed or cert is expired.
        :rtype: bool

"""
        if self.broken:
            return False

# Let's Encrypt doesn't update OCSP for expired certificates,
        # so don't check OCSP if the cert is expired.
        # https://github.com/certbot/certbot/issues/7152
        now = datetime.now(pytz.UTC)
        if crypto_util.notAfter(cert_path) <= now:
            return False

url, host = _determine_ocsp_server(cert_path)
        if not host or not url:
            return False

if self.use_openssl_binary:
            return self._check_ocsp_openssl_bin(cert_path, chain_path, host, url, timeout)
        return _check_ocsp_cryptography(cert_path, chain_path, url, timeout)

def _check_ocsp_openssl_bin(self, cert_path: str, chain_path: str,
                                host: str, url: str, timeout: int) -> bool:
        # Minimal implementation of proxy selection logic as seen in, e.g., cURL
        # Some things that won't work, but may well be in use somewhere:
        # - username and password for proxy authentication
        # - proxies accepting TLS connections
        # - proxy exclusion through NO_PROXY
        env_http_proxy = getenv('http_proxy')
        env_HTTP_PROXY = getenv('HTTP_PROXY')
        proxy_host = None
        if env_http_proxy is not None or env_HTTP_PROXY is not None:
            proxy_host = env_http_proxy if env_http_proxy is not None else env_HTTP_PROXY
        if proxy_host is None:
            url_opts = ["-url", url]
        else:
            if proxy_host.startswith('http://'):
                proxy_host = proxy_host[len('http://'):]
            url_opts = ["-host", proxy_host, "-path", url]
        # jdkasten thanks "Bulletproof SSL and TLS - Ivan Ristic" for documenting this!
        cmd = ["openssl", "ocsp",
               "-no_nonce",
               "-issuer", chain_path,
               "-cert", cert_path,
               "-CAfile", chain_path,
               "-verify_other", chain_path,
               "-trust_other",
               "-timeout", str(timeout),
               "-header"] + self.host_args(host) + url_opts
        logger.debug("Querying OCSP for %s", cert_path)
        logger.debug(" ".join(cmd))
        try:
            output, err = util.run_script(cmd, log=logger.debug)
        except errors.SubprocessError:
            logger.info("OCSP check failed for %s (are we offline?)", cert_path)
            return False
        return _translate_ocsp_query(cert_path, output, err)

def _determine_ocsp_server(cert_path: str) -> Tuple[Optional[str], Optional[str]]:
    """Extract the OCSP server host from a certificate.

:param str cert_path: Path to the cert we're checking OCSP for
    :rtype tuple:
    :returns: (OCSP server URL or None, OCSP server host or None)

"""
    with open(cert_path, 'rb') as file_handler:
        cert = x509.load_pem_x509_certificate(file_handler.read(), default_backend())
    try:
        extension = cert.extensions.get_extension_for_class(x509.AuthorityInformationAccess)
        ocsp_oid = x509.AuthorityInformationAccessOID.OCSP
        descriptions = [description for description in extension.value
                        if description.access_method == ocsp_oid]

url = descriptions[0].access_location.value
    except (x509.ExtensionNotFound, IndexError):
        logger.info("Cannot extract OCSP URI from %s", cert_path)
        return None, None

url = url.rstrip()
    host = url.partition("://")[2].rstrip("/")

if host:
        return url, host
    logger.info("Cannot process OCSP host from URL (%s) in certificate at %s", url, cert_path)
    return None, None

def _check_ocsp_cryptography(cert_path: str, chain_path: str, url: str, timeout: int) -> bool:
    # Retrieve OCSP response
    with open(chain_path, 'rb') as file_handler:
        issuer = x509.load_pem_x509_certificate(file_handler.read(), default_backend())
    with open(cert_path, 'rb') as file_handler:
        cert = x509.load_pem_x509_certificate(file_handler.read(), default_backend())
    builder = ocsp.OCSPRequestBuilder()
    builder = builder.add_certificate(cert, issuer, hashes.SHA1())
    request = builder.build()
    request_binary = request.public_bytes(serialization.Encoding.DER)
    try:
        response = requests.post(url, data=request_binary,
                                 headers={'Content-Type': 'application/ocsp-request'},
                                 timeout=timeout)
    except requests.exceptions.RequestException:
        logger.info("OCSP check failed for %s (are we offline?)", cert_path, exc_info=True)
        return False
    if response.status_code != 200:
        logger.info("OCSP check failed for %s (HTTP status: %d)", cert_path, response.status_code)
        return False

response_ocsp = ocsp.load_der_ocsp_response(response.content)

# Check OCSP response validity
    if response_ocsp.response_status != ocsp.OCSPResponseStatus.SUCCESSFUL:
        logger.warning("Invalid OCSP response status for %s: %s",
                     cert_path, response_ocsp.response_status)
        return False

# Check OCSP signature
    try:
        _check_ocsp_response(response_ocsp, request, issuer, cert_path)
    except UnsupportedAlgorithm as e:
        logger.warning(str(e))
    except errors.Error as e:
        logger.warning(str(e))
    except InvalidSignature:
        logger.warning('Invalid signature on OCSP response for %s', cert_path)
    except AssertionError as error:
        logger.warning('Invalid OCSP response for %s: %s.', cert_path, str(error))
    else:
        # Check OCSP certificate status
        logger.debug("OCSP certificate status for %s is: %s",
                     cert_path, response_ocsp.certificate_status)
        return response_ocsp.certificate_status == ocsp.OCSPCertStatus.REVOKED

return False

def _check_ocsp_response(response_ocsp: 'ocsp.OCSPResponse', request_ocsp: 'ocsp.OCSPRequest',
                         issuer_cert: x509.Certificate, cert_path: str) -> None:
    """Verify that the OCSP is valid for several criteria"""
    # Assert OCSP response corresponds to the certificate we are talking about
    if response_ocsp.serial_number != request_ocsp.serial_number:
        raise AssertionError('the certificate in response does not correspond '
                             'to the certificate in request')

# Assert signature is valid
    _check_ocsp_response_signature(response_ocsp, issuer_cert, cert_path)

# Assert issuer in response is the expected one
    if (not isinstance(response_ocsp.hash_algorithm, type(request_ocsp.hash_algorithm))
            or response_ocsp.issuer_key_hash != request_ocsp.issuer_key_hash
            or response_ocsp.issuer_name_hash != request_ocsp.issuer_name_hash):
        raise AssertionError('the issuer does not correspond to issuer of the certificate.')

# In following checks, two situations can occur:
    #   * nextUpdate is set, and requirement is thisUpdate < now < nextUpdate
    #   * nextUpdate is not set, and requirement is thisUpdate < now
    # NB1: We add a validity period tolerance to handle clock time inconsistencies,
    #      value is 5 min like for OpenSSL.
    # NB2: Another check is to verify that thisUpdate is not too old, it is optional
    #      for OpenSSL, so we do not do it here.
    # See OpenSSL implementation as a reference:
    # https://github.com/openssl/openssl/blob/ef45aa14c5af024fcb8bef1c9007f3d1c115bd85/crypto/ocsp/ocsp_cl.c#L338-L391
    # thisUpdate/nextUpdate are expressed in UTC/GMT time zone
    now = datetime.now(pytz.UTC).replace(tzinfo=None)
    # The this_update and next_update attributes were deprecated in cryptography's 43.0.0 release.
    # Updating the code and tests to (also) work with the new API is being tracked in
    # https://github.com/certbot/certbot/issues/10053.
    with warnings.catch_warnings():
        warnings.filterwarnings('ignore', message='Properties that return.*datetime object')
        if not response_ocsp.this_update:
            raise AssertionError('param thisUpdate is not set.')
        if response_ocsp.this_update > now + timedelta(minutes=5):
            raise AssertionError('param thisUpdate is in the future.')
        if response_ocsp.next_update and response_ocsp.next_update < now - timedelta(minutes=5):
            raise AssertionError('param nextUpdate is in the past.')

def _check_ocsp_response_signature(response_ocsp: 'ocsp.OCSPResponse',
                                   issuer_cert: x509.Certificate, cert_path: str) -> None:
    """Verify an OCSP response signature against certificate issuer or responder"""
    def _key_hash(cert: x509.Certificate) -> bytes:
        return x509.SubjectKeyIdentifier.from_public_key(cert.public_key()).digest

if (response_ocsp.responder_name == issuer_cert.subject
            or response_ocsp.responder_key_hash == _key_hash(issuer_cert)):
        # Case where the OCSP responder is also the certificate issuer
        logger.debug('OCSP response for certificate %s is signed by the certificate\'s issuer.',
                     cert_path)
        responder_cert = issuer_cert
    else:
        # Case where the OCSP responder is not the certificate issuer
        logger.debug('OCSP response for certificate %s is delegated to an external responder.',
                     cert_path)

responder_certs = [cert for cert in response_ocsp.certificates
                           if response_ocsp.responder_name == cert.subject or \
                              response_ocsp.responder_key_hash == _key_hash(cert)]
        if not responder_certs:
            raise AssertionError('no matching responder certificate could be found')

# We suppose here that the ACME server support only one certificate in the OCSP status
        # request. This is currently the case for LetsEncrypt servers.
        # See https://github.com/letsencrypt/boulder/issues/2331
        responder_cert = responder_certs[0]

if responder_cert.issuer != issuer_cert.subject:
            raise AssertionError('responder certificate is not signed '
                                 'by the certificate\'s issuer')

try:
            extension = responder_cert.extensions.get_extension_for_class(x509.ExtendedKeyUsage)
            delegate_authorized = x509.oid.ExtendedKeyUsageOID.OCSP_SIGNING in extension.value
        except (x509.ExtensionNotFound, IndexError):
            delegate_authorized = False
        if not delegate_authorized:
            raise AssertionError('responder is not authorized by issuer to sign OCSP responses')

# Following line may raise UnsupportedAlgorithm
        chosen_cert_hash = responder_cert.signature_hash_algorithm
        assert chosen_cert_hash # always present for RSA and ECDSA certificates.
        # For a delegate OCSP responder, we need first check that its certificate is effectively
        # signed by the certificate issuer.
        crypto_util.verify_signed_payload(issuer_cert.public_key(), responder_cert.signature,
                                          responder_cert.tbs_certificate_bytes, chosen_cert_hash)

# Following line may raise UnsupportedAlgorithm
    chosen_response_hash = response_ocsp.signature_hash_algorithm
    # We check that the OSCP response is effectively signed by the responder
    # (an authorized delegate one or the certificate issuer itself).
    if not chosen_response_hash:
        raise AssertionError("no signature hash algorithm defined")
    crypto_util.verify_signed_payload(responder_cert.public_key(), response_ocsp.signature,
                                      response_ocsp.tbs_response_bytes, chosen_response_hash)

def _translate_ocsp_query(cert_path: str, ocsp_output: str, ocsp_errors: str) -> bool:
    """Parse openssl's weird output to work out what it means."""

states = ("good", "revoked", "unknown")
    patterns = [r"{0}: (WARNING.*)?{1}".format(cert_path, s) for s in states]
    good, revoked, unknown = (re.search(p, ocsp_output, flags=re.DOTALL) for p in patterns)

warning = good.group(1) if good else None

if ("Response verify OK" not in ocsp_errors) or (good and warning) or unknown:
        logger.info("Revocation status for %s is unknown", cert_path)
        logger.debug("Uncertain output:\n%s\nstderr:\n%s", ocsp_output, ocsp_errors)
        return False
    elif good and not warning:
        return False
    elif revoked:
        warning = revoked.group(1)
        if warning:
            logger.info("OCSP revocation warning: %s", warning)
        return True
    else:
        logger.warning("Unable to properly parse OCSP output: %s\nstderr:%s",
                       ocsp_output, ocsp_errors)
        return False

🔐 Sid Gifari File Manager Pro