1.7.8

SSH_HOST_KEYS.md

@@ -1,12 +1,64 @@
 # Trusted SSH host keys
 
+## v1.7.9 — web enrollment and migration
+
+Administrators can open a server page and use **SSH host key → Inspect SSH key**,
+including when statistics/protocol loading fails. Unknown and changed keys have
+distinct warnings; the server list no longer equates TCP reachability with trusted
+SSH. Inspection performs SSH key exchange only: no username, password, private
+key or agent authentication is sent.
+
+Server-list refreshes check TCP reachability and local pin presence only, not
+live SSH identity (`ssh_verified: false`); use **Inspect SSH key** to check the
+current fingerprint. Refreshing a list never starts an extra SSH key exchange.
+Inspection uses the same pinned-algorithm preference as Paramiko SSHClient,
+so an existing RSA pin is not falsely reported changed merely because the
+server also offers Ed25519 or ECDSA.
+
+Compare the displayed algorithm and SHA256 fingerprint with an independent VPS
+console (use the matching `/etc/ssh/ssh_host_*_key.pub`, not necessarily Ed25519).
+For replacement, investigate why the old fingerprint changed. Tick the explicit
+verification checkbox, then **Trust key and retry**. The page reloads and retries
+normal operations with a fresh connection and reset cooldown. No VPN clients,
+VPN keys or configurations are changed.
+
+Confirmation is signed, expires after five minutes, binds the server UID,
+host, port and exact candidate key, and re-probes before saving. Restarting the
+panel invalidates outstanding confirmations. A changed candidate or endpoint
+requires a fresh inspection. Only admins (not support users) may use these POST
+endpoints; session/Bearer access controls and same-origin browser checks apply.
+
+Confirmations also bind the previous trust entries: a concurrent administrator
+change requires re-inspection rather than silently overwriting that decision.
+Endpoint edits are serialized with confirmation in the panel process. Modern
+browsers' same-origin Fetch Metadata permits TLS-terminating reverse proxies;
+older browsers require the proxy to preserve the external request origin.
+
+The default file is `ssh_known_hosts` beside the resolved `DATA_FILE`: beside
+`app.py` for source installs, beside the executable for frozen installs, and
+`/app/data/ssh_known_hosts` for the supplied Docker Compose deployment. Override
+with `SSH_KNOWN_HOSTS_FILE`. A missing file can be created by enrollment; its
+parent must be writable. Read-only mounts/files produce an explicit error; make
+the persistent trust location writable if web enrollment is desired.
+
+Updates are serialized within the panel process and use atomic file replacement.
+Other endpoints (including aliases on shared lines) are preserved; hashed entries
+and nonstandard ports are supported. Conflicting system `~/.ssh/known_hosts`
+entries are never silently overridden: correct them independently as the panel
+OS user, then inspect again. Run one panel writer per trust file; external editors
+or multiple processes are not coordinated by the in-process lock.
+
+Existing manually enrolled entries keep working. The manual procedure below is
+still supported for read-only/operator-managed trust. `RejectPolicy` remains in
+force; there is no automatic trust-on-first-use fallback.
+
 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.
+**Upgrade prerequisite:** enroll every managed SSH endpoint manually before upgrade
+or use the administrator recovery controls after upgrade. Until enrollment,
+operations fail closed. 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.py

@@ -32,7 +32,7 @@ from fastapi.staticfiles import StaticFiles
 from fastapi.templating import Jinja2Templates
 from fastapi import FastAPI, Request, Query, UploadFile, File, Form
 from starlette.middleware.sessions import SessionMiddleware
-from pydantic import BaseModel, Field
+from pydantic import BaseModel, Field, StrictBool, constr
 from typing import Optional, List, Dict
 import uvicorn
 import httpx
@@ -43,6 +43,7 @@ except ImportError:
     CaptchaGenerator = None
 
 from managers.ssh_manager import SSHManager
+from managers import ssh_trust
 from managers.awg_manager import AWGManager, normalize_special_junk
 from managers.xray_manager import XrayManager
 from managers.wireguard_manager import WireGuardManager
@@ -201,6 +202,7 @@ else:
 DATA_FILE = os.path.abspath(os.path.expanduser(
     os.environ.get('DATA_FILE') or os.path.join(application_path, 'data.json')
 ))
+ssh_trust.configure(DATA_FILE)
 CURRENT_VERSION = "v1.7.8"
 
 # Custom protocol instance names: the rename modal caps input at 64 chars.
@@ -3143,6 +3145,63 @@ def _check_admin(request):
     return None
 
 
