Revize bbcb7c89

Re #8472 - Implemented create_ca method in CertificateService and added test verifying this method's validity

src/services/certificate_service.py

@@ -1 @@
-from src.constants import ROOT_CA_ID
+from src.constants import ROOT_CA_ID, INTERMEDIATE_CA_ID
 from src.dao.certificate_repository import CertificateRepository
 from src.model.certificate import Certificate
 from src.model.private_key import PrivateKey
@@ -8 +8 @@
 import time
 
 DATE_FORMAT = "%d.%m.%Y %H:%M:%S"
+CA_EXTENSIONS = "basicConstraints=critical,CA:TRUE"
 
 
 class CertificateService:
@@ -16 +17 @@
         self.cryptography_service = cryptography_service
         self.certificate_repository = certificate_repository
 
+    # TODO usages present in method parameters but not in class diagram
     def create_root_ca(self, key: PrivateKey, subject: Subject, extensions: str = "", config: str = ""):
         # create a new self signed  certificate
         cert_pem = self.cryptography_service.create_sscrt(subject, key.private_key, key_pass=key.password,
@@ -40 +42 @@
 
         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):
+        extensions = extensions + "\n" + CA_EXTENSIONS
+        # 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,
+                                                        subject_key_pass=subject_key.password,
+                                                        issuer_key_pass=issuer_key.password, extensions=extensions,
+                                                        days=days)
+
+        # 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, subject.common_name, not_before_formatted, not_after_formatted, cert_pem,
+                                  subject_key.private_key_id, INTERMEDIATE_CA_ID, 0, {})
+
+        # store the wrapper into the repository
+        created_id = self.certificate_repository.create(certificate)
+
+        # assign the generated ID to the inserted certificate
+        certificate.certificate_id = created_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 @@
+import subprocess
+
 from src.model.subject import Subject
 
 
+def export_crt(crt):
+    return subprocess.check_output(["openssl", "x509", "-noout", "-text", "-in", "-"],
+                                   input=bytes(crt, encoding="utf-8"), stderr=subprocess.STDOUT).decode()
+
+
 def test_create_and_get_pk(private_key_service, certificate_service):
     private_key = private_key_service.create_new_key(passphrase="foobar")
     private_key_loaded = private_key_service.get_key(private_key.private_key_id)
@@ -10 +17 @@
     assert private_key.password == private_key_loaded.password
 
 
-def test_create_and_get_cert(private_key_service, certificate_service):
+def test_create_and_get_root_ca(private_key_service, certificate_service):
     private_key = private_key_service.create_new_key(passphrase="foobar")
 
     cert = certificate_service.create_root_ca(private_key,
@@ -18 +25 @@
 
     cert_loaded = certificate_service.get_certificate(cert.certificate_id)
 
+    # 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
+
     assert cert.certificate_id == cert_loaded.certificate_id
     assert cert.common_name == cert_loaded.common_name
     assert cert.valid_from == cert_loaded.valid_from
@@ -27 +39 @@
     assert cert.type_id == cert_loaded.type_id
     assert cert.parent_id == cert_loaded.parent_id
     assert cert.usages == 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()
+
+    root_ca = certificate_service.create_root_ca(root_ca_private_key,
+                                                 Subject(common_name="RootFoo", organization_unit="Department of Foo"))
+
+    inter_cert = certificate_service.create_ca(inter_ca_private_key, Subject(common_name="Intermediate CA"), root_ca,
+                                             root_ca_private_key)
+
+    inter_cert_loaded = certificate_service.get_certificate(inter_cert.certificate_id)
+
+    # 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
+
+    assert inter_cert_loaded.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
+    assert inter_cert.pem_data == inter_cert_loaded.pem_data
+    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.usages == inter_cert_loaded.usages