Simplify logic

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)

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)

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)

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