+class SSHTrustConfirmation(BaseModel):
+    token: constr(min_length=1, max_length=4096)
+    confirmed: StrictBool
+
+
+def _ssh_trust_admin(request):
+    # Keep session/Bearer authorization; reject cross-origin browser writes.
+    origin = request.headers.get('origin')
+    fetch_site = request.headers.get('sec-fetch-site')
+    # Modern browsers supply unforgeable-by-JS same-origin metadata even
+    # behind TLS-terminating proxies. Do not trust arbitrary Forwarded headers.
+    if origin and fetch_site != 'same-origin' and origin.rstrip('/') != str(request.base_url).rstrip('/'):
+        return False
+    if fetch_site in ('cross-site', 'same-site'):
+        return False
+    user = _check_admin(request)
+    return bool(user and user['role'] == 'admin')
+
+
+@app.post('/api/servers/{server_id}/ssh-trust/inspect', tags=["Servers"])
+def api_ssh_trust_inspect(request: Request, server_id: int):
+    if not _ssh_trust_admin(request):
+        return JSONResponse({'error': 'Forbidden'}, status_code=403)
+    servers = load_data()['servers']
+    if not 0 <= server_id < len(servers):
+        return JSONResponse({'error': 'Server not found'}, status_code=404)
+    try:
+        return ssh_trust.inspect(servers[server_id])
+    except Exception as exc:
+        return JSONResponse({'error': str(exc)}, status_code=409)
+
+
+@app.post('/api/servers/{server_id}/ssh-trust/confirm', tags=["Servers"])
+def api_ssh_trust_confirm(request: Request, server_id: int, req: SSHTrustConfirmation):
+    if not _ssh_trust_admin(request):
+        return JSONResponse({'error': 'Forbidden'}, status_code=403)
+    if not req.confirmed:
+        return JSONResponse({'error': 'Explicit confirmation required'}, status_code=400)
+    servers = load_data()['servers']
+    if not 0 <= server_id < len(servers):
+        return JSONResponse({'error': 'Server not found'}, status_code=404)
+    server = servers[server_id]
+    try:
+        def unchanged():
+            current = load_data()['servers']
+            return (server_id < len(current) and
+                    all(current[server_id].get(k) == server.get(k)
+                        for k in ('uid', 'host', 'ssh_port')))
+        # Serialize the final endpoint check and trust write with server edits.
+        with data_manager.lock:
+            result = ssh_trust.enroll(server, req.token, validate_endpoint=unchanged)
+            drop_ssh(server)
+        return result
+    except Exception as exc:
+        return JSONResponse({'error': str(exc)}, status_code=409)
+
+
 @app.post('/api/servers/add', tags=["Servers"])
 def api_add_server(request: Request, req: AddServerRequest):
     if not _check_admin(request):
@@ -3265,7 +3324,11 @@ async def api_server_ping(request: Request, server_id: int):
             await writer.wait_closed()
         except Exception:
             pass
-        return {'alive': True, 'ms': ms}
+        try:
+            trust = await asyncio.to_thread(ssh_trust.saved_status, server)
+            return {'alive': True, 'ms': ms, 'ssh_trust': trust, 'ssh_verified': False}
+        except Exception:
+            return {'alive': True, 'ms': ms, 'ssh_trust': 'error'}
     except asyncio.TimeoutError:
         return {'alive': False, 'error': 'timeout', 'ms': None}
     except Exception as e:

managers/ssh_manager.py

@@ -11,6 +11,7 @@ import logging
 import shlex
 import os
 from contextlib import nullcontext
+from managers import ssh_trust
 
 logger = logging.getLogger(__name__)
 
@@ -134,9 +135,7 @@ class SSHManager:
         # 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')
+        known_hosts = ssh_trust.trust_path()
         if configured or os.path.isfile(known_hosts):
             self.client.load_host_keys(known_hosts)
         self.client.set_missing_host_key_policy(paramiko.RejectPolicy())

managers/ssh_trust.py

