Add private key password option to ssl adapters

- It is now possible to use password protected private keys
  in both builtin and openssl ssl-adapters

- Added also positive and negative unit test cases

- With reference to #1583

Author: jatalahd

cheroot/ssl/__init__.py

@@ -20,12 +20,14 @@ def __init__(
         private_key,
         certificate_chain=None,
         ciphers=None,
+        private_key_password=None,
     ):
-        """Set up certificates, private key ciphers and reset context."""
+        """Set up certificates, private key, ciphers and reset context."""
         self.certificate = certificate
         self.private_key = private_key
         self.certificate_chain = certificate_chain
         self.ciphers = ciphers
+        self.private_key_password = private_key_password
         self.context = None
 
     @abstractmethod

cheroot/ssl/__init__.pyi

@@ -6,6 +6,7 @@ class Adapter(ABC):
     private_key: Any
     certificate_chain: Any
     ciphers: Any
+    private_key_password: str | bytes | None
     context: Any
     @abstractmethod
     def __init__(
@@ -14,6 +15,7 @@ class Adapter(ABC):
         private_key,
         certificate_chain: Any | None = ...,
         ciphers: Any | None = ...,
+        private_key_password: str | bytes | None = ...,
     ): ...
     @abstractmethod
     def bind(self, sock): ...

cheroot/ssl/builtin.py

@@ -76,10 +76,19 @@ def _loopback_for_cert_thread(context, server):
             ssl_sock.send(b'0000')
 
 
-def _loopback_for_cert(certificate, private_key, certificate_chain):
+def _loopback_for_cert(
+    certificate,
+    private_key,
+    certificate_chain,
+    private_key_password=None,
+):
     """Create a loopback connection to parse a cert with a private key."""
     context = ssl.create_default_context(cafile=certificate_chain)
-    context.load_cert_chain(certificate, private_key)
+    context.load_cert_chain(
+        certificate,
+        private_key,
+        password=private_key_password,
+    )
     context.check_hostname = False
     context.verify_mode = ssl.CERT_NONE
 
@@ -112,15 +121,25 @@ def _loopback_for_cert(certificate, private_key, certificate_chain):
         server.close()
 
 
-def _parse_cert(certificate, private_key, certificate_chain):
+def _parse_cert(
+    certificate,
+    private_key,
+    certificate_chain,
+    private_key_password=None,
+):
     """Parse a certificate."""
     # loopback_for_cert uses socket.socketpair which was only
     # introduced in Python 3.0 for *nix and 3.5 for Windows
     # and requires OS support (AttributeError, OSError)
     # it also requires a private key either in its own file
     # or combined with the cert (SSLError)
     with suppress(AttributeError, ssl.SSLError, OSError):
-        return _loopback_for_cert(certificate, private_key, certificate_chain)
+        return _loopback_for_cert(
+            certificate,
+            private_key,
+            certificate_chain,
+            private_key_password=private_key_password,
+        )
 
     # KLUDGE: using an undocumented, private, test method to parse a cert
     # unfortunately, it is the only built-in way without a connection
@@ -153,6 +172,9 @@ class BuiltinSSLAdapter(Adapter):
     ciphers = None
     """The ciphers list of SSL."""
 
+    private_key_password = None
+    """Optional passphrase for password protected private key."""
+
     # from mod_ssl/pkg.sslmod/ssl_engine_vars.c ssl_var_lookup_ssl_cert
     CERT_KEY_TO_ENV = {
         'version': 'M_VERSION',
@@ -208,6 +230,7 @@ def __init__(
         private_key,
         certificate_chain=None,
         ciphers=None,
+        private_key_password=None,
     ):
         """Set up context in addition to base class properties if available."""
         if ssl is None:
@@ -218,19 +241,29 @@ def __init__(
             private_key,
             certificate_chain,
             ciphers,
+            private_key_password,
         )
 
         self.context = ssl.create_default_context(
             purpose=ssl.Purpose.CLIENT_AUTH,
             cafile=certificate_chain,
         )
-        self.context.load_cert_chain(certificate, private_key)
+        self.context.load_cert_chain(
+            certificate,
+            private_key,
+            password=private_key_password,
+        )
         if self.ciphers is not None:
             self.context.set_ciphers(ciphers)
 
         self._server_env = self._make_env_cert_dict(
             'SSL_SERVER',
-            _parse_cert(certificate, private_key, self.certificate_chain),
+            _parse_cert(
+                certificate,
+                private_key,
+                self.certificate_chain,
+                private_key_password=private_key_password,
+            ),
         )
         if not self._server_env:
             return

cheroot/ssl/builtin.pyi

@@ -13,6 +13,7 @@ class BuiltinSSLAdapter(Adapter):
         private_key,
         certificate_chain: Any | None = ...,
         ciphers: Any | None = ...,
+        private_key_password: str | bytes | None = ...,
     ) -> None: ...
     @property
     def context(self): ...

