Revize 20b33bd4

Re #8571 - certificate_service.py revocation support end tests

src/services/certificate_service.py

@@ -2 +2 @@
 
 from injector import inject
 
-from src.constants import ROOT_CA_ID, INTERMEDIATE_CA_ID, CA_ID, CERTIFICATE_ID
+from src.config.configuration import Configuration
+from src.constants import ROOT_CA_ID, INTERMEDIATE_CA_ID, CA_ID, CERTIFICATE_ID, CERTIFICATE_STATES, \
+    CERTIFICATE_REVOCATION_REASONS
 from src.dao.certificate_repository import CertificateRepository
 from src.model.certificate import Certificate
 from src.model.private_key import PrivateKey
@@ -13 +15 @@
 
 DATE_FORMAT = "%d.%m.%Y %H:%M:%S"
 CA_EXTENSIONS = "basicConstraints=critical,CA:TRUE"
+CRL_EXTENSION = "crlDistributionPoints=URI:"
+OCSP_EXTENSION = "authorityInfoAccess=OCSP;URI:"
+STATUS_REVOKED = "revoked"
+STATUS_VALID = "valid"
 
 
 class CertificateService:
 
     @inject
-    def __init__(self, cryptography_service: CryptographyService, certificate_repository: CertificateRepository):
+    def __init__(self, cryptography_service: CryptographyService,
+                 certificate_repository: CertificateRepository,
+                 configuration: Configuration):
         self.cryptography_service = cryptography_service
         self.certificate_repository = certificate_repository