@@ -0,0 +1,172 @@
+"""Fail-closed SSH trust enrollment. Probes perform key exchange, never auth."""
+import base64
+import hashlib
+import hmac
+import json
+import os
+import secrets
+import socket
+import stat
+import sys
+import tempfile
+import threading
+import time
+
+import paramiko
+
+LOCK = threading.RLock()
+_SECRET = secrets.token_bytes(32)
+_DATA_FILE = None
+
+
+def configure(data_file):
+    global _DATA_FILE
+    _DATA_FILE = data_file
+
+
+def trust_path():
+    default_dir = (os.path.dirname(sys.executable) if getattr(sys, 'frozen', False)
+                   else os.path.dirname(os.path.dirname(__file__)))
+    data = os.environ.get('DATA_FILE') or _DATA_FILE or os.path.join(default_dir, 'data.json')
+    return os.path.abspath(os.path.expanduser(os.environ.get('SSH_KNOWN_HOSTS_FILE') or
+                                            os.path.join(os.path.dirname(data), 'ssh_known_hosts')))
+
+
+def identity(server):
+    host = server.get('host', '')
+    port = server.get('ssh_port', 22)
+    if not isinstance(host, str) or not host or any(c.isspace() for c in host) or any(c in host for c in ',|[]'):
+        raise ValueError('Invalid SSH host')
+    if isinstance(port, bool) or not isinstance(port, int) or not 1 <= port <= 65535:
+        raise ValueError('Invalid SSH port')
+    return host, port
+
+
+def host_name(host, port):
+    return host if port == 22 else f'[{host}]:{port}'
+
+
+def fingerprint(key):
+    return 'SHA256:' + base64.b64encode(hashlib.sha256(key.asbytes()).digest()).decode().rstrip('=')
+
+
+def system_keys():
+    client = paramiko.SSHClient()
+    client.load_system_host_keys()
+    return client._system_host_keys
+
+
+def panel_keys():
+    keys = paramiko.HostKeys()
+    if os.path.exists(trust_path()):
+        keys.load(trust_path())
+    return keys
+
+
+def probe(host, port):
+    # No username, password, private key, SSH agent or auth method enters here.
+    with socket.create_connection((host, port), timeout=10) as sock:
+        transport = paramiko.Transport(sock)
+        try:
+            # Match SSHClient.connect: system pins take precedence, and its
+            # first stored key type is moved to the front of the KEX offers.
+            name = host_name(host, port)
+            with LOCK:
+                effective = system_keys().lookup(name) or panel_keys().lookup(name)
+                if effective:
+                    preferred = next(iter(effective))
+                    options = transport.get_security_options()
+                    options.key_types = [preferred] + [k for k in options.key_types if k != preferred]
+            transport.start_client(timeout=10)
+            return transport.get_remote_server_key()
+        finally:
+            transport.close()
+
+
+def _token(payload):
+    body = base64.urlsafe_b64encode(json.dumps(payload, sort_keys=True).encode()).decode()
+    return body + '.' + hmac.new(_SECRET, body.encode(), hashlib.sha256).hexdigest()
+
+
+def _trust_snapshot(name):
+    return [[[k.get_name(), k.get_base64()] for k in (source.lookup(name) or {}).values()]
+            for source in (system_keys(), panel_keys())]
+
+
+def saved_status(server):
+    """Report saved pin presence without network I/O (not live verification)."""
+    host, port = identity(server)
+    name = host_name(host, port)
+    with LOCK:
+        return 'trusted' if system_keys().lookup(name) or panel_keys().lookup(name) else 'unknown'
+
+
+def inspect(server):
+    host, port = identity(server)
+    key = probe(host, port)
+    name = host_name(host, port)
+    with LOCK:
+        system, local = system_keys().lookup(name), panel_keys().lookup(name)
+        effective = system or local
+        old = [{'algorithm': k.get_name(), 'fingerprint': fingerprint(k)}
+               for source in (system, local) if source for k in source.values()]
+        status = 'unknown' if not effective else ('trusted' if effective.get(key.get_name()) == key else 'changed')
+        result = {'host': host, 'port': port, 'algorithm': key.get_name(),
+                  'fingerprint': fingerprint(key), 'old_fingerprints': old, 'status': status}
+        result['token'] = _token({'host': host, 'port': port, 'uid': server.get('uid'),
+                                  'key': key.get_base64(), 'algorithm': key.get_name(),
+                                  'trust': _trust_snapshot(name),
+                                  'expires': time.time() + 300})
+        return result
+
+
+def enroll(server, token, validate_endpoint=None):
+    host, port = identity(server)
+    try:
+        body, signature = token.split('.')
+        expected = hmac.new(_SECRET, body.encode(), hashlib.sha256).hexdigest()
+        if not hmac.compare_digest(signature, expected):
+            raise ValueError()
+        payload = json.loads(base64.urlsafe_b64decode(body))
+        if (payload['expires'] < time.time() or payload['host'] != host or
+                payload['port'] != port or payload['uid'] != server.get('uid')):
+            raise ValueError()
+    except Exception as exc:
+        raise ValueError('Invalid or expired SSH confirmation; inspect again') from exc
+    key = probe(host, port)
+    if key.get_base64() != payload['key'] or key.get_name() != payload['algorithm']:
+        raise ValueError('SSH candidate changed; inspect and verify again')
+    name = host_name(host, port)
+    with LOCK:
+        if validate_endpoint is not None and not validate_endpoint():
+            raise ValueError('Server endpoint changed; probe again')
+        system = system_keys().lookup(name)
+        if system and system.get(key.get_name()) != key:
+            raise ValueError('Conflicting system known_hosts entry; administrator must correct it independently first')
+        if payload.get('trust') != _trust_snapshot(name):
+            raise ValueError('Trust changed; probe again')
+        path = trust_path()
+        if os.path.exists(path) and not os.stat(path).st_mode & stat.S_IWRITE:
+            raise PermissionError('SSH trust file is read-only')
+        keys = panel_keys()
+        # Remove all entries for this endpoint, including hashed names and old
+        # algorithms, preserving aliases on shared lines and unrelated hosts.
+        for entry in list(keys._entries):
+            entry.hostnames = [h for h in entry.hostnames
+                               if not keys._hostname_matches(name, paramiko.hostkeys.HostKeyEntry([h], entry.key))]
+        keys._entries = [entry for entry in keys._entries if entry.hostnames]
+        keys.add(name, key.get_name(), key)
+        directory = os.path.dirname(path)
+        os.makedirs(directory, exist_ok=True)
+        fd, temporary = tempfile.mkstemp(prefix='.ssh-trust-', dir=directory)
+        os.close(fd)
+        try:
+            os.chmod(temporary, 0o600)
+            keys.save(temporary)
+            with open(temporary, 'rb+') as stream:
+                os.fsync(stream.fileno())
+            os.replace(temporary, path)
+        finally:
+            if os.path.exists(temporary):
+                os.unlink(temporary)
+    return {'status': 'success'}

templates/index.html

