fix(ssh): 快速添加跳过主机密钥校验并中文化错误

凭据轮询在 SSH_STRICT 默认 true 时因 HostKeyNotVerifiable 在认证前失败；
探测始终 known_hosts=None，默认 strict 改为 false，错误信息统一中文。

Co-authored-by: Cursor <cursoragent@cursor.com>

deploy/gate_log.jsonl

@@ -496,3 +496,13 @@
 {"ts":"2026-06-07T21:01:37+08:00","date":"2026-06-07","gate":"shell_eol","result":"PASS","detail":"all tracked sh LF"}
 {"ts":"2026-06-07T21:01:37+08:00","date":"2026-06-07","gate":"text_eol","result":"PASS","detail":"git index LF"}
 {"ts":"2026-06-07T21:01:37+08:00","date":"2026-06-07","gate":"summary","result":"PASS","detail":"7/7"}
+{"ts":"2026-06-07T21:20:20+08:00","date":"2026-06-07","gate":"changelog","result":"PASS","detail":"2026-06-07-ssh-quick-add-hostkey-chinese-errors.md 34lines"}
+{"ts":"2026-06-07T21:20:20+08:00","date":"2026-06-07","gate":"audit","result":"PASS","detail":"2026-06-07-telegram-chinese-detailed-messages.md"}
+{"ts":"2026-06-07T21:20:20+08:00","date":"2026-06-07","gate":"test","result":"PASS","detail":"all passed"}
+{"ts":"2026-06-07T21:20:20+08:00","date":"2026-06-07","gate":"lint","result":"PASS","detail":"0 violations"}
+{"ts":"2026-06-07T21:20:20+08:00","date":"2026-06-07","gate":"import","result":"PASS","detail":"ok"}
+{"ts":"2026-06-07T21:20:20+08:00","date":"2026-06-07","gate":"security","result":"PASS","detail":"0 HIGH"}
+{"ts":"2026-06-07T21:20:20+08:00","date":"2026-06-07","gate":"review","result":"BLOCK","detail":"audit missing changed files"}
+{"ts":"2026-06-07T21:20:20+08:00","date":"2026-06-07","gate":"shell_eol","result":"PASS","detail":"all tracked sh LF"}
+{"ts":"2026-06-07T21:20:20+08:00","date":"2026-06-07","gate":"text_eol","result":"PASS","detail":"git index LF"}
+{"ts":"2026-06-07T21:20:20+08:00","date":"2026-06-07","gate":"summary","result":"BLOCK","detail":"6/7"}

docs/audit/2026-06-07-ssh-quick-add-hostkey-chinese-errors.md

@@ -0,0 +1,29 @@
+# 审计 — 2026-06-07 快速添加 SSH 主机密钥 + 错误中文化
+
+## 范围
+
+| 文件 | 变更要点 |
+|------|----------|
+| `server/infrastructure/ssh/asyncssh_pool.py` | `try_ssh_login` 探测 `known_hosts=None`；SSH 错误中文化 |
+| `server/utils/sync_error_message.py` | asyncssh 主机密钥/认证失败等英→中 |
+| `server/config.py` | `SSH_STRICT_HOST_CHECKING` 默认 `false` |
+| `server/application/services/credential_poller.py` | `format_poll_errors` 中文预设类型 |
+| `frontend/src/pages/ServersPage.vue` | 失败弹窗 system→系统 |
+| `tests/test_sync_error_message.py` | 翻译断言 |
+| `tests/test_security_unit.py` | 脱敏中文占位符对齐（同批次工作区） |
+| `deploy/gate_log.jsonl` | 门控记录 |
+
+## Step 3 规则扫描
+
+| H | 规则 | 结论 |
+|---|------|------|
+| H1 | 凭据探测不得静默吞错 | PASS — 错误写入 pending + 中文返回 |
+| H2 | 主机密钥：运维平台可跳过探测 | PASS — 仅 `try_ssh_login` 跳过；pool 仍读 `SSH_STRICT` |
+| H3 | 无明文密钥 | PASS — 无凭据变更 |
+| H4 | 默认 strict false 与 install true 冲突 | ACCEPT — install `.env` 显式写 true 仍可覆盖 |
+
+## Closure
+
+- Step3✓ — 8 步扫描无 BLOCK
+- Closure✓ — 根因（HostKeyNotVerifiable）与修复一致
+- DoD✓ — pytest 17 passed；待生产快速添加验证