cheroot/ssl/pyopenssl.py

@@ -293,12 +293,16 @@ class pyOpenSSLAdapter(Adapter):
     ciphers = None
     """The ciphers list of TLS."""
 
+    private_key_password = None
+    """Optional passphrase for password protected private key."""
+
     def __init__(
         self,
         certificate,
         private_key,
         certificate_chain=None,
         ciphers=None,
+        private_key_password=None,
     ):
         """Initialize OpenSSL Adapter instance."""
         if SSL is None:
@@ -309,6 +313,7 @@ def __init__(
             private_key,
             certificate_chain,
             ciphers,
+            private_key_password,
         )
 
         self._environ = None
@@ -328,13 +333,28 @@ def wrap(self, sock):
         # closing so we can't reliably access protocol/client cert for the env
         return sock, self._environ.copy()
 
+    def _password_callback(
+        self,
+        _password_max_length,
+        _verify_twice,
+        password,
+    ):
+        """Pass a passphrase to password protected private key."""
+        if not password:
+            return b''
+        if not isinstance(password, bytes):
+            return password.encode('utf-8')
+        return password
+
     def get_context(self):
         """Return an ``SSL.Context`` from self attributes.
 
         Ref: :py:class:`SSL.Context <pyopenssl:OpenSSL.SSL.Context>`
         """
         # See https://code.activestate.com/recipes/442473/
         c = SSL.Context(SSL.SSLv23_METHOD)
+        if self.private_key_password is not None:
+            c.set_passwd_cb(self._password_callback, self.private_key_password)
         c.use_privatekey_file(self.private_key)
         if self.certificate_chain:
             c.load_verify_locations(self.certificate_chain)

cheroot/ssl/pyopenssl.pyi

@@ -31,9 +31,16 @@ class pyOpenSSLAdapter(Adapter):
         private_key,
         certificate_chain: Any | None = ...,
         ciphers: Any | None = ...,
+        private_key_password: str | bytes | None = ...,
     ) -> None: ...
     def bind(self, sock): ...
     def wrap(self, sock): ...
+    def _password_callback(
+        self,
+        _password_max_length: int,
+        _verify_twice: bool,
+        password: bytes | str,
+    ) -> bytes: ...
     def get_environ(self): ...
     def makefile(self, sock, mode: str = ..., bufsize: int = ...): ...
     def get_context(self) -> SSL.Context: ...

cheroot/test/test_ssl.py

@@ -4,18 +4,29 @@
 import http.client
 import json
 import os
+import secrets
 import ssl
+import string
 import subprocess
 import sys
 import threading
 import time
 import traceback
+import uuid
+from pathlib import Path
 
 import pytest
 
 import OpenSSL.SSL
 import requests
 import trustme
+from cryptography.hazmat.backends import default_backend
+from cryptography.hazmat.primitives.serialization import (
+    BestAvailableEncryption,
+    Encoding,
+    PrivateFormat,
+    load_pem_private_key,
+)
 
 from .._compat import (
     IS_ABOVE_OPENSSL10,
@@ -29,7 +40,7 @@
     ntob,
     ntou,
 )
