mirror of
https://github.com/certbot/certbot.git
synced 2026-07-23 16:22:34 +02:00
Remove openssl-based OCSP checking
Use exclusively the `cryptography` package for OCSP checking.
This commit is contained in:
Jacob Hoffman-Andrews
@@ -24,107 +24,6 @@ ocsp: Use -help for summary.
|
||||
"""
|
||||
|
||||
|
||||
class OCSPTestOpenSSL(unittest.TestCase):
|
||||
"""
|
||||
OCSP revocation tests using OpenSSL binary.
|
||||
"""
|
||||
|
||||
def setUp(self):
|
||||
from certbot import ocsp
|
||||
with mock.patch('certbot.ocsp.subprocess.run') as mock_run:
|
||||
with mock.patch('certbot.util.exe_exists') as mock_exists:
|
||||
mock_run.stderr = out
|
||||
mock_exists.return_value = True
|
||||
self.checker = ocsp.RevocationChecker(enforce_openssl_binary_usage=True)
|
||||
|
||||
@mock.patch('certbot.ocsp.logger.info')
|
||||
@mock.patch('certbot.ocsp.subprocess.run')
|
||||
@mock.patch('certbot.util.exe_exists')
|
||||
def test_init(self, mock_exists, mock_run, mock_log):
|
||||
mock_run.return_value.stderr = out
|
||||
mock_exists.return_value = True
|
||||
|
||||
from certbot import ocsp
|
||||
checker = ocsp.RevocationChecker(enforce_openssl_binary_usage=True)
|
||||
assert mock_run.call_count == 1
|
||||
assert checker.host_args("x") == ["Host=x"]
|
||||
|
||||
mock_run.return_value.stderr = out.partition("\n")[2]
|
||||
checker = ocsp.RevocationChecker(enforce_openssl_binary_usage=True)
|
||||
assert checker.host_args("x") == ["Host", "x"]
|
||||
assert checker.broken is False
|
||||
|
||||
mock_exists.return_value = False
|
||||
mock_run.call_count = 0
|
||||
checker = ocsp.RevocationChecker(enforce_openssl_binary_usage=True)
|
||||
assert mock_run.call_count == 0
|
||||
assert mock_log.call_count == 1
|
||||
assert checker.broken is True
|
||||
|
||||
@mock.patch('certbot.ocsp._determine_ocsp_server')
|
||||
@mock.patch('certbot.ocsp.crypto_util.notAfter')
|
||||
@mock.patch('certbot.util.run_script')
|
||||
def test_ocsp_revoked(self, mock_run, mock_na, mock_determine):
|
||||
now = datetime.now(pytz.UTC)
|
||||
cert_obj = mock.MagicMock()
|
||||
cert_obj.cert_path = "x"
|
||||
cert_obj.chain_path = "y"
|
||||
mock_na.return_value = now + timedelta(hours=2)
|
||||
|
||||
self.checker.broken = True
|
||||
mock_determine.return_value = ("", "")
|
||||
assert self.checker.ocsp_revoked(cert_obj) is False
|
||||
|
||||
self.checker.broken = False
|
||||
mock_run.return_value = tuple(openssl_happy[1:])
|
||||
assert self.checker.ocsp_revoked(cert_obj) is False
|
||||
assert mock_run.call_count == 0
|
||||
|
||||
mock_determine.return_value = ("http://x.co", "x.co")
|
||||
assert self.checker.ocsp_revoked(cert_obj) is False
|
||||
mock_run.side_effect = errors.SubprocessError("Unable to load certificate launcher")
|
||||
assert self.checker.ocsp_revoked(cert_obj) is False
|
||||
assert mock_run.call_count == 2
|
||||
|
||||
# cert expired
|
||||
mock_na.return_value = now
|
||||
mock_determine.return_value = ("", "")
|
||||
count_before = mock_determine.call_count
|
||||
assert self.checker.ocsp_revoked(cert_obj) is False
|
||||
assert mock_determine.call_count == count_before
|
||||
|
||||
def test_determine_ocsp_server(self):
|
||||
cert_path = test_util.vector_path('ocsp_certificate.pem')
|
||||
|
||||
from certbot import ocsp
|
||||
result = ocsp._determine_ocsp_server(cert_path)
|
||||
assert ('http://ocsp.test4.buypass.com', 'ocsp.test4.buypass.com') == result
|
||||
|
||||
@mock.patch('certbot.ocsp.logger')
|
||||
@mock.patch('certbot.util.run_script')
|
||||
def test_translate_ocsp(self, mock_run, mock_log):
|
||||
# pylint: disable=protected-access
|
||||
mock_run.return_value = openssl_confused
|
||||
from certbot import ocsp
|
||||
assert ocsp._translate_ocsp_query(*openssl_happy) is False
|
||||
assert ocsp._translate_ocsp_query(*openssl_confused) is False
|
||||
assert mock_log.debug.call_count == 1
|
||||
assert mock_log.warning.call_count == 0
|
||||
mock_log.debug.call_count = 0
|
||||
assert ocsp._translate_ocsp_query(*openssl_unknown) is False
|
||||
assert mock_log.debug.call_count == 1
|
||||
assert mock_log.warning.call_count == 0
|
||||
assert ocsp._translate_ocsp_query(*openssl_expired_ocsp) is False
|
||||
assert mock_log.debug.call_count == 2
|
||||
assert ocsp._translate_ocsp_query(*openssl_broken) is False
|
||||
assert mock_log.warning.call_count == 1
|
||||
mock_log.info.call_count = 0
|
||||
assert ocsp._translate_ocsp_query(*openssl_revoked) is True
|
||||
assert mock_log.info.call_count == 0
|
||||
assert ocsp._translate_ocsp_query(*openssl_expired_ocsp_revoked) is True
|
||||
assert mock_log.info.call_count == 1
|
||||
|
||||
|
||||
class OSCPTestCryptography(unittest.TestCase):
|
||||
"""
|
||||
OCSP revokation tests using Cryptography >= 2.4.0
|
||||
@@ -330,59 +229,5 @@ def _construct_mock_ocsp_response(certificate_status, response_status):
|
||||
)
|
||||
|
||||
|
||||
# pylint: disable=line-too-long
|
||||
openssl_confused = ("", """
|
||||
/etc/letsencrypt/live/example.org/cert.pem: good
|
||||
This Update: Dec 17 00:00:00 2016 GMT
|
||||
Next Update: Dec 24 00:00:00 2016 GMT
|
||||
""",
|
||||
"""
|
||||
Response Verify Failure
|
||||
139903674214048:error:27069065:OCSP routines:OCSP_basic_verify:certificate verify error:ocsp_vfy.c:138:Verify error:unable to get local issuer certificate
|
||||
""")
|
||||
|
||||
openssl_happy = ("blah.pem", """
|
||||
blah.pem: good
|
||||
This Update: Dec 20 18:00:00 2016 GMT
|
||||
Next Update: Dec 27 18:00:00 2016 GMT
|
||||
""",
|
||||
"Response verify OK")
|
||||
|
||||
openssl_revoked = ("blah.pem", """
|
||||
blah.pem: revoked
|
||||
This Update: Dec 20 01:00:00 2016 GMT
|
||||
Next Update: Dec 27 01:00:00 2016 GMT
|
||||
Revocation Time: Dec 20 01:46:34 2016 GMT
|
||||
""",
|
||||
"""Response verify OK""")
|
||||
|
||||
openssl_unknown = ("blah.pem", """
|
||||
blah.pem: unknown
|
||||
This Update: Dec 20 18:00:00 2016 GMT
|
||||
Next Update: Dec 27 18:00:00 2016 GMT
|
||||
""",
|
||||
"Response verify OK")
|
||||
|
||||
openssl_broken = ("", "tentacles", "Response verify OK")
|
||||
|
||||
openssl_expired_ocsp = ("blah.pem", """
|
||||
blah.pem: WARNING: Status times invalid.
|
||||
140659132298912:error:2707307D:OCSP routines:OCSP_check_validity:status expired:ocsp_cl.c:372:
|
||||
good
|
||||
This Update: Apr 6 00:00:00 2016 GMT
|
||||
Next Update: Apr 13 00:00:00 2016 GMT
|
||||
""",
|
||||
"""Response verify OK""")
|
||||
|
||||
openssl_expired_ocsp_revoked = ("blah.pem", """
|
||||
blah.pem: WARNING: Status times invalid.
|
||||
140659132298912:error:2707307D:OCSP routines:OCSP_check_validity:status expired:ocsp_cl.c:372:
|
||||
revoked
|
||||
This Update: Apr 6 00:00:00 2016 GMT
|
||||
Next Update: Apr 13 00:00:00 2016 GMT
|
||||
""",
|
||||
"""Response verify OK""")
|
||||
|
||||
|
||||
if __name__ == '__main__':
|
||||
sys.exit(pytest.main(sys.argv[1:] + [__file__])) # pragma: no cover
|
||||
|
||||
@@ -2,9 +2,6 @@
|
||||
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
|
||||
|
||||
@@ -20,8 +17,6 @@ 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__)
|
||||
@@ -29,26 +24,6 @@ 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.
|
||||
|
||||
@@ -72,9 +47,6 @@ class RevocationChecker:
|
||||
: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
|
||||
@@ -86,48 +58,8 @@ class RevocationChecker:
|
||||
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.
|
||||
|
||||
@@ -299,29 +231,3 @@ def _check_ocsp_response_signature(response_ocsp: 'ocsp.OCSPResponse',
|
||||
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
|
||||
|
||||
Reference in New Issue
Block a user