From 3c3928a6901f4fa52a91147f8f92d5333c8385cc Mon Sep 17 00:00:00 2001 From: Adrien Ferrand Date: Mon, 24 Jun 2019 21:54:20 +0200 Subject: [PATCH] Simplify logic --- .../certbot_tests/context.py | 31 +++++- .../certbot_tests/test_main.py | 44 ++++----- .../certbot_integration_tests/utils/misc.py | 30 +----- .../utils/ocsp_server.py | 94 +++++++------------ 4 files changed, 84 insertions(+), 115 deletions(-) diff --git a/certbot-ci/certbot_integration_tests/certbot_tests/context.py b/certbot-ci/certbot_integration_tests/certbot_tests/context.py index 34072bd5c..e4cd841f5 100644 --- a/certbot-ci/certbot_integration_tests/certbot_tests/context.py +++ b/certbot-ci/certbot_integration_tests/certbot_tests/context.py @@ -1,15 +1,20 @@ """Module to handle the context of integration tests.""" import os import shutil +import subprocess import sys import tempfile -from certbot_integration_tests.utils import certbot_call +import requests + +from certbot_integration_tests.utils import certbot_call, misc, ocsp_server +from certbot_integration_tests.utils.constants import MOCK_OCSP_SERVER_PORT class IntegrationTestsContext(object): """General fixture describing a certbot integration tests context""" def __init__(self, request): + self._subprocesses = [] self.request = request if hasattr(request.config, 'slaveinput'): # Worker node @@ -51,6 +56,9 @@ class IntegrationTestsContext(object): def cleanup(self): """Cleanup the integration test context.""" + for process in self._subprocesses: + process.terminate() + process.wait() shutil.rmtree(self.workspace) def certbot(self, args, force_renew=True): @@ -77,3 +85,24 @@ class IntegrationTestsContext(object): :return: the well-formed domain suitable for redirection on """ return '{0}.{1}.wtf'.format(subdomain, self.worker_id) + + def mock_ocsp_server(self, cert_path, ocsp_status): + """ + Start a mock OCSP server to check OSCP statuses with Pebble. + :param cert_path: the path to the cert whose OCSP status is checked + :param ocsp_status: the OCSP status to return + """ + root_url = self.directory_url.replace('/dir', '') + + issuer_key_path = os.path.join(self.workspace, 'ocsp_key.pem') + issuer_cert_path = os.path.join(self.workspace, 'ocsp_cert.pem') + with open(issuer_key_path, 'w') as file_h: + misc.ignore_https_warnings() + file_h.write(requests.get(root_url + '/intermediate-key', verify=False).content) + with open(issuer_cert_path, 'w') as file_h: + misc.ignore_https_warnings() + file_h.write(requests.get(root_url + '/intermediate', verify=False).content) + + process = subprocess.Popen([sys.executable, ocsp_server.__file__, cert_path, + issuer_cert_path, issuer_key_path, ocsp_status]) + self._subprocesses.append(process) diff --git a/certbot-ci/certbot_integration_tests/certbot_tests/test_main.py b/certbot-ci/certbot_integration_tests/certbot_tests/test_main.py index ef08e6f51..be871b42f 100644 --- a/certbot-ci/certbot_integration_tests/certbot_tests/test_main.py +++ b/certbot-ci/certbot_integration_tests/certbot_tests/test_main.py @@ -3,7 +3,6 @@ from __future__ import print_function import os import re -import requests import shutil import subprocess import time @@ -556,35 +555,26 @@ def test_ocsp_status_stale(context): def test_ocsp_status_live(context): """Test retrieval of OCSP statuses for live config""" - mock_ocsp_server = None - if context.acme_server == 'pebble': - mock_ocsp_server = misc.mock_ocsp_server(context.directory_url, context.workspace) - cert = context.get_domain('ocsp-check') cert_path = join(context.config_dir, 'live', cert, 'cert.pem') - try: - # OSCP 1: Check live certificate OCSP status (VALID) - context.certbot(['--domains', cert]) - if mock_ocsp_server: - requests.put(mock_ocsp_server[1], json={'cert_path': cert_path, 'ocsp_status': 'GOOD'}) - output = context.certbot(['certificates']) + # OSCP 1: Check live certificate OCSP status (VALID) + context.certbot(['--domains', cert]) + if context.acme_server == 'pebble': + context.mock_ocsp_server(cert_path, 'GOOD') + output = context.certbot(['certificates']) - assert output.count('VALID') == 1, 'Expected {0} to be VALID'.format(cert) - assert output.count('EXPIRED') == 0, 'Did not expect {0} to be EXPIRED'.format(cert) + assert output.count('VALID') == 1, 'Expected {0} to be VALID'.format(cert) + assert output.count('EXPIRED') == 0, 'Did not expect {0} to be EXPIRED'.format(cert) - # OSCP 2: Check live certificate OCSP status (REVOKED) - context.certbot(['revoke', '--cert-name', cert, '--no-delete-after-revoke']) - if mock_ocsp_server: - requests.put(mock_ocsp_server[1], json={'cert_path': cert_path, 'ocsp_status': 'REVOKED'}) - # Sometimes in oldest tests (using openssl binary and not cryptography), the OCSP status is - # not seen immediately by Certbot as invalid. Waiting few seconds solves this transient issue. - time.sleep(5) - output = context.certbot(['certificates']) + # OSCP 2: Check live certificate OCSP status (REVOKED) + context.certbot(['revoke', '--cert-name', cert, '--no-delete-after-revoke']) + if context.acme_server == 'pebble': + context.mock_ocsp_server(cert_path, 'REVOKED') + # Sometimes in oldest tests (using openssl binary and not cryptography), the OCSP status is + # not seen immediately by Certbot as invalid. Waiting few seconds solves this transient issue. + time.sleep(5) + output = context.certbot(['certificates']) - assert output.count('INVALID') == 1, 'Expected {0} to be INVALID'.format(cert) - assert output.count('REVOKED') == 1, 'Expected {0} to be REVOKED'.format(cert) - finally: - if mock_ocsp_server: - mock_ocsp_server[0].terminate() - mock_ocsp_server[0].wait() + assert output.count('INVALID') == 1, 'Expected {0} to be INVALID'.format(cert) + assert output.count('REVOKED') == 1, 'Expected {0} to be REVOKED'.format(cert) diff --git a/certbot-ci/certbot_integration_tests/utils/misc.py b/certbot-ci/certbot_integration_tests/utils/misc.py index ddb04f7ec..0aad90790 100644 --- a/certbot-ci/certbot_integration_tests/utils/misc.py +++ b/certbot-ci/certbot_integration_tests/utils/misc.py @@ -23,10 +23,6 @@ from cryptography.hazmat.primitives.asymmetric import ec from cryptography.hazmat.primitives.serialization import Encoding, PrivateFormat, NoEncryption from six.moves import socketserver, SimpleHTTPServer - -from certbot_integration_tests.utils import ocsp_server -from certbot_integration_tests.utils.constants import MOCK_OCSP_SERVER_PORT - RSA_KEY_TYPE = 'rsa' ECDSA_KEY_TYPE = 'ecdsa' @@ -41,7 +37,7 @@ def check_until_timeout(url): for _ in range(0, 150): time.sleep(1) try: - _ignore_https_warnings() + ignore_https_warnings() if requests.get(url, verify=False).status_code == 200: return except requests.exceptions.ConnectionError: @@ -286,28 +282,8 @@ def load_sample_data_path(workspace): return copied -def mock_ocsp_server(directory_url, workspace): - root_url = directory_url.replace('/dir', '') - - key_path = os.path.join(workspace, 'ocsp_key.pem') - cert_path = os.path.join(workspace, 'ocsp_cert.pem') - with open(key_path, 'w') as file_h: - _ignore_https_warnings() - file_h.write(requests.get(root_url + '/intermediate-key', verify=False).content) - with open(cert_path, 'w') as file_h: - _ignore_https_warnings() - file_h.write(requests.get(root_url + '/intermediate', verify=False).content) - - environ = os.environ.copy() - environ['ISSUER_KEY_PATH'] = key_path - environ['ISSUER_CERT_PATH'] = cert_path - - process = subprocess.Popen([sys.executable, ocsp_server.__file__], env=environ) - - return process, 'http://127.0.0.1:{0}'.format(MOCK_OCSP_SERVER_PORT) - - -def _ignore_https_warnings(): +def ignore_https_warnings(): + """Invoke logic to ignore HTTPS warning while invoking request""" try: import urllib3 urllib3.disable_warnings(urllib3.exceptions.InsecureRequestWarning) diff --git a/certbot-ci/certbot_integration_tests/utils/ocsp_server.py b/certbot-ci/certbot_integration_tests/utils/ocsp_server.py index 598709d1c..e9aa1f1eb 100644 --- a/certbot-ci/certbot_integration_tests/utils/ocsp_server.py +++ b/certbot-ci/certbot_integration_tests/utils/ocsp_server.py @@ -1,79 +1,53 @@ import datetime -import json -import os +import sys from cryptography.hazmat.backends import default_backend from cryptography.hazmat.primitives import serialization, hashes from cryptography import x509 from cryptography.x509 import ocsp -from flask import Flask, request +from six.moves import BaseHTTPServer +from certbot_integration_tests.utils.misc import GracefulTCPServer from certbot_integration_tests.utils.constants import MOCK_OCSP_SERVER_PORT -app = Flask(__name__) -app.debug = False -certificates_map = {} +def _create_ocsp_handler(cert_path, issuer_cert_path, issuer_key_path, ocsp_status_text): + class ProxyHandler(BaseHTTPServer.BaseHTTPRequestHandler): + def do_POST(self): + with open(issuer_cert_path, 'rb') as file_h1: + issuer_cert = x509.load_pem_x509_certificate(file_h1.read(), default_backend()) + with open(issuer_key_path, 'rb') as file_h2: + issuer_key = serialization.load_pem_private_key(file_h2.read(), None, default_backend()) + with open(cert_path, 'rb') as file_h3: + cert = x509.load_pem_x509_certificate(file_h3.read(), default_backend()) + ocsp_status = getattr(ocsp.OCSPCertStatus, ocsp_status_text) -@app.route('/', methods=['GET']) -def heartbeat(): - return 'Done', 200 + now = datetime.datetime.utcnow() + revocation_time = now if ocsp_status == ocsp.OCSPCertStatus.REVOKED else None + revocation_reason = x509.ReasonFlags.unspecified if ocsp_status == ocsp.OCSPCertStatus.REVOKED else None + builder = ocsp.OCSPResponseBuilder() + builder = builder.add_response( + cert=cert, issuer=issuer_cert, algorithm=hashes.SHA1(), + cert_status=ocsp_status, + this_update=now, + next_update=now + datetime.timedelta(hours=1), + revocation_time=revocation_time, revocation_reason=revocation_reason + ).responder_id(ocsp.OCSPResponderEncoding.NAME, issuer_cert) -@app.route('/', methods=['PUT']) -def register_certificate(): - config = json.loads(request.get_data()) - cert_path = config['cert_path'] - ocsp_status = config['ocsp_status'] + response = builder.sign(issuer_key, hashes.SHA256()) - with open(cert_path, 'rb') as file_h3: - cert = x509.load_pem_x509_certificate(file_h3.read(), default_backend()) - serial_number = cert.serial_number - certificates_map[serial_number] = (cert_path, ocsp_status) + self.send_response(200) + self.end_headers() + self.wfile.write(response.public_bytes(serialization.Encoding.DER)) - return 'Done', 200 - - -@app.route('/', methods=['POST']) -def status_certificate(): - raw = request.get_data() - ocsp_request = ocsp.load_der_ocsp_request(raw) - - serial_number = ocsp_request.serial_number - if serial_number not in certificates_map: - response = ocsp.OCSPResponseBuilder.build_unsuccessful( - ocsp.OCSPResponseStatus.UNAUTHORIZED) - else: - config = certificates_map[serial_number] - cert_path = config[0] - ocsp_status = getattr(ocsp.OCSPCertStatus, config[1]) - - with open(os.environ['ISSUER_CERT_PATH'], 'rb') as file_h1: - issuer_cert = x509.load_pem_x509_certificate(file_h1.read(), default_backend()) - with open(os.environ['ISSUER_KEY_PATH'], 'rb') as file_h2: - issuer_key = serialization.load_pem_private_key(file_h2.read(), None, default_backend()) - with open(cert_path, 'rb') as file_h3: - cert = x509.load_pem_x509_certificate(file_h3.read(), default_backend()) - - now = datetime.datetime.utcnow() - - revocation_time = now if ocsp_status == ocsp.OCSPCertStatus.REVOKED else None - revocation_reason = x509.ReasonFlags.unspecified if ocsp_status == ocsp.OCSPCertStatus.REVOKED else None - - builder = ocsp.OCSPResponseBuilder() - builder = builder.add_response( - cert=cert, issuer=issuer_cert, algorithm=hashes.SHA1(), - cert_status=ocsp_status, - this_update=now, - next_update=now + datetime.timedelta(hours=1), - revocation_time=revocation_time, revocation_reason=revocation_reason - ).responder_id(ocsp.OCSPResponderEncoding.NAME, issuer_cert) - - response = builder.sign(issuer_key, hashes.SHA256()) - - return response.public_bytes(serialization.Encoding.DER), 200 + return ProxyHandler if __name__ == '__main__': - app.run(host='0.0.0.0', port=int(os.environ.get('OCSP_PORT', MOCK_OCSP_SERVER_PORT))) + httpd = GracefulTCPServer(('', MOCK_OCSP_SERVER_PORT), _create_ocsp_handler(*sys.argv[1:5])) + try: + httpd.handle_request() + except KeyboardInterrupt: + pass