account-manager/scripts/db/credentials_repo.py

# -*- 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),))