Revize ef65f488

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

Added return type specification to read_all method in CertificateRepository

src/dao/certificate_repository.py

@@ -113 +113 @@
         else:
             return None
 
-    def read_all(self, filter_type: int = None):
+    def read_all(self, filter_type: int = None) -> List[Certificate]:
         """
         Reads (selects) all certificates (with type).
 

src/services/certificate_service.py

@@ +1 @@
+from typing import List
+
 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
@@ -123 +125 @@
     def get_certificate(self, unique_id: int) -> Certificate:
         return self.certificate_repository.read(unique_id)
 
-    def get_certificates(self, cert_type=None) -> Certificate:
+    def get_certificates(self, cert_type=None) -> List[Certificate]:
         return self.certificate_repository.read_all(cert_type)
+
+    def get_chain_of_trust(self, from_id: int, to_id: int = -1, exclude_root=True) -> List[Certificate]:
+        start_cert = self.certificate_repository.read(from_id)
+
+        if start_cert is None or (start_cert.type_id == ROOT_CA_ID and exclude_root):
+            return []
+
+        current_cert = start_cert
+        chain_of_trust = [current_cert]
+
+        # TODO could possibly be simplified
+        if start_cert.type_id == ROOT_CA_ID:
+            return chain_of_trust
+
+        while True:
+            parent_cert = self.certificate_repository.read(current_cert.parent_id)
+
+            # check whether parent certificate
+            if parent_cert is None or parent_cert.type_id == ROOT_CA_ID:
+                if not exclude_root:
+                    chain_of_trust.append(parent_cert)
+                break
+
+            chain_of_trust.append(parent_cert)
+
+            if parent_cert.certificate_id == to_id:
+                break
+
+            current_cert = parent_cert
+
+        return chain_of_trust

src/services/cryptography.py

@@ -260 +260 @@
             found = re.findall(r"\s?([^c=\s]+)\s?=\s?([^,\n]+)", match.group(1))
             subj = Subject()
             for key, value in found:
-                print(f"{key}:{value}")
                 if key == "C":
                     subj.country = value
                 elif key == "ST":

tests/integration_tests/services/certificate_key_service_test.py

@@ -149 +149 @@
     assert 1 == len(certificate_service_unique.get_certificates(cert_type=ROOT_CA_ID))
     assert 1 == len(certificate_service_unique.get_certificates(cert_type=INTERMEDIATE_CA_ID))
     assert 1 == len(certificate_service_unique.get_certificates(cert_type=CERTIFICATE_ID))
+
+
+def test_get_chain_of_trust(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"))
+    print(root_ca_cert.certificate_id)
+
+    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})
+    print(inter_ca_cert.certificate_id)
+    print(inter_ca_cert.parent_id)
+
+    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})
+    print(cert.certificate_id)
+    print(cert.parent_id)
+
+    cot = certificate_service.get_chain_of_trust(cert.certificate_id)
+    assert len(cot) == 2
+    assert [cert.certificate_id, inter_ca_cert.certificate_id] == [cot[0].certificate_id, cot[1].certificate_id]
+
+    cot = certificate_service.get_chain_of_trust(cert.certificate_id, root_ca_cert.private_key_id)
+    assert len(cot) == 2
+    assert [cert.certificate_id, inter_ca_cert.certificate_id] == [cot[0].certificate_id, cot[1].certificate_id]
+
+    cot = certificate_service.get_chain_of_trust(cert.certificate_id, inter_ca_cert.private_key_id)
+    assert len(cot) == 2
+    assert [cert.certificate_id, inter_ca_cert.certificate_id] == [cot[0].certificate_id, cot[1].certificate_id]
+
+    cot = certificate_service.get_chain_of_trust(cert.certificate_id, exclude_root=False)
+    assert len(cot) == 3
+    assert [cert.certificate_id, inter_ca_cert.certificate_id, root_ca_cert.certificate_id] == [cot[0].certificate_id,
+                                                                                                cot[1].certificate_id,
+                                                                                                cot[2].certificate_id]
+
+    # starting from intermediate certificate
+    cot = certificate_service.get_chain_of_trust(inter_ca_cert.certificate_id)
+    assert len(cot) == 1
+    assert [inter_ca_cert.certificate_id] == [cot[0].certificate_id]
+
+    cot = certificate_service.get_chain_of_trust(inter_ca_cert.certificate_id, root_ca_cert.private_key_id)
+    assert len(cot) == 1
+    assert [inter_ca_cert.certificate_id] == [cot[0].certificate_id]
+
+    cot = certificate_service.get_chain_of_trust(inter_ca_cert.certificate_id, exclude_root=False)
+    assert len(cot) == 2
+    assert [inter_ca_cert.certificate_id, root_ca_cert.certificate_id] == [cot[0].certificate_id,
+                                                                           cot[1].certificate_id]
+
+    # starting from intermediate certificate
+    cot = certificate_service.get_chain_of_trust(root_ca_cert.certificate_id)
+    assert len(cot) == 0
+
+    cot = certificate_service.get_chain_of_trust(root_ca_cert.certificate_id, root_ca_cert.private_key_id)
+    assert len(cot) == 0
+
+    cot = certificate_service.get_chain_of_trust(root_ca_cert.certificate_id, exclude_root=False)
+    assert len(cot) == 1
+    assert [root_ca_cert.certificate_id] == [cot[0].certificate_id]