Port to Python cryptography module

README.md

@@ -10,13 +10,11 @@ If you prefer to configure OMERO manually see the examples in these documents:
 
 ## Installation
 
-Install `openssl` if it's not already on your system.
-Then activate your OMERO.server virtualenv and run:
+Activate your OMERO.server virtualenv and run:
 ```
 pip install omero-certificates
 ```
 
-
 ## Usage
 
 Set the `OMERODIR` environment variable to the location of OMERO.server.
@@ -26,11 +24,6 @@ Run:
 omero certificates
 ```
 ```
-OpenSSL 1.1.1d  10 Sep 2019
-Generating RSA private key, 2048 bit long modulus (2 primes)
-.+++++
-.............................+++++
-e is 65537 (0x010001)
 certificates created: /OMERO/certs/server.key /OMERO/certs/server.pem /OMERO/certs/server.p12
 ```
 to update your OMERO.server configuration and to generate or update your self-signed certificates.
@@ -47,17 +40,31 @@ The original values can be found on https://docs.openmicroscopy.org/omero/5.6.0/
 Certificates will be stored under `{omero.data.dir}/certs` by default.
 Set `omero.glacier2.IceSSL.DefaultDir` to change this.
 
-If you see a warning message such as
+For full information see the output of:
 ```
-Can't load ./.rnd into RNG
+omero certificates --help
 ```
-it should be safe to ignore.
 
-For full information see the output of:
+## Upgrading
+
+Since version 0.3.0 this plugin uses portable RFC 4514 (supercedes RFC 2253)
+formatted strings for the `omero.certificates.owner` configuration option.  If
+you have ran `omero certificates` before you may have OpenSSL command line
+formatted strings in your configuration that should be updated before you can
+run `omero certificates` again.  In most cases this means taking a string such
+as `/L=OMERO/O=OMERO.server` and reformatting it to
+`L=OMERO,O=OMERO.server`; remove the leading `/` and replace separator `/`'s
+with `,`'s.
+
+You can see the RFC 4514 compatible string for the `Issuer` and `Subject`
+of your existing certificate by running:
 ```
