diff --git a/asyncssh/connection.py b/asyncssh/connection.py
index 6d56bb47..c49d0658 100644
--- a/asyncssh/connection.py
+++ b/asyncssh/connection.py
@@ -21,6 +21,7 @@
 """SSH connection handlers"""
 
 import asyncio
+import copy
 import functools
 import getpass
 import inspect
@@ -3749,7 +3750,7 @@ async def host_based_auth_requested(self) -> \
                 self._saved_rsa_key = None
             else:
                 try:
-                    key = self._client_host_keys.pop(0)
+                    key = copy.copy(self._client_host_keys.pop(0))
                 except IndexError:
                     key = None
                     break
@@ -3843,7 +3844,7 @@ async def public_key_auth_requested(self) -> Optional[SSHKeyPair]:
                 key.algorithm = key.sig_algorithm + b'-cert-v01@openssh.com'
                 self._saved_rsa_key = None
             else:
-                key = self._client_keys.pop(0)
+                key = copy.copy(self._client_keys.pop(0))
 
             if self._choose_signature_alg(key):
                 if key.algorithm == b'ssh-rsa-cert-v01@openssh.com' and \
@@ -5991,6 +5992,9 @@ def choose_server_host_key(self,
             keypair = self._server_host_keys.get(alg)
             if keypair:
                 if alg != keypair.algorithm:
+                    # Server host keys are shared by all connections on
+                    # a listener, so set the algorithm on a copy
+                    keypair = copy.copy(keypair)
                     keypair.set_sig_algorithm(alg)
 
                 self._server_host_key = keypair
diff --git a/tests/test_connection.py b/tests/test_connection.py
index adbe75d1..5b8ebcd0 100644
--- a/tests/test_connection.py
+++ b/tests/test_connection.py
@@ -631,6 +631,35 @@ async def test_reserved_server_host_keys(self):
                                send_server_host_keys=True):
             pass
 
+    @asynctest
+    async def test_server_host_key_signature_alg_per_connection(self):
+        """Test connections on a listener using different RSA signature algs"""
+
+        server_conns = []
+
+        class _RecordingServer(Server):
+            """Server which records its connections"""
+
+            def connection_made(self, conn):
+                """Record the server connection"""
+
+                super().connection_made(conn)
+                server_conns.append(conn)
+
+        async with self.listen(server_factory=_RecordingServer) as server:
+            port = server.get_port()
+
+            async with self.connect(port=port,
+                                    known_hosts=(['skey.pub'], [], []),
+                                    server_host_key_algs=['rsa-sha2-256']):
+                async with self.connect(port=port,
+                                        known_hosts=(['skey.pub'], [], []),
+                                        server_host_key_algs=['rsa-sha2-512']):
+                    self.assertEqual(
+                        [conn.get_server_host_key().sig_algorithm
+                         for conn in server_conns],
+                        [b'rsa-sha2-256', b'rsa-sha2-512'])
+
     @asynctest
     async def test_get_server_host_key(self):
         """Test retrieving a server host key"""
diff --git a/tests/test_connection_auth.py b/tests/test_connection_auth.py
index c7e585b7..42ff8f1c 100644
--- a/tests/test_connection_auth.py
+++ b/tests/test_connection_auth.py
@@ -785,6 +785,19 @@ async def test_client_host_signature_algs(self):
                                     signature_algs=[alg]):
                 pass
 
+    @asynctest
+    async def test_shared_client_host_key_signature_alg(self):
+        """Test that signing doesn't change a shared client host key"""
+
+        skey = asyncssh.load_keypairs('skey')[0]
+        sig_algorithm = skey.sig_algorithm
+
+        for alg in ('rsa-sha2-256', 'rsa-sha2-512'):
+            async with self.connect(username='user', client_host_keys=[skey],
+                                    client_username='user',
+                                    signature_algs=[alg]):
+                self.assertEqual(skey.sig_algorithm, sig_algorithm)
+
     @asynctest
     async def test_no_server_signature_algs(self):
         """Test a server which doesn't advertise signature algorithms"""
@@ -1362,6 +1375,18 @@ async def test_public_key_signature_algs(self):
                                     client_keys='ckey', signature_algs=[alg]):
                 pass
 
+    @asynctest
+    async def test_shared_client_key_signature_alg(self):
+        """Test that signing doesn't change a shared client key"""
+
+        ckey = asyncssh.load_keypairs('ckey')[0]
+        sig_algorithm = ckey.sig_algorithm
+
+        for alg in ('rsa-sha2-256', 'rsa-sha2-512'):
+            async with self.connect(username='ckey', agent_path=None,
+                                    client_keys=[ckey], signature_algs=[alg]):
+                self.assertEqual(ckey.sig_algorithm, sig_algorithm)
+
     @asynctest
     async def test_no_server_signature_algs(self):
         """Test a server which doesn't advertise signature algorithms"""