Revize c4b733ef

Re #8577 - Minor refactoring (crl -> crl_ocsp).

src/services/crl/ca_index_file_line_generator.py

@@ -1 @@
-import time
-from time import struct_time
-
-from src.model.certificate import Certificate
-from src.model.subject import Subject
-
-SRL_LEN = 8  # number of hex digits in serial number
-TAB_CHAR = "\t"
-INDEX_FILE_DATE_ENTRY_FORMAT = "%y%m%d%H%M%SZ"
-
-
-def get_index_file_time_entry(date: struct_time):
-    # convert the time to the format of openssl CA index file
-    return time.strftime(INDEX_FILE_DATE_ENTRY_FORMAT, date)
-
-
-def get_distinguished_name(subject: Subject):
-    # convert subject class instance to the distinguished name in the openssl CA index file format
-    return "".join([f"/{key}={value}" if value is not None else "" for key, value in subject.to_dict().items()])
-
-
-def create_index_file_revoked_line(revoked_certificate: Certificate, subject: Subject, revocation_date: struct_time,
-                                   valid_to: struct_time) -> str:
-    # converts the given certificate as well as the subject and revocation / valid_to dates to a line of openssl CA
-    # index file format
-    items = [
-        # certificate status flag (R stands for revoked)
-        "R",
-        # followed by the expiration date field
-        f"{get_index_file_time_entry(valid_to)}",
-        # followed by the revocation date field
-        f"{get_index_file_time_entry(revocation_date)},{revoked_certificate.revocation_reason}",
-        # followed by the serial number of the certificate in hex format
-        __get_serial(revoked_certificate.certificate_id),
-        # certificate filename ("unknown" literal used for unknown file names)
-        "unknown",
-        # certificate distinguished name
-        get_distinguished_name(subject)
-    ]
-
-    return TAB_CHAR.join(items)
-
-def create_index_file_valid_line(certificate: Certificate, subject: Subject, valid_to: struct_time) -> str:
-    # converts the given certificate as well as the subject and revocation / valid_to dates to a line of openssl CA
-    # index file format
-    items = [
-        # certificate status flag (R stands for revoked)
-        "V",
-        # followed by the expiration date field
-        f"{get_index_file_time_entry(valid_to)}",
-        # followed by the revocation date field
-        f"",
-        # followed by the serial number of the certificate in hex format
-        __get_serial(certificate.certificate_id),
-        # certificate filename ("unknown" literal used for unknown file names)
-        "unknown",
-        # certificate distinguished name
-        get_distinguished_name(subject)
-    ]
-
-    return TAB_CHAR.join(items)
-
-def __get_serial(cert_id) -> str:
-    srl = hex(cert_id).replace("0x", "")
-    srl = "0"*(SRL_LEN - len(srl)) + srl  # generate exactly SRL_LEN digits
-    return srl.upper()

src/services/crl/crl_service.py

