#8 Provider/Tunnel 稳定持久 id：凭证与可编辑字段解耦

app.py

@@ -20,7 +20,7 @@
 from mpconf import config_store
 from shellui.bridge_protocol import ACTION_OPEN_PATH, ACTION_RECONNECT_PROXY
 from capture.capture import DEFAULT_CAPTURE_DIR, DEFAULT_CAPTURE_PORT
-from mpconf.config import load_config, save_config, merge_config, DEFAULT_CONFIG
+from mpconf.config import IdentityMigrationError, load_config, save_config, merge_config, DEFAULT_CONFIG
 from shellui.log_window import LogBuffer, show_log_window
 from shellui.webview_window import show_config_window
 from shellui.menu_builder import MenuBuilder, MenuState, _status_color_for_connection
@@ -67,7 +67,16 @@ def _setup_logging():
 
 class MagicProxyApp(rumps.App):
     def __init__(self):
-        cfg = load_config()
+        try:
+            cfg = load_config()
+        except IdentityMigrationError as exc:
+            # 迁移可行动错误（显式重复 id）：绝不带病运行——弹窗给出
+            # 处置指引后退出，原配置文件未被动过
+            rumps.alert(
+                "Magic AI Router",
+                f"配置包含重复的隧道 id，无法安全启动。\n\n{exc}\n\n"
+                "请打开配置文件修正重复 id 后重启应用。")
+            raise SystemExit(1)
         self._config = merge_config(cfg)
         self._stats = Stats()
         self.VERSION = VERSION

docs/adr/002-config-representation-and-masking.md

@@ -59,3 +59,9 @@ Suanpan provider 的 `api_key` 发给设置窗（WKWebView）时，不再发掩
 - `on_sp_saved` 回调只在完整提交后触发；失败返回结构化阶段（validate/journal/mp/sp/keychain）且错误不含 secret。
 - 跨文件提交崩溃由启动时 `recover()` 幂等重放 journal 补齐。
 - invalid 主文件不覆盖最后已知良好的 `.bak`；首创建与保存共用 0600/0700 权限路径。
+
+## 增补（2026-08-20，issue #8）：稳定 id 与凭证所有权
+
+- Tunnel 持不可变 `id`（`t-<sha1(user@host:port)[:10]>`，装载期确定性迁移赋值；重复身份/重复 id 抛可行动错误，不猜 secret 归属）；Keychain 账户优先 `tunnel:<id>`，无 id 时 legacy `user@host:port` 仅作迁移期回退读。重命名/改地址不改 id 不丢密码。
+- Provider 持 `id`（`p-<sha1(name)[:10]>`）；api_key 的 keep/replace/clear（`_restore_key`）按 id 匹配旧值——重命名保住 key，id 不同绝不按名串接。legacy 无 id 旧档按名回退。
+- 旧 Keychain 条目迁移：password 隧道在下次提交时经 ConfigStateStore.prepare 的 re-pin 迁到 id 账户。re-pin 只在 id==当前身份哈希时读 legacy（身份编辑过的隧道绝不读——legacy 账户可能属于别的实体，不猜归属）；收敛后 legacy-only 删除随事务完成。删除隧道时双账户清理随事务执行。迁移错误（重复身份/重复 id）为 IdentityMigrationError：不触发 .bak 隔离、UI 侧降级为 _load_error 提示，绝不静默猜测 secret 归属。

mpconf/config.py

@@ -46,6 +46,54 @@
 }
 
 
