| Server IP : 74.208.127.88 / Your IP : 18.224.64.24 Web Server : Apache/2.4.41 (Ubuntu) System : Linux ubuntu 5.4.0-163-generic #180-Ubuntu SMP Tue Sep 5 13:21:23 UTC 2023 x86_64 User : www-data ( 33) PHP Version : 7.4.3-4ubuntu2.29 Disable Function : pcntl_alarm,pcntl_fork,pcntl_waitpid,pcntl_wait,pcntl_wifexited,pcntl_wifstopped,pcntl_wifsignaled,pcntl_wifcontinued,pcntl_wexitstatus,pcntl_wtermsig,pcntl_wstopsig,pcntl_signal,pcntl_signal_get_handler,pcntl_signal_dispatch,pcntl_get_last_error,pcntl_strerror,pcntl_sigprocmask,pcntl_sigwaitinfo,pcntl_sigtimedwait,pcntl_exec,pcntl_getpriority,pcntl_setpriority,pcntl_async_signals,pcntl_unshare, MySQL : OFF | cURL : ON | WGET : ON | Perl : ON | Python : OFF | Sudo : ON | Pkexec : ON Directory : /lib/python3/dist-packages/certbot/ |
Upload File : |
"""Tools for checking certificate revocation."""
import logging
import re
from datetime import datetime, timedelta
from subprocess import Popen, PIPE
try:
# Only cryptography>=2.5 has ocsp module
# and signature_hash_algorithm attribute in OCSPResponse class
from cryptography.x509 import ocsp # pylint: disable=import-error
getattr(ocsp.OCSPResponse, 'signature_hash_algorithm')
except (ImportError, AttributeError): # pragma: no cover
ocsp = None # type: ignore
from cryptography import x509
from cryptography.hazmat.backends import default_backend
from cryptography.hazmat.primitives import serialization
from cryptography.hazmat.primitives import hashes # type: ignore
from cryptography.exceptions import UnsupportedAlgorithm, InvalidSignature
import pytz
import requests
from acme.magic_typing import Optional, Tuple # pylint: disable=unused-import, no-name-in-module
from certbot import crypto_util
from certbot import errors
from certbot.storage import RenewableCert # pylint: disable=unused-import
from certbot import util
logger = logging.getLogger(__name__)
class RevocationChecker(object):
"""This class figures out OCSP checking on this system, and performs it."""
def __init__(self, enforce_openssl_binary_usage=False):
self.broken = False
self.use_openssl_binary = enforce_openssl_binary_usage or not ocsp
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 = Popen(["openssl", "ocsp", "-header", "var", "val"],
stdout=PIPE, stderr=PIPE, universal_newlines=True)
_out, err = test_host_format.communicate()
if "Missing =" in err:
self.host_args = lambda host: ["Host=" + host]
else:
self.host_args = lambda host: ["Host", host]
def ocsp_revoked(self, cert):
# type: (RenewableCert) -> bool
"""Get revoked status for a particular cert version.
.. todo:: Make this a non-blocking call
:param `.storage.RenewableCert` cert: Certificate object
:returns: True if revoked; False if valid or the check failed or cert is expired.
:rtype: bool
"""
cert_path, chain_path = cert.cert, cert.chain
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 = pytz.UTC.fromutc(datetime.utcnow())
if cert.target_expiry <= 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)
return _check_ocsp_cryptography(cert_path, chain_path, url)
def _check_ocsp_openssl_bin(self, cert_path, chain_path, host, url):
# type: (str, str, str, str) -> bool
# jdkasten thanks "Bulletproof SSL and TLS - Ivan Ristic" for documenting this!
cmd = ["openssl", "ocsp",
"-no_nonce",
"-issuer", chain_path,
"-cert", cert_path,
"-url", url,
"-CAfile", chain_path,
"-verify_other", chain_path,
"-trust_other",
"-header"] + self.host_args(host)
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):
# type: (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 cert at %s", url, cert_path)
return None, None
def _check_ocsp_cryptography(cert_path, chain_path, url):
# type: (str, str, str) -> 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'})
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.error("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.error(str(e))
except errors.Error as e:
logger.error(str(e))
except InvalidSignature:
logger.error('Invalid signature on OCSP response for %s', cert_path)
except AssertionError as error:
logger.error('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, request_ocsp, issuer_cert, cert_path):
"""Verify that the OCSP is valid for serveral criterias"""
# 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
now = datetime.utcnow() # thisUpdate/nextUpdate are expressed in UTC/GMT time zone
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, issuer_cert, cert_path):
"""Verify an OCSP response signature against certificate issuer or responder"""
if response_ocsp.responder_name == issuer_cert.subject:
# 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 cert.subject == response_ocsp.responder_name]
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_hash = responder_cert.signature_hash_algorithm
# 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_hash)
# Following line may raise UnsupportedAlgorithm
chosen_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).
crypto_util.verify_signed_payload(responder_cert.public_key(), response_ocsp.signature,
response_ocsp.tbs_response_bytes, chosen_hash)
def _translate_ocsp_query(cert_path, ocsp_output, ocsp_errors):
"""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