diff --git a/app.py b/app.py index 03e8270..a037fba 100644 --- a/app.py +++ b/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 diff --git a/docs/adr/002-config-representation-and-masking.md b/docs/adr/002-config-representation-and-masking.md index 4a9bfb2..d4436c2 100644 --- a/docs/adr/002-config-representation-and-masking.md +++ b/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-`，装载期确定性迁移赋值；重复身份/重复 id 抛可行动错误，不猜 secret 归属）；Keychain 账户优先 `tunnel:`，无 id 时 legacy `user@host:port` 仅作迁移期回退读。重命名/改地址不改 id 不丢密码。 +- Provider 持 `id`（`p-`）；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 归属。 diff --git a/mpconf/config.py b/mpconf/config.py index 8641a8e..67bd325 100644 --- a/mpconf/config.py +++ b/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-——同身份恒同 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: diff --git a/mpconf/config_state.py b/mpconf/config_state.py index 8754275..0c1f33a 100644 --- a/mpconf/config_state.py +++ b/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) diff --git a/services/config_server.py b/services/config_server.py index 519f0df..acef2e7 100644 --- a/services/config_server.py +++ b/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", []): diff --git a/shellui/config_ui.html b/shellui/config_ui.html index 4fbe838..2305748 100644 --- a/shellui/config_ui.html +++ b/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 @@

写入 Claude Code 配置

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 diff --git a/suanpan/config.py b/suanpan/config.py index 30592f1..0a2f7e5 100644 --- a/suanpan/config.py +++ b/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-）。 + + 重复 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"), diff --git a/sysctl/keychain.py b/sysctl/keychain.py index 0db3514..f9157d7 100644 --- a/sysctl/keychain.py +++ b/sysctl/keychain.py @@ -14,17 +14,25 @@ def _account(tunnel: dict) -> str: + """凭证账户名：优先稳定 id（issue #8）；无 id 时为 legacy 推导。""" + stable_id = tunnel.get("id") + if stable_id: + return f"tunnel:{stable_id}" + return _legacy_account(tunnel) + + +def _legacy_account(tunnel: dict) -> str: user = tunnel.get("ssh_user", "") host = tunnel.get("ssh_host", "") port = tunnel.get("ssh_port", 22) return f"{user}@{host}:{port}" -def _base_query(tunnel: dict) -> dict: +def _base_query(tunnel: dict, account: str | None = None) -> dict: return { Security.kSecClass: Security.kSecClassGenericPassword, Security.kSecAttrService: SERVICE, - Security.kSecAttrAccount: _account(tunnel), + Security.kSecAttrAccount: account or _account(tunnel), } @@ -33,8 +41,9 @@ def set_password(tunnel: dict, password: str) -> bool: return False try: # Replace any existing entry so -U semantics (update-in-place) hold. - Security.SecItemDelete(_base_query(tunnel)) - attrs = dict(_base_query(tunnel)) + final_account = _account(tunnel) + Security.SecItemDelete(_base_query(tunnel, final_account)) + attrs = _base_query(tunnel, final_account) attrs[Security.kSecValueData] = password.encode("utf-8") status = Security.SecItemAdd(attrs, None) ok = status[0] == Security.errSecSuccess if isinstance(status, tuple) \ @@ -51,24 +60,40 @@ def set_password(tunnel: dict, password: str) -> bool: def get_password(tunnel: dict) -> str: if not tunnel.get("ssh_host"): return "" + accounts = [_account(tunnel)] + legacy = _legacy_account(tunnel) + if legacy not in accounts: + accounts.append(legacy) # 迁移期回退读（issue #8） try: - query = dict(_base_query(tunnel)) - query[Security.kSecReturnData] = True - query[Security.kSecMatchLimit] = Security.kSecMatchLimitOne - status, data = Security.SecItemCopyMatching(query, None) - if status == Security.errSecSuccess and data is not None: - return bytes(data).decode("utf-8") + for account in accounts: + query = _base_query(tunnel, account) + query[Security.kSecReturnData] = True + query[Security.kSecMatchLimit] = Security.kSecMatchLimitOne + status, data = Security.SecItemCopyMatching(query, None) + if status == Security.errSecSuccess and data is not None: + return bytes(data).decode("utf-8") except Exception as e: # noqa: BLE001 logger.warning("Keychain get failed: %s", type(e).__name__) return "" +def delete_legacy_password(tunnel: dict) -> bool: + """仅清 legacy 账户（user@host:port）——re-pin 收敛用，不动 id 账户。""" + try: + Security.SecItemDelete(_base_query(tunnel, _legacy_account(tunnel))) + return True + except Exception as e: # noqa: BLE001 + logger.warning("Keychain legacy delete failed: %s", type(e).__name__) + return False + + def delete_password(tunnel: dict) -> bool: """删除隧道密码。返回是否成功（条目本就不存在视为成功）。""" if not tunnel.get("ssh_host"): return True try: - Security.SecItemDelete(_base_query(tunnel)) + for account in {_account(tunnel), _legacy_account(tunnel)}: + Security.SecItemDelete(_base_query(tunnel, account)) return True except Exception as e: # noqa: BLE001 logger.warning("Keychain delete failed: %s", type(e).__name__) diff --git a/tests/js/model.test.mjs b/tests/js/model.test.mjs index 21f9c3e..4cfb9d0 100644 --- a/tests/js/model.test.mjs +++ b/tests/js/model.test.mjs @@ -667,6 +667,12 @@ test("ccBackupNote renders both backup branches", () => { assert.equal(L.ccBackupNote({ ok: false }), ""); }); +test("validateConfig blocks save when sp has _load_error (#8)", () => { + const errs = L.validateConfig( + L.normalizeState({ sp: { _load_error: "重复 id" } })); + assert.ok(errs.some((e) => e.includes("已阻止保存"))); +}); + // ── 保存流（saveFlow）：两阶段保存状态机（架构候选 1）────────────── // 事故回归：99999 端口写库、幽灵 api_key 保存都发生在这条流上。 diff --git a/tests/test_config_state.py b/tests/test_config_state.py index acc3646..72c7a4e 100644 --- a/tests/test_config_state.py +++ b/tests/test_config_state.py @@ -378,6 +378,98 @@ def test_corrupt_journal_recovered_and_cleared(self): "损坏 journal 必须清除，不得永久残留") + def test_repin_guard_blocks_crosswired_identity(self): + """Y 改址到 X 旧地址：id 与身份哈希不符 → 绝不读 legacy（防串线）。""" + ops = [] + class KC: + def get_password(self, t): + ops.append("get") + return "x-secret" + def set_password(self, t, pw): + ops.append(("set", t.get("id"))) + return True + def delete_password(self, t): + return True + def delete_legacy_password(self, t): + ops.append(("del-legacy",)) + return True + d = tempfile.TemporaryDirectory() + self.addCleanup(d.cleanup) + store = ConfigStateStore( + mp_path=str(Path(d.name) / "m.json"), + sp_path=str(Path(d.name) / "s.yaml"), keychain=KC()) + # id=哈希(旧地址)，现身份=新地址 → 守卫拒绝 + from mpconf.config import stable_tunnel_id + old_addr_id = stable_tunnel_id("u", "old.example.com", 22) + plan = store.prepare(mp={"tunnels": [ + {"id": old_addr_id, "name": "y", "ssh_user": "u", + "ssh_host": "new.example.com", "ssh_port": 22, + "auth_type": "password"}]}) + store.commit(plan) + self.assertNotIn("get", ops, "身份编辑过 → 不读 legacy，不猜归属") + + def test_deleted_tunnel_secrets_cleaned_both_accounts(self): + ops = [] + class KC: + def get_password(self, t): + return "" + def set_password(self, t, pw): + return True + def delete_password(self, t): + ops.append(("del-all", t.get("id"))) + return True + def delete_legacy_password(self, t): + ops.append(("del-legacy", t.get("id"))) + return True + d = tempfile.TemporaryDirectory() + self.addCleanup(d.cleanup) + store = ConfigStateStore( + mp_path=str(Path(d.name) / "m.json"), + sp_path=str(Path(d.name) / "s.yaml"), keychain=KC()) + # 先建档含 t-gone，再保存不含它 + Path(store.mp_path).write_text(json.dumps( + {"tunnels": [{"id": "t-gone", "name": "g", "ssh_host": "h", + "auth_type": "password", + "ssh_user": "u", "ssh_port": 22}]})) + plan = store.prepare(mp={"tunnels": []}) + self.assertTrue(plan.ok) + store.commit(plan) + self.assertIn(("del-all", "t-gone"), ops, "删除隧道双账户清理") + + + def test_prepare_rejects_sp_with_load_error(self): + d = tempfile.TemporaryDirectory() + self.addCleanup(d.cleanup) + store = ConfigStateStore( + mp_path=str(Path(d.name) / "m.json"), + sp_path=str(Path(d.name) / "s.yaml")) + plan = store.prepare(sp={"_load_error": "装载失败", "providers": {}}) + self.assertFalse(plan.ok) + self.assertIn("已阻止保存", plan.errors[0]) + + def test_restore_strips_load_error_marker(self): + d = tempfile.TemporaryDirectory() + self.addCleanup(d.cleanup) + store = ConfigStateStore( + mp_path=str(Path(d.name) / "m.json"), + sp_path=str(Path(d.name) / "s.yaml")) + Path(store.sp_path).write_text( + yaml.dump({"listen_port": 9527, "providers": { + "a": {"id": "p-1", "base_url": "https://a.test", + "api_key": "sk-1", "models": ["m"]}}})) + plan = store.prepare(sp={ + "_load_error": "x", "listen_port": 9527, + "providers": {"a": {"id": "p-1", "base_url": "https://a.test", + "api_key": None, "api_key_set": True, + "models": ["m"]}}}) + # _load_error 在此之前已被拒——restore 剥离由内部直测 + self.assertFalse(plan.ok) # 拒收优先 + # 直测 restore 剥离 + cleaned = store._restore_masked_sp_keys( + {"_load_error": "x", "providers": {}, "api_key": None}) + self.assertNotIn("_load_error", cleaned) + + if __name__ == "__main__": unittest.main() @@ -427,3 +519,222 @@ def delete_password(self, t): self.assertFalse(result.ok) self.assertEqual(result.stage, "keychain") self.assertTrue(any("旧密码清理失败" in e for e in result.errors)) + + +class TestStableIdentityMigration(unittest.TestCase): + """issue #8 S2：load/prepare 期确定性 id 迁移；重命名不改 id。""" + + def test_legacy_tunnel_gets_deterministic_id(self): + from mpconf.config import assign_stable_ids + tunnels = [{"name": "a", "ssh_user": "u", "ssh_host": "h", + "ssh_port": 22}] + ids = assign_stable_ids(tunnels) + self.assertTrue(tunnels[0]["id"].startswith("t-")) + self.assertEqual(len(tunnels[0]["id"]), 12) + self.assertEqual(ids, 1, "恰一个迁移") + # 确定性：同身份重算同 id + again = [{"name": "a", "ssh_user": "u", "ssh_host": "h", + "ssh_port": 22}] + assign_stable_ids(again) + self.assertEqual(again[0]["id"], tunnels[0]["id"]) + + def test_existing_id_untouched_and_stable_across_edits(self): + from mpconf.config import assign_stable_ids + tunnels = [{"id": "t-keepme1234", "ssh_user": "u", + "ssh_host": "h", "ssh_port": 22}] + assign_stable_ids(tunnels) + self.assertEqual(tunnels[0]["id"], "t-keepme1234") + tunnels[0]["ssh_host"] = "changed.example.com" # 编辑地址 + tunnels[0]["name"] = "renamed" + assign_stable_ids(tunnels) + self.assertEqual(tunnels[0]["id"], "t-keepme1234", + "重命名/改地址不改变 id") + + def test_duplicate_legacy_identity_gets_deterministic_suffix(self): + """legacy 同身份双隧道本合法——确定性序数后缀，不拒启。""" + from mpconf.config import assign_stable_ids + dup = [{"name": "a", "ssh_user": "u", "ssh_host": "h", "ssh_port": 22}, + {"name": "b", "ssh_user": "u", "ssh_host": "h", "ssh_port": 22}] + n = assign_stable_ids(dup) + self.assertEqual(n, 2) + self.assertNotEqual(dup[0]["id"], dup[1]["id"]) + again = [{"name": "a", "ssh_user": "u", "ssh_host": "h", "ssh_port": 22}, + {"name": "b", "ssh_user": "u", "ssh_host": "h", "ssh_port": 22}] + assign_stable_ids(again) + self.assertEqual([t["id"] for t in again], [t["id"] for t in dup], + "确定性：重算同序") + + def test_duplicate_explicit_ids_fails_actionable(self): + from mpconf.config import assign_stable_ids + dup = [{"id": "t-same", "ssh_user": "u1", "ssh_host": "h", "ssh_port": 22}, + {"id": "t-same", "ssh_user": "u2", "ssh_host": "h", "ssh_port": 22}] + with self.assertRaises(ValueError) as ctx: + assign_stable_ids(dup) + self.assertIn("t-same", str(ctx.exception)) + + +class TestProviderIdSemantics(unittest.TestCase): + """issue #8 S3：rename 后 keep/replace/clear 三态按 id 正确。""" + + def _roundtrip(self, old_providers, new_providers): + import tempfile, os + from pathlib import Path + from suanpan.config import save_config_dict, load_config_raw + with tempfile.TemporaryDirectory() as d: + path = str(Path(d) / "s.yaml") + Path(path).write_text(yaml.dump( + {"providers": old_providers, "listen_port": 9527}, + allow_unicode=True)) + ok, err = save_config_dict( + {"providers": new_providers, "listen_port": 9527, + "api_key_set": False}, path) + self.assertTrue(ok, err) + return load_config_raw(path)["providers"] + + def test_rename_keeps_key_via_id(self): + saved = self._roundtrip( + old_providers={"old-name": {"id": "p-stable1", "base_url": + "https://a.test", "api_key": "sk-1", + "models": ["m"]}}, + new_providers={"new-name": {"id": "p-stable1", "base_url": + "https://a.test", "api_key": None, + "api_key_set": True, + "models": ["m"]}}) + self.assertEqual(saved["new-name"]["api_key"], "sk-1", + "重命名后 keep 语义按 id 恢复") + + def test_replace_by_id_wins_over_name_match(self): + saved = self._roundtrip( + old_providers={"a": {"id": "p-1", "base_url": "https://a.test", + "api_key": "sk-old", "models": ["m"]}}, + new_providers={"a": {"id": "p-2", "base_url": "https://a.test", + "api_key": None, "api_key_set": True, + "models": ["m"]}}) + self.assertIsNone(saved["a"]["api_key"], + "id 不同≠同一实体：不串接他者的 key") + + def test_clear_still_clears(self): + saved = self._roundtrip( + old_providers={"a": {"id": "p-1", "base_url": "https://a.test", + "api_key": "sk-old", "models": ["m"]}}, + new_providers={"a": {"id": "p-1", "base_url": "https://a.test", + "api_key": "", "api_key_set": False, + "models": ["m"]}}) + self.assertIsNone(saved["a"]["api_key"]) + + def test_load_migration_assigns_deterministic_provider_id(self): + from suanpan.config import assign_provider_ids, load_config_raw + import tempfile + from pathlib import Path + with tempfile.TemporaryDirectory() as d: + path = Path(d) / "s.yaml" + path.write_text( + 'providers:\n glm:\n base_url: https://a.test\n' + ' models: [m]\nlisten_port: 9527\n') + cfg = load_config_raw(path) + # 装载即赋（load_config_raw 内联 assign_provider_ids） + self.assertTrue(cfg["providers"]["glm"]["id"].startswith("p-"), + "装载即含确定性 id") + n2 = assign_provider_ids(cfg) + self.assertEqual(n2, 0, "幂等") + + def test_duplicate_provider_ids_fail_actionable(self): + from suanpan.config import assign_provider_ids + cfg = {"providers": { + "a": {"id": "p-dup", "base_url": "https://a.test"}, + "b": {"id": "p-dup", "base_url": "https://b.test"}}} + with self.assertRaises(ValueError) as ctx: + assign_provider_ids(cfg) + self.assertIn("p-dup", str(ctx.exception)) + + +import yaml # noqa: E402 — 测试用例内 dump 需要 + + +class TestTunnelSecretRepin(unittest.TestCase): + """issue #8：id 稳定后 legacy 密码随下次提交 re-pin 到 id 账户。""" + + def test_password_tunnel_without_new_pw_repins_legacy_secret(self): + ops = [] + class KC: + def get_password(self, t): + ops.append(("get", t.get("id") or "legacy")) + return "legacy-pw" if not t.get("id") else "" + def set_password(self, t, pw): + ops.append(("set", t.get("id"), pw)) + return True + def delete_password(self, t): + ops.append(("del", t.get("id"))) + return True + def delete_legacy_password(self, t): + ops.append(("del-legacy",)) + return True + d = tempfile.TemporaryDirectory() + self.addCleanup(d.cleanup) + store = ConfigStateStore( + mp_path=str(Path(d.name) / "m.json"), + sp_path=str(Path(d.name) / "s.yaml"), keychain=KC()) + from mpconf.config import stable_tunnel_id + real_id = stable_tunnel_id("u", "h", 22) + plan = store.prepare(mp={"tunnels": [ + {"id": real_id, "name": "a", "ssh_user": "u", + "ssh_host": "h", "ssh_port": 22, "auth_type": "password"}]}) + self.assertTrue(plan.ok) + result = store.commit(plan) + self.assertTrue(result.ok, result.errors) + self.assertIn(("set", real_id, "legacy-pw"), ops, + "旧密码经 legacy 回退读取后 re-pin 到 id 账户") + + +class TestRound3Holes(unittest.TestCase): + """三审两洞：mp _load_error 双带 + 显式分支身份登记。""" + + def test_explicit_id_tunnel_registers_identity(self): + """已迁移 A（id=hash 身份）+ 手工加同身份无 id B → B 不再静默同 id。""" + from mpconf.config import assign_stable_ids, stable_tunnel_id, IdentityMigrationError + real_id = stable_tunnel_id("u", "h", 22) + tunnels = [ + {"id": real_id, "name": "a", "ssh_user": "u", + "ssh_host": "h", "ssh_port": 22}, + {"name": "b", "ssh_user": "u", "ssh_host": "h", "ssh_port": 22}, + ] + n = assign_stable_ids(tunnels) + self.assertEqual(n, 1) # 只有 B 被迁移 + self.assertNotEqual(tunnels[1]["id"], tunnels[0]["id"], + "B 得序数后缀 id，绝不与 A 静默同 id") + + def test_fresh_assign_collision_detected(self): + from mpconf.config import assign_stable_ids, IdentityMigrationError + # 两条同身份隧道：第一条得 ordinal=1 id；第二条 #2 后缀 + # 但若 #2 id 撞显式 id —— 分配后查重兜底 + tunnels = [ + {"name": "a", "ssh_user": "u", "ssh_host": "h", "ssh_port": 22}, + {"name": "b", "ssh_user": "u", "ssh_host": "h", "ssh_port": 22}, + ] + n = assign_stable_ids(tunnels) + self.assertEqual(n, 2) # 正常路径不受影响 + + def test_prepare_rejects_mp_with_load_error(self): + d = tempfile.TemporaryDirectory() + self.addCleanup(d.cleanup) + store = ConfigStateStore( + mp_path=str(Path(d.name) / "m.json"), + sp_path=str(Path(d.name) / "s.yaml")) + plan = store.prepare(mp={"_load_error": "装载失败", "tunnels": []}) + self.assertFalse(plan.ok) + self.assertIn("已阻止保存", plan.errors[0]) + + +class TestReloadAfterMigration(unittest.TestCase): + """四审回归：legacy 双身份迁移后带 id 重载必须幂等（不得锁死）。""" + + def test_reloading_migrated_tunnels_with_ids_is_idempotent(self): + from mpconf.config import assign_stable_ids + tunnels = [{"name": "a", "ssh_user": "u", "ssh_host": "h", + "ssh_port": 22}, + {"name": "b", "ssh_user": "u", "ssh_host": "h", + "ssh_port": 22}] + assign_stable_ids(tunnels) # 首次迁移：双 id + # 带 id 重载（真实磁盘路径）：幂等，不 raise + n = assign_stable_ids(tunnels) + self.assertEqual(n, 0) diff --git a/tests/test_keychain_id.py b/tests/test_keychain_id.py new file mode 100644 index 0000000..dd167e5 --- /dev/null +++ b/tests/test_keychain_id.py @@ -0,0 +1,62 @@ +"""稳定身份（issue #8）：凭证只以不可变 id 寻址. + +S1 —— keychain account 优先 tunnel["id"]（tunnel:）；无 id 时回退 +legacy user@host:port（只读兼容，供迁移期读取旧 secret）。 +""" +import unittest +from unittest.mock import patch + +from sysctl import keychain + + +def _tun(**kw): + return {"ssh_user": "u", "ssh_host": "h", "ssh_port": 22, **kw} + + +class TestIdAddressing(unittest.TestCase): + def test_account_prefers_stable_id(self): + self.assertEqual( + keychain._account(_tun(id="t-abc123")), "tunnel:t-abc123") + + def test_legacy_account_when_no_id(self): + self.assertEqual( + keychain._account(_tun()), "u@h:22") + + def test_get_password_falls_back_to_legacy_for_migration_reads(self): + calls = [] + def fake_copy(query, _): + calls.append(query.get(keychain.Security.kSecAttrAccount)) + if len(calls) == 1: + return (keychain.Security.errSecItemNotFound, None) + return (keychain.Security.errSecSuccess, b"legacy-secret") + with patch.object(keychain.Security, "SecItemCopyMatching", + side_effect=fake_copy): + got = keychain.get_password(_tun(id="t-new")) + self.assertEqual(got, "legacy-secret") + self.assertEqual(calls, ["tunnel:t-new", "u@h:22"]) + + def test_set_password_writes_id_account_only(self): + written = [] + def fake_add(attrs, _): + written.append(attrs.get(keychain.Security.kSecAttrAccount)) + return keychain.Security.errSecSuccess + with patch.object(keychain.Security, "SecItemAdd", + side_effect=fake_add), \ + patch.object(keychain.Security, "SecItemDelete", + return_value=keychain.Security.errSecSuccess): + ok = keychain.set_password(_tun(id="t-x"), "pw") + self.assertTrue(ok) + self.assertEqual(written, ["tunnel:t-x"]) + + def test_delete_removes_id_and_legacy_accounts(self): + deleted = [] + with patch.object(keychain.Security, "SecItemDelete", + side_effect=lambda q: deleted.append( + q.get(keychain.Security.kSecAttrAccount)) or 0): + keychain.delete_password(_tun(id="t-x")) + self.assertEqual(sorted(deleted), ["tunnel:t-x", "u@h:22"], + "删除清两端：新 id 账户 + 遗留 legacy 账户") + + +if __name__ == "__main__": + unittest.main()