Obtain S/MIME signature using cryptography library (#129)

* Obtain S/MIME signature using cryptography library

The governance and permission files must be S/MIME signed. The biggest
obstacle to adopting the cryptography library is that it doesn't expose
an S/MIME API (see pyca/cryptography#1621 for
more information).

Use the OpenSSL API directly (via cryptography's hazmat layer) to obtain
S/MIME signatures, and prove the functionality out by using it to sign
the governance files.

Progress on #109

Signed-off-by: Kyle Fazzari <kyle@canonical.com>

* Resolve flake8 import ordering issues

Signed-off-by: Kyle Fazzari <kyle@canonical.com>

sros2/sros2/api/__init__.py

@@ -23,6 +23,7 @@
 
 from cryptography import x509
 from cryptography.hazmat.backends import default_backend as cryptography_backend
+from cryptography.hazmat.bindings.openssl.binding import Binding as SSLBinding
 from cryptography.hazmat.primitives import hashes
 from cryptography.hazmat.primitives import serialization
 from cryptography.hazmat.primitives.asymmetric import ec
@@ -265,11 +266,21 @@ def create_governance_file(path, domain_id):
 
 
 def create_signed_governance_file(signed_gov_path, gov_path, ca_cert_path, ca_key_path):
-    openssl_executable = find_openssl_executable()
-    check_openssl_version(openssl_executable)
-    run_shell_command(
-        '%s smime -sign -in %s -text -out %s -signer %s -inkey %s' %
-        (openssl_executable, gov_path, signed_gov_path, ca_cert_path, ca_key_path))
+    # Load the CA cert and key from disk
+    with open(ca_cert_path, 'rb') as cert_file:
+        cert = x509.load_pem_x509_certificate(
+            cert_file.read(), cryptography_backend())
+
+    with open(ca_key_path, 'rb') as key_file:
+        private_key = serialization.load_pem_private_key(
+            key_file.read(), None, cryptography_backend())
+
+    # Get the contents of the governance file (which we're about to sign)
+    with open(gov_path, 'rb') as f:
+        content = f.read()
+
+    with open(signed_gov_path, 'wb') as f:
+        f.write(_sign_bytes(cert, private_key, content))
 
 
 def create_keystore(keystore_path):
@@ -587,3 +598,42 @@ def generate_artifacts(keystore_path=None, identity_names=[], policy_files=[]):
             create_permissions_from_policy_element(
                 keystore_path, identity_name, policy_element)
     return True
+
+
+def _sign_bytes(cert, key, byte_string):
+    # Using two flags here to get the output required:
+    #   - PKCS7_DETACHED: Use cleartext signing
+    #   - PKCS7_TEXT: Set the MIME headers for text/plain
+    flags = SSLBinding.lib.PKCS7_DETACHED
+    flags |= SSLBinding.lib.PKCS7_TEXT
+
+    # Convert the byte string into a buffer for SSL
+    bio_in = SSLBinding.lib.BIO_new_mem_buf(byte_string, len(byte_string))
+    try:
+        pkcs7 = SSLBinding.lib.PKCS7_sign(
+            cert._x509, key._evp_pkey, SSLBinding.ffi.NULL, bio_in, flags)
+    finally:
+        # Free the memory allocated for the buffer
+        SSLBinding.lib.BIO_free(bio_in)
+
+    # PKCS7_sign consumes the buffer; allocate a new one again to get it into the final document
+    bio_in = SSLBinding.lib.BIO_new_mem_buf(byte_string, len(byte_string))
+    try:
+        # Allocate a buffer for the output document
+        bio_out = SSLBinding.lib.BIO_new(SSLBinding.lib.BIO_s_mem())
+        try:
+            # Write the final document out to the buffer
+            SSLBinding.lib.SMIME_write_PKCS7(bio_out, pkcs7, bio_in, flags)
+
+            # Copy the output document back to python-managed memory
+            result_buffer = SSLBinding.ffi.new('char**')
+            buffer_length = SSLBinding.lib.BIO_get_mem_data(bio_out, result_buffer)
+            output = SSLBinding.ffi.buffer(result_buffer[0], buffer_length)[:]
+        finally:
+            # Free the memory required for the output buffer
+            SSLBinding.lib.BIO_free(bio_out)
+    finally:
+        # Free the memory allocated for the input buffer
+        SSLBinding.lib.BIO_free(bio_in)
+
+    return output

sros2/test/sros2/commands/security/verbs/test_create_keystore.py

@@ -63,10 +63,17 @@ def check_ca_key_pem(path):
             public = key.public_key()
             assert public.curve.name == 'secp256r1'
 
+    def check_governance_p7s(path):
+        # Would really like to verify the signature, but ffi just can't use
+        # that part of the OpenSSL API
+        with open(path, 'r') as f:
+            lines = f.readlines()
+            assert lines[0] == 'MIME-Version: 1.0\n'
+
     with tempfile.TemporaryDirectory() as keystore_dir:
         assert cli.main(argv=['security', 'create_keystore', keystore_dir]) == 0
         expected_files = (
-            ('governance.p7s', None),
+            ('governance.p7s', check_governance_p7s),
             ('index.txt', check_index_txt),
             ('ca.cert.pem', check_ca_cert_pem),
             ('ca_conf.cnf', check_ca_conf),