diff --git a/SSH_HOST_KEYS.md b/SSH_HOST_KEYS.md
new file mode 100644
index 0000000..ad6e8ae
--- /dev/null
+++ b/SSH_HOST_KEYS.md
@@ -0,0 +1,62 @@
+# Trusted SSH host keys
+
+SSH connections reject unknown or changed host keys before password/public-key
+authentication. There is no automatic first-use enrollment. Existing SSH user
+keys and passwords are still supported; server host trust is a separate check.
+
+**Upgrade prerequisite:** enroll every managed SSH endpoint before deploying a
+release containing this change. Otherwise operations against unenrolled servers
+fail closed. This is deliberate; do not restore `AutoAddPolicy` as a workaround.
+
+Paramiko loads the panel process user's `~/.ssh/known_hosts` and a persistent
+panel trust file, defaulting to `ssh_known_hosts` beside `DATA_FILE` (normally
+`/app/data/ssh_known_hosts` in Docker). Set `SSH_KNOWN_HOSTS_FILE` to use another
+file. An explicitly configured missing, unreadable or malformed file is an error.
+Bind-mount an operator-managed file read-only, or place it in the existing
+persistent data volume; do not rely on an ephemeral container layer. Restrict
+write access to trusted operators and ensure the panel user can read it.
+
+## Initial enrollment
+
+1. Obtain the endpoint's public host key and SHA256 fingerprint using an
+   independent trusted channel, such as the VPS provider console or an existing
+   independently verified administrative SSH session. On the endpoint:
+
+   ```sh
+   ssh-keygen -lf /etc/ssh/ssh_host_ed25519_key.pub -E sha256
+   cat /etc/ssh/ssh_host_ed25519_key.pub
+   ```
+
+   These are **public host keys**, never client private keys.
+2. Confirm the fingerprint independently. `ssh-keyscan` can collect a candidate
+   but cannot authenticate it: never pipe its output directly into trusted
+   known_hosts without comparing fingerprints through the independent channel.
+3. Add an OpenSSH known_hosts line to a private candidate trust file:
+
+   ```text
+   node.example.test ssh-ed25519 <verified-public-key-base64>
+   [node.example.test]:2222 ssh-ed25519 <verified-public-key-base64>
+   ```
+
+   Use the exact configured SSH hostname/IP and port. A non-default port requires
+   `[host]:port`. Enroll all aliases actually used by the panel. Hostnames do not
+   replace fingerprint verification; DNS changes cannot enroll another key.
+4. Validate with `ssh-keygen -lf <candidate> -E sha256`, preserve a backup, and
+   install the verified file atomically with restricted permissions. Mount or
+   configure it for the panel, then test a read-only SSH operation.
+
+## Controlled rotation and failures
+
+A changed key is rejected; do not approve it based on the error message alone.
+Verify the new fingerprint independently, schedule rotation, back up the trust
+file, replace the relevant verified entry atomically, and reconnect/restart the
+panel during the agreed window. Existing pooled SSH transports keep the host
+identity verified at their original handshake until disconnected; changing the
+file does not revoke an already-open transport. Paramiko reloads trust on each
+new handshake. Remove stale entries from both system and panel trust stores;
+system entries can take precedence over the panel file.
+
+If verification fails or the endpoint is unavailable, the operation fails and
+the normal connection backoff applies. No trust file is rewritten by the panel.
+Restore only a previously verified backup when rolling back trust configuration.
+No VPN configuration, host key or user SSH key is regenerated by this feature.
diff --git a/managers/ssh_manager.py b/managers/ssh_manager.py
index 911485c..4f0c803 100644
--- a/managers/ssh_manager.py
+++ b/managers/ssh_manager.py
@@ -9,6 +9,7 @@ import time
 import threading
 import logging
 import shlex
+import os
 from contextlib import nullcontext
 
 logger = logging.getLogger(__name__)
@@ -97,6 +98,13 @@ class SSHManager:
                             f"SSH connect to {self.host} attempt {attempt} "
                             f"failed: {e}")
                         self._disconnect_locked()
+                    except Exception:
+                        # Host-key/authentication/configuration failures are
+                        # not transient TCP failures: never retry or accept a
+                        # different key, and do not leave a half-open client.
+                        self._disconnect_locked()
+                        self._record_connect_failure()
+                        raise
                 if last_exc is not None:
                     self._record_connect_failure()
                     raise last_exc
@@ -121,7 +129,17 @@ class SSHManager:
     def _connect_once(self):
         """Single TCP+SSH handshake attempt (caller holds _conn_lock)."""
         self.client = paramiko.SSHClient()
