Revize c0aed2f5

Re #8472 - Added create_sscrt method that creates a self signed certificate

Added 5 unit tests verifying the validity of the certificate generation.

proj/model/subject.py

@@ +1 @@
+class Subject:
+
+    def __init__(self, common_name=None, country=None, locality=None, state=None, organization=None,
+                 organization_unit=None, email_address=None):
+        self.common_name = common_name
+        self.country = country
+        self.locality = locality
+        self.state = state
+        self.organization = organization
+        self.organization_unit = organization_unit
+        self.email_address = email_address

proj/services/cryptography.py

@@ +1 @@
+import subprocess
+
+# encryption method to be used when generating private keys
+from proj.utils.temporary_file import TemporaryFile
+
+PRIVATE_KEY_ENCRYPTION_METHOD = "-aes256"
+
+# openssl executable name
+OPENSSL_EXECUTABLE = "openssl"
+
+
+class CryptographyService:
+
+    @staticmethod
+    def _run_for_output(args=None, stdin=None, executable=OPENSSL_EXECUTABLE):
+        """
+        Launches a new process in which the given executable is run. STDIN and process arguments can be set.
+        If the process ends with a non-zero then <CryptographyException> is raised.
+        :param args: Arguments to be passed to the program.
+        :param stdin: String input to be passed to the stdin of the created process.
+        :param executable: Executable to be run (defaults to openssl)
+        :return: If the process ends with a zero return code then the STDOUT of the process is returned as a byte array.
+        """
+        if args is None:
+            args = []
+        try:
+            # prepend the name of the executable
+            args.insert(0, executable)
+
+            # create a new process
+            proc = subprocess.Popen(args, stdin=subprocess.PIPE if stdin is not None else None, stdout=subprocess.PIPE,
+                                    stderr=subprocess.PIPE)
+
+            out, err = proc.communicate(stdin)
+
+            if proc.returncode != 0:
+                # if the process did not result in zero result code, then raise an exception
+                if err is not None:
+                    raise CryptographyException(executable, args, err.decode())
+                else:
+                    raise CryptographyException(executable, args,
+                                                f""""Execution resulted in non-zero argument""")
+
+            return out
+        except FileNotFoundError:
+            raise CryptographyException(executable, args, f""""{executable}" not found in the current PATH.""")
+
+    def create_private_key(self, passphrase=None):
+        """
+        Creates a private key with the option to encrypt it using a passphrase.
+        :param passphrase: A passphrase to be used when encrypting the key (if none is passed then the key is not
+        encrypted at all). Empty passphrase ("") also results in a key that is not encrypted.
+        :return: A text representation of the generated private key.
+        """
+        if passphrase is None or len(passphrase) == 0:
+            return self._run_for_output(["genrsa", "2048"]).decode()
+        else:
+            return self._run_for_output(
+                ["genrsa", PRIVATE_KEY_ENCRYPTION_METHOD, "-passout", f"pass:{passphrase}", "2048"]).decode()
+
+    def create_sscrt(self, key, subject, config="", extensions="", key_passphrase=None):
+        """
+        Creates a root CA
+
+        :param key: private key of the CA to be used
+        :param subject: an instance of <Subject> representing the subject to be added to the certificate
+        :param config: string containing the configuration to be used
+        :param extensions: name of the section in the configuration representing extensions
+        :param key_passphrase: passphrase of the private key
+
+        :return: byte array containing the generated certificate
+        """
+        assert key is not None
+        assert subject is not None
+
+        subj_dict = {}
+        if subject.common_name is not None:
+            subj_dict["CN"] = subject.common_name
+        if subject.country is not None:
+            subj_dict["C"] = subject.country
+        if subject.locality is not None:
+            subj_dict["L"] = subject.locality
+        if subject.state is not None:
+            subj_dict["ST"] = subject.state
+        if subject.organization is not None:
+            subj_dict["O"] = subject.organization
+        if subject.organization_unit is not None:
+            subj_dict["OU"] = subject.organization_unit
+        if subject.email_address is not None:
+            subj_dict["emailAddress"] = subject.email_address
+
+        # merge the subject into a "subj" parameter
+        subj = "".join([f"/{key}={value}" for key, value in subj_dict.items()])
+
+        with TemporaryFile("openssl.conf", config) as conf_path:
+            args = ["req", "-x509", "-new", "-subj", subj,
+                    "-key", "-"]
+            if len(config) > 0:
+                args.extend(["-config", conf_path])
+
+            if len(extensions) > 0:
+                args.extend(["-extensions", extensions])
+
+            # it would be best to not send the pass phrase at all, but for some reason pytest then prompts for
+            # the pass phrase (this does not happen when run from pycharm)
+
+            # if key_passphrase is not None:
+            args.extend(["-passin", f"pass:{key_passphrase}"])
+
+            return self._run_for_output(args, stdin=bytes(key, encoding="utf-8")).decode()
+
+
+class CryptographyException(Exception):
+
+    def __init__(self, executable, args, message):
+        self.executable = executable
+        self.args = args
+        self.message = message
+
+    def __str__(self):
+        return f"""
+        EXECUTABLE: {self.executable}
+        ARGS: {self.args}
+        MESSAGE: {self.message}
+        """

