Move full logic of mock at the acme server config

certbot-ci/certbot_integration_tests/certbot_tests/context.py

@@ -1,14 +1,10 @@
 """Module to handle the context of integration tests."""
 import os
 import shutil
-import subprocess
 import sys
 import tempfile
 
-import requests
-
-from certbot_integration_tests.utils import certbot_call, misc, ocsp_server
-from certbot_integration_tests.utils.constants import MOCK_OCSP_SERVER_PORT
+from certbot_integration_tests.utils import certbot_call
 
 
 class IntegrationTestsContext(object):
@@ -85,24 +81,3 @@ 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

@@ -556,12 +556,9 @@ def test_ocsp_status_stale(context):
 def test_ocsp_status_live(context):
     """Test retrieval of OCSP statuses for live config"""
     cert = context.get_domain('ocsp-check')
-    cert_path = join(context.config_dir, 'live', cert, 'cert.pem')
 
     # 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)
@@ -569,8 +566,6 @@ def test_ocsp_status_live(context):
 
     # 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)

certbot-ci/certbot_integration_tests/utils/acme_server.py

@@ -13,7 +13,7 @@ from os.path import join
 
 import requests
 
-from certbot_integration_tests.utils import misc, proxy, pebble_artifacts
+from certbot_integration_tests.utils import misc, proxy, pebble_artifacts, ocsp_server
 from certbot_integration_tests.utils.constants import *
 
 
@@ -134,6 +134,8 @@ class ACMEServer(object):
             [challtestsrv_path, '-management', ':{0}'.format(CHALLTESTSRV_PORT), '-defaultIPv6', '""',
              '-defaultIPv4', '127.0.0.1', '-http01', '""', '-tlsalpn01', '""', '-https01', '""'])
 
+        self._launch_process([sys.executable, ocsp_server.__file__])
+
         # Wait for the ACME CA server to be up.
         print('=> Waiting for pebble instance to respond...')
         misc.check_until_timeout(self.acme_xdist['directory_url'])

certbot-ci/certbot_integration_tests/utils/constants.py

@@ -5,4 +5,5 @@ CHALLTESTSRV_PORT = 8055
 BOULDER_V1_DIRECTORY_URL = 'http://localhost:4000/directory'
 BOULDER_V2_DIRECTORY_URL = 'http://localhost:4001/directory'
 PEBBLE_DIRECTORY_URL = 'https://localhost:14000/dir'
+PEBBLE_MANAGEMENT_URL = 'https://localhost:15000'
 MOCK_OCSP_SERVER_PORT = 4002

certbot-ci/certbot_integration_tests/utils/ocsp_server.py

@@ -1,5 +1,8 @@
 import datetime
 import sys
+from dateutil import parser
+
+import requests
 
 from cryptography.hazmat.backends import default_backend
 from cryptography.hazmat.primitives import serialization, hashes
@@ -8,23 +11,32 @@ from cryptography.x509 import ocsp
 from six.moves import BaseHTTPServer
 
 from certbot_integration_tests.utils.misc import GracefulTCPServer
-from certbot_integration_tests.utils.constants import MOCK_OCSP_SERVER_PORT
+from certbot_integration_tests.utils.constants import MOCK_OCSP_SERVER_PORT, PEBBLE_MANAGEMENT_URL
 
 
-def _create_ocsp_handler(cert_path, issuer_cert_path, issuer_key_path, ocsp_status_text):
+def _create_ocsp_handler():
     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())
+            request = requests.get(PEBBLE_MANAGEMENT_URL + '/intermediate-key', verify=False)
+            issuer_key = serialization.load_pem_private_key(request.content, None, default_backend())
 
-            ocsp_status = getattr(ocsp.OCSPCertStatus, ocsp_status_text)
+            request = requests.get(PEBBLE_MANAGEMENT_URL + '/intermediate', verify=False)
+            issuer_cert = x509.load_pem_x509_certificate(request.content, default_backend())
+
+            try:
+                content_len = int(self.headers.getheader('content-length', 0))
+            except AttributeError:
+                content_len = int(self.headers.get('Content-Length'))
+
+            cert = x509.load_pem_x509_certificate(self.rfile.read(content_len), default_backend())
+            request = requests.get('{0}/cert-status-by-serial/{1}'.format(
+                PEBBLE_MANAGEMENT_URL, str(hex(cert.serial_number)).replace('0x', '')), verify=False)
+            data = request.json()
+
+            ocsp_status = getattr(ocsp.OCSPCertStatus, data['Status'].upper())
 
             now = datetime.datetime.utcnow()
-            revocation_time = now if ocsp_status == ocsp.OCSPCertStatus.REVOKED else None
+            revocation_time = parser.parse(data['RevokedAt']) if ocsp_status == ocsp.OCSPCertStatus.REVOKED else None
             revocation_reason = x509.ReasonFlags.unspecified if ocsp_status == ocsp.OCSPCertStatus.REVOKED else None
 
             builder = ocsp.OCSPResponseBuilder()
@@ -46,8 +58,8 @@ def _create_ocsp_handler(cert_path, issuer_cert_path, issuer_key_path, ocsp_stat
 
 
 if __name__ == '__main__':
-    httpd = GracefulTCPServer(('', MOCK_OCSP_SERVER_PORT), _create_ocsp_handler(*sys.argv[1:5]))
+    httpd = GracefulTCPServer(('', MOCK_OCSP_SERVER_PORT), _create_ocsp_handler())
     try:
-        httpd.handle_request()
+        httpd.serve_forever()
     except KeyboardInterrupt:
         pass