feat(cli): add init-master-key / export-credentials / import-credentials commands

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

scripts/cli/app.py

@@ -13,6 +13,11 @@ Machine commands output single-line JSON where appropriate.
 """
 import sys
 
+from service.account_crypto_commands import (
+    cmd_account_export_credentials,
+    cmd_account_import_credentials,
+    cmd_account_init_master_key,
+)
 from service.account_service import (
     cmd_account_add_secret,
     cmd_account_add_web,
@@ -211,6 +216,12 @@ def main() -> None:
                 print("ERROR:CLI_BAD_ARGS account open 需要 <id>。")
                 sys.exit(1)
             cmd_open(av[3])
+        elif sub == "init-master-key":
+            cmd_account_init_master_key(av)
+        elif sub == "export-credentials":
+            cmd_account_export_credentials(av)
+        elif sub == "import-credentials":
+            cmd_account_import_credentials(av)
         else:
             print("ERROR:CLI_BAD_ARGS 未知的 account 子命令。")
             print(_USAGE)

scripts/service/account_crypto_commands.py

@@ -0,0 +1,304 @@
+# -*- coding: utf-8 -*-
+"""CLI handlers for master key init and credential export/import (stage B)."""
+from __future__ import annotations
+
+import json
+import os
+import sys
+import time
+from datetime import datetime
+from typing import Any, Optional
+
+from db.accounts_repo import _unix_to_iso_local
+from db.connection import get_conn, init_db
+from db.credentials_repo import insert_credential_encrypted
+from service.crypto import CredentialDecryptError, decrypt_secret
+from service.master_key import (
+    ENV_MASTER_KEY,
+    MasterKeyExistsError,
+    MasterKeyInvalidError,
+    generate_master_key,
+    normalize_master_key_input,
+    write_master_key_file,
+)
+from service.secret_store import mask_secret, read_secret
+
+
+def _has_flag(argv: list[str], flag: str) -> bool:
+    return flag in argv
+
+
+def _get_opt(argv: list[str], flag: str, default=None):
+    if flag in argv:
+        idx = argv.index(flag)
+        if idx + 1 < len(argv):
+            return argv[idx + 1]
+    return default
+
+
+def _key_preview(key_bytes: bytes) -> str:
+    s = key_bytes.strip().decode("ascii")
+    if len(s) <= 8:
+        return "****"
+    return s[:4] + "****" + s[-4:]
+
+
+def cmd_account_init_master_key(argv: list[str]) -> None:
+    """account init-master-key [--from-env] [--print] [--rotate]"""
+    from_env = _has_flag(argv, "--from-env")
+    do_print = _has_flag(argv, "--print")
+    rotate = _has_flag(argv, "--rotate")
+
+    try:
+        if from_env:
+            raw = (os.environ.get(ENV_MASTER_KEY) or "").strip()
+            if not raw:
+                raise MasterKeyInvalidError("ACCOUNT_MANAGER_MASTER_KEY is not set")
+            key = normalize_master_key_input(raw)
+        else:
+            key = generate_master_key()
+
+        if rotate:
+            print(
+                "[init-master-key] WARNING: --rotate overwrites master.key; "
+                "ensure you have an export backup.",
+                file=sys.stderr,
+            )
+
+        path_written: Optional[str] = None
+        if not do_print:
+            path_written = write_master_key_file(key, allow_overwrite=rotate)
+
+        if do_print:
+            line = json.dumps(
+                {"success": True, "key": key.decode("ascii")},
+                ensure_ascii=False,
+            )
+            print(line)
+            return
+
+        assert path_written is not None
+        preview = _key_preview(key)
+        print(
+            json.dumps(
+                {"success": True, "path": path_written, "key_preview": preview},
+                ensure_ascii=False,
+            )
+        )
+    except MasterKeyExistsError as e:
+        print(
+            json.dumps(
+                {
+                    "success": False,
+                    "error": {"code": e.code, "message": str(e)},
+                },
+                ensure_ascii=False,
+            )
+        )
+    except MasterKeyInvalidError as e:
+        print(
+            json.dumps(
+                {
+                    "success": False,
+                    "error": {"code": e.code, "message": str(e)},
+                },
+                ensure_ascii=False,
+            )
+        )
+
+
+def cmd_account_export_credentials(argv: list[str]) -> None:
+    """account export-credentials --plaintext"""
+    if not _has_flag(argv, "--plaintext"):
+        print(
+            json.dumps(
+                {
+                    "success": False,
+                    "error": {
+                        "code": "ERR_EXPORT_REQUIRES_PLAINTEXT_FLAG",
+                        "message": "export requires explicit --plaintext acknowledgement",
+                    },
+                },
+                ensure_ascii=False,
+            )
+        )
+        return
+
+    print(
+        "[export] WARNING: plaintext credentials written to stdout. Handle with extreme care.",
+        file=sys.stderr,
+    )
+
+    init_db()
+    conn = get_conn()
+    try:
+        cur = conn.cursor()
+        cur.execute(
+            """
+            SELECT id, account_id, platform_key, credential_label, credential_type,
+                   secret_storage, secret_ref, secret_mask, secret_ciphertext,
+                   expires_at, extra_json, created_at, updated_at
+            FROM credentials
+            ORDER BY id ASC
+            """
+        )
+        rows = cur.fetchall()
+        out: list[dict[str, Any]] = []
+        for r in rows:
+            cid = r[0]
+            storage = (r[5] or "").strip().lower()
+            secret_ref = r[6] or ""
+            secret_mask = r[7] or ""
+            ct = r[8]
+            extra_raw = r[10]
+            try:
+                extra_obj = json.loads(extra_raw) if isinstance(extra_raw, str) else {}
+            except json.JSONDecodeError:
+                extra_obj = {}
+
+            plaintext: Optional[str] = None
+            extra_out = dict(extra_obj)
+
+            if storage == "none":
+                plaintext = None
+            elif storage == "local_encrypted":
+                try:
+                    if not ct:
+                        raise CredentialDecryptError("missing ciphertext")
+                    plaintext = decrypt_secret(str(ct))
+                except CredentialDecryptError as e:
+                    print(
+                        f"[export] decrypt_failed id={cid} code={e.code} msg={e}",
+                        file=sys.stderr,
+                    )
+                    plaintext = None
+            elif storage == "env":
+                val = os.environ.get(secret_ref)
+                if val is None:
+                    plaintext = None
+                    extra_out["export_note"] = "env var missing at export time"
+                else:
+                    plaintext = val
+            elif storage == "windows_credential":
+                try:
+                    plaintext = read_secret("windows_credential", secret_ref)
+                except (ValueError, OSError) as e:
+                    plaintext = None
+                    extra_out["export_note"] = f"windows_credential read failed: {e}"
+            else:
+                extra_out["export_note"] = f"unsupported storage at export: {storage}"
+
+            item = {
+                "id": cid,
+                "account_id": r[1],
+                "platform_key": r[2],
+                "credential_label": r[3],
+                "credential_type": r[4],
+                "secret_storage": r[5],
+                "plaintext": plaintext,
+                "secret_mask": secret_mask,
+                "expires_at": _unix_to_iso_local(r[9]),
+                "extra_json": extra_out,
+                "created_at": _unix_to_iso_local(r[11]),
+                "updated_at": _unix_to_iso_local(r[12]),
+            }
+            out.append(item)
+
+        print(json.dumps(out, ensure_ascii=False))
+    finally:
+        conn.close()
+
+
+def _iso_maybe_to_unix(val: Any) -> Optional[int]:
+    if val is None:
+        return None
+    if isinstance(val, (int, float)):
+        return int(val)
+    s = str(val).strip()
+    if not s:
+        return None
+    try:
+        dt = datetime.fromisoformat(s)
+        return int(dt.timestamp())
+    except ValueError:
+        return None
+
+
+def cmd_account_import_credentials(argv: list[str]) -> None:
+    """account import-credentials [--replace-all] < stdin JSON array"""
+    replace_all = _has_flag(argv, "--replace-all")
+
+    raw = sys.stdin.read()
+    try:
+        data = json.loads(raw)
+    except json.JSONDecodeError as e:
+        print(f"[import] fatal: invalid JSON stdin: {e}", file=sys.stderr)
+        return
+
+    if not isinstance(data, list):
+        print("[import] fatal: stdin JSON must be an array", file=sys.stderr)
+        return
+
+    init_db()
+    conn = get_conn()
+    inserted = 0
+    skipped = 0
+    try:
+        cur = conn.cursor()
+        if replace_all:
+            cur.execute(
+                "DELETE FROM credentials WHERE secret_storage = ?",
+                ("local_encrypted",),
+            )
+        now = int(time.time())
+        for idx, row in enumerate(data):
+            if not isinstance(row, dict):
+                print(f"[import] skip line {idx}: not an object", file=sys.stderr)
+                skipped += 1
+                continue
+            pt = row.get("plaintext")
+            if pt is None or str(pt).strip() == "":
+                skipped += 1
+                continue
+            try:
+                plat = str(row.get("platform_key") or "").strip()
+                label = str(row.get("credential_label") or "").strip()
+                ctype = str(row.get("credential_type") or "").strip()
+                if not plat or not label or not ctype:
+                    raise ValueError("missing platform_key/credential_label/credential_type")
+
+                account_id = row.get("account_id")
+                aid = int(account_id) if account_id is not None else None
+
+                expires_at = _iso_maybe_to_unix(row.get("expires_at"))
+                ex = row.get("extra_json")
+                if isinstance(ex, dict):
+                    extra_json = json.dumps(ex, ensure_ascii=False)
+                elif isinstance(ex, str):
+                    extra_json = ex
+                else:
+                    extra_json = "{}"
+
+                insert_credential_encrypted(
+                    conn,
+                    account_id=aid,
+                    platform_key=plat,
+                    credential_label=label,
+                    credential_type=ctype,
+                    plaintext=str(pt),
+                    expires_at=expires_at,
+                    extra_json=extra_json,
+                    now=now,
+                )
+                inserted += 1
+            except Exception as e:
+                print(f"[import] skip line {idx}: {e}", file=sys.stderr)
+                skipped += 1
+        conn.commit()
+    finally:
+        conn.close()
+
+    print(
+        f"[import] inserted {inserted} credentials, skipped {skipped} lines (no plaintext).",
+        file=sys.stderr,
+    )