proxy-god/proxy_chain_manager/secrets_store.py

"""Windows DPAPI-based secret storage.

Provides ``encrypt(plaintext: str) -> str`` and ``decrypt(token: str) -> str``
backed by ``CryptProtectData`` / ``CryptUnprotectData``.

Why DPAPI:
  - Built into Windows; no third-party crypto dependency.
  - Key material is tied to the *current user account* (or the machine, when
    ``CRYPTPROTECT_LOCAL_MACHINE`` is used) — no passphrase to remember.
  - Data encrypted on disk cannot be decrypted by another user / machine.

Token format:  ``DPAPI:v1:<base64-blob>``

Plain strings that don't start with ``DPAPI:`` are returned unchanged by
``decrypt()`` so the loader can safely walk old plaintext files and re-encrypt
them on the next save (migration is free).

Falls back gracefully on non-Windows or when DPAPI is unavailable: ``encrypt``
returns the original string verbatim (logged once).
"""
from __future__ import annotations

import base64
import ctypes
import logging
import sys
from ctypes import wintypes

log = logging.getLogger(__name__)

_TOKEN_PREFIX = "DPAPI:v1:"


# ── Windows API binding ──────────────────────────────────────────────────────

class _DATA_BLOB(ctypes.Structure):
    _fields_ = [
        ("cbData", wintypes.DWORD),
        ("pbData", ctypes.POINTER(ctypes.c_char)),
    ]


def _blob(data: bytes) -> _DATA_BLOB:
    buf = ctypes.create_string_buffer(data, len(data))
    return _DATA_BLOB(len(data), ctypes.cast(buf, ctypes.POINTER(ctypes.c_char)))


def _is_windows() -> bool:
    return sys.platform == "win32"


_warned_unavailable = False


def _warn_unavailable(reason: str) -> None:
    global _warned_unavailable
    if not _warned_unavailable:
        log.warning("DPAPI secret store unavailable: %s — secrets will be stored in plaintext.", reason)
        _warned_unavailable = True


# ── Public API ───────────────────────────────────────────────────────────────

def is_available() -> bool:
    """Return True if the current process can use DPAPI."""
    if not _is_windows():
        return False
    try:
        ctypes.windll.crypt32  # noqa: B018  — just confirm the DLL is importable
        return True
    except Exception:  # noqa: BLE001
        return False


def encrypt(plaintext: str) -> str:
    """Encrypt *plaintext* with current-user DPAPI.

    Returns ``DPAPI:v1:<base64>`` on success.  Returns the original plaintext
    on any failure (callers can therefore always trust the return value as the
    string to persist).
    """
    if not plaintext:
        return ""
    # Already encrypted — pass through.
    if plaintext.startswith(_TOKEN_PREFIX):
        return plaintext
    if not is_available():
        _warn_unavailable("not running on Windows")
        return plaintext
    try:
        data = plaintext.encode("utf-8")
        in_blob = _blob(data)
        out_blob = _DATA_BLOB()
        ok = ctypes.windll.crypt32.CryptProtectData(
            ctypes.byref(in_blob),
            "ProxyGodSecret",          # description (ignored by us)
            None,                       # optional entropy
            None,                       # reserved
            None,                       # prompt struct
            0,                          # flags
            ctypes.byref(out_blob),
        )
        if not ok:
            err = ctypes.get_last_error()
            _warn_unavailable(f"CryptProtectData failed (err={err})")
            return plaintext
        try:
            enc = ctypes.string_at(out_blob.pbData, out_blob.cbData)
        finally:
            ctypes.windll.kernel32.LocalFree(out_blob.pbData)
        return _TOKEN_PREFIX + base64.b64encode(enc).decode("ascii")
    except Exception as exc:  # noqa: BLE001
        _warn_unavailable(f"unexpected error: {exc}")
        return plaintext


def decrypt(token: str) -> str:
    """Decrypt a token produced by :func:`encrypt`.

    Plain strings (not starting with the ``DPAPI:`` prefix) are returned
    verbatim — this is intentional, it lets the loader transparently read
    legacy plaintext files and re-encrypt on save.
    """
    if not token:
        return ""
    if not token.startswith(_TOKEN_PREFIX):
        return token  # legacy plaintext
    if not is_available():
        _warn_unavailable("not running on Windows — cannot decrypt")
        return ""
    try:
        b64 = token[len(_TOKEN_PREFIX):]
        data = base64.b64decode(b64.encode("ascii"))
        in_blob = _blob(data)
        out_blob = _DATA_BLOB()
        ok = ctypes.windll.crypt32.CryptUnprotectData(
            ctypes.byref(in_blob),
            None,                       # ppszDataDescr
            None,                       # entropy
            None,                       # reserved
            None,                       # prompt struct
            0,                          # flags
            ctypes.byref(out_blob),
        )
        if not ok:
            err = ctypes.get_last_error()
            log.warning("CryptUnprotectData failed (err=%s) — secret stayed encrypted.", err)
            return ""
        try:
            plain = ctypes.string_at(out_blob.pbData, out_blob.cbData)
        finally:
            ctypes.windll.kernel32.LocalFree(out_blob.pbData)
        return plain.decode("utf-8", errors="replace")
    except Exception as exc:  # noqa: BLE001
        log.warning("DPAPI decrypt error: %s", exc)
        return ""


def is_encrypted(token: str) -> bool:
    return bool(token) and token.startswith(_TOKEN_PREFIX)