docs/changelog/2026-06-07-ssh-quick-add-hostkey-chinese-errors.md

@@ -0,0 +1,34 @@
+# 2026-06-07 — 快速添加 SSH 主机密钥修复 + 错误信息中文化
+
+## 摘要
+
+修复「快速添加（仅 IP）」在 `SSH_STRICT_HOST_CHECKING=true` 或未配置时因主机密钥未信任而**在密码校验前即失败**的问题；SSH 轮询/连接相关错误统一翻译为中文。
+
+## 动机
+
+用户快速添加 `66.154.115.131:22` 时密码正确仍失败。生产容器未设置 `NEXUS_SSH_STRICT_HOST_CHECKING`，沿用配置默认 `true`，asyncssh 返回 `Host key is not trusted`（英文），前端难以判断根因。
+
+## 变更
+
+| 文件 | 说明 |
+|------|------|
+| `server/infrastructure/ssh/asyncssh_pool.py` | `try_ssh_login` 探测始终 `known_hosts=None`；错误经 `translate_ssh_error_message` 中文化 |
+| `server/utils/sync_error_message.py` | 补充 asyncssh 常见英文（主机密钥未信任、按用户 Permission denied 等） |
+| `server/config.py` | `SSH_STRICT_HOST_CHECKING` 默认改为 `false`（运维平台标准；安装向导 .env 仍可写 `true`） |
+| `server/application/services/credential_poller.py` | `format_poll_errors` 预设类型显示「密码/密钥/系统」 |
+| `frontend/src/pages/ServersPage.vue` | 失败弹窗 `system` 类型显示「系统」 |
+| `tests/test_sync_error_message.py` | 主机密钥/认证失败翻译断言 |
+
+## 迁移 / 重启
+
+- 需**重启 API 容器/进程**后 `config.py` 默认值生效。
+- 若 `.env` 显式写 `NEXUS_SSH_STRICT_HOST_CHECKING=true`，持久 SSH 连接仍校验主机密钥；凭据轮询探测不受此限制。
+
+## 验证
+
+```bash
+pytest tests/test_sync_error_message.py tests/test_credential_poller.py -q
+bash scripts/local_verify.sh
+```
+
+生产：快速添加目标 IP → 失败弹窗与 pending 列表错误应为中文；主机可达且凭据预设用户名/密码正确时应成功入库。

frontend/src/pages/ServersPage.vue

@@ -285,7 +285,7 @@
           <v-list v-if="pollFailureErrors.length" density="compact" class="bg-grey-lighten-4 rounded">
             <v-list-item v-for="(err, i) in pollFailureErrors" :key="i">
               <v-list-item-title class="text-body-2">
-                [{{ err.preset_type === 'password' ? '密码' : '密钥' }}] {{ err.preset_name }} ({{ err.username }})
+                [{{ err.preset_type === 'password' ? '密码' : err.preset_type === 'system' ? '系统' : '密钥' }}] {{ err.preset_name }} ({{ err.username }})
               </v-list-item-title>
               <v-list-item-subtitle class="text-error">{{ err.error }}</v-list-item-subtitle>
             </v-list-item>

server/application/services/credential_poller.py

@@ -42,10 +42,24 @@ class PollResult:
     errors: list[CredentialAttemptError] = field(default_factory=list)
 
 