@@ -380,7 +380,12 @@
             const res = await fetch(`/api/servers/${idx}/ping`, { credentials: 'same-origin' });
             const data = await res.json();
             dot.classList.remove('ping-pending');
-            if (data.alive) {
+            if (data.alive && data.ssh_trust !== 'trusted') {
+                dot.classList.add('ping-fail');
+                const labels = {unknown: _('ssh_trust_unknown'), changed: _('ssh_trust_changed'), error: _('ssh_trust_error')};
+                dot.title = labels[data.ssh_trust] || labels.error;
+                if (ms) ms.textContent = dot.title;
+            } else if (data.alive) {
                 dot.classList.add('ping-ok');
                 dot.title = `${data.ms} ms`;
                 if (ms) ms.textContent = `${data.ms} ms`;

templates/server.html

@@ -150,6 +150,49 @@
 {% endblock %}
 
 {% block content %}
+{% if current_user and current_user.role == 'admin' %}
+<section class="card" id="ssh-trust-panel" style="margin-bottom: 20px; padding: 20px;">
+    <h2>{{ _('ssh_trust_title') }}</h2>
+    <p>{{ _('ssh_trust_warning') }}</p>
+    <button class="btn btn-secondary" type="button" onclick="inspectSSHTrust()">{{ _('ssh_trust_inspect') }}</button>
+    <pre id="ssh-trust-details" aria-live="polite" style="white-space: pre-wrap; overflow-wrap: anywhere;"></pre>
+    <div id="ssh-trust-confirmation" hidden>
+        <label><input id="ssh-trust-verified" type="checkbox"> {{ _('ssh_trust_verified') }}</label>
+        <button id="ssh-trust-save" class="btn btn-primary" type="button" onclick="confirmSSHTrust()">{{ _('ssh_trust_save') }}</button>
+    </div>
+</section>
+<script>
+    let sshTrustToken = null;
+    async function inspectSSHTrust() {
+        sshTrustToken = null;
+        document.getElementById('ssh-trust-confirmation').hidden = true;
+        document.getElementById('ssh-trust-verified').checked = false;
+        const details = document.getElementById('ssh-trust-details');
+        try {
+            const data = await apiCall('/api/servers/{{ server_id }}/ssh-trust/inspect', 'POST');
+            const labels = {unknown: {{ _('ssh_trust_unknown') | tojson }}, changed: {{ _('ssh_trust_changed') | tojson }}, trusted: {{ _('ssh_trust_trusted') | tojson }}};
+            details.textContent = labels[data.status] + '\n' + data.host + ':' + data.port + '\n' +
+                {{ _('ssh_trust_new') | tojson }} + ': ' + data.algorithm + ' ' + data.fingerprint + '\n' +
+                {{ _('ssh_trust_old') | tojson }} + ':\n' + data.old_fingerprints.map(k => k.algorithm + ' ' + k.fingerprint).join('\n');
+            sshTrustToken = data.token;
+            document.getElementById('ssh-trust-confirmation').hidden = data.status === 'trusted';
+        } catch (error) { details.textContent = error.message; }
+    }
+    async function confirmSSHTrust() {
+        if (!sshTrustToken || !document.getElementById('ssh-trust-verified').checked) return;
+        const button = document.getElementById('ssh-trust-save');
+        button.disabled = true;
+        try {
+            await apiCall('/api/servers/{{ server_id }}/ssh-trust/confirm', 'POST', {token: sshTrustToken, confirmed: true});
+            location.reload();
+        } catch (error) {
+            sshTrustToken = null;
+            document.getElementById('ssh-trust-confirmation').hidden = true;
+            document.getElementById('ssh-trust-details').textContent = error.message;
+        } finally { button.disabled = false; }
+    }
+</script>
+{% endif %}
 <a href="/" class="back-link page-back-link">{{ _('back_to_servers') }}</a>
 
 <!-- Server Info Header -->

tests/test_ssh_enrollment.py

@@ -0,0 +1,168 @@
+"""Enrollment never authenticates; confirmations bind endpoint and exact key."""
+import os
+import tempfile
+import unittest
+from pathlib import Path
+from unittest import mock
+
+import paramiko
+from managers import ssh_trust
+from tests.test_ssh_host_keys import endpoint
+
+
+class EnrollmentTests(unittest.TestCase):
+    @classmethod
+    def setUpClass(cls):
+        cls.key = paramiko.RSAKey.generate(1024)
+        cls.other = paramiko.RSAKey.generate(1024)
+
+    def setUp(self):
+        self.tmp = tempfile.TemporaryDirectory()
+        self.addCleanup(self.tmp.cleanup)
+        self.path = Path(self.tmp.name) / 'known_hosts'
+        self.env = mock.patch.dict(os.environ, {'SSH_KNOWN_HOSTS_FILE': str(self.path)})
+        self.env.start()
+        self.addCleanup(self.env.stop)
+        self.system = mock.patch.object(ssh_trust, 'system_keys', return_value=paramiko.HostKeys())
+        self.system.start()
+        self.addCleanup(self.system.stop)
+        self.server = {'host': 'example.test', 'ssh_port': 2222, 'uid': 'one'}
+
+    def candidate(self):
+        with mock.patch.object(ssh_trust, 'probe', return_value=self.key):
+            return ssh_trust.inspect(self.server)
+
+    def confirm(self, candidate, server=None):
+        with mock.patch.object(ssh_trust, 'probe', return_value=self.key):
+            return ssh_trust.enroll(server or self.server, candidate['token'])
+
+    def test_probe_real_handshake_without_auth(self):
+        with endpoint(self.key) as (port, service):
+            self.assertEqual(ssh_trust.probe('127.0.0.1', port), self.key)
+            self.assertEqual(service.auth_calls, 0)
+
+    def test_probe_matches_client_pinned_algorithm_preference(self):
+        # ECDSA normally wins over RSA, but SSHClient prefers the stored RSA pin.
+        preferred = paramiko.ECDSAKey.generate()
+        with endpoint([self.key, preferred]) as (port, service):
+            keys = paramiko.HostKeys()
+            keys.add(f'[127.0.0.1]:{port}', 'ssh-rsa', self.key)
+            keys.save(str(self.path))
+            result = ssh_trust.inspect({'host': '127.0.0.1', 'ssh_port': port})
+            self.assertEqual(result['status'], 'trusted')
+            self.assertEqual(result['algorithm'], 'ssh-rsa')
+            self.assertEqual(service.auth_calls, 0)
+
+    def test_confirmation_rejects_concurrent_trust_change(self):
+        candidate = self.candidate()
+        keys = paramiko.HostKeys()
+        keys.add('[example.test]:2222', 'ssh-rsa', self.other)
+        keys.save(str(self.path))
+        with self.assertRaisesRegex(ValueError, 'Trust changed'):
+            self.confirm(candidate)
+        self.assertEqual(paramiko.HostKeys(str(self.path)).lookup('[example.test]:2222')['ssh-rsa'], self.other)
+
+    def test_list_status_never_contacts_server(self):
+        with mock.patch.object(ssh_trust, 'probe', side_effect=AssertionError('network forbidden')):
+            self.assertEqual(ssh_trust.saved_status(self.server), 'unknown')
+            keys = paramiko.HostKeys()
+            keys.add('[example.test]:2222', 'ssh-rsa', self.key)
+            keys.save(str(self.path))
+            self.assertEqual(ssh_trust.saved_status(self.server), 'trusted')
+
+    def test_unknown_persistent_nonstandard_preserves_other_hosts(self):
+        keys = paramiko.HostKeys()
+        keys.add('unrelated', self.other.get_name(), self.other)
+        keys.save(str(self.path))
+        candidate = self.candidate()
+        self.assertEqual(candidate['status'], 'unknown')
+        self.assertEqual(candidate['old_fingerprints'], [])
+        self.assertTrue(candidate['fingerprint'].startswith('SHA256:'))
+        self.confirm(candidate)
+        saved = paramiko.HostKeys(str(self.path))
+        self.assertEqual(saved['unrelated'][self.other.get_name()], self.other)
+        self.assertEqual(saved['[example.test]:2222'][self.key.get_name()], self.key)
+        self.assertEqual(self.candidate()['status'], 'trusted')
+
+    def test_changed_hashed_entry(self):
+        keys = paramiko.HostKeys()
+        keys.add(paramiko.HostKeys.hash_host('[example.test]:2222'), self.other.get_name(), self.other)
+        keys.save(str(self.path))
+        candidate = self.candidate()
+        self.assertEqual(candidate['status'], 'changed')
+        self.assertEqual(candidate['old_fingerprints'][0]['fingerprint'], ssh_trust.fingerprint(self.other))
+        self.confirm(candidate)
+        self.assertTrue(paramiko.HostKeys(str(self.path)).check('[example.test]:2222', self.key))
+
+    def test_race_and_endpoint_change_rejected(self):
+        candidate = self.candidate()
+        with mock.patch.object(ssh_trust, 'probe', return_value=self.other):
+            with self.assertRaises(ValueError):
+                ssh_trust.enroll(self.server, candidate['token'])
+        with self.assertRaises(ValueError):
+            self.confirm(candidate, dict(self.server, host='elsewhere'))
+        self.assertFalse(self.path.exists())
+
+    def test_invalid_expired_token(self):
+        with self.assertRaises(ValueError):
+            ssh_trust.enroll(self.server, 'invalid')
+        candidate = self.candidate()
+        with mock.patch.object(ssh_trust.time, 'time', return_value=10**12):
+            with self.assertRaises(ValueError):
+                self.confirm(candidate)
+
+    def test_endpoint_edited_during_probe_is_rejected(self):
+        candidate = self.candidate()
+        with mock.patch.object(ssh_trust, 'probe', return_value=self.key):
+            with self.assertRaisesRegex(ValueError, 'endpoint'):
+                ssh_trust.enroll(self.server, candidate['token'], validate_endpoint=lambda: False)
+        self.assertFalse(self.path.exists())
+
+    def test_system_conflict_and_read_only_fail(self):
+        candidate = self.candidate()
+        system = paramiko.HostKeys()
+        system.add('[example.test]:2222', self.other.get_name(), self.other)
+        with mock.patch.object(ssh_trust, 'system_keys', return_value=system):
+            with self.assertRaisesRegex(ValueError, 'system'):
+                self.confirm(candidate)
+        self.path.touch()
+        self.path.chmod(0o444)
+        try:
+            with self.assertRaises(PermissionError):
+                self.confirm(candidate)
+        finally:
+            self.path.chmod(0o600)
+
+    def test_malformed_endpoint(self):
+        for host, port in [('x\nline', 22), ('', 22), ('x', 0), ('x', 65536)]:
+            with self.assertRaises(ValueError):
+                ssh_trust.inspect({'host': host, 'ssh_port': port})
+
+    def test_concurrent_writes_preserve_hosts_and_aliases(self):
+        from concurrent.futures import ThreadPoolExecutor
+        self.path.write_text('example.test,alias ssh-rsa ' + self.other.get_base64() + '\n')
+        servers = [dict(self.server, host='node' + str(i), ssh_port=22) for i in range(8)]
+        with mock.patch.object(ssh_trust, 'probe', return_value=self.key):
+            candidates = [ssh_trust.inspect(server) for server in servers]
+            with ThreadPoolExecutor(max_workers=8) as executor:
+                list(executor.map(lambda pair: ssh_trust.enroll(pair[0], pair[1]['token']), zip(servers, candidates)))
+            server = dict(self.server, ssh_port=22)
+            ssh_trust.enroll(server, ssh_trust.inspect(server)['token'])
+        saved = paramiko.HostKeys(str(self.path))
+        self.assertEqual(saved['alias']['ssh-rsa'], self.other)
+        for server in servers:
+            self.assertTrue(saved.check(server['host'], self.key))
+
+    def test_failed_atomic_replace_preserves_file(self):
+        self.path.write_text('other ssh-rsa ' + self.other.get_base64() + '\n')
+        before = self.path.read_bytes()
+        with mock.patch.object(ssh_trust.os, 'replace', side_effect=PermissionError('read-only mount')):
+            with self.assertRaises(PermissionError):
+                self.confirm(self.candidate())
+        self.assertEqual(self.path.read_bytes(), before)
+        self.assertEqual(list(self.path.parent.glob('.ssh-trust-*')), [])
+
+    def test_path_uses_application_data(self):
+        with mock.patch.dict(os.environ, {}, clear=True):
+            ssh_trust.configure(str(Path(self.tmp.name) / 'custom' / 'data.json'))
+            self.assertEqual(ssh_trust.trust_path(), str(Path(self.tmp.name) / 'custom' / 'ssh_known_hosts'))

tests/test_ssh_enrollment_http.py

@@ -0,0 +1,76 @@
+"""HTTP authorization and explicit confirmation, without loading real state."""
+import ast
+import threading
+import unittest
+from pathlib import Path
+from types import SimpleNamespace
+from unittest import mock
+
+from fastapi import FastAPI, Request
+from fastapi.responses import JSONResponse
+from fastapi.testclient import TestClient
+from pydantic import BaseModel, StrictBool, constr
+from managers import ssh_trust
+
+
+class EnrollmentHTTPTests(unittest.TestCase):
+    def setUp(self):
+        tree = ast.parse((Path(__file__).resolve().parents[1] / 'app.py').read_text(encoding='utf-8'))
+        names = {'SSHTrustConfirmation', '_ssh_trust_admin', 'api_ssh_trust_inspect', 'api_ssh_trust_confirm'}
+        nodes = [n for n in tree.body if getattr(n, 'name', '') in names]
+        self.assertEqual(len(nodes), 4, 'SSH trust routes and request model must exist')
+        self.user = {'role': 'admin'}
+        self.server = {'host': 'example.test', 'ssh_port': 22, 'uid': 'a'}
+        self.trust = mock.Mock()
+        self.trust.inspect.return_value = {'status': 'unknown', 'token': 'abc'}
+        self.trust.enroll.return_value = {'status': 'success'}
+        self.drop = mock.Mock()
+        app = FastAPI()
+        self.data_lock = threading.RLock()
+        ns = dict(app=app, Request=Request, JSONResponse=JSONResponse, BaseModel=BaseModel,
+                  StrictBool=StrictBool, constr=constr, ssh_trust=self.trust,
+                  _check_admin=lambda request: self.user, load_data=lambda: {'servers': [self.server]},
+                  drop_ssh=self.drop, data_manager=SimpleNamespace(lock=self.data_lock))
+        exec(compile(ast.Module(body=nodes, type_ignores=[]), '<trust routes>', 'exec'), ns)
+        self.client = TestClient(app)
+
+    def test_admin_only(self):
+        for user in [None, {'role': 'support'}, {'role': 'user'}]:
+            self.user = user
+            for action in ['inspect', 'confirm']:
+                response = self.client.post('/api/servers/0/ssh-trust/' + action,
+                                            json={'token': 'abc', 'confirmed': True})
+                self.assertEqual(response.status_code, 403)
+        self.trust.inspect.assert_not_called()
+        self.trust.enroll.assert_not_called()
+
+    def test_confirmation_and_reset(self):
+        self.assertEqual(self.client.post('/api/servers/0/ssh-trust/inspect').status_code, 200)
+        self.assertEqual(self.client.post('/api/servers/0/ssh-trust/confirm', json={'token': 'abc', 'confirmed': False}).status_code, 400)
+        self.trust.enroll.assert_not_called()
+        self.assertEqual(self.client.post('/api/servers/0/ssh-trust/confirm', json={'token': 'abc', 'confirmed': True}).status_code, 200)
+        self.drop.assert_called_once_with(self.server)
+
+    def test_confirmation_serializes_endpoint_mutations(self):
+        def enroll(*args, **kwargs):
+            self.assertTrue(self.data_lock._is_owned())
+            return {'status': 'success'}
+        self.trust.enroll.side_effect = enroll
+        response = self.client.post('/api/servers/0/ssh-trust/confirm',
+                                    json={'token': 'abc', 'confirmed': True})
+        self.assertEqual(response.status_code, 200, response.text)
+
+    def test_errors(self):
+        # Browser Fetch Metadata is not script-settable. TLS termination may
+        # leave request.base_url as HTTP while the browser origin is HTTPS.
+        response = self.client.post('/api/servers/0/ssh-trust/inspect',
+                                    headers={'Origin': 'https://testserver', 'Sec-Fetch-Site': 'same-origin'})
+        self.assertEqual(response.status_code, 200)
+        self.assertEqual(self.client.post('/api/servers/0/ssh-trust/inspect', headers={'Origin': 'https://evil.example'}).status_code, 403)
+        self.assertEqual(self.client.post('/api/servers/-1/ssh-trust/inspect').status_code, 404)
+        for body in [{}, {'token': '', 'confirmed': True}, {'token': 'abc', 'confirmed': 'yes'}]:
+            self.assertEqual(self.client.post('/api/servers/0/ssh-trust/confirm', json=body).status_code, 422)
+        self.trust.enroll.side_effect = ValueError('candidate changed')
+        response = self.client.post('/api/servers/0/ssh-trust/confirm', json={'token': 'abc', 'confirmed': True})
+        self.assertEqual(response.status_code, 409)
+        self.drop.assert_not_called()

tests/test_ssh_host_keys.py

@@ -43,7 +43,8 @@ def endpoint(key):
             conn, _ = listener.accept()
             transport = paramiko.Transport(conn)
             transports.append(transport)
-            transport.add_server_key(key)
+            for host_key in key if isinstance(key, (list, tuple)) else [key]:
+                transport.add_server_key(host_key)
             transport.start_server(server=server)
             stop.wait(4)
         except (OSError, EOFError, paramiko.SSHException):

tests/test_ssh_trust_ui.py

@@ -0,0 +1,24 @@
+"""Recovery controls must not depend on successful SSH stats loading."""
+from pathlib import Path
+import unittest
+
+ROOT = Path(__file__).resolve().parents[1]
+
+class TrustUITests(unittest.TestCase):
+    def test_recovery_is_in_static_server_page(self):
+        source = (ROOT / 'templates/server.html').read_text(encoding='utf-8')
+        self.assertIn('id="ssh-trust-panel"', source)
+        self.assertIn('ssh-trust/inspect', source)
+        self.assertIn('ssh-trust/confirm', source)
+        self.assertIn('ssh_trust_warning', source)
+        self.assertIn('confirmed:', source)
+        self.assertIn('location.reload()', source)
+
+    def test_listing_does_not_call_tcp_ping_ssh_success(self):
+        app_source = (ROOT / 'app.py').read_text(encoding='utf-8')
+        self.assertIn('asyncio.to_thread(ssh_trust.saved_status, server)', app_source)
+        self.assertNotIn('asyncio.to_thread(ssh_trust.inspect, server)', app_source)
+        source = (ROOT / 'templates/index.html').read_text(encoding='utf-8')
+        self.assertIn('data.ssh_trust', source)
+        self.assertIn('ssh_trust_unknown', source)
+        self.assertIn('ssh_trust_changed', source)

translations/en.json

@@ -763,5 +763,16 @@
   "rknblock_disabled_short": "disabled",
   "rknblock_opt_block_in_short": "inbound",
   "rknblock_opt_block_out_short": "outbound",
-  "rknblock_opt_log_off_short": "logging"
+  "rknblock_opt_log_off_short": "logging",
+    "ssh_trust_title": "SSH host key",
+    "ssh_trust_warning": "Verify the fingerprint independently using the server/provider console (ssh-keygen -lf /etc/ssh/ssh_host_ed25519_key.pub). A changed key may indicate an attack. Never trust a key only because this page shows it.",
+    "ssh_trust_inspect": "Inspect SSH key",
+    "ssh_trust_verified": "I verified this exact fingerprint through an independent console.",
+    "ssh_trust_save": "Trust key and retry",
+    "ssh_trust_unknown": "Unknown SSH key",
+    "ssh_trust_changed": "SSH key changed",
+    "ssh_trust_trusted": "SSH key trusted",
+    "ssh_trust_new": "Candidate key",
+    "ssh_trust_old": "Previously trusted keys",
+    "ssh_trust_error": "SSH trust check failed"
 }

translations/fa.json

@@ -763,5 +763,16 @@
   "rknblock_disabled_short": "خاموش",
   "rknblock_opt_block_in_short": "ورودی",
   "rknblock_opt_block_out_short": "خروجی",
-  "rknblock_opt_log_off_short": "ثبت رویداد"
+  "rknblock_opt_log_off_short": "ثبت رویداد",
+    "ssh_trust_title": "کلید میزبان SSH",
+    "ssh_trust_warning": "اثر انگشت را مستقل از این صفحه در کنسول سرور یا ارائهدهنده بررسی کنید (ssh-keygen -lf /etc/ssh/ssh_host_ed25519_key.pub). تغییر کلید ممکن است نشانه حمله باشد. فقط به دلیل نمایش کلید در این صفحه به آن اعتماد نکنید.",
+    "ssh_trust_inspect": "بررسی کلید SSH",
+    "ssh_trust_verified": "این اثر انگشت دقیق را در کنسول مستقل تأیید کردم.",
+    "ssh_trust_save": "اعتماد به کلید و تلاش مجدد",
+    "ssh_trust_unknown": "کلید SSH ناشناخته",
+    "ssh_trust_changed": "کلید SSH تغییر کرده",
+    "ssh_trust_trusted": "کلید SSH مورد اعتماد",
+    "ssh_trust_new": "کلید پیشنهادی",
+    "ssh_trust_old": "کلیدهای مورد اعتماد قبلی",
+    "ssh_trust_error": "بررسی اعتماد SSH ناموفق بود"
 }

