starred/certbot-certbot-1
1
0
Fork 0
mirror of https://github.com/certbot/certbot.git synced 2026-08-02 03:22:14 +02:00
Code Issues Packages Projects Releases Wiki Activity

Use built-in support for OCSP in cryptography >= 2.5 (#6603)

Browse Source 
In response to #6594. [Fixes #6594.]

To execute OCSP requests, certbot relies currently on a openssl binary execution. If openssl is not present in the PATH, the OCSP check will be silently ignored. Since version 2.4, cryptography has support for OCSP requests, without the need to have openssl binary available locally.

This PR takes advantage of it, and will use the built-in support of OCSP in cryptography for versions >= 2.4. Otherwise, fallback is done do a direct call to openssl binary, allowing oldest requirements to still work with legacy cryptography versions.

Update: requirement is now cryptography >= 2.5, to avoid to rely on a private method from cryptography.

* Implement logic using cryptography

* Working OSCP using pure cryptography

* Fix openssl usage in unit tests

* Reduce verbosity

* Add tests

* Improve naive skipIf

* Test resiliency

* Update ocsp.py

* Validate OCSP response. Unify OCSP URL get

* Improve resiliency checks, correct lint/mypy

* Improve hash selection

* Fix warnings when calling openssl bin

* Load OCSP tests assets as vectors.

* Update ocsp.py

* Protect against invalid ocsp response.

* Add checks to OCSP response

* Add more control on ocsp response

* Be lenient about assertion that next_update must be in the future, similarly to openssl.

* Construct a more advanced OCSP response mock to trigger more logic in ocsp module.

* Add test

* Refactor signature process to use crypto_util

* Fallback for cryptography 2.4

* Avoid a collision with a meteor.

* Correct method signature documentation

* Relax OCSP update interval

* Trigger built-in ocsp logic from cryptography with 2.5+

* Update pinned version of cryptography

* Update certbot/ocsp.py

Co-Authored-By: adferrand <adferrand@users.noreply.github.com>

* Update ocsp.py

* Update ocsp_test.py

* Update CHANGELOG.md

* Update CHANGELOG.md
This commit is contained in:
Adrien Ferrand
2019-02-05 10:45:15 -08:00
committed by Brad Warren
parent 9671985885
commit 2ddaf3db04
9 changed files with 451 additions and 125 deletions
Download Patch File Download Diff File Expand all files Collapse all files
certbot/ocsp.py
+153 -44
View File
@@ -1,53 +1,79 @@
"""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 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 import util
logger = logging.getLogger(__name__)
class RevocationChecker(object):
"This class figures out OCSP checking on this system, and performs it."
"""This class figures out OCSP checking on this system, and performs it."""
def __init__(self):
def __init__(self, enforce_openssl_binary_usage=False):
self.broken = False
self.use_openssl_binary = enforce_openssl_binary_usage or not ocsp
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]
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_path, chain_path):
# type: (str, str) -> bool
"""Get revoked status for a particular cert version.
.. todo:: Make this a non-blocking call
:param str cert_path: Path to certificate
:param str chain_path: Path to intermediate cert
:rtype bool or None:
:returns: True if revoked; False if valid or the check failed
:rtype: bool
"""
if self.broken:
return False
url, host = self.determine_ocsp_server(cert_path)
if not host:
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)
else:
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",
@@ -65,33 +91,117 @@ class RevocationChecker(object):
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(self, cert_path):
"""Extract the OCSP server host from a certificate.
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)
: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)
"""
try:
url, _err = util.run_script(
["openssl", "x509", "-in", cert_path, "-noout", "-ocsp_uri"],
log=logger.debug)
except errors.SubprocessError:
logger.info("Cannot extract OCSP URI from %s", cert_path)
return None, 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
else:
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)
response = requests.post(url, data=request_binary,
headers={'Content-Type': 'application/ocsp-request'})
if response.status_code != 200:
logger.info("OCSP check failed for %s (are we offline?)", cert_path)
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)
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):
"""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)
# 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.')
# Assert nextUpdate is in the future, and that thisUpdate is not too old
if response_ocsp.next_update:
if response_ocsp.next_update < datetime.now() - timedelta(minutes=5):
raise AssertionError('next update is in the past.')
interval = response_ocsp.next_update - response_ocsp.this_update
if datetime.now() - response_ocsp.this_update > interval + timedelta(minutes=5):
raise AssertionError('this update is too old.')
def _check_ocsp_response_signature(response_ocsp, issuer_cert):
"""Verify an OCSP response signature against certificate issuer"""
# Following line may raise UnsupportedAlgorithm
chosen_hash = response_ocsp.signature_hash_algorithm
crypto_util.verify_signed_payload(issuer_cert.public_key(), response_ocsp.signature,
response_ocsp.tbs_response_bytes, chosen_hash)
url = url.rstrip()
host = url.partition("://")[2].rstrip("/")
if host:
return url, host
else:
logger.info("Cannot process OCSP host from URL (%s) in cert at %s", url, cert_path)
return None, None
def _translate_ocsp_query(cert_path, ocsp_output, ocsp_errors):
"""Parse openssl's weird output to work out what it means."""
@@ -102,7 +212,7 @@ def _translate_ocsp_query(cert_path, ocsp_output, ocsp_errors):
warning = good.group(1) if good else None
if (not "Response verify OK" in ocsp_errors) or (good and warning) or unknown:
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
@@ -115,6 +225,5 @@ def _translate_ocsp_query(cert_path, ocsp_output, ocsp_errors):
return True
else:
logger.warning("Unable to properly parse OCSP output: %s\nstderr:%s",
ocsp_output, ocsp_errors)
ocsp_output, ocsp_errors)
return False