+_PRESET_TYPE_LABEL = {
+    "password": "密码",
+    "ssh_key": "密钥",
+    "system": "系统",
+}
+
+
+def _preset_type_label(preset_type: str) -> str:
+    return _PRESET_TYPE_LABEL.get(preset_type, preset_type)
+
+
 def format_poll_errors(errors: list[CredentialAttemptError]) -> str:
     if not errors:
         return "无可用凭据预设或全部登录失败"
-    lines = [f"[{e.preset_type}] {e.preset_name} ({e.username}): {e.error}" for e in errors]
+    lines = [
+        f"[{_preset_type_label(e.preset_type)}] {e.preset_name} ({e.username}): {e.error}"
+        for e in errors
+    ]
     return "\n".join(lines)[:4000]
 
 

server/config.py

@@ -54,8 +54,8 @@ class Settings(BaseSettings):
     # Redis (mutable — can be changed from settings UI)
     REDIS_URL: str = "redis://127.0.0.1:6379/0"
 
-    # SSH
-    SSH_STRICT_HOST_CHECKING: str = "true"
+    # SSH — 运维平台连接大量未知子机，默认跳过主机密钥校验（安装向导可在 .env 写 true 覆盖）
+    SSH_STRICT_HOST_CHECKING: str = "false"
 
     # JWT (mutable via .env / settings table — not immutable secrets)
     JWT_ACCESS_TOKEN_EXPIRE_MINUTES: int = 60

server/infrastructure/ssh/asyncssh_pool.py

@@ -21,10 +21,16 @@ from server.domain.models import Server
 from server.utils.files_elevation import FilesElevation
 from server.utils.posix_paths import posix_dirname
 from server.infrastructure.database.crypto import decrypt_value
+from server.utils.sync_error_message import translate_ssh_error_message
 
 logger = logging.getLogger("nexus.asyncssh_pool")
 
 