proj/tests/services/cryptography_test.py

@@ +1 @@
+import pytest
+import subprocess
+
+from proj.model.subject import Subject
+from proj.services.cryptography import CryptographyService, CryptographyException
+
+
[e-mail]
+def service():
+    # provide a CryptographyService fixture
+    return CryptographyService()
+
+
+def test_private_key(service):
+    private_key = service.create_private_key()
+
+    # verify the private key
+    subprocess.check_output(["openssl", "rsa", "-in", "-", "-check"], input=bytes(private_key, encoding="utf-8"),
+                            stderr=subprocess.STDOUT)
+
+
+def test_encrypted_private_key(service):
+    private_key = service.create_private_key(passphrase="foobar")
+
+    # verify the private key providing a correct passphrase
+    subprocess.check_output(["openssl", "rsa", "-in", "-", "-passin", "pass:foobar", "-check"],
+                            input=bytes(private_key, encoding="utf-8"), stderr=subprocess.STDOUT)
+
+
+def test_encrypted_private_key_incorrect_pass(service):
+    private_key = service.create_private_key(passphrase="foobar")
+
+    # incorrect passphrase provided
+    with pytest.raises(subprocess.CalledProcessError):
+        subprocess.check_output(["openssl", "rsa", "-in", "-", "-passin", "pass:bazbaz", "-check"],
+                                input=bytes(private_key, encoding="utf-8"), stderr=subprocess.STDOUT)
+
+
+def test_create_sscrt(service):
+    # create a self signed certificate using configuration and extensions
+    private_key = service.create_private_key(passphrase="foobar")
+
+    # distinguished_name is always required
+    config = """
+    # Simple Root CA
+
+    [ req ]
+    distinguished_name      = ca_dn                 # DN section
+
+    [ ca_dn ]
+
+    [ root_ca_ext ]
+    keyUsage                = critical,keyCertSign,cRLSign
+    basicConstraints        = critical,CA:true
+    subjectKeyIdentifier    = hash
+    authorityKeyIdentifier  = keyid:always
+    """
+
+    cert = service.create_sscrt(private_key,
+                                Subject(common_name="Topnax",
+                                        country="CZ",
+                                        locality="My Locality",
+                                        state="My state",
+                                        organization="Mysterious Org.",
+                                        organization_unit="Department of Mysteries",
+                                        email_address="[e-mail]"),
+                                config=config,
+                                extensions="root_ca_ext",
+                                key_passphrase="foobar")
+
+    cert_printed = subprocess.check_output(["openssl", "x509", "-noout", "-text", "-in", "-"],
+                                           input=bytes(cert, encoding="utf-8"), stderr=subprocess.STDOUT).decode()
+
+    assert "Certificate Sign, CRL Sign" in cert_printed
+    assert "X509v3 Key Usage: critical" in cert_printed
+    assert "CA:TRUE" in cert_printed
+
+    assert "Issuer: CN = Topnax, C = CZ, L = My Locality, ST = My state, O = Mysterious Org., OU = Department of Mysteries, emailAddress = [e-mail]" in cert_printed
+    assert "Subject: CN = Topnax, C = CZ, L = My Locality, ST = My state, O = Mysterious Org., OU = Department of Mysteries, emailAddress = [e-mail]" in cert_printed
+
+
+def test_create_sscrt_config_without_extensions(service):
+    # create a self signed certificate without specifying extensions
+    private_key = service.create_private_key()
+
+    config = """
+    # Simple Root CA
+
+    [ req ]
+    distinguished_name      = ca_dn                 # DN section
+
+    [ ca_dn ]
+
+    """
+
+    cert = service.create_sscrt(private_key, Subject(common_name="Topnax", country="CZ"), config=config)
+
+    cert_printed = subprocess.check_output(["openssl", "x509", "-noout", "-text", "-in", "-"],
+                                           input=bytes(cert, encoding="utf-8"), stderr=subprocess.STDOUT).decode()
+
+    # TODO pass something in the configuration that can be asserted
+    assert "Issuer: CN = Topnax, C = CZ" in cert_printed
+    assert "Subject: CN = Topnax, C = CZ" in cert_printed
+
+
+def test_create_sscrt_plain(service):
+    # create a self signed certificate without configuration
+    private_key = service.create_private_key()
+
+    cert = service.create_sscrt(private_key, Subject(common_name="Topnax", country="CZ"))
+
+    cert_printed = subprocess.check_output(["openssl", "x509", "-noout", "-text", "-in", "-"],
+                                           input=bytes(cert, encoding="utf-8"), stderr=subprocess.STDOUT).decode()
+
+    assert "Issuer: CN = Topnax, C = CZ" in cert_printed
+    assert "Subject: CN = Topnax, C = CZ" in cert_printed
+
+
+def test_create_sscrt_passphrase(service):
+    # create a self signed certificate with a PK that is protected by a passphrase
+    private_key = service.create_private_key(passphrase="foobar")
+
+    cert = service.create_sscrt(private_key, Subject(common_name="Topnax", country="CZ"), key_passphrase="foobar")
+
+    cert_printed = subprocess.check_output(["openssl", "x509", "-noout", "-text", "-in", "-"],
+                                           input=bytes(cert, encoding="utf-8"), stderr=subprocess.STDOUT).decode()
+
+    assert "Issuer: CN = Topnax, C = CZ" in cert_printed
+    assert "Subject: CN = Topnax, C = CZ" in cert_printed
+
+
+def test_create_sscrt_incorrect_passphrase(service):
+    # make an attempt to create a self signed certificate using a private key with specifying wrong key passphrase or
+    # no passphrase at all
+    private_key = service.create_private_key(passphrase="foobar")
+
+    # incorrect passphrase provided when using a protected private key
+    with pytest.raises(CryptographyException) as e:
+        service.create_sscrt(private_key, Subject(common_name="Topnax", country="CZ"), key_passphrase="bazfoo")
+    assert "bad decrypt" in e.value.message
+
+    # no passphrase provided when using a protected private key
+    with pytest.raises(CryptographyException) as e:
+        service.create_sscrt(private_key, Subject(common_name="Topnax", country="CZ"))
+    assert "bad decrypt" in e.value.message

