add Connection.use_(certificate|privatekey) (#1121)

* add `Connection.use_(certificate|privatekey)`

* bump minimum cryptography version

* deduplicate tests

* black!

* max line length

Author: mhils

CHANGELOG.rst

@@ -29,6 +29,8 @@ Changes:
 - Add ``OpenSSL.SSL.Connection.set_verify`` and ``OpenSSL.SSL.Connection.get_verify_mode``
   to override the context object's verification flags.
   `#1073 <https://github.com/pyca/pyopenssl/pull/1073>`_
+- Add ``OpenSSL.SSL.Connection.use_certificate`` and ``OpenSSL.SSL.Connection.use_privatekey``
+  to set a certificate per connection (and not just per context) `#1121 <https://github.com/pyca/pyopenssl/pull/1121>`_.
 
 22.0.0 (2022-01-29)
 -------------------

setup.py

@@ -96,7 +96,7 @@ def find_meta(meta):
         package_dir={"": "src"},
         install_requires=[
             # Fix cryptographyMinimum in tox.ini when changing this!
-            "cryptography>=37.0.2,<39",
+            "cryptography>=38.0.0,<39",
         ],
         extras_require={
             "test": ["flaky", "pretend", "pytest>=3.0.1"],

src/OpenSSL/SSL.py

@@ -962,6 +962,7 @@ def use_certificate(self, cert):
         :param cert: The X509 object
         :return: None
         """
+        # Mirrored at Connection.use_certificate
         if not isinstance(cert, X509):
             raise TypeError("cert must be an X509 instance")
 
@@ -1023,6 +1024,7 @@ def use_privatekey(self, pkey):
         :param pkey: The PKey object
         :return: None
         """
+        # Mirrored at Connection.use_privatekey
         if not isinstance(pkey, PKey):
             raise TypeError("pkey must be a PKey instance")
 
@@ -1788,6 +1790,36 @@ def get_verify_mode(self):
         """
         return _lib.SSL_get_verify_mode(self._ssl)
 
+    def use_certificate(self, cert):
+        """
+        Load a certificate from a X509 object
+
+        :param cert: The X509 object
+        :return: None
+        """
+        # Mirrored from Context.use_certificate
+        if not isinstance(cert, X509):
+            raise TypeError("cert must be an X509 instance")
+
+        use_result = _lib.SSL_use_certificate(self._ssl, cert._x509)
+        if not use_result:
+            _raise_current_error()
+
+    def use_privatekey(self, pkey):
+        """
+        Load a private key from a PKey object
+
+        :param pkey: The PKey object
+        :return: None
+        """
+        # Mirrored from Context.use_privatekey
+        if not isinstance(pkey, PKey):
+            raise TypeError("pkey must be a PKey instance")
+
+        use_result = _lib.SSL_use_PrivateKey(self._ssl, pkey._pkey)
+        if not use_result:
+            self._context._raise_passphrase_exception()
+
     def set_ciphertext_mtu(self, mtu):
         """
         For DTLS, set the maximum UDP payload size (*not* including IP/UDP

tests/test_ssl.py

@@ -29,6 +29,7 @@
     socket,
 )
 from sys import getfilesystemencoding, platform
+from typing import Union
 from warnings import simplefilter
 from weakref import ref
 
@@ -621,17 +622,6 @@ def test_type(self):
         """
         assert is_consistent_type(Context, "Context", TLSv1_METHOD)
 
-    def test_use_privatekey(self):
-        """
-        `Context.use_privatekey` takes an `OpenSSL.crypto.PKey` instance.
-        """
-        key = PKey()
-        key.generate_key(TYPE_RSA, 1024)
-        ctx = Context(SSLv23_METHOD)
-        ctx.use_privatekey(key)
-        with pytest.raises(TypeError):
-            ctx.use_privatekey("")
-
     def test_use_privatekey_file_missing(self, tmpfile):
         """
         `Context.use_privatekey_file` raises `OpenSSL.SSL.Error` when passed
@@ -685,37 +675,6 @@ def test_use_privatekey_file_unicode(self, tmpfile):
             FILETYPE_PEM,
         )
 
-    def test_use_certificate_wrong_args(self):
-        """
-        `Context.use_certificate_wrong_args` raises `TypeError` when not passed
-        exactly one `OpenSSL.crypto.X509` instance as an argument.
-        """
-        ctx = Context(SSLv23_METHOD)
-        with pytest.raises(TypeError):
-            ctx.use_certificate("hello, world")
-
-    def test_use_certificate_uninitialized(self):
-        """
-        `Context.use_certificate` raises `OpenSSL.SSL.Error` when passed a
-        `OpenSSL.crypto.X509` instance which has not been initialized
-        (ie, which does not actually have any certificate data).
-        """
-        ctx = Context(SSLv23_METHOD)
-        with pytest.raises(Error):
-            ctx.use_certificate(X509())
-
-    def test_use_certificate(self):
-        """
-        `Context.use_certificate` sets the certificate which will be
-        used to identify connections created using the context.
-        """
-        # TODO
-        # Hard to assert anything.  But we could set a privatekey then ask
-        # OpenSSL if the cert and key agree using check_privatekey.  Then as
-        # long as check_privatekey works right we're good...
-        ctx = Context(SSLv23_METHOD)
-        ctx.use_certificate(load_certificate(FILETYPE_PEM, root_cert_pem))
-
     def test_use_certificate_file_wrong_args(self):
         """
         `Context.use_certificate_file` raises `TypeError` if the first
@@ -2180,6 +2139,76 @@ def test_construction(self):
         assert isinstance(new_session, Session)
 
 
+@pytest.fixture(params=["context", "connection"])
+def ctx_or_conn(request) -> Union[Context, Connection]:
+    ctx = Context(SSLv23_METHOD)
+    if request.param == "context":
+        return ctx
+    else:
+        return Connection(ctx, None)
+
+
+class TestContextConnection:
+    """
+    Unit test for methods that are exposed both by Connection and Context
+    objects.
+    """
+
+    def test_use_privatekey(self, ctx_or_conn):
+        """
+        `use_privatekey` takes an `OpenSSL.crypto.PKey` instance.
+        """
+        key = PKey()
+        key.generate_key(TYPE_RSA, 1024)
+
+        ctx_or_conn.use_privatekey(key)
+        with pytest.raises(TypeError):
+            ctx_or_conn.use_privatekey("")
+
+    def test_use_privatekey_wrong_key(self, ctx_or_conn):
+        """
+        `use_privatekey` raises `OpenSSL.SSL.Error` when passed a
+        `OpenSSL.crypto.PKey` instance which has not been initialized.
+        """
+        key = PKey()
+        key.generate_key(TYPE_RSA, 1024)
+        ctx_or_conn.use_certificate(
+            load_certificate(FILETYPE_PEM, root_cert_pem)
+        )
+        with pytest.raises(Error):
+            ctx_or_conn.use_privatekey(key)
+
+    def test_use_certificate(self, ctx_or_conn):
+        """
+        `use_certificate` sets the certificate which will be
+        used to identify connections created using the context.
+        """
+        # TODO
+        # Hard to assert anything.  But we could set a privatekey then ask
+        # OpenSSL if the cert and key agree using check_privatekey.  Then as
+        # long as check_privatekey works right we're good...
+        ctx_or_conn.use_certificate(
+            load_certificate(FILETYPE_PEM, root_cert_pem)
+        )
+
+    def test_use_certificate_wrong_args(self, ctx_or_conn):
+        """
+        `use_certificate_wrong_args` raises `TypeError` when not passed
+        exactly one `OpenSSL.crypto.X509` instance as an argument.
+        """
+        with pytest.raises(TypeError):
+            ctx_or_conn.use_certificate("hello, world")
+
+    def test_use_certificate_uninitialized(self, ctx_or_conn):
+        """
+        `use_certificate` raises `OpenSSL.SSL.Error` when passed a
+        `OpenSSL.crypto.X509` instance which has not been initialized
+        (ie, which does not actually have any certificate data).
+        """
+        with pytest.raises(Error):
+            ctx_or_conn.use_certificate(X509())
+
+
 class TestConnection:
     """
     Unit tests for `OpenSSL.SSL.Connection`.

tox.ini

@@ -10,7 +10,7 @@ extras =
 deps =
     coverage>=4.2
     cryptographyMain: git+https://github.com/pyca/cryptography.git
-    cryptographyMinimum: cryptography==37.0.2
+    cryptographyMinimum: cryptography==38.0.0
     randomorder: pytest-randomly
 setenv =
     # Do not allow the executing environment to pollute the test environment