-        self.client.set_missing_host_key_policy(paramiko.AutoAddPolicy())
+        self.client.load_system_host_keys()
+        # Persist trust alongside DATA_FILE (the standard Docker data volume).
+        # Explicitly configured files must exist and parse successfully; never
+        # fall back to trust-on-first-use when trust configuration is broken.
+        configured = os.environ.get('SSH_KNOWN_HOSTS_FILE')
+        known_hosts = configured or os.path.join(
+            os.path.dirname(os.environ.get('DATA_FILE', '/app/data/data.json')),
+            'ssh_known_hosts')
+        if configured or os.path.isfile(known_hosts):
+            self.client.load_host_keys(known_hosts)
+        self.client.set_missing_host_key_policy(paramiko.RejectPolicy())
 
         kwargs = {
             'hostname': self.host,
diff --git a/tests/test_ssh_host_keys.py b/tests/test_ssh_host_keys.py
new file mode 100644
index 0000000..5d4a7c4
--- /dev/null
+++ b/tests/test_ssh_host_keys.py
@@ -0,0 +1,196 @@
+"""Real loopback SSH handshakes: host trust must precede authentication."""
+import contextlib
+import io
+import os
+import socket
+import tempfile
+import threading
+import unittest
+from pathlib import Path
+from unittest import mock
+
+import paramiko
+
+from managers.ssh_manager import SSHManager
+
+
+class Server(paramiko.ServerInterface):
+    def __init__(self):
+        self.auth_calls = 0
+
+    def check_auth_password(self, username, password):
+        self.auth_calls += 1
+        return paramiko.AUTH_SUCCESSFUL
+
+    def check_auth_publickey(self, username, key):
+        self.auth_calls += 1
+        return paramiko.AUTH_SUCCESSFUL
+
+
+@contextlib.contextmanager
+def endpoint(key):
+    listener = socket.socket()
+    listener.bind(('127.0.0.1', 0))
+    listener.listen(1)
+    listener.settimeout(4)
+    port = listener.getsockname()[1]
+    server = Server()
+    stop = threading.Event()
+    transports = []
+
+    def serve():
+        try:
+            conn, _ = listener.accept()
+            transport = paramiko.Transport(conn)
+            transports.append(transport)
+            transport.add_server_key(key)
+            transport.start_server(server=server)
+            stop.wait(4)
+        except (OSError, EOFError, paramiko.SSHException):
+            pass  # A rejected host key intentionally closes the transport.
+        finally:
+            for transport in transports:
+                transport.close()
+
+    thread = threading.Thread(target=serve, daemon=True)
+    thread.start()
+    try:
+        yield port, server
+    finally:
+        stop.set()
+        listener.close()
+        for transport in transports:
+            transport.close()
+        thread.join(5)
+        if thread.is_alive():
+            raise AssertionError('loopback SSH fixture did not stop')
+
+
+class HostKeyTests(unittest.TestCase):
+    @classmethod
+    def setUpClass(cls):
+        cls.key = paramiko.ECDSAKey.generate()
+        cls.other = paramiko.ECDSAKey.generate()
+
+    def setUp(self):
+        self.tmp = tempfile.TemporaryDirectory()
+        self.addCleanup(self.tmp.cleanup)
+        self.store = Path(self.tmp.name) / 'known_hosts'
+        self.store.write_text('')
+        self.patch = mock.patch.dict(os.environ, {'SSH_KNOWN_HOSTS_FILE': str(self.store)})
+        self.patch.start()
+        self.addCleanup(self.patch.stop)
+
+    def trust(self, port, key):
+        keys = paramiko.HostKeys()
+        keys.add('[127.0.0.1]:%d' % port, key.get_name(), key)
+        keys.save(str(self.store))
+
+    def manager(self, port, **kwargs):
+        return SSHManager('127.0.0.1', port, 'synthetic', **kwargs)
+
+    def test_unknown_key_rejected_before_auth_and_store_unchanged(self):
+        before = self.store.read_bytes()
+        with endpoint(self.key) as (port, server):
+            manager = self.manager(port, password='synthetic')
+            with self.assertRaises(paramiko.SSHException):
+                manager.connect()
+            self.assertEqual(server.auth_calls, 0)
+            self.assertIsNone(manager.client)
+            self.assertEqual(manager._connect_fail_count, 1)
+        self.assertEqual(self.store.read_bytes(), before)
+
+    def test_trusted_key_accepts_password_auth(self):
+        with endpoint(self.key) as (port, server):
+            self.trust(port, self.key)
+            manager = self.manager(port, password='synthetic')
+            try:
+                self.assertTrue(manager.connect())
+                self.assertGreater(server.auth_calls, 0)
+            finally:
+                manager.force_disconnect()
+
+    def test_trusted_key_accepts_private_key_auth(self):
+        auth_key = paramiko.RSAKey.generate(1024)
+        buffer = io.StringIO()
+        auth_key.write_private_key(buffer)
+        with endpoint(self.key) as (port, server):
+            self.trust(port, self.key)
+            manager = self.manager(port, private_key=buffer.getvalue())
+            try:
+                self.assertTrue(manager.connect())
+                self.assertGreater(server.auth_calls, 0)
+            finally:
+                manager.force_disconnect()
+
+    def test_mitm_changed_key_rejected_before_auth(self):
+        with endpoint(self.other) as (port, server):
+            self.trust(port, self.key)
+            before = self.store.read_bytes()
+            manager = self.manager(port, password='synthetic')
+            with self.assertRaises(paramiko.BadHostKeyException):
+                manager.connect()
+            self.assertEqual(server.auth_calls, 0)
+            self.assertIsNone(manager.client)
+            self.assertEqual(self.store.read_bytes(), before)
+
+    def test_rotation_requires_explicit_trusted_store_update(self):
+        with endpoint(self.other) as (port, server):
+            self.trust(port, self.key)
+            manager = self.manager(port, password='synthetic')
+            with self.assertRaises(paramiko.BadHostKeyException):
+                manager.connect()
+            self.assertEqual(server.auth_calls, 0)
+        # Operator independently verifies the replacement key before this edit.
+        with endpoint(self.other) as (port, server):
+            self.trust(port, self.other)
+            manager = self.manager(port, password='synthetic')
+            try:
+                self.assertTrue(manager.connect())
+                self.assertGreater(server.auth_calls, 0)
+            finally:
+                manager.force_disconnect()
+
+    def test_missing_explicit_store_fails_before_connect(self):
+        self.store.unlink()
+        with mock.patch.object(paramiko.SSHClient, 'connect') as connect:
+            with self.assertRaises(OSError):
+                self.manager(2222, password='synthetic').connect()
+            connect.assert_not_called()
+
+    def test_malformed_store_fails_before_connect(self):
+        self.store.write_text('synthetic ssh-rsa not-valid-base64!!!\n')
+        with mock.patch.object(paramiko.SSHClient, 'connect') as connect:
+            with self.assertRaises(paramiko.hostkeys.InvalidHostKey):
+                self.manager(2222, password='synthetic').connect()
+            connect.assert_not_called()
+
+    def test_system_trust_and_persistent_default_are_loaded(self):
+        data = Path(self.tmp.name) / 'data.json'
+        default_store = data.with_name('ssh_known_hosts')
+        default_store.write_text('')
+        with mock.patch.dict(os.environ, {'DATA_FILE': str(data)}), \
+                mock.patch.dict(os.environ) as env, \
+                mock.patch('managers.ssh_manager.paramiko.SSHClient') as client:
+            env.pop('SSH_KNOWN_HOSTS_FILE', None)
+            manager = self.manager(2222, password='synthetic')
+            manager.connect()
+            client.return_value.load_system_host_keys.assert_called_once_with()
+            client.return_value.load_host_keys.assert_called_once_with(str(default_store))
+            policy = client.return_value.set_missing_host_key_policy.call_args.args[0]
+            self.assertIsInstance(policy, paramiko.RejectPolicy)
+            manager.force_disconnect()
+
+    def test_connection_failure_keeps_store(self):
+        before = self.store.read_bytes()
+        with mock.patch.object(paramiko.SSHClient, 'connect', side_effect=OSError('synthetic reset')):
+            manager = self.manager(2222, password='synthetic')
+            with self.assertRaises(OSError):
+                manager.connect()
+            self.assertIsNone(manager.client)
+            self.assertEqual(manager._connect_fail_count, 1)
+        self.assertEqual(self.store.read_bytes(), before)
+
+
+if __name__ == '__main__':
+    unittest.main()
diff --git a/tests/test_ssh_transport_race.py b/tests/test_ssh_transport_race.py
index e4454ac..337881d 100644
--- a/tests/test_ssh_transport_race.py
+++ b/tests/test_ssh_transport_race.py
@@ -37,6 +37,12 @@ class _FakeClient:
     def set_missing_host_key_policy(self, policy):
         pass
 
+    def load_system_host_keys(self):
+        pass
+
+    def load_host_keys(self, filename):
+        pass
+
     def connect(self, **kwargs):
         pass