Revize 4c19a9b1

Re #8472 - Implemented create_end_cert method and added an integration test validating it

src/services/certificate_service.py

@@ -1 @@
-from src.constants import ROOT_CA_ID, INTERMEDIATE_CA_ID, CA_ID
+from src.constants import ROOT_CA_ID, INTERMEDIATE_CA_ID, CA_ID, CERTIFICATE_ID
 from src.dao.certificate_repository import CertificateRepository
 from src.model.certificate import Certificate
 from src.model.private_key import PrivateKey
@@ -26 +26 @@
         # 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)
-
-        # parse the generated pem for subject and notBefore/notAfter fields
-        subj, not_before, not_after = self.cryptography_service.parse_cert_pem(cert_pem)
-
-        # format the parsed date
-        not_before_formatted = time.strftime(DATE_FORMAT, not_before)
-        not_after_formatted = time.strftime(DATE_FORMAT, not_after)
-
         # specify CA usage
         usages[CA_ID] = True
 
-        # create a certificate wrapper
-        certificate = Certificate(-1, subject.common_name, not_before_formatted, not_after_formatted, cert_pem,
-                                  key.private_key_id, ROOT_CA_ID, 0, usages)
+        # wrap into Certificate class
+        certificate = self.__create_wrapper(cert_pem, key.private_key_id, subject.common_name, usages, 0,
+                                            ROOT_CA_ID)
 
         # store the wrapper into the repository
         created_id = self.certificate_repository.create(certificate)
@@ -49 +41 @@
 
         return certificate
 
+    def __create_wrapper(self, cert_pem, private_key_id, common_name, usages, parent_id, cert_type):
+        # parse the generated pem for subject and notBefore/notAfter fields
+        subj, not_before, not_after = self.cryptography_service.parse_cert_pem(cert_pem)
+        # format the parsed date
+        not_before_formatted = time.strftime(DATE_FORMAT, not_before)
+        not_after_formatted = time.strftime(DATE_FORMAT, not_after)
+
+        # create a certificate wrapper
+        certificate = Certificate(-1, common_name, not_before_formatted, not_after_formatted, cert_pem,
+                                  private_key_id, cert_type, parent_id, usages)
+
+        print(f"CW: {parent_id} {certificate.parent_id}")
+        return certificate
+
     # TODO config parameter present in class diagram but not here (unused)
     def create_ca(self, subject_key: PrivateKey, subject: Subject, issuer_cert: Certificate, issuer_key: PrivateKey,
                   extensions: str = "", days: int = 30, usages=None):
@@ -63 +69 @@
                                                         issuer_key_pass=issuer_key.password, extensions=extensions,
                                                         days=days)
 
+        # specify CA usage
+        usages[CA_ID] = True
+
+        # wrap into Certificate class
+        self.__create_wrapper(cert_pem, subject_key.private_key_id, subject.common_name, usages,
+                              issuer_cert.certificate_id, INTERMEDIATE_CA_ID)
+
         # parse the generated pem for subject and notBefore/notAfter fields
         subj, not_before, not_after = self.cryptography_service.parse_cert_pem(cert_pem)
 
@@ -85 +98 @@
 
         return certificate
 
+    def create_end_cert(self, subject_key: PrivateKey, subject: Subject, issuer_cert: Certificate,
+                        issuer_key: PrivateKey,
+                        extensions: str = "", days: int = 30, usages=None):
+        if usages is None:
+            usages = {}
+
+        # generate a new certificate
+        cert_pem = self.cryptography_service.create_crt(subject, subject_key.private_key, issuer_cert.pem_data,
+                                                        issuer_key.private_key,
+                                                        subject_key_pass=subject_key.password,
+                                                        issuer_key_pass=issuer_key.password, extensions=extensions,
+                                                        days=days)
+
+        print(f"test cert {issuer_cert.certificate_id}")
+        # wrap the generated certificate using Certificate class
+        certificate = self.__create_wrapper(cert_pem, subject_key.private_key_id, subject.common_name, usages,
+                                            issuer_cert.certificate_id, CERTIFICATE_ID)
+
+        print(f"parent cert {certificate.parent_id}")
+
+        created_id = self.certificate_repository.create(certificate)
+
+        certificate.certificate_id = created_id
+
+        print(f"parent cert {certificate.parent_id}")
+
+        return certificate
+
     def get_certificate(self, unique_id: int) -> Certificate:
         return self.certificate_repository.read(unique_id)

tests/integration_tests/services/certificate_key_service_test.py

@@ -1 +1 @@
 import subprocess
 
