Revize 79a61171

Re #8577 - Added OCSP integration tests.

tests/integration_tests/rest_api/crl_ocsp_test.py

@@ +1 @@
+from src.services.cryptography import CryptographyService as cs
+from src.utils.temporary_file import TemporaryFile
+run = cs._CryptographyService__run_for_output
+import re
+import base64
+
 
 def make_root_ca(server, title="Root CA s.r.o."):
     return server.post("/api/certificates", content_type="application/json", json={
@@ -106 +112 @@
     assert ret.status_code == 404
 
 
+def test_ocsp_valid(server):
+    ret = make_root_ca(server)
+    root_idx = ret.json["data"]
+    end_1 = make_end_cert(server, root_idx, "end1")
+    end_2 = make_end_cert(server, root_idx, "end2")
+
+    root_contents = server.get(f"/api/certificates/{root_idx}")      .json["data"]
+    end_1_index = end_1.json['data']
+    end_1_contents = server.get(f"/api/certificates/{end_1_index}")  .json["data"]
+    end_2_index = end_2.json['data']
+    end_2_contents = server.get(f"/api/certificates/{end_2_index}")  .json["data"]
+
+    def call_ocsp_service_post(ocsp_req):
+        retval = server.post(f"/api/ocsp/{root_idx}", data=ocsp_req, content_type="application/ocsp-request")
+        assert retval.status_code == 200
+        ocsp_response_decoded = str(
+            run(["ocsp", "-respin", "-", "-text", "-CAfile", root_cert], proc_input=retval.data),
+            encoding='utf-8')
+        cert_status = re.findall("Cert Status: ([a-z]*)", ocsp_response_decoded)[0]
+        return cert_status
+
+    def call_ocsp_service_get(ocsp_req):
+        retval = server.get(f"/api/ocsp/{root_idx}/{str(base64.b64encode(ocsp_req), encoding='utf-8')}", content_type="application/ocsp-request")
+        assert retval.status_code == 200
+        ocsp_response_decoded = str(
+            run(["ocsp", "-respin", "-", "-text", "-CAfile", root_cert], proc_input=retval.data),
+            encoding='utf-8')
+        cert_status = re.findall("Cert Status: ([a-z]*)", ocsp_response_decoded)[0]
+        return cert_status
+
+    # check both OCSP methods and assert that the certificate is valid
+    def assert_good(ocsp_req):
+        cert_status_1 = call_ocsp_service_post(ocsp_req)
+        cert_status_2 = call_ocsp_service_get(ocsp_req)
+        assert cert_status_1 == cert_status_2 == "good"
+
+    # check both OCSP methods and assert that the certificate is revoked
+    def assert_revoked(ocsp_req):
+        cert_status_1 = call_ocsp_service_post(ocsp_req)
+        cert_status_2 = call_ocsp_service_post(ocsp_req)
+        assert cert_status_1 == cert_status_2 == "revoked"
+
+    with TemporaryFile("end_1_cert.pem", end_1_contents) as end_1_cert_pem, \
+         TemporaryFile("end_2_cert.pem", end_2_contents) as end_2_cert_pem, \
+         TemporaryFile("issuer.pem", root_contents) as root_cert:
+
+        end_cert_1_contents = str(run(["x509", "-in", end_1_cert_pem, "-text", "-noout"]), encoding='utf-8')
+        serial_1 = re.findall("Serial Number: ([0-9]+) ", end_cert_1_contents)[0]
+
+        end_cert_2_contents = str(run(["x509", "-in", end_2_cert_pem, "-text", "-noout"]), encoding='utf-8')
+        serial_2 = re.findall("Serial Number: ([0-9]+) ", end_cert_2_contents)[0]
+
+        # test serial (first)
+        ocsp_request = run(["ocsp", "-issuer", root_cert, "-serial", serial_1, "-reqout", "-"])
+        assert_good(ocsp_request)
+
+        # test contents directly (second)
+        ocsp_request = run(["ocsp", "-issuer", root_cert, "-cert", end_2_cert_pem, "-reqout", "-"])
+        assert_good(ocsp_request)
+
+        # revoke first
+        ret = server.patch(f"/api/certificates/{end_1_index}", json={"status": "revoked"})
+        assert ret.status_code == 200
+
+        # test serial (first)
+        ocsp_request = run(["ocsp", "-issuer", root_cert, "-serial", serial_1, "-reqout", "-"])
+        assert_revoked(ocsp_request)
+
+        # test contents directly (first)
+        ocsp_request = run(["ocsp", "-issuer", root_cert, "-cert", end_1_cert_pem, "-reqout", "-"])
+        assert_revoked(ocsp_request)
+
+        # test serial (second)
+        ocsp_request = run(["ocsp", "-issuer", root_cert, "-serial", serial_2, "-reqout", "-"])
+        assert_good(ocsp_request)
+
+        # test contents directly (second)
+        ocsp_request = run(["ocsp", "-issuer", root_cert, "-cert", end_2_cert_pem, "-reqout", "-"])
+        assert_good(ocsp_request)
+
+
+def test_ocsp_invalid_1(server):
+    assert server.post(f"/api/ocsp/8008135").status_code == 400
+
+
+def test_ocsp_invalid_2(server):
+    assert server.get(f"/api/ocsp/8008135/look_at_me_im_binary").status_code == 404
+
+
+def test_ocsp_invalid_3(server):
+    assert server.get(f"/api/ocsp/8008135").status_code == 405
+