161 lines
5.6 KiB
Python
161 lines
5.6 KiB
Python
"""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)
|