add Connection.use_(certificate|privatekey)

CHANGELOG.rst

@@ -25,6 +25,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)
 -------------------

src/OpenSSL/SSL.py

@@ -954,6 +954,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")
 
@@ -1015,6 +1016,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")
 
@@ -1780,6 +1782,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

@@ -625,6 +625,7 @@ def test_use_privatekey(self):
         """
         `Context.use_privatekey` takes an `OpenSSL.crypto.PKey` instance.
         """
+        # Mirrored at TestConnection.test_use_privatekey
         key = PKey()
         key.generate_key(TYPE_RSA, 1024)
         ctx = Context(SSLv23_METHOD)
@@ -709,6 +710,7 @@ def test_use_certificate(self):
         `Context.use_certificate` sets the certificate which will be
         used to identify connections created using the context.
         """
+        # Mirrored at TestConnection.test_use_certificate
         # 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
@@ -2206,6 +2208,31 @@ def test_type(self):
         ctx = Context(SSLv23_METHOD)
         assert is_consistent_type(Connection, "Connection", ctx, None)
 
+    def test_use_privatekey(self):
+        """
+        `Connection.use_privatekey` takes an `OpenSSL.crypto.PKey` instance.
+        """
+        # Mirrored from TestContext.test_use_privatekey
+        key = PKey()
+        key.generate_key(TYPE_RSA, 1024)
+        ctx = Context(SSLv23_METHOD)
+        connection = Connection(ctx, None)
+        connection.use_privatekey(key)
+        with pytest.raises(TypeError):
+            connection.use_privatekey("")
+
+    def test_use_certificate(self):
+        """
+        `Connection.use_certificate` sets the certificate which will be
+        used to identify connections created using the context.
+        """
+        # Mirrored from TestContext.test_use_certificate
+        ctx = Context(SSLv23_METHOD)
+        connection = Connection(ctx, None)
+        connection.use_certificate(
+            load_certificate(FILETYPE_PEM, root_cert_pem)
+        )
+
     @pytest.mark.parametrize("bad_context", [object(), "context", None, 1])
     def test_wrong_args(self, bad_context):
         """