translations/fr.json

@@ -763,5 +763,16 @@
   "rknblock_disabled_short": "désactivé",
   "rknblock_opt_block_in_short": "entrant",
   "rknblock_opt_block_out_short": "sortant",
-  "rknblock_opt_log_off_short": "journalisation"
+  "rknblock_opt_log_off_short": "journalisation",
+    "ssh_trust_title": "Clé hôte SSH",
+    "ssh_trust_warning": "Vérifiez indépendamment l’empreinte via la console du serveur/fournisseur (ssh-keygen -lf /etc/ssh/ssh_host_ed25519_key.pub). Un changement peut signaler une attaque. Ne faites pas confiance à une clé uniquement parce que cette page l’affiche.",
+    "ssh_trust_inspect": "Examiner la clé SSH",
+    "ssh_trust_verified": "J’ai vérifié cette empreinte exacte via une console indépendante.",
+    "ssh_trust_save": "Faire confiance et réessayer",
+    "ssh_trust_unknown": "Clé SSH inconnue",
+    "ssh_trust_changed": "Clé SSH modifiée",
+    "ssh_trust_trusted": "Clé SSH approuvée",
+    "ssh_trust_new": "Clé proposée",
+    "ssh_trust_old": "Anciennes clés approuvées",
+    "ssh_trust_error": "Échec de vérification SSH"
 }