@@ -1 @@
-from datetime import datetime
-
-from injector import inject
-
-from src.dao.certificate_repository import CertificateRepository
-from src.dao.private_key_repository import PrivateKeyRepository
-from src.exceptions.certificate_not_found_exception import CertificateNotFoundException
-from src.exceptions.private_key_not_found_exception import PrivateKeyNotFoundException
-from src.services.crl.ca_index_file_line_generator import create_index_file_revoked_line, create_index_file_valid_line
-from src.services.cryptography import CryptographyService
-from src.utils.temporary_file import TemporaryFile
-
-
-class CrlService:
-    @inject
-    def __init__(self,
-                 certificate_repository: CertificateRepository,
-                 key_repository: PrivateKeyRepository,
-                 cryptography_service: CryptographyService
-                 ):
-        self.key_repository = key_repository
-        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
-        """
-        # get issuing certificate
-        certificate = self.certificate_repository.read(ca_id)
-        if certificate is None:
-            raise CertificateNotFoundException(ca_id)
-
-        # get subject and notAfter of the issuer
-        subject, _, not_after = self.cryptography_service.parse_cert_pem(certificate.pem_data)
-
-        index_lines = [create_index_file_valid_line(certificate, subject, not_after)]
-        # 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 revocation date from unix timestamp to struct_time
-                                                  datetime.utcfromtimestamp(int(certificate.revocation_date)).timetuple(),
-                                                  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)
-
-    def create_index(self, ca_id) -> str:
-        """
-        Queries the certificate repository and looks for all certificates issued 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 issued (=child) certificates should be put into the index file
-        :return: a str representing the content of a CA index file
-        """
-        # get issuing certificate
-        certificate = self.certificate_repository.read(ca_id)
-        if certificate is None:
-            raise CertificateNotFoundException(ca_id)
-
-        # get subject and notAfter of the issuer
-        subject, _, not_after = self.cryptography_service.parse_cert_pem(certificate.pem_data)
-
-        index_lines = [create_index_file_valid_line(certificate, subject, not_after)]
-        # iterate over revoked certificates of the CA given by an ID
-        for certificate in self.certificate_repository.get_all_issued_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)
-            if len(certificate.revocation_reason) > 0:
-                line = create_index_file_revoked_line(certificate,
-                                                      subject,
-                                                      # parse revocation date from unix timestamp to struct_time
-                                                      datetime.utcfromtimestamp(
-                                                          int(certificate.revocation_date)).timetuple(),
-                                                      not_after)
-            else:
-                line = create_index_file_valid_line(certificate, subject, 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)
-
-    def generate_crl_response(self, ca_id: int) -> str:
-        """
-        Generate a CRL for the given certificate authority
-        that contains all revoked certificates
-
-        :param ca_id: ID of a CA whose CRL shall be generated
-        :return: CRL in PEM format
-        """
-        # get cert and check if the requested CA exists and if not throw an exception
-        cert = self.certificate_repository.read(ca_id)
-        if cert is None:
-            raise CertificateNotFoundException(ca_id)
-
-        # get key and check if it exists
-        key = self.key_repository.read(cert.private_key_id)
-        if key is None:
-            raise PrivateKeyNotFoundException(ca_id)
-
-        # Create an index file and call cryptography service to generate CRL
-        with TemporaryFile("crl.index", f"{self.create_revoked_index(ca_id)}\n") as index_path:
-            crl_content = self.cryptography_service.generate_crl(cert, key, index_path)
-
-        return crl_content
-
-    def generate_ocsp_response(self, ca_id: int, der_ocsp_request: bytes):
-        # get cert and check if the requested CA exists and if not throw an exception
-        cert = self.certificate_repository.read(ca_id)
-        if cert is None:
-            raise CertificateNotFoundException(ca_id)
-
-        # get key and check if it exists
-        key = self.key_repository.read(cert.private_key_id)
-        if key is None:
-            raise PrivateKeyNotFoundException(ca_id)
-
-        # Create an index file and call cryptography service to generate the OCSP response
-        with TemporaryFile("crl.index", f"{self.create_index(ca_id)}\n") as index_path:
-            ocsp_content = self.cryptography_service.generate_ocsp(cert, key, index_path, der_ocsp_request)
-
-        return ocsp_content

src/services/crl_ocsp/ca_index_file_line_generator.py

@@ +1 @@
+import time
+from time import struct_time
+
+from src.model.certificate import Certificate
+from src.model.subject import Subject
+
+SRL_LEN = 8  # number of hex digits in serial number
+TAB_CHAR = "\t"
+INDEX_FILE_DATE_ENTRY_FORMAT = "%y%m%d%H%M%SZ"
+
+
+def get_index_file_time_entry(date: struct_time):
+    # convert the time to the format of openssl CA index file
+    return time.strftime(INDEX_FILE_DATE_ENTRY_FORMAT, date)
+
+
+def get_distinguished_name(subject: Subject):
+    # convert subject class instance to the distinguished name in the openssl CA index file format
+    return "".join([f"/{key}={value}" if value is not None else "" for key, value in subject.to_dict().items()])
+
+
+def create_index_file_revoked_line(revoked_certificate: Certificate, subject: Subject, revocation_date: struct_time,
+                                   valid_to: struct_time) -> str:
+    # converts the given certificate as well as the subject and revocation / valid_to dates to a line of openssl CA
+    # index file format
+    items = [
+        # certificate status flag (R stands for revoked)
+        "R",
+        # followed by the expiration date field
+        f"{get_index_file_time_entry(valid_to)}",
+        # followed by the revocation date field
+        f"{get_index_file_time_entry(revocation_date)},{revoked_certificate.revocation_reason}",
+        # followed by the serial number of the certificate in hex format
+        __get_serial(revoked_certificate.certificate_id),
+        # certificate filename ("unknown" literal used for unknown file names)
+        "unknown",
+        # certificate distinguished name
+        get_distinguished_name(subject)
+    ]
+
+    return TAB_CHAR.join(items)
+
+def create_index_file_valid_line(certificate: Certificate, subject: Subject, valid_to: struct_time) -> str:
+    # converts the given certificate as well as the subject and revocation / valid_to dates to a line of openssl CA
+    # index file format
+    items = [
+        # certificate status flag (R stands for revoked)
+        "V",
+        # followed by the expiration date field
+        f"{get_index_file_time_entry(valid_to)}",
+        # followed by the revocation date field
+        f"",
+        # followed by the serial number of the certificate in hex format
+        __get_serial(certificate.certificate_id),
+        # certificate filename ("unknown" literal used for unknown file names)
+        "unknown",
+        # certificate distinguished name
+        get_distinguished_name(subject)
+    ]
+
+    return TAB_CHAR.join(items)
+
+def __get_serial(cert_id) -> str:
+    srl = hex(cert_id).replace("0x", "")
+    srl = "0"*(SRL_LEN - len(srl)) + srl  # generate exactly SRL_LEN digits
+    return srl.upper()

src/services/crl_ocsp/crl_ocsp_service.py

@@ +1 @@
+from datetime import datetime
+
+from injector import inject
+
+from src.dao.certificate_repository import CertificateRepository
+from src.dao.private_key_repository import PrivateKeyRepository
+from src.exceptions.certificate_not_found_exception import CertificateNotFoundException
+from src.exceptions.unknown_exception import UnknownException
+from src.services.crl.ca_index_file_line_generator import create_index_file_revoked_line, create_index_file_valid_line
+from src.services.cryptography import CryptographyService
+from src.utils.temporary_file import TemporaryFile
+
+
+class CrlOcspService:
+    @inject
+    def __init__(self,
+                 certificate_repository: CertificateRepository,
+                 key_repository: PrivateKeyRepository,
+                 cryptography_service: CryptographyService
+                 ):
+        self.key_repository = key_repository
+        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
+        """
+        # get issuing certificate
+        certificate = self.certificate_repository.read(ca_id)
+        if certificate is None:
+            raise CertificateNotFoundException(ca_id)
+
+        # get subject and notAfter of the issuer
+        subject, _, not_after = self.cryptography_service.parse_cert_pem(certificate.pem_data)
+
+        index_lines = [create_index_file_valid_line(certificate, subject, not_after)]
+        # 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 revocation date from unix timestamp to struct_time
+                                                  datetime.utcfromtimestamp(int(certificate.revocation_date)).timetuple(),
+                                                  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)
+
+    def create_index(self, ca_id) -> str:
+        """
+        Queries the certificate repository and looks for all certificates issued 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 issued (=child) certificates should be put into the index file
+        :return: a str representing the content of a CA index file
+        """
+        # get issuing certificate
+        certificate = self.certificate_repository.read(ca_id)
+        if certificate is None:
+            raise CertificateNotFoundException(ca_id)
+
+        # get subject and notAfter of the issuer
+        subject, _, not_after = self.cryptography_service.parse_cert_pem(certificate.pem_data)
+
+        index_lines = [create_index_file_valid_line(certificate, subject, not_after)]
+        # iterate over revoked certificates of the CA given by an ID
+        for certificate in self.certificate_repository.get_all_issued_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)
+            if len(certificate.revocation_reason) > 0:
+                line = create_index_file_revoked_line(certificate,
+                                                      subject,
+                                                      # parse revocation date from unix timestamp to struct_time
+                                                      datetime.utcfromtimestamp(
+                                                          int(certificate.revocation_date)).timetuple(),
+                                                      not_after)
+            else:
+                line = create_index_file_valid_line(certificate, subject, 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)
+
+    def generate_crl_response(self, ca_id: int) -> str:
+        """
+        Generate a CRL for the given certificate authority
+        that contains all revoked certificates
+
+        :param ca_id: ID of a CA whose CRL shall be generated
+        :return: CRL in PEM format
+        """
+        # get cert and check if the requested CA exists and if not throw an exception
+        cert, key = self.check_ca_and_key(ca_id)
+
+        # Create an index file and call cryptography service to generate CRL
+        with TemporaryFile("crl.index", f"{self.create_revoked_index(ca_id)}\n") as index_path:
+            crl_content = self.cryptography_service.generate_crl(cert, key, index_path)
+
+        return crl_content
+
+    def generate_ocsp_response(self, ca_id: int, der_ocsp_request: bytes):
+        """
+        Generates an OCSP Response from a DER encoded OCSP Request given a CA id.
+        :param ca_id: certificate authority ID
+        :param der_ocsp_request: DER encoded OCSP Request
+        :return: DER encoded OCSP Response
+        """
+        cert, key = self.check_ca_and_key(ca_id)
+
+        # Create an index file and call cryptography service to generate the OCSP response
+        with TemporaryFile("crl.index", f"{self.create_index(ca_id)}\n") as index_path:
+            ocsp_content = self.cryptography_service.generate_ocsp(cert, key, index_path, der_ocsp_request)
+
+        return ocsp_content
+
+    def check_ca_and_key(self, ca_id):
+        """
+        Checks whether a certificate with a given ID exists, if it does, returns its contents, including the
+        associated key.
+        :param ca_id: certificate authority ID
+        :return: certificate data & associated key
+        """
+        # get cert and check if the requested CA exists and if not throw an exception
+        cert = self.certificate_repository.read(ca_id)
+        if cert is None:
+            raise CertificateNotFoundException(ca_id)
+        # get key and check if it exists
+        key = self.key_repository.read(cert.private_key_id)
+        if key is None:
+            raise UnknownException("Corrupted database, expected PrivateKey not found.")
+            # TODO log this
+        return cert, key