diff --git a/scripts/service/crypto.py b/scripts/service/crypto.py new file mode 100644 index 0000000..b51957e --- /dev/null +++ b/scripts/service/crypto.py @@ -0,0 +1,36 @@ +# -*- coding: utf-8 -*- +"""Fernet 加密 / 解密的薄封装。 + +不暴露 Fernet 对象;调用方只看 encrypt/decrypt 字符串。 +""" +from __future__ import annotations + +from cryptography.fernet import Fernet, InvalidToken + +from service.master_key import load_master_key +from service.secret_store import mask_secret + + +class CredentialDecryptError(Exception): + """Decryption failed (bad key or corrupted ciphertext).""" + + code = "ERR_CREDENTIAL_DECRYPT_FAILED" + + +def encrypt_secret(plaintext: str) -> str: + """Encrypt using current master key; returns url-safe token string.""" + key = load_master_key() + f = Fernet(key) + tok = f.encrypt(plaintext.encode("utf-8")) + return tok.decode("ascii") + + +def decrypt_secret(ciphertext: str) -> str: + """Decrypt Fernet token string to plaintext.""" + key = load_master_key() + f = Fernet(key) + try: + raw = f.decrypt(ciphertext.encode("ascii")) + return raw.decode("utf-8") + except InvalidToken as e: + raise CredentialDecryptError("Fernet decrypt failed") from e diff --git a/scripts/service/master_key.py b/scripts/service/master_key.py new file mode 100644 index 0000000..d462ab4 --- /dev/null +++ b/scripts/service/master_key.py @@ -0,0 +1,142 @@ +# -*- coding: utf-8 -*- +"""master key 加载与生成。 + +加载优先级(高到低): + 1. 环境变量 ACCOUNT_MANAGER_MASTER_KEY(base64-urlsafe 字符串,44 字符) + 2. 文件 {DATA_ROOT}/{USER_ID}/account-manager/master.key + 3. 都没有 → raise MasterKeyMissingError,错误码 ERR_MASTER_KEY_MISSING + +不再自动生成——必须显式 `account init-master-key`。 +""" +from __future__ import annotations + +import base64 +import os + +from cryptography.fernet import Fernet + +from util.runtime_paths import get_skill_data_dir + +ENV_MASTER_KEY = "ACCOUNT_MANAGER_MASTER_KEY" +_MASTER_FILENAME = "master.key" + + +class MasterKeyMissingError(Exception): + """Raised when no valid master key can be resolved.""" + + code = "ERR_MASTER_KEY_MISSING" + + +class MasterKeyExistsError(Exception): + """Raised when master.key already exists and overwrite is not allowed.""" + + code = "ERR_MASTER_KEY_EXISTS" + + +class MasterKeyInvalidError(Exception): + """Raised when a supplied master key string cannot be used as a Fernet key.""" + + code = "ERR_MASTER_KEY_INVALID" + + +def master_key_path() -> str: + """Absolute path to master.key alongside account-manager.db.""" + return os.path.join(get_skill_data_dir(), _MASTER_FILENAME) + + +def _try_fernet_key(key_material: bytes) -> bytes | None: + """Return url-safe key bytes accepted by Fernet, or None if invalid.""" + try: + raw = key_material.strip() + if not raw: + return None + # Accept stored form: ascii urlsafe base64 string as bytes + Fernet(raw) + return raw + except Exception: + return None + + +def _try_decode_manual_b64(s: str) -> bytes | None: + """If string decodes to 32 raw bytes, re-encode as Fernet url-safe key.""" + try: + pad = "=" * ((4 - len(s) % 4) % 4) + decoded = base64.urlsafe_b64decode(s + pad) + if len(decoded) != 32: + return None + key_bytes = base64.urlsafe_b64encode(decoded) + Fernet(key_bytes) + return key_bytes + except Exception: + return None + + +def _normalize_key_string(s: str) -> bytes | None: + s = (s or "").strip() + if not s: + return None + fk = _try_fernet_key(s.encode("ascii")) + if fk is not None: + return fk + return _try_decode_manual_b64(s) + + +def load_master_key() -> bytes: + """Resolve master key (env → file); raises MasterKeyMissingError if unavailable.""" + env_val = (os.environ.get(ENV_MASTER_KEY) or "").strip() + if env_val: + kb = _normalize_key_string(env_val) + if kb is not None: + return kb + + path = master_key_path() + try: + with open(path, "rb") as f: + raw = f.read() + txt = raw.strip().decode("ascii", errors="strict") + kb = _normalize_key_string(txt) + if kb is not None: + return kb + except OSError: + pass + except (UnicodeDecodeError, ValueError): + pass + + raise MasterKeyMissingError("No valid ACCOUNT_MANAGER_MASTER_KEY or master.key") + + +def normalize_master_key_input(s: str) -> bytes: + """Validate user-supplied key text; raises MasterKeyInvalidError if unusable.""" + kb = _normalize_key_string((s or "").strip()) + if kb is None: + raise MasterKeyInvalidError("invalid master key material") + return kb + + +def generate_master_key() -> bytes: + """Return a new Fernet key (44-char url-safe base64 as bytes).""" + return Fernet.generate_key() + + +def write_master_key_file(key: bytes, *, allow_overwrite: bool = False) -> str: + """Write key bytes to master_key_path(); chmod 0o600 best-effort.""" + path = master_key_path() + os.makedirs(os.path.dirname(path), exist_ok=True) + if os.path.exists(path) and not allow_overwrite: + raise MasterKeyExistsError("master.key already exists; use --rotate to overwrite") + data = key.strip() + with open(path, "wb") as f: + f.write(data) + try: + os.chmod(path, 0o600) + except OSError: + pass + return path + + +def is_master_key_available() -> bool: + try: + load_master_key() + return True + except MasterKeyMissingError: + return False diff --git a/scripts/service/secret_store.py b/scripts/service/secret_store.py index 3e52249..e3754a2 100644 --- a/scripts/service/secret_store.py +++ b/scripts/service/secret_store.py @@ -135,6 +135,12 @@ def read_secret(storage: str, secret_ref: str) -> str: ) return val + elif storage_n == "local_encrypted": + raise ValueError( + "local_encrypted backend cannot be read via secret_ref. " + "Use credentials_repo.read_credential_plaintext() instead." + ) + else: log.error("secret_resolve_failed unsupported storage=%s", storage) raise ValueError(f"Unsupported secret storage type: {storage}") @@ -304,3 +310,14 @@ def _win_cred_delete_impl(target: str) -> None: err = ctypes.get_last_error() if err != 1168: raise OSError(f"CredDeleteW failed with error {err}") + + +def encrypt_secret_for_storage(secret_value: str) -> str: + """加密 secret 用于 secret_storage='local_encrypted'。 + + 返回密文字符串,调用方负责写到 credentials.secret_ciphertext 列。 + """ + from service.crypto import encrypt_secret + + return encrypt_secret(secret_value) +