-from ..server import HTTPServer, get_ssl_adapter_class
+from ..server import Gateway, HTTPServer, get_ssl_adapter_class
 from ..testing import (
     ANY_INTERFACE_IPV4,
     ANY_INTERFACE_IPV6,
@@ -118,6 +129,14 @@ def make_tls_http_server(bind_addr, ssl_adapter, request):
     return httpserver
 
 
+@pytest.fixture
+def private_key_password():
+    """Provide random 10 character password for private key."""
+    return ''.join(
+        secrets.choice(string.ascii_letters + string.digits) for _ in range(10)
+    )
+
+
 @pytest.fixture
 def tls_http_server(request):
     """Provision a server creator as a fixture."""
@@ -158,6 +177,36 @@ def tls_certificate_private_key_pem_path(tls_certificate):
         yield cert_key_pem
 
 
+@pytest.fixture
+def tls_certificate_passwd_private_key_pem_path(
+    tls_certificate,
+    private_key_password,
+):
+    """Provide a certificate private key PEM file path via fixture."""
+    key_as_bytes = tls_certificate.private_key_pem.bytes()
+    private_key_object = load_pem_private_key(
+        key_as_bytes,
+        password=None,
+        backend=default_backend(),
+    )
+    encrypted_key_as_bytes = private_key_object.private_bytes(
+        encoding=Encoding.PEM,
+        format=PrivateFormat.PKCS8,
+        encryption_algorithm=BestAvailableEncryption(
+            password=private_key_password.encode('utf-8'),
+        ),
+    )
+
+    with tls_certificate.private_key_pem.tempfile() as tf:
+        keyfile_temp_path = Path(tf)
+    keyfile_temp_path.write_bytes(encrypted_key_as_bytes)
+
+    yield keyfile_temp_path
+
+    # removes the temporary file in teardown
+    Path.unlink(keyfile_temp_path, missing_ok=True)
+
+
 def _thread_except_hook(exceptions, args):
     """Append uncaught exception ``args`` in threads to ``exceptions``."""
     if issubclass(args.exc_type, SystemExit):
@@ -726,3 +775,96 @@ def test_http_over_https_error(
         'The underlying error is {underlying_error!r}'.format(**locals())
     )
     assert expected_error_text in err_text
+
+
+@pytest.mark.parametrize(
+    'adapter_type',
+    (
+        'builtin',
+        'pyopenssl',
+    )
+    * 2,
+)
+def test_ssl_adapters_with_private_key_password(
+    private_key_password,
+    tls_certificate_chain_pem_path,
+    tls_certificate_passwd_private_key_pem_path,
+    adapter_type,
+):
+    """Check that server starts using ssl adapter with password-protected private key."""
+    httpserver = HTTPServer(
+        bind_addr=(ANY_INTERFACE_IPV4, EPHEMERAL_PORT),
+        gateway=Gateway,
+    )
+
+    tls_adapter_cls = get_ssl_adapter_class(name=adapter_type)
+    tls_adapter = tls_adapter_cls(
+        certificate=tls_certificate_chain_pem_path,
+        private_key=tls_certificate_passwd_private_key_pem_path,
+        private_key_password=secrets.choice(
+            [
+                private_key_password,
+                private_key_password.encode('utf-8'),
+            ],
+        ),
+    )
+
+    httpserver.ssl_adapter = tls_adapter
+    httpserver.prepare()
+
+    assert httpserver.ready
+    assert httpserver.requests._threads
+    for thr in httpserver.requests._threads:
+        assert thr.ready
+
+    httpserver.stop()
+
+
+@pytest.mark.parametrize(
+    'adapter_type',
+    ('builtin',),
+)
+def test_builtin_adapter_with_false_key_password(
+    tls_certificate_chain_pem_path,
+    tls_certificate_passwd_private_key_pem_path,
+    adapter_type,
+):
+    """Check that builtin ssl-adapter initialization fails when wrong private key password given."""
+    tls_adapter_cls = get_ssl_adapter_class(name=adapter_type)
+    with pytest.raises(ssl.SSLError, match=r'\[SSL\] PEM.+'):
+        tls_adapter_cls(
+            certificate=tls_certificate_chain_pem_path,
+            private_key=tls_certificate_passwd_private_key_pem_path,
+            private_key_password=str(uuid.uuid4()),
+        )
+
+
+@pytest.mark.parametrize(
+    'adapter_type',
+    ('pyopenssl',),
+)
+def test_openssl_adapter_with_false_key_password(
+    tls_certificate_chain_pem_path,
+    tls_certificate_passwd_private_key_pem_path,
+    adapter_type,
+):
+    """Check that server init fails when wrong private key password given."""
+    httpserver = HTTPServer(
+        bind_addr=(ANY_INTERFACE_IPV4, EPHEMERAL_PORT),
+        gateway=Gateway,
+    )
+
+    tls_adapter_cls = get_ssl_adapter_class(name=adapter_type)
+    tls_adapter = tls_adapter_cls(
+        certificate=tls_certificate_chain_pem_path,
+        private_key=tls_certificate_passwd_private_key_pem_path,
+        private_key_password=str(uuid.uuid4()),
+    )
+
+    httpserver.ssl_adapter = tls_adapter
+
+    with pytest.raises(OpenSSL.SSL.Error, match=r'.+bad decrypt.+'):
+        httpserver.prepare()
+
+    assert not httpserver.requests._threads
+    assert not httpserver.ready