+        self.configuration = configuration
 
     # TODO usages present in method parameters but not in class diagram
     def create_root_ca(self, key: PrivateKey, subject: Subject, extensions: str = "", config: str = "",
@@ -38 +47 @@
         if usages is None:
             usages = {}
 
+        cert_id = self.certificate_repository.get_next_id()
+        extensions = extensions + "\n" + CRL_EXTENSION + " " + self.__get_crl_endpoint(cert_id)
+        extensions = extensions + "\n" + OCSP_EXTENSION + " " + self.__get_ocsp_endpoint(cert_id)
+
         # create a new self signed  certificate
         cert_pem = self.cryptography_service.create_sscrt(subject, key.private_key, key_pass=key.password,
                                                           extensions=extensions, config=config, days=days)
@@ -98 +111 @@
             usages = {}
 
         extensions = extensions + "\n" + CA_EXTENSIONS
+        # Add CRL and OCSP distribution point to certificate extensions
+        cert_id = self.certificate_repository.get_next_id()
+        extensions = extensions + "\n" + CRL_EXTENSION + " " + self.__get_crl_endpoint(cert_id)
+        extensions = extensions + "\n" + OCSP_EXTENSION + " " + self.__get_ocsp_endpoint(cert_id)
+
         # TODO implement AIA URI via extensions
         cert_pem = self.cryptography_service.create_crt(subject, subject_key.private_key, issuer_cert.pem_data,
                                                         issuer_key.private_key,
@@ -248 +266 @@
         # TODO delete children?
         return self.certificate_repository.delete(unique_id)
 
+    def set_certificate_revocation_status(self, id, status, reason="unspecified"):
+        """
+        Set certificate status to 'valid' or 'revoked'.
+        If the new status is revoked a reason can be provided -> default is unspecified
+        :param reason: reason for revocation
+        :param id: identifier of the certificate whose status is to be changed
+        :param status: new status of the certificate
+        """
+        if status not in CERTIFICATE_STATES:
+            raise CertificateStatusInvalidException(status)
+        if reason not in CERTIFICATE_REVOCATION_REASONS:
+            raise RevocationReasonInvalidException(reason)
+
+        if status == STATUS_VALID:
+            self.certificate_repository.clear_certificate_revocation(id)
+        elif status == STATUS_REVOKED:
+            revocation_timestamp = int(time.time())
+            self.certificate_repository.set_certificate_revoked(id, str(revocation_timestamp), reason)
+
     def get_subject_from_certificate(self, certificate: Certificate) -> Subject:
         """
         Get Subject distinguished name from a Certificate
@@ -265 +302 @@
         :return: a string containing the extracted public key in PEM format
         """
         return self.cryptography_service.extract_public_key_from_certificate(certificate.pem_data)
+
+    def __get_crl_endpoint(self, ca_identifier: int) -> str:
+        """
+        Get URL address of CRL distribution endpoint based on
+        issuer's ID
+
+        :param ca_identifier: ID of issuing authority
+        :return: CRL endpoint for the given CA
+        """
+        return self.configuration.base_server_url + "/api/crl/" + str(ca_identifier)
+
+    def __get_ocsp_endpoint(self, ca_identifier: int) -> str:
+        """
+        Get URL address of OCSP distribution endpoint based on
+        issuer's ID
+
+        :param ca_identifier: ID of issuing authority
+        :return: OCSP endpoint for the given CA
+        """
+        return self.configuration.base_server_url + "/api/ocsp/" + str(ca_identifier)
+
+
+class RevocationReasonInvalidException(Exception):
+    """
+    Exception that denotes that the caller was trying to revoke
+    a certificate using an invalid revocation reason
+    """
+
+    def __init__(self, reason):
+        self.reason = reason
+
+    def __str__(self):
+        return f"Revocation reason '{self.reason}' is not valid."
+
+
+class CertificateStatusInvalidException(Exception):
+    """
+    Exception that denotes that the caller was trying to set
+    a certificate to an invalid state
+    """
+
+    def __init__(self, status):
+        self.status = status
+
+    def __str__(self):
+        return f"Certificate status '{self.status}' is not valid."

tests/integration_tests/services/certificate_service_test.py

@@ -27 +27 @@
 
     # verify that the loaded certificate is a CA
     cert_loaded_printed = export_crt(cert_loaded.pem_data)
-    assert """            X509v3 Basic Constraints: critical
-                CA:TRUE""" in cert_loaded_printed
+    expected = """            X509v3 Basic Constraints: critical
+                CA:TRUE""".replace("\r", "").replace("\n", "")
+    assert expected in cert_loaded_printed.replace("\n", "").replace("\r", "")
 
     assert cert.certificate_id == cert_loaded.certificate_id
     assert cert.common_name == cert_loaded.common_name
@@ -62 +63 @@
 
     # verify that the loaded certificate is a CA
     cert_loaded_printed = export_crt(inter_cert_loaded.pem_data)
-    assert """            X509v3 Basic Constraints: critical
-                CA:TRUE""" in cert_loaded_printed
+    expected = """            X509v3 Basic Constraints: critical
+                CA:TRUE""".replace("\n", "").replace("\r", "")
+    assert expected in cert_loaded_printed.replace("\n", "").replace("\r", "")
 
     assert inter_cert.certificate_id == inter_cert_loaded.certificate_id
     assert inter_cert.common_name == inter_cert_loaded.common_name

tests/integration_tests/services/conftest.py

@@ -4 +4 @@
 
 import pytest
 
-from src.config.configuration import test_configuration
+from src.config.configuration import test_configuration, Configuration
 from src.config.connection_provider import ConnectionProvider
 from src.dao.certificate_repository import CertificateRepository
 from src.dao.private_key_repository import PrivateKeyRepository
@@ -36 +36 @@
     return CryptographyService()
 
 
[e-mail]
+def configuration():
+    return Configuration()
+
+
 @pytest.fixture
 def private_key_service(private_key_repository, cryptography_service):
     return KeyService(cryptography_service, private_key_repository)
 
 
 @pytest.fixture
-def certificate_service(certificate_repository, cryptography_service):
-    return CertificateService(cryptography_service, certificate_repository)
+def certificate_service(certificate_repository, cryptography_service, configuration):
+    return CertificateService(cryptography_service, certificate_repository, configuration)
 
 
 @pytest.fixture
@@ -73 +78 @@
 
 
 @pytest.fixture
-def certificate_service_unique(certificate_repository_unique, cryptography_service_unique):
-    return CertificateService(cryptography_service_unique, certificate_repository_unique)
+def certificate_service_unique(certificate_repository_unique, cryptography_service_unique, configuration):
+    return CertificateService(cryptography_service_unique, certificate_repository_unique, configuration)

tests/unit_tests/services/cryptography/create_crt_test.py

@@ -80 +80 @@
     # create a CA using the root CA
     inter_ca = service.create_crt(Subject(common_name="bar", country="CZ"), inter_key, root_ca, root_key,
                                   subject_key_pass=inter_key_passphrase, issuer_key_pass=root_key_passphrase,
-                                  extensions="authorityInfoAccess = caIssuers;URI:bar.cz/baz/cert\nbasicConstraints=critical,CA:TRUE")
+                                  extensions="authorityInfoAccess=caIssuers;URI:bar.cz/baz/cert\n"
+                                             "basicConstraints=critical,CA:TRUE\n"
+                                             "crlDistributionPoints=URI:http://localhost/api/crl/0\n"
+                                             "authorityInfoAccess=OCSP;URI:http://localhost/api/ocsp/0\n")
 
     inter_ca_printed = export_crt(inter_ca)
 
     # assert fields
     assert "Issuer: CN = foo" in inter_ca_printed
     assert "Subject: CN = bar, C = CZ" in inter_ca_printed
+    assert "X509v3 CRL Distribution Points:" in inter_ca_printed
+    assert "URI:http://localhost/api/crl/0" in inter_ca_printed
+    assert "Authority Information Access:" in inter_ca_printed
+    assert "OCSP - URI:http://localhost/api/ocsp/0" in inter_ca_printed
 
     # assert extensions
     expected_extensions = """        X509v3 extensions:
-            Authority Information Access: 
-                CA Issuers - URI:bar.cz/baz/cert
-
             X509v3 Basic Constraints: critical
-                CA:TRUE"""
+                CA:TRUE
+            X509v3 CRL Distribution Points: 
+
+                Full Name:
+                  URI:http://localhost/api/crl/0
+
+            Authority Information Access: 
+                OCSP - URI:http://localhost/api/ocsp/0"""
+    expected_extensions = expected_extensions.replace("\n", "").replace("\r", "")
+    inter_ca_printed = inter_ca_printed.replace("\n", "").replace("\r", "")
     assert expected_extensions in inter_ca_printed

tests/unit_tests/services/cryptography/self_signed_cert_test.py

@@ -16 +16 @@
 
     [ req ]
     distinguished_name      = ca_dn                 # DN section
+    extensions = root_ca_ext
 
     [ ca_dn ]
 
-    [ root_ca_ext ]
-    keyUsage                = critical,keyCertSign,cRLSign
-    basicConstraints        = critical,CA:true
-    subjectKeyIdentifier    = hash
-    authorityKeyIdentifier  = keyid:always
     """
 
     cert = service.create_sscrt(Subject(common_name="Topnax",
@@ -33 +29 @@
                                         organization="Mysterious Org.",
                                         organization_unit="Department of Mysteries",
                                         email_address="[e-mail]"), private_key, config=config,
-                                extensions="root_ca_ext", key_pass="foobar")
+                                extensions="basicConstraints=critical,CA:TRUE\n"
+                                           "keyUsage=critical,keyCertSign,cRLSign\n"
+                                           "crlDistributionPoints=URI:http://localhost/api/crl/0\n"
+                                           "authorityInfoAccess=OCSP;URI:http://localhost/api/ocsp/0\n",
+                                key_pass="foobar")
 
     cert_printed = subprocess.check_output(["openssl", "x509", "-noout", "-text", "-in", "-"],
                                            input=bytes(cert, encoding="utf-8"), stderr=subprocess.STDOUT).decode()
@@ -41 +41 @@
     assert "Certificate Sign, CRL Sign" in cert_printed
     assert "X509v3 Key Usage: critical" in cert_printed
     assert "CA:TRUE" in cert_printed
+    assert "X509v3 CRL Distribution Points:" in cert_printed
+    assert "URI:http://localhost/api/crl/0" in cert_printed
+    assert "Authority Information Access:" in cert_printed
+    assert "OCSP - URI:http://localhost/api/ocsp/0" in cert_printed
 
     assert "Issuer: CN = Topnax, C = CZ, L = My Locality, ST = My state, O = Mysterious Org., OU = Department of Mysteries, emailAddress = [e-mail]" in cert_printed
     assert "Subject: CN = Topnax, C = CZ, L = My Locality, ST = My state, O = Mysterious Org., OU = Department of Mysteries, emailAddress = [e-mail]" in cert_printed

tests/unit_tests/services/cryptography/sign_csr_test.py

@@ -91 +91 @@
 
             X509v3 Basic Constraints: critical
                 CA:TRUE"""
+    expected_extensions = expected_extensions.replace("\n", "").replace("\r", "")
+    inter_ca_printed = inter_ca_printed.replace("\n", "").replace("\r", "")
     assert expected_extensions in inter_ca_printed