translations/ru.json

@@ -763,5 +763,16 @@
   "rknblock_disabled_short": "выключено",
   "rknblock_opt_block_in_short": "входящие",
   "rknblock_opt_block_out_short": "исходящие",
-  "rknblock_opt_log_off_short": "логирование"
+  "rknblock_opt_log_off_short": "логирование",
+    "ssh_trust_title": "Ключ SSH сервера",
+    "ssh_trust_warning": "Независимо проверьте отпечаток через консоль сервера/провайдера (ssh-keygen -lf /etc/ssh/ssh_host_ed25519_key.pub). Изменение ключа может означать атаку. Не доверяйте ключу только потому, что он показан здесь.",
+    "ssh_trust_inspect": "Проверить ключ SSH",
+    "ssh_trust_verified": "Я сверил именно этот отпечаток через независимую консоль.",
+    "ssh_trust_save": "Доверять ключу и повторить",
+    "ssh_trust_unknown": "Неизвестный ключ SSH",
+    "ssh_trust_changed": "Ключ SSH изменился",
+    "ssh_trust_trusted": "Ключ SSH доверенный",
+    "ssh_trust_new": "Новый ключ",
+    "ssh_trust_old": "Ранее доверенные ключи",
+    "ssh_trust_error": "Ошибка проверки доверия SSH"
 }

translations/zh.json

@@ -763,5 +763,16 @@
   "rknblock_disabled_short": "已关闭",
   "rknblock_opt_block_in_short": "入站",
   "rknblock_opt_block_out_short": "出站",
-  "rknblock_opt_log_off_short": "日志"
+  "rknblock_opt_log_off_short": "日志",
+    "ssh_trust_title": "SSH 主机密钥",
+    "ssh_trust_warning": "请通过服务器或提供商控制台独立核对指纹（ssh-keygen -lf /etc/ssh/ssh_host_ed25519_key.pub）。密钥变更可能意味着攻击。不要仅因为此页面显示密钥就信任它。",
+    "ssh_trust_inspect": "检查 SSH 密钥",
+    "ssh_trust_verified": "我已通过独立控制台核对这一确切指纹。",
+    "ssh_trust_save": "信任密钥并重试",
+    "ssh_trust_unknown": "未知 SSH 密钥",
+    "ssh_trust_changed": "SSH 密钥已变更",
+    "ssh_trust_trusted": "SSH 密钥已信任",
+    "ssh_trust_new": "候选密钥",
+    "ssh_trust_old": "以前信任的密钥",
+    "ssh_trust_error": "SSH 信任检查失败"
 }