Revize 6c098d6e

Re #8472 - Added make_csr method that makes a CSR

Added 2 unit tests testing the added method.
Moved CryptographyService tests into a separate files

proj/services/cryptography.py

@@ -12 +12 @@
 class CryptographyService:
 
     @staticmethod
-    def _run_for_output(args=None, stdin=None, executable=OPENSSL_EXECUTABLE):
+    def subject_to_param_format(subject):
+        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 format
+        return "".join([f"/{key}={value}" for key, value in subj_dict.items()])
+
+    @staticmethod
+    def _run_for_output(args=None, proc_input=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 proc_input: 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.
         """
@@ -28 +49 @@
             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,
+            proc = subprocess.Popen(args, stdin=subprocess.PIPE if proc_input is not None else None, stdout=subprocess.PIPE,
                                     stderr=subprocess.PIPE)
 
-            out, err = proc.communicate(stdin)
+            out, err = proc.communicate(proc_input)
 
             if proc.returncode != 0:
                 # if the process did not result in zero result code, then raise an exception
@@ -73 +94 @@
         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()])
+        subj = self.subject_to_param_format(subject)
 
         with TemporaryFile("openssl.conf", config) as conf_path:
             args = ["req", "-x509", "-new", "-subj", subj,
@@ -107 +111 @@
             # 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()
+            return self._run_for_output(args, proc_input=bytes(key, encoding="utf-8")).decode()
+
+    def make_csr(self, subject, subject_key, subject_key_pass=None):
+        """
+        Makes a CSR (Certificate Signing Request)
+
+        :param subject: an instance of <Subject> representing the subject to be added to the CSR
+        :param subject_key: the private key of the subject to be used to generate the CSR
+        :param subject_key_pass: passphrase of the subject's private key
+        :return: byte array containing the generated certificate signing request
+        """
+
+        subj_param = self.subject_to_param_format(subject)
+
+        args = ["req", "-new", "-subj", subj_param, "-key", "-"]
+
+        if subject_key_pass is not None:
+            args.extend(["-passin", f"pass:{subject_key_pass}"])
+
+        return self._run_for_output(args, proc_input=bytes(subject_key, encoding="utf-8")).decode()
 
 
 class CryptographyException(Exception):

proj/tests/services/cryptography/conftest.py

@@ +1 @@
+import pytest
+
+from proj.services.cryptography import CryptographyService
+
+
[e-mail]
+def service():
+    # provide a CryptographyService fixture
+    return CryptographyService()

proj/tests/services/cryptography/make_csr_test.py

@@ +1 @@
+import subprocess
+
+from proj.model.subject import Subject
+
+
+def get_csr_pem(csr):
+    return subprocess.check_output(["openssl", "req", "-noout", "-text", "-verify", "-in", "-"],
+                                   input=bytes(csr, encoding="utf-8"), stderr=subprocess.STDOUT).decode()
+
+
+def test_make_csr(service):
+    private_key = service.create_private_key()
+
+    subject = Subject(common_name="foo", country="CZ")
+    csr = service.make_csr(subject, private_key)
+
+    assert "Subject: CN = foo, C = CZ" in get_csr_pem(csr)
+
+
+def test_make_csr_pkey_passphrase(service):
+    private_key = service.create_private_key(passphrase="foobar")
+
+    subject = Subject(common_name="foo", country="CZ", organization_unit="Mysterious Unit")
+    csr = service.make_csr(subject, private_key, subject_key_pass="foobar")
+
+    assert "Subject: CN = foo, C = CZ, OU = Mysterious Unit" in get_csr_pem(csr)

proj/tests/services/cryptography/private_keys_test.py

@@ +1 @@
+import pytest
+import subprocess
+
+
+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)

proj/tests/services/cryptography/self_signed_cert_test.py

@@ +1 @@
+import pytest
+import subprocess
+
+from proj.model.subject import Subject
+from proj.services.cryptography import CryptographyService, CryptographyException
+
+
+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/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