proj/utils/temporary_file.py

@@ +1 @@
+import os
+
+
+# allows the caller, when using "with" keyword, access a temporary file with the desired content that is deleted
+# when "with" context is destroyed
+class TemporaryFile(object):
+    def __init__(self, file_name, content):
+        self.file_name = file_name
+        self.content = content
+
+    def __enter__(self):
+        # if content was passed, write it to a new file specified by the given file name
+        if len(self.content) > 0:
+            with open(self.file_name, "w") as file:
+                file.write(self.content)
+        return self.file_name
+
+    def __exit__(self, type_, value, traceback):
+        if len(self.content) > 0:
+            # if content was passed then the file has been created => delete it.
+            os.remove(self.file_name)

services/cryptography.py

@@ -1 @@
-import subprocess
-
-# encryption method to be used when generating private keys
-PRIVATE_KEY_ENCRYPTION_METHOD = "-aes256"
-
-# openssl executable name
-OPENSSL_EXECUTABLE = "openssl"
-
-
-class CryptographyService:
-
-    @staticmethod
-    def _run_for_output(args=None, stdin=None, executable=OPENSSL_EXECUTABLE):
-        """
-        Launches a new process in which the given executable is run. STDIN and process arguments can be set.
-        If the process ends with a non-zero then <CryptographyException> is raised.
-        :param args: Arguments to be passed to the program.
-        :param stdin: Arguments to be passed to the process.
-        :param executable: Executable to be run (defaults to openssl)
-        :return: If the process ends with a zero return code then the STDOUT of the process is returned as a byte array.
-        """
-        if args is None:
-            args = []
-        try:
-            # prepend the name of the executable
-            args.insert(0, executable)
-
-            # create a new process
-            proc = subprocess.Popen(args, stdin=stdin, stdout=subprocess.PIPE,
-                                    stderr=subprocess.PIPE)
-
-            # get process result
-            out, err = proc.communicate()
-
-            if proc.returncode != 0:
-                # if the process did not result in zero result code, then raise an exception
-                if err is not None:
-                    raise CryptographyException(err.decode())
-                else:
-                    raise CryptographyException(
-                        f""""{executable} with parameters {args} resulted in non-zero argument""")
-
-            return out
-        except FileNotFoundError:
-            raise CryptographyException(f""""{executable}" not found in the current PATH.""")
-
-    def create_private_key(self, passphrase=None):
-        """
-        Creates a private key with the option to encrypt it using a passphrase.
-        :param passphrase: A passphrase to be used when encrypting the key (if none is passed then the key is not
-        encrypted at all). Empty passphrase ("") also results in a key that is not encrypted.
-        :return: A text representation of the generated private key.
-        """
-        if passphrase is None or len(passphrase) == 0:
-            return self._run_for_output(["genrsa", "2048"]).decode()
-        else:
-            return self._run_for_output(
-                ["genrsa", PRIVATE_KEY_ENCRYPTION_METHOD, "-passout", f"pass:{passphrase}", "2048"]).decode()
-
-
-class CryptographyException(Exception):
-
-    def __init__(self, message):
-        self.message = message

