Revize 7313994f

Re #8575 - Implemented CrlService that allows the caller to generate a CA database index of revoked certificates

Added an integration test verifying its validity.
Renamed NOT_BEFORE_AFTER_DATE_FORMAT constant to VALID_FROM_TO_DATE_FORMAT in certificate_service.py.
Removed debug print in cryptography.py

src/services/certificate_service.py

@@ -15 +15 @@
 
 import time
 
-NOT_BEFORE_AFTER_DATE_FORMAT = "%d.%m.%Y %H:%M:%S"
+VALID_FROM_TO_DATE_FORMAT = "%d.%m.%Y %H:%M:%S"
 CA_EXTENSIONS = "basicConstraints=critical,CA:TRUE"
 CRL_EXTENSION = "crlDistributionPoints=URI:"
 OCSP_EXTENSION = "authorityInfoAccess=OCSP;URI:"
@@ -86 +86 @@
         # TODO this could be improved in the future in such way that calling openssl is not required to parse the dates
         subj, not_before, not_after = self.cryptography_service.parse_cert_pem(cert_pem)
         # format the parsed date
-        not_before_formatted = time.strftime(NOT_BEFORE_AFTER_DATE_FORMAT, not_before)
-        not_after_formatted = time.strftime(NOT_BEFORE_AFTER_DATE_FORMAT, not_after)
+        not_before_formatted = time.strftime(VALID_FROM_TO_DATE_FORMAT, not_before)
+        not_after_formatted = time.strftime(VALID_FROM_TO_DATE_FORMAT, not_after)
 
         # create a certificate wrapper
         certificate = Certificate(-1, subj.common_name, not_before_formatted, not_after_formatted, cert_pem,
@@ -136 +136 @@
         subj, not_before, not_after = self.cryptography_service.parse_cert_pem(cert_pem)
 
         # format the parsed date
-        not_before_formatted = time.strftime(NOT_BEFORE_AFTER_DATE_FORMAT, not_before)
-        not_after_formatted = time.strftime(NOT_BEFORE_AFTER_DATE_FORMAT, not_after)
+        not_before_formatted = time.strftime(VALID_FROM_TO_DATE_FORMAT, not_before)
+        not_after_formatted = time.strftime(VALID_FROM_TO_DATE_FORMAT, not_after)
 
         # specify CA usage
         usages[CA_ID] = True

src/services/crl/crl_service.py

@@ +1 @@
+import time
+
+from injector import inject
+
+from src.dao.certificate_repository import CertificateRepository
+from src.services.certificate_service import VALID_FROM_TO_DATE_FORMAT
+from src.services.crl.ca_index_file_line_generator import create_index_file_revoked_line
+from src.services.cryptography import CryptographyService
+
+
+class CrlService:
+    @inject
+    def __init__(self,
+                 certificate_repository: CertificateRepository,
+                 cryptography_service: CryptographyService
+                 ):
+        self.certificate_repository = certificate_repository
+        self.cryptography_service = cryptography_service
+
+    def create_revoked_index(self, ca_id) -> str:
+        """
+        Queries the certificate repository and looks for all certificates revoked by the certificate authority given
+        by the passed ID. Found certificates are then put into a string representing the CA's database index file.
+        
+        :param ca_id: ID of the CA whose revoked certificates should be put into the index file
+        :return: a str representing the content of a CA index file
+        """
+
+        index_lines = []
+        # iterate over revoked certificates of the CA given by an ID
+        for certificate in self.certificate_repository.get_all_revoked_by(ca_id):
+            # extract the complete subject information and not_after date field
+            subject, _, not_after = self.cryptography_service.parse_cert_pem(certificate.pem_data)
+            line = create_index_file_revoked_line(certificate,
+                                                  subject,
+                                                  # parse valid_to date to a date struct
+                                                  time.strptime(certificate.valid_to, VALID_FROM_TO_DATE_FORMAT),
+                                                  not_after)
+
+            # append it to the list of lines
+            index_lines.append(line)
+
+        # join all lines with a new line
+        return "\n".join(index_lines)

src/services/cryptography.py

@@ -155 +155 @@
             # waiting for the passphrase to be typed in
             args.extend(["-passin", f"pass:{key_pass}"])
 
-            print(args)
-
             return self.__run_for_output(args, proc_input=bytes(key, encoding="utf-8")).decode()
 
     def __create_csr(self, subject, key, key_pass=""):

tests/integration_tests/services/conftest.py

@@ -10 +10 @@
 from src.dao.private_key_repository import PrivateKeyRepository
 from src.db.init_queries import SCHEMA_SQL, DEFAULT_VALUES_SQL
 from src.services.certificate_service import CertificateService
+from src.services.crl.crl_service import CrlService
 from src.services.cryptography import CryptographyService
 from src.services.key_service import KeyService
 
@@ -80 +81 @@
 @pytest.fixture
 def certificate_service_unique(certificate_repository_unique, cryptography_service_unique, configuration):
     return CertificateService(cryptography_service_unique, certificate_repository_unique, configuration)
+
+
[e-mail]
+def crl_service_unique(certificate_repository_unique, cryptography_service):
+    return CrlService(certificate_repository_unique, cryptography_service)

tests/integration_tests/services/crl_service_test.py

@@ +1 @@
+import time
+
+from src.model.subject import Subject
+from src.services.certificate_service import VALID_FROM_TO_DATE_FORMAT
+
+CA_INDEX_DATE_FORMAT = "%y%m%d%H%M%SZ"
+
+
+def convert_valid_to_date_to_ca_index_format(date):
+    return convert_date_to_ca_index_format(time.strptime(date, VALID_FROM_TO_DATE_FORMAT))
+
+
+def convert_date_to_ca_index_format(date):
+    return time.strftime(CA_INDEX_DATE_FORMAT, date)
+
+
+def test_set_certificate_revoked(certificate_service_unique, private_key_service_unique, crl_service_unique,
+                                 cryptography_service):
+    root_ca_private_key = private_key_service_unique.create_new_key(passphrase="foobar")
+    inter_ca_private_key = private_key_service_unique.create_new_key(passphrase="barfoo")
+
+    root_ca_cert = certificate_service_unique.create_root_ca(root_ca_private_key,
+                                                             Subject(common_name="RootFoo",
+                                                                     organization_unit="Department of Foo"))
+
+    # create a CA
+    foo_ca = certificate_service_unique.create_ca(inter_ca_private_key, Subject(common_name="Foo CA", locality="Brno"),
+                                                  root_ca_cert,
+                                                  root_ca_private_key)
+
+    # create another CA
+    bar_ca = certificate_service_unique.create_ca(inter_ca_private_key,
+                                                  Subject(common_name="Bar CA", country="CZ", locality="Pilsen"),
+                                                  root_ca_cert,
+                                                  root_ca_private_key)
+
+    # create another CA
+    bar_baz_ca = certificate_service_unique.create_ca(inter_ca_private_key, Subject(common_name="BarBaz CA"),
+                                                      root_ca_cert,
+                                                      root_ca_private_key)
+
+    # create a certificate
+    baz_cert = certificate_service_unique.create_end_cert(inter_ca_private_key,
+                                                          Subject(common_name="Baz CA", state="ST"),
+                                                          root_ca_cert,
+                                                          root_ca_private_key)
+
+    # revoke first created intermediate CA
+    certificate_service_unique.set_certificate_revocation_status(foo_ca.certificate_id, "revoked", "unspecified")
+    # revoke second created intermediate CA
+    certificate_service_unique.set_certificate_revocation_status(bar_ca.certificate_id, "revoked", "keyCompromise")
+    # revoke the created end certificate (non-CA)
+    certificate_service_unique.set_certificate_revocation_status(baz_cert.certificate_id, "revoked",
+                                                                 "privilegeWithdrawn")
+
+    # split lines
+    out = crl_service_unique.create_revoked_index(root_ca_cert.certificate_id)
+
+    # convert revoked date fields
+    revoked_dates = [
+        convert_date_to_ca_index_format(cryptography_service.parse_cert_pem(foo_ca.pem_data)[2]),
+        convert_date_to_ca_index_format(cryptography_service.parse_cert_pem(bar_ca.pem_data)[2]),
+        convert_date_to_ca_index_format(cryptography_service.parse_cert_pem(baz_cert.pem_data)[2]),
+    ]
+
+    # convert valid_to date fields
+    valid_to_dates = [
+        convert_valid_to_date_to_ca_index_format(foo_ca.valid_to),
+        convert_valid_to_date_to_ca_index_format(bar_ca.valid_to),
+        convert_valid_to_date_to_ca_index_format(baz_cert.valid_to),
+    ]
+
+    expected_lines = [
+        f"R	{valid_to_dates[0]}	{revoked_dates[0]},unspecified	02	unknown	/CN=Foo CA/L=Brno",
+        f"R	{valid_to_dates[1]}	{revoked_dates[1]},keyCompromise	03	unknown	/CN=Bar CA/C=CZ/L=Pilsen",
+        f"R	{valid_to_dates[2]}	{revoked_dates[2]},privilegeWithdrawn	05	unknown	/CN=Baz CA/ST=ST"
+    ]
+
+    assert out == "\n".join(expected_lines)