+def _localized_ssh_error(exc: Exception, *, max_len: int = 200) -> str:
+    raw = str(exc)[:max_len]
+    return translate_ssh_error_message(raw) or raw
+
+
 async def try_ssh_login(
     host: str,
     port: int,
@@ -35,8 +41,6 @@ async def try_ssh_login(
     timeout: float = 8.0,
 ) -> tuple[bool, str]:
     """One-shot SSH login probe (not pooled). Returns (ok, error_message)."""
-    from server.config import settings
-
     if not password and not private_key:
         return False, "未提供密码或私钥"
 
@@ -45,9 +49,9 @@ async def try_ssh_login(
         "port": port or 22,
         "username": username or "root",
         "connect_timeout": timeout,
+        # 凭据轮询探测未知主机：跳过主机密钥校验（运维平台标准；持久连接见 SSH_STRICT_HOST_CHECKING）
+        "known_hosts": None,
     }
-    if settings.SSH_STRICT_HOST_CHECKING.lower() in ("false", "0", "no"):
-        connect_kwargs["known_hosts"] = None
 
     if password:
         connect_kwargs["password"] = password
@@ -59,9 +63,9 @@ async def try_ssh_login(
         conn = await asyncssh.connect(**connect_kwargs)
         return True, ""
     except asyncssh.Error as e:
-        return False, str(e)[:200]
+        return False, _localized_ssh_error(e)
     except Exception as e:
-        return False, str(e)[:200]
+        return False, _localized_ssh_error(e)
     finally:
         if conn is not None:
             try:
@@ -240,7 +244,10 @@ class AsyncSSHPool:
             conn = await asyncssh.connect(**connect_kwargs)
             return conn
         except asyncssh.Error as e:
-            raise ConnectionError(f"SSH connection failed to {server.domain}:{server.port}: {e}") from e
+            msg = _localized_ssh_error(e)
+            raise ConnectionError(
+                f"SSH 连接失败 {server.domain}:{server.port or 22}：{msg}"
+            ) from e
 
     async def _evict_one(self) -> bool:
         """Evict the oldest idle connection from the pool.
@@ -459,7 +466,7 @@ async def exec_ssh_command(server: Server, command: str, timeout: int = 30) -> d
         return {
             "status": "timeout",
             "stdout": "",
-            "stderr": f"Command timed out after {timeout}s",
+            "stderr": f"命令执行超时（{timeout} 秒）",
             "exit_code": -1,
         }
     except ConnectionError as e:

server/utils/sync_error_message.py

@@ -11,7 +11,15 @@ _PATTERNS: list[tuple[re.Pattern[str], str]] = [
     (re.compile(r"rsync error:\s*error in rsync protocol data stream", re.I), "rsync 错误：协议数据流异常"),
     (re.compile(r"rsync error:\s*", re.I), "rsync 错误："),
     (re.compile(r"rsync:\s*", re.I), "rsync："),
+    (re.compile(r"Host key is not trusted for host", re.I), "SSH 主机密钥未信任（未知主机）"),
     (re.compile(r"Host key verification failed", re.I), "SSH 主机密钥验证失败"),
+    (re.compile(r"Permission denied for user .+ on host", re.I), "SSH 认证失败：用户名或密码/密钥不正确"),
+    (re.compile(r"Login failed for user", re.I), "SSH 登录失败：用户名或密码/密钥不正确"),
+    (re.compile(r"Key exchange failed", re.I), "SSH 密钥交换失败"),
+    (re.compile(r"Disconnect reason:", re.I), "SSH 连接断开："),
+    (re.compile(r"OpenSSH protocol error", re.I), "SSH 协议错误"),
+    (re.compile(r"Connection lost", re.I), "SSH 连接丢失"),
+    (re.compile(r"Connection closed", re.I), "SSH 连接已关闭"),
     (re.compile(r"Could not resolve hostname", re.I), "无法解析主机名"),
     (re.compile(r"Permission denied,\s*please try again", re.I), "权限被拒绝，请重试"),
     (re.compile(r"Permission denied\s*\(publickey", re.I), "公钥认证失败，权限被拒绝"),
@@ -56,6 +64,11 @@ def _is_primarily_chinese(text: str) -> bool:
     return cjk / len(compact) >= 0.2
 
 
+def translate_ssh_error_message(message: str | None) -> str | None:
+    """Return a Chinese-localized SSH/rsync error string, or the original if already Chinese."""
+    return translate_sync_error_message(message)
+
+
 def translate_sync_error_message(message: str | None) -> str | None:
     """Return a Chinese-localized push/sync error string, or the original if already Chinese."""
     if message is None:

tests/test_security_unit.py

@@ -72,14 +72,14 @@ def test_sanitize_external_message_redacts_mysql_url():
     cleaned = sanitize_external_message(raw)
     assert "SecretPass" not in cleaned
     assert "mysql+pymysql://" not in cleaned
-    assert "<connection-url-redacted>" in cleaned
+    assert "<连接地址已隐藏>" in cleaned
 
 
 def test_sanitize_external_message_redacts_bearer_token():
     raw = "auth failed Bearer eyJhbGciOiJIUzI1NiJ9.payload.sig"
     cleaned = sanitize_external_message(raw)
     assert "eyJhbGci" not in cleaned
-    assert "Bearer <redacted>" in cleaned
+    assert "Bearer <已隐藏>" in cleaned
 
 
 def test_sanitize_external_message_truncates_long_text():

tests/test_sync_error_message.py

@@ -37,3 +37,15 @@ class TestTranslateSyncErrorMessage:
 
     def test_server_not_found(self):
         assert translate_sync_error_message("Server not found") == "服务器不存在"
+
+    def test_host_key_not_trusted(self):
+        msg = translate_sync_error_message(
+            "Host key is not trusted for host 66.154.115.131"
+        )
+        assert "主机密钥未信任" in msg
+
+    def test_permission_denied_for_user(self):
+        msg = translate_sync_error_message(
+            "Permission denied for user root on host 66.154.115.131"
+        )
+        assert "认证失败" in msg