tests/services/cryptography_test.py

@@ -1 @@
-import pytest
-from services.cryptography import CryptographyService, CryptographyException
-import subprocess
-
-
[e-mail]
-def service():
-    # provide a CryptographyService fixture
-    return CryptographyService()
-
-
-def test_private_key(service):
-    private_key = service.create_private_key()
-
-    # verify the private key
-    subprocess.check_output(["openssl", "rsa", "-in", "-", "-check"], input=bytes(private_key, encoding="utf-8"),
-                            stderr=subprocess.STDOUT)
-
-
-def test_encrypted_private_key(service):
-    private_key = service.create_private_key(passphrase="foobar")
-
-    # verify the private key providing a correct passphrase
-    subprocess.check_output(["openssl", "rsa", "-in", "-", "-passin", "pass:foobar", "-check"],
-                            input=bytes(private_key, encoding="utf-8"), stderr=subprocess.STDOUT)
-
-
-def test_encrypted_private_key_incorrect_pass(service):
-    private_key = service.create_private_key(passphrase="foobar")
-
-    # incorrect passphrase provided
-    with pytest.raises(subprocess.CalledProcessError) as e:
-        subprocess.check_output(["openssl", "rsa", "-in", "-", "-passin", "pass:bazbaz", "-check"],
-                                input=bytes(private_key, encoding="utf-8"), stderr=subprocess.STDOUT)
-    # no passphrase provided
-    with pytest.raises(subprocess.CalledProcessError) as e:
-        subprocess.check_output(["openssl", "rsa", "-in", "-", "-check"],
-                                input=bytes(private_key, encoding="utf-8"), stderr=subprocess.STDOUT)