-from src.constants import SSL_ID, CA_ID
+from src.constants import SSL_ID, CA_ID, AUTHENTICATION_ID, INTERMEDIATE_CA_ID, ROOT_CA_ID, CERTIFICATE_ID, SIGNATURE_ID
 from src.model.subject import Subject
 
 
@@ -25 +25 @@
                                               Subject(common_name="FooName", organization_unit="Department of Foo"),
                                               usages={SSL_ID: True})
 
-    assert CA_ID in cert.usages
-    assert SSL_ID in cert.usages
+    assert ROOT_CA_ID == cert.type_id
+
+    assert cert.usages[CA_ID]
+    assert cert.usages[SSL_ID]
+    assert cert.usages[AUTHENTICATION_ID] is False
+    assert cert.usages[SIGNATURE_ID] is False
 
     cert_loaded = certificate_service.get_certificate(cert.certificate_id)
 
@@ -56 +60 @@
     inter_cert = certificate_service.create_ca(inter_ca_private_key, Subject(common_name="Intermediate CA"), root_ca,
                                                root_ca_private_key, usages={SSL_ID: True})
 
-    assert CA_ID in inter_cert.usages
-    assert SSL_ID in inter_cert.usages
+    assert INTERMEDIATE_CA_ID == inter_cert.type_id
+
+    assert inter_cert.usages[CA_ID]
+    assert inter_cert.usages[SSL_ID]
+    assert inter_cert.usages[AUTHENTICATION_ID] is False
+    assert inter_cert.usages[SIGNATURE_ID] is False
 
     inter_cert_loaded = certificate_service.get_certificate(inter_cert.certificate_id)
 
@@ -66 +74 @@
     assert """            X509v3 Basic Constraints: critical
                 CA:TRUE""" in cert_loaded_printed
 
-    assert inter_cert_loaded.certificate_id == inter_cert_loaded.certificate_id
+    assert inter_cert.certificate_id == inter_cert_loaded.certificate_id
     assert inter_cert.common_name == inter_cert_loaded.common_name
     assert inter_cert.valid_from == inter_cert_loaded.valid_from
     assert inter_cert.valid_to == inter_cert_loaded.valid_to
@@ -74 +82 @@
     assert inter_cert.private_key_id == inter_cert_loaded.private_key_id
     assert inter_cert.type_id == inter_cert_loaded.type_id
     assert inter_cert.parent_id == root_ca.certificate_id
+    assert inter_cert_loaded.parent_id == root_ca.certificate_id
     assert inter_cert.usages == inter_cert_loaded.usages
+
+
+def test_create_and_get_cert(private_key_service, certificate_service):
+    root_ca_private_key = private_key_service.create_new_key(passphrase="foobar")
+    inter_ca_private_key = private_key_service.create_new_key(passphrase="barfoo")
+    end_cert_private_key = private_key_service.create_new_key(passphrase="foofoo")
+
+    root_ca_cert = certificate_service.create_root_ca(root_ca_private_key,
+                                                      Subject(common_name="RootFoo",
+                                                              organization_unit="Department of Foo"))
+
+    inter_ca_cert = certificate_service.create_ca(inter_ca_private_key, Subject(common_name="Intermediate CA"),
+                                                  root_ca_cert,
+                                                  root_ca_private_key, usages={SSL_ID: True})
+
+    cert = certificate_service.create_end_cert(end_cert_private_key,
+                                               Subject("Foo Child", email_address="[e-mail]"), inter_ca_cert,
+                                               inter_ca_private_key, usages={AUTHENTICATION_ID: True})
+    assert CERTIFICATE_ID == cert.type_id
+
+    assert cert.usages[AUTHENTICATION_ID]
+    assert cert.usages[SSL_ID] is False
+    assert cert.usages[SIGNATURE_ID] is False
+    assert cert.usages[CA_ID] is False
+
+    cert_loaded = certificate_service.get_certificate(cert.certificate_id)
+
+    assert cert.certificate_id == cert_loaded.certificate_id
+    assert cert.common_name == cert_loaded.common_name
+    assert cert.valid_from == cert_loaded.valid_from
+    assert cert.valid_to == cert_loaded.valid_to
+    assert cert.pem_data == cert_loaded.pem_data
+    assert cert.private_key_id == cert_loaded.private_key_id
+    assert cert.type_id == cert_loaded.type_id
+    assert cert.parent_id == inter_ca_cert.certificate_id
+    assert cert_loaded.parent_id == inter_ca_cert.certificate_id
+    assert cert.usages == cert_loaded.usages