-omero certificates --help
+openssl x509 -in /path/to/cert.pem -text -nameopt rfc2253
 ```
 
+You can review the RFC in full for more specific details:
+- https://tools.ietf.org/html/rfc4514.html
+
 ## Developer notes
 
 This project uses [setuptools-scm](https://pypi.org/project/setuptools-scm/).

omero_certificates/certificates.py

@@ -1,12 +1,23 @@
 #!/usr/bin/env python
 # -*- coding: utf-8 -*-
 """
-Wrap openssl to manage self-signed certificates
+Wrap cryptography to manage self-signed certificates
 """
 
 import logging
 import os
-import subprocess
+import re
+from datetime import datetime, timedelta
+from cryptography import x509
+from cryptography.hazmat.primitives import serialization
+from cryptography.hazmat.primitives.asymmetric import rsa
+from cryptography.hazmat.primitives.hashes import SHA256, SHA1
+from cryptography.hazmat.primitives.serialization import (
+    Encoding,
+    NoEncryption,
+    PrivateFormat,
+    pkcs12,
+)
 from omero.config import ConfigXml
 
 log = logging.getLogger(__name__)
@@ -30,7 +41,7 @@ def set_if_empty(cfgkey, default):
         os.path.join(cfgdict.get("omero.data.dir", "/OMERO"), "certs"),
     )
     set_if_empty("omero.certificates.commonname", "localhost")
-    set_if_empty("omero.certificates.owner", "/L=OMERO/O=OMERO.server")
+    set_if_empty("omero.certificates.owner", "L=OMERO,O=OMERO.server")
     set_if_empty("omero.certificates.key", "server.key")
     set_if_empty("omero.glacier2.IceSSL.CertFile", "server.p12")
     set_if_empty("omero.glacier2.IceSSL.CAs", "server.pem")
@@ -45,81 +56,97 @@ def set_if_empty(cfgkey, default):
     return cfgdict
 
 
-def run_openssl(args):
-    command = ["openssl"] + args
-    log.info("Executing: %s", " ".join(command))
-    subprocess.run(command)
-
-
 def create_certificates(omerodir):
     cfgmap = update_config(omerodir)
     certdir = cfgmap["omero.glacier2.IceSSL.DefaultDir"]
 
     cn = cfgmap["omero.certificates.commonname"]
     owner = cfgmap["omero.certificates.owner"]
-    days = "365"
+    days = 365
     pkcs12path = os.path.join(certdir, cfgmap["omero.glacier2.IceSSL.CertFile"])
     keypath = os.path.join(certdir, cfgmap["omero.certificates.key"])
     certpath = os.path.join(certdir, cfgmap["omero.glacier2.IceSSL.CAs"])
     password = cfgmap["omero.glacier2.IceSSL.Password"]
 
-    try:
-        run_openssl(["version"])
-    except subprocess.CalledProcessError as e:
-        msg = "openssl version failed, is it installed?"
-        log.fatal("%s: %s", msg, e)
-        raise
-
     os.makedirs(certdir, exist_ok=True)
     created_files = []
 
     # Private key
     if os.path.exists(keypath):
         log.info("Using existing key: %s", keypath)
+        with open(keypath, "rb") as pem_openssl_key:
+            rsa_private_key = serialization.load_pem_private_key(
+                pem_openssl_key.read(),
+                password=None,
+            )
     else:
         log.info("Creating self-signed CA key: %s", keypath)
-        run_openssl(["genrsa", "-out", keypath, "2048"])
+        # Do what `openssl genrsa -out <keypath> <numbits>` would do
+        rsa_private_key = rsa.generate_private_key(public_exponent=65537, key_size=2048)
+        with open(keypath, "wb") as pem_openssl_key:
+            pem_openssl_key.write(
+                rsa_private_key.private_bytes(
+                    Encoding.PEM,
+                    PrivateFormat.TraditionalOpenSSL,  # Essentially PKCS#1
+                    NoEncryption(),
+                )
+            )
         created_files.append(keypath)
 
     # Self-signed certificate
     log.info("Creating self-signed certificate: %s", certpath)
-    run_openssl(
-        [
-            "req",
-            "-new",
-            "-x509",
-            "-subj",
-            "{}/CN={}".format(owner, cn),
-            "-days",
-            days,
-            "-key",
-            keypath,
-            "-out",
-            certpath,
-            "-extensions",
-            "v3_ca",
-        ]
+    # Do what `openssl req -x509 ...` would do
+    utcnow = datetime.utcnow()
+    try:
+        if owner.startswith("/"):
+            log.warn(
+                f"'omero.certificates.owner' configuration setting '{owner}' not a "
+                "valid RFC 4514 string!  Attempting to convert."
+            )
+            owner = re.sub(r"\s*/\s*", r",", owner.lstrip("/"))
+        subject = issuer = x509.Name.from_rfc4514_string("{},CN={}".format(owner, cn))
+    except ValueError:
+        return (
+            f"'omero.certificates.owner' configuration setting '{owner}' not a "
+            "valid RFC 4514 string!  Are you upgrading?  See "
+            "https://pypi.org/project/omero-certificates/ for help."
+        )
+    cert = (
+        x509.CertificateBuilder()
+        .subject_name(subject)
+        .issuer_name(issuer)
+        .not_valid_before(utcnow)
+        .not_valid_after(utcnow + timedelta(days=days))
+        .public_key(rsa_private_key.public_key())
+        .serial_number(x509.random_serial_number())
+        .sign(rsa_private_key, SHA256())
     )
+    with open(certpath, "wb") as pem_cert:
+        pem_cert.write(cert.public_bytes(Encoding.PEM))
     created_files.append(certpath)
 
     # PKCS12 format
     log.info("Creating PKCS12 bundle: %s", pkcs12path)
-    run_openssl(
-        [
-            "pkcs12",
-            "-export",
-            "-out",
-            pkcs12path,
-            "-inkey",
-            keypath,
-            "-in",
-            certpath,
-            "-name",
-            "server",
-            "-password",
-            "pass:{}".format(password),
-        ]
-    )
+    # Do what `openssl pkcs12 ...` would do
+    with open(pkcs12path, "wb") as p12:
+        # Maintain compatibility with OpenSSL < 3.0.0, the macOS security
+        # framework and Windows.
+        encryption = (
+            PrivateFormat.PKCS12.encryption_builder()
+            .kdf_rounds(50000)
+            .key_cert_algorithm(pkcs12.PBES.PBESv1SHA1And3KeyTripleDESCBC)
+            .hmac_hash(SHA1())
+            .build(password.encode("utf-8"))
+        )
+        p12.write(
+            pkcs12.serialize_key_and_certificates(
+                b"server",
+                rsa_private_key,
+                cert,
+                None,
+                encryption,
+            )
+        )
     created_files.append(pkcs12path)
 
     return "certificates created: " + " ".join(created_files)

setup.py

@@ -18,7 +18,10 @@
     url="https://github.com/ome/omero-certificates",
     packages=["omero_certificates", "omero.plugins"],
     setup_requires=["setuptools_scm"],
-    install_requires=["omero-py>=5.6.0"],
+    install_requires=[
+        "omero-py>=5.6.0",
+        "cryptography>=38.0.0",
+    ],
     use_scm_version={"write_to": "omero_certificates/_version.py"},
     classifiers=[
         "Environment :: Console",

tests/unit/test_certificates.py

@@ -1,6 +1,12 @@
 import os
-import subprocess
 
+from cryptography import x509
+from cryptography.hazmat.primitives.asymmetric.rsa import RSAPrivateKey
+from cryptography.hazmat.primitives.serialization.pkcs12 import (
+    PKCS12Certificate,
+    load_pkcs12,
+)
+from cryptography.x509.oid import NameOID
 from omero.config import ConfigXml
 from omero_certificates.certificates import create_certificates, update_config
 
@@ -31,7 +37,7 @@ def test_config_from_empty(self, tmpdir):
             "omero.glacier2.IceSSL.Protocols": "TLS1_0,TLS1_1,TLS1_2",
             "omero.certificates.commonname": "localhost",
             "omero.certificates.key": "server.key",
-            "omero.certificates.owner": "/L=OMERO/O=OMERO.server",
+            "omero.certificates.owner": "L=OMERO,O=OMERO.server",
         }
 
     def test_config_keep_existing(self, tmpdir):
@@ -58,6 +64,26 @@ def test_config_keep_existing(self, tmpdir):
             "omero.certificates.owner": "/L=universe/O=42",
         }
 
+    def assert_pkcs12(self, f):
+        p12 = load_pkcs12(f.read(), b"secret")
+        assert p12.key
+        assert isinstance(p12.key, RSAPrivateKey)
+        assert p12.key.key_size == 2048
+
+        assert p12.cert
+        assert isinstance(p12.cert, PKCS12Certificate)
+        certificate = p12.cert.certificate
+        assert certificate
+        assert isinstance(certificate, x509.Certificate)
+        subject = certificate.subject
+        assert len(subject) == 3
+        (cn,) = subject.get_attributes_for_oid(NameOID.COMMON_NAME)
+        assert cn.value == "localhost"
+        (l,) = subject.get_attributes_for_oid(NameOID.LOCALITY_NAME)
+        assert l.value == "OMERO"
+        (o,) = subject.get_attributes_for_oid(NameOID.ORGANIZATION_NAME)
+        assert o.value == "OMERO.server"
+
     def test_create_certificates(self, tmpdir):
         (tmpdir / "etc" / "grid").ensure(dir=True)
         omerodir = str(tmpdir)
@@ -75,25 +101,26 @@ def test_create_certificates(self, tmpdir):
         for filename in ("server.key", "server.p12", "server.pem"):
             assert os.path.isfile(os.path.join(datadir, "certs", filename))
 
-        out = subprocess.check_output(
-            [
-                "openssl",
-                "pkcs12",
-                "-in",
-                os.path.join(datadir, "certs", "server.p12"),
-                "-passin",
-                "pass:secret",
-                "-passout",
-                "pass:secret",
-            ]
-        )
-        out = out.decode().splitlines()
-        for line in (
-            "subject=L = OMERO, O = OMERO.server, CN = localhost",
-            "issuer=L = OMERO, O = OMERO.server, CN = localhost",
-            "-----BEGIN CERTIFICATE-----",
-            "-----END CERTIFICATE-----",
-            "-----BEGIN ENCRYPTED PRIVATE KEY-----",
-            "-----END ENCRYPTED PRIVATE KEY-----",
-        ):
-            assert line in out
+        with open(os.path.join(datadir, "certs", "server.p12"), "rb") as f:
+            self.assert_pkcs12(f)
+
+    def test_create_certificates_from_existing_0_2_0(self, tmpdir):
+        (tmpdir / "etc" / "grid").ensure(dir=True)
+        omerodir = str(tmpdir)
+        datadir = str(tmpdir / "OMERO")
+        configxml = ConfigXml(os.path.join(omerodir, "etc", "grid", "config.xml"))
+        configxml["omero.data.dir"] = datadir
+        configxml["omero.certificates.owner"] = "/L=OMERO/O=OMERO.server"
+        configxml.close()
+
+        m = create_certificates(omerodir)
+        assert m.startswith("certificates created: ")
+
+        cfg = get_config(omerodir)
+        assert cfg["omero.glacier2.IceSSL.DefaultDir"] == os.path.join(datadir, "certs")
+
+        for filename in ("server.key", "server.p12", "server.pem"):
+            assert os.path.isfile(os.path.join(datadir, "certs", filename))
+
+        with open(os.path.join(datadir, "certs", "server.p12"), "rb") as f:
+            self.assert_pkcs12(f)