From 2cd7a49e683f1ea579ca191b40375d50821b7751 Mon Sep 17 00:00:00 2001
From: Maximilian Hils <git@maximilianhils.com>
Date: Wed, 11 May 2022 15:39:35 +0200
Subject: [PATCH] add `Connection.use_(certificate|privatekey)`

---
 CHANGELOG.rst      |  3 +++
 src/OpenSSL/SSL.py | 32 ++++++++++++++++++++++++++++++++
 tests/test_ssl.py  | 27 +++++++++++++++++++++++++++
 3 files changed, 62 insertions(+)

diff --git a/CHANGELOG.rst b/CHANGELOG.rst
index 1f7b06313..6c879cad3 100644
--- a/CHANGELOG.rst
+++ b/CHANGELOG.rst
@@ -18,6 +18,9 @@ Deprecations:
 Changes:
 ^^^^^^^^
 
+- 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)
 -------------------
 
diff --git a/src/OpenSSL/SSL.py b/src/OpenSSL/SSL.py
index 3e6ee1b34..403e9002a 100644
--- a/src/OpenSSL/SSL.py
+++ b/src/OpenSSL/SSL.py
@@ -948,6 +948,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")
 
@@ -1009,6 +1010,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")
 
@@ -1727,6 +1729,36 @@ def get_servername(self):
 
         return _ffi.string(name)
 
+    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
diff --git a/tests/test_ssl.py b/tests/test_ssl.py
index a02dc4bf4..452f74882 100644
--- a/tests/test_ssl.py
+++ b/tests/test_ssl.py
@@ -613,6 +613,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)
@@ -697,6 +698,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
@@ -2194,6 +2196,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):
         """