+class IdentityMigrationError(ValueError):
+    """稳定 id 迁移的可行动错误（重复身份/重复 id）——绝不与文件损坏
+    混同：不触发 .bak 隔离，原样上抛（issue #8）。"""
+
+
+def stable_tunnel_id(user: str, host: str, port) -> str:
+    """确定性 id：t-<sha1(user@host:port)[:10]>——同身份恒同 id（issue #8）。"""
+    import hashlib
+    basis = f"{user or ''}@{host or ''}:{port or 22}"
+    return "t-" + hashlib.sha1(basis.encode("utf-8")).hexdigest()[:10]
+
+
+def assign_stable_ids(tunnels) -> int:
+    """为无 id 的隧道赋确定性 id；重复身份/重复 id 抛可行动错误。
+
+    返回迁移数量。已有 id 一律不动（重命名/改地址不影响）。
+    """
+    seen_ids, seen_identity = {}, {}
+    migrated = 0
+    for t in tunnels or []:
+        ident = f"{t.get('ssh_user', '')}@{t.get('ssh_host', '')}:{t.get('ssh_port', 22)}"
+        if t.get("id"):
+            if t["id"] in seen_ids:
+                raise IdentityMigrationError(
+                    f"隧道配置存在重复 id：{t['id']}（请修正配置文件后重试）")
+            seen_ids[t["id"]] = ident
+            seen_identity[ident] = True
+            continue
+        ordinal = 2 if ident in seen_identity else 1
+        if ordinal > 1:
+            # legacy 同身份双隧道（如 key+password 并存）本合法——确定性
+            # 序数后缀区分 id；两隧道仍共享同一 legacy 凭证槽（与迁移前
+            # 行为一致），不猜归属。显式手写重复 id 才致命。
+            logger.warning("隧道重复身份 %s：以序数后缀区分 id", ident)
+        seen_identity[ident] = True
+        import hashlib as _hl
+        basis = f"{t.get('ssh_user', '')}@{t.get('ssh_host', '')}:{t.get('ssh_port', 22)}"
+        if ordinal > 1:
+            basis += f"#{ordinal}"
+        t["id"] = "t-" + _hl.sha1(basis.encode("utf-8")).hexdigest()[:10]
+        if t["id"] in seen_ids:
+            raise IdentityMigrationError(
+                f"隧道配置存在重复 id：{t['id']}（请修正配置文件后重试）")
+        seen_ids[t["id"]] = ident
+        migrated += 1
+    return migrated
+
+
 def load_config(path=None):
     """Load and migrate config; returns merged dict or None."""
     p = path or get_path("mp")
@@ -56,6 +104,9 @@ def load_config(path=None):
             cfg = json.load(f)
         before = json.dumps(cfg, sort_keys=True)
         migrated = _migrate(cfg)
+        # issue #8：迁移错误（重复身份/id）是可行动错误——绝不与损坏
+        # 混同进 .bak 隔离；原样上抛让编排层给出可行动提示
+        assign_stable_ids(migrated.get("tunnels") or [])
         if json.dumps(migrated, sort_keys=True) != before and not save_config(migrated, p):
             # The migrated dict is already clean in memory, but the file on
             # disk is still the pre-migration version — it may hold plaintext
@@ -71,6 +122,11 @@ def load_config(path=None):
                     "Migrated config could not be written AND the old file "
                     "could not be isolated")
         return migrated
+    except IdentityMigrationError as e:
+        # 迁移可行动错误：不隔离、不改写——上抛（JSONDecodeError 等仍走
+        # 既有损坏隔离路径）
+        logger.error("配置迁移失败（需人工处理，原文件未动）：%s", e)
+        raise
     except (json.JSONDecodeError, OSError, TypeError, ValueError) as e:
         backup = p + ".bak"
         try:

mpconf/config_state.py

@@ -141,19 +141,50 @@ def prepare(self, mp=None, sp=None) -> CommitPlan:
         # merge 默认值必须在校验之后：merge_config 会把非法端口/负保留
         # 静默重置为默认，前置会让 mp 侧数值约束在真实入口永不触发
         if mp_c is not None:
+            if mp_c.get("_load_error"):
+                return CommitPlan(False, [
+                    f"配置装载失败，已阻止保存以防覆盖：{mp_c['_load_error']}"])
             from mpconf.config import merge_config
             mp_c = merge_config(mp_c)
+        if sp_c is not None:
+            if sp_c.get("_load_error"):
+                return CommitPlan(False, [
+                    f"配置装载失败，已阻止保存以防覆盖：{sp_c['_load_error']}"])
+            # 掩码 key 恢复（原 save_config_dict 语义——live PUT 唯一保存
+            # 路径在此）：按 id 匹配旧档恢复真实 key；legacy 无 id 档按名
+            sp_c = self._restore_masked_sp_keys(sp_c)
         kc_sets, kc_dels = [], []
         if mp_c is not None:
             import copy
             mp_c = copy.deepcopy(mp_c)
+            # 删除的隧道（id 在旧档、不在候选）：双账户清理 secret
+            old_mp = self._read_mp_current() or {}
+            new_ids = {t.get("id") for t in mp_c.get("tunnels") or []
+                       if isinstance(t, dict)}
+            for t in old_mp.get("tunnels") or []:
+                if isinstance(t, dict) and t.get("id") and t["id"] not in new_ids:
+                    kc_dels.append(("all", t))
             for t in mp_c.get("tunnels") or []:
                 t.pop("has_password", None)      # 服务端注入的只读字段
                 t.pop("capture_active", None)
                 pw = t.pop("password", None)
-                # 显式切换离 password 才删（部分载荷不得静默清密）
                 if pw:
                     kc_sets.append((dict(t), pw))
+                elif (t.get("auth_type") == "password" and t.get("id")
+                      and self._keychain is not None):
+                    # issue #8 re-pin——只在 id==当前身份哈希时读 legacy：
+                    # 身份编辑过的隧道 id 与地址已脱钩，legacy 账户可能
+                    # 属于别的实体（Y 改址到 X 旧地址会串走 X 的密码），
+                    # 绝不猜测归属。收敛：写入 id 账户 + legacy-only 删除。
+                    from mpconf.config import stable_tunnel_id
+                    if t["id"] == stable_tunnel_id(
+                            t.get("ssh_user", ""), t.get("ssh_host", ""),
+                            t.get("ssh_port", 22)):
+                        legacy = {k: v for k, v in t.items() if k != "id"}
+                        old_pw = self._keychain.get_password(legacy)
+                        if old_pw:
+                            kc_sets.append((dict(t), old_pw))
+                            kc_dels.append(("legacy-only", legacy))
                 elif "auth_type" in t and t.get("auth_type") != "password":
                     kc_dels.append(dict(t))
         return CommitPlan(True, [], mp_c, sp_c, kc_sets, kc_dels)
@@ -180,6 +211,46 @@ def _atomic_install(self, path, text):
         if not config_store.atomic_write(path, text):  # 唯一安全写入口
             raise OSError(f"atomic_write failed: {path}")
 
+    def _restore_masked_sp_keys(self, sp_c: dict) -> dict:
+        """api_key_set 掩码契约：UI 回传 api_key=null+api_key_set=true 表示
+        保留旧 key——按 id（或 legacy 名）从当前磁盘档恢复真实值。"""
+        import copy
+        sp_c = copy.deepcopy(sp_c)
+        sp_c.pop("_load_error", None)  # 装载错误标记永不落盘
+        try:
+            with open(self.sp_path) as f:
+                old = yaml.safe_load(f) or {}
+        except (OSError, yaml.YAMLError):
+            old = {}
+        old_by_id = {p.get("id"): p for p in (old.get("providers") or {}).values()
+                     if isinstance(p, dict) and p.get("id")}
+        old_by_name = old.get("providers") or {}
+        for name, p in sp_c.get("providers", {}).items():
+            if not isinstance(p, dict):
+                continue
+            old_p = old_by_id.get(p.get("id"))
+            if old_p is None:
+                legacy = old_by_name.get(name)
+                if isinstance(legacy, dict) and not legacy.get("id"):
+                    old_p = legacy
+            keep = bool(p.pop("api_key_set", False))
+            new_key = p.get("api_key")
+            p["api_key"] = (old_p or {}).get("api_key") if (keep and not new_key) \
+                else (new_key or None)
+        top_keep = bool(sp_c.pop("api_key_set", False))
+        top_new = sp_c.get("api_key")
+        sp_c["api_key"] = old.get("api_key") if (top_keep and not top_new) \
+            else (top_new or None)
+        return sp_c
+
+    def _read_mp_current(self) -> dict | None:
+        try:
+            with open(self.mp_path) as f:
+                data = json.load(f)
+            return data if isinstance(data, dict) else None
+        except (OSError, ValueError):
+            return None
+
     def _current_text(self, path):
         try:
             with open(path) as f:
@@ -254,10 +325,17 @@ def commit(self, plan, on_committed=None) -> SaveResult:
                         f"隧道 {tunnel.get('name', '?')} 的密码保存到钥匙串失败")
                     break
             if not keychain_errors:
-                for tunnel in plan.keychain_dels:
-                    if not self._keychain.delete_password(tunnel):
+                for entry in plan.keychain_dels:
+                    if isinstance(entry, tuple):
+                        mode, tunnel = entry
+                    else:
+                        mode, tunnel = "all", entry
+                    ok = (self._keychain.delete_legacy_password(tunnel)
+                          if mode == "legacy-only"
+                          else self._keychain.delete_password(tunnel))
+                    if not ok:
                         keychain_errors.append(
-                            f"隧道 {tunnel.get('name', '?')} 的旧密码清理失败")
+                            f"隧道 {tunnel.get('name', tunnel.get('ssh_host', '?'))} 的旧密码清理失败")
         if keychain_errors:
             self._rollback(payload)  # 文件回到旧内容：不暴露部分新状态
             return SaveResult(False, "keychain", keychain_errors)

services/config_server.py

@@ -42,7 +42,15 @@
 
 
 def _read_mp():
-    cfg = merge_config(load_config())
+    try:
+        cfg = merge_config(load_config())
+    except Exception:
+        # 迁移可行动错误等：降级为带 _load_error 的空态供 UI 提示，
+        # /api/state 不 500（UI 保存被 validateConfig/prepare 双带阻断）
+        from mpconf.config import IdentityMigrationError
+        import logging as _lg
+        _lg.getLogger("magic-proxy.config_server").exception("_read_mp degraded")
+        return {"_load_error": "Magic Proxy 配置装载失败，已阻止保存以防覆盖"}
     if not cfg:
         return {}
     for t in cfg.get("tunnels", []):

shellui/config_ui.html

@@ -360,7 +360,7 @@
   if(view==='tunnel')return{
     current_tunnel:numberDefault(mp.current_tunnel,0),
     tunnels:(mp.tunnels||[]).map(t=>({
-      name:t.name||'',ssh_user:t.ssh_user||'',ssh_host:t.ssh_host||'',
+      id:t.id||'',name:t.name||'',ssh_user:t.ssh_user||'',ssh_host:t.ssh_host||'',
       ssh_port:numberDefault(t.ssh_port,22),auth_type:t.auth_type||'key',
       ssh_key:t.ssh_key||'',ssh_compression:t.ssh_compression!==false,
       password:t.password||(t.has_password?'__saved_password__':''),
@@ -385,6 +385,7 @@
     Object.keys(sp.providers||{}).sort().forEach(name=>{
       const p=sp.providers[name]||{};
       providers[name]={
+        id:p.id||'',
         base_url:p.base_url||'',
         api_key:p.api_key||(p.api_key_set?'__saved_api_key__':''),
         api_key_env:p.api_key_env||'',auth_header:p.auth_header||'',
@@ -407,6 +408,8 @@
 function validateConfig(S){
   // Pre-save validation; returns a list of human-readable error strings.
   const errors=[];
+  if(S&&S.sp&&S.sp._load_error)errors.push('Suanpan 配置装载失败，已阻止保存以防覆盖：'+S.sp._load_error);
+  if(S&&S.mp&&S.mp._load_error)errors.push('Magic Proxy 配置装载失败，已阻止保存以防覆盖：'+S.mp._load_error);
   (S.mp.tunnels||[]).forEach((t,i)=>{
     if(!t.ssh_host)errors.push(`隧道 ${i+1}: 未填写地址`);
     if(t.ssh_port&&(t.ssh_port<1||t.ssh_port>65535))errors.push(`隧道 ${i+1}: SSH 端口无效`);
@@ -1318,6 +1321,7 @@ <h3 id="cc-modal-title">写入 Claude Code 配置</h3>
   const sw=d.querySelector('[data-pf="enabled"]');p.enabled=sw&&sw.getAttribute('aria-checked')==='true';
   const swAN=d.querySelector('[data-pf="anthropic_native"]');p.anthropic_native=swAN&&swAN.getAttribute('aria-checked')==='true';
   if(newName!==activeProvider){
+    if(S.sp.providers[newName]){toast('供应商名称「'+newName+'」已被占用，未重命名',true);return;}
     const oldName=activeProvider;
     const np={};for(const[k2,v2]of Object.entries(S.sp.providers))np[k2===activeProvider?newName:k2]=v2;S.sp.providers=np;activeProvider=newName;
     // Update rule references

suanpan/config.py

@@ -20,6 +20,9 @@
 class ProviderConfig(BaseModel):
     model_config = ConfigDict(validate_assignment=True)
 
+    # 稳定身份（issue #8）：不可变、无业务含义；显示名可随意改，
+    # api_key 的 keep/replace/clear 恢复按 id 匹配旧值。
+    id: str | None = None
     base_url: str
     api_key: str | None = None
     api_key_env: str | None = None
@@ -155,6 +158,32 @@ def _restore_key(new_val, old_val, keep):
     return new_val or None
 
 
+from mpconf.config import IdentityMigrationError  # noqa: E402
+
+
+def assign_provider_ids(cfg: dict) -> int:
+    """为无 id 的 provider 赋确定性 id（p-<sha1(name)[:10]>）。
+
+    重复 id 抛可行动错误——不猜测 secret 归属（issue #8）。
+    """
+    import hashlib
+    seen, migrated = set(), 0
+    for name, p in (cfg.get("providers") or {}).items():
+        if not isinstance(p, dict):
+            continue
+        pid = p.get("id")
+        if pid:
+            if pid in seen:
+                raise IdentityMigrationError(
+                    f"供应商存在重复 id：{pid}（请修正配置文件后重试）")
+            seen.add(pid)
+            continue
+        p["id"] = "p-" + hashlib.sha1(name.encode("utf-8")).hexdigest()[:10]
+        seen.add(p["id"])
+        migrated += 1
+    return migrated
+
+
 def load_config_raw(path: Path | str) -> dict:
     """Read raw (unmasked) config dict from YAML. Returns {} on any error."""
     p = Path(path)
@@ -164,6 +193,10 @@ def load_config_raw(path: Path | str) -> dict:
         data = yaml.safe_load(p.read_text())
     except Exception:
         return {}
+    if isinstance(data, dict):
+        # issue #8：装载即赋幂等 id；重复 id 等可行动错误上抛——
+        # 「任何错误返回 {}」的旧契约对迁移错误失真，读方按需捕获
+        assign_provider_ids(data)
     return data if isinstance(data, dict) else {}
 
 
@@ -172,8 +205,13 @@ def load_config_masked(path: Path | str) -> dict:
 
     Each provider (and the top-level key) is rewritten to ``api_key: None``
     plus ``api_key_set: bool`` telling the UI whether a key is saved.
+    迁移可行动错误降级为 ``{"_load_error": msg}``——UI 可展示，服务不 500。
     """
-    cfg = load_config_raw(path)
+    try:
+        cfg = load_config_raw(path)
+    except ValueError as e:
+        return {"_load_error": str(e), "providers": {}, "rules": [],
+                "router": {}}
     for p in cfg.get("providers", {}).values():
         if isinstance(p, dict):
             p["api_key_set"] = bool(p.get("api_key"))
@@ -190,11 +228,21 @@ def save_config_dict(data: dict, path: Path | str) -> tuple[bool, str | None]:
     Returns (ok, error_msg).
     """
     old = load_config_raw(path)
-    old_providers = old.get("providers", {})
+    old_by_id = {p.get("id"): p
+                 for p in old.get("providers", {}).values()
+                 if isinstance(p, dict) and p.get("id")}
+    old_by_name = old.get("providers", {})
     for name, p in data.get("providers", {}).items():
+        # id 命中 → 同一实体（重命名后 key 仍恢复）。id 未命中时仅当旧档
+        # 无 id（legacy）才按名回退——有 id 而不同 ≠ 同一实体，绝不串接。
+        old_p = old_by_id.get(p.get("id"))
+        if old_p is None:
+            legacy = old_by_name.get(name, {})
+            if isinstance(legacy, dict) and not legacy.get("id"):
+                old_p = legacy
         p["api_key"] = _restore_key(
             p.get("api_key"),
-            old_providers.get(name, {}).get("api_key"),
+            (old_p or {}).get("api_key"),
             bool(p.pop("api_key_set", False)))
     data["api_key"] = _restore_key(
         data.get("api_key"), old.get("api_key"),