diff --git a/scripts/db/accounts_repo.py b/scripts/db/accounts_repo.py index 613b7ed..b6d171e 100644 --- a/scripts/db/accounts_repo.py +++ b/scripts/db/accounts_repo.py @@ -431,19 +431,20 @@ def insert_credential( expires_at: Optional[int], extra_json: Optional[str], now: int, + secret_ciphertext: Optional[str] = None, ) -> int: cur = conn.cursor() cur.execute( """ INSERT INTO credentials (account_id, platform_key, credential_label, credential_type, - secret_storage, secret_ref, secret_mask, expires_at, status, + secret_storage, secret_ref, secret_mask, secret_ciphertext, expires_at, status, extra_json, created_at, updated_at) - VALUES (?, ?, ?, ?, ?, ?, ?, ?, 'active', ?, ?, ?) + VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, 'active', ?, ?, ?) """, ( account_id, platform_key, credential_label, credential_type, - secret_storage, secret_ref, secret_mask, expires_at, + secret_storage, secret_ref, secret_mask, secret_ciphertext, expires_at, extra_json or "{}", now, now, ), ) diff --git a/scripts/db/credentials_repo.py b/scripts/db/credentials_repo.py new file mode 100644 index 0000000..cbc7eb9 --- /dev/null +++ b/scripts/db/credentials_repo.py @@ -0,0 +1,206 @@ +# -*- coding: utf-8 -*- +"""Credentials repository. + +包含加密路径（secret_storage='local_encrypted'）与现存非加密路径 +（none/env/windows_credential）的统一访问。 +""" +from __future__ import annotations + +import json +from typing import Any, Optional + +from db.accounts_repo import _unix_to_iso_local, insert_credential as _insert_credential_raw +from service.crypto import CredentialDecryptError, decrypt_secret +from service.secret_store import delete_secret, mask_secret, read_secret + +CREDENTIAL_TYPES = frozenset( + { + "password", + "api_key", + "api_token", + "bearer_token", + "oauth_client", + "refresh_token", + "browser_profile", + "note", + } +) +ALLOWED_SECRET_STORAGE = frozenset({"none", "env", "windows_credential", "local_encrypted"}) + + +class CredentialNotEncryptedError(Exception): + """Credential has no decryptable plaintext via this API (e.g. storage=none).""" + + code = "ERR_CREDENTIAL_NOT_ENCRYPTED" + + +def _validate_credential_type(credential_type: str) -> None: + if credential_type not in CREDENTIAL_TYPES: + raise ValueError(f"invalid credential_type: {credential_type!r}") + + +def _validate_storage(secret_storage: str) -> None: + if secret_storage not in ALLOWED_SECRET_STORAGE: + raise ValueError(f"invalid secret_storage: {secret_storage!r}") + + +def insert_credential_encrypted( + conn, + *, + account_id: Optional[int], + platform_key: str, + credential_label: str, + credential_type: str, + plaintext: str, + expires_at: Optional[int], + extra_json: Optional[str], + now: int, +) -> int: + _validate_credential_type(credential_type) + if plaintext is None or str(plaintext).strip() == "": + raise ValueError("plaintext required for encrypted credential insert") + from service.crypto import encrypt_secret + + ciphertext = encrypt_secret(str(plaintext)) + mask = mask_secret(str(plaintext)) + return _insert_credential_raw( + conn, + account_id, + platform_key, + credential_label, + credential_type, + "local_encrypted", + None, + mask, + expires_at, + extra_json, + now, + secret_ciphertext=ciphertext, + ) + + +def insert_credential_external( + conn, + *, + account_id: Optional[int], + platform_key: str, + credential_label: str, + credential_type: str, + secret_storage: str, + secret_ref: Optional[str], + secret_mask: Optional[str], + expires_at: Optional[int], + extra_json: Optional[str], + now: int, +) -> int: + _validate_credential_type(credential_type) + _validate_storage(secret_storage) + if secret_storage == "local_encrypted": + raise ValueError("use insert_credential_encrypted for local_encrypted") + return _insert_credential_raw( + conn, + account_id, + platform_key, + credential_label, + credential_type, + secret_storage, + secret_ref, + secret_mask, + expires_at, + extra_json, + now, + secret_ciphertext=None, + ) + + +def read_credential_plaintext(conn, credential_id: int) -> str: + cur = conn.cursor() + cur.execute( + """ + SELECT secret_storage, secret_ref, secret_ciphertext + FROM credentials WHERE id = ? + """, + (int(credential_id),), + ) + row = cur.fetchone() + if not row: + raise ValueError(f"credential not found: id={credential_id}") + + storage = (row[0] or "").strip().lower() + secret_ref = row[1] or "" + secret_ciphertext = row[2] + + if storage == "local_encrypted": + if not secret_ciphertext: + raise CredentialDecryptError("missing secret_ciphertext") + return decrypt_secret(str(secret_ciphertext)) + + if storage == "none": + raise CredentialNotEncryptedError("credential storage is none") + + if storage == "env": + return read_secret("env", secret_ref) + + if storage == "windows_credential": + return read_secret("windows_credential", secret_ref) + + raise ValueError(f"unsupported secret_storage: {storage!r}") + + +def list_credentials_by_account(conn, account_id: int) -> list[dict[str, Any]]: + cur = conn.cursor() + cur.execute( + """ + SELECT id, account_id, platform_key, credential_label, credential_type, + secret_storage, secret_ref, secret_mask, expires_at, status, + last_used_at, extra_json, created_at, updated_at + FROM credentials + WHERE account_id = ? + ORDER BY updated_at DESC, id DESC + """, + (int(account_id),), + ) + rows = cur.fetchall() + out: list[dict[str, Any]] = [] + for r in rows: + ex = r[11] + parsed_ex: Any + try: + parsed_ex = json.loads(ex) if isinstance(ex, str) else (ex or {}) + except json.JSONDecodeError: + parsed_ex = {} + out.append( + { + "id": r[0], + "account_id": r[1], + "platform_key": r[2], + "credential_label": r[3], + "credential_type": r[4], + "secret_storage": r[5], + "secret_ref": r[6] or "", + "secret_mask": r[7] or "", + "expires_at": _unix_to_iso_local(r[8]), + "status": r[9], + "last_used_at": _unix_to_iso_local(r[10]), + "extra_json": parsed_ex, + "created_at": _unix_to_iso_local(r[12]), + "updated_at": _unix_to_iso_local(r[13]), + } + ) + return out + + +def delete_credential(conn, credential_id: int) -> None: + cur = conn.cursor() + cur.execute( + "SELECT secret_storage, secret_ref FROM credentials WHERE id = ?", + (int(credential_id),), + ) + row = cur.fetchone() + if not row: + return + storage = (row[0] or "").strip().lower() + secret_ref = row[1] or "" + if storage == "windows_credential" and secret_ref: + delete_secret("windows_credential", secret_ref) + cur.execute("DELETE FROM credentials WHERE id = ?", (int(credential_id),))