rex/credential_vault.py
Rama master · réplica de lectura del repositorio original.
"""OS-backed credential vault for desktop secrets (S4).
Secrets (API keys, tokens, passwords) are encrypted at rest using the
Windows Data Protection API (DPAPI) via ``pywin32``. Only ciphertext and
non-secret metadata ever touch disk; normal config files
(``rex_config.json``, ``gui_settings.json``, ``.env``) must only ever hold
an opaque reference (the vault *key*), never the secret value.
Two scopes are supported, mirroring ``rex.runtime_paths``' existing
household/private split:
- ``"household"`` — shared, installation-wide secrets (API keys, HA token).
Matches today's behavior for every existing unscoped credential consumer.
- ``"user"`` — bound to one validated Rex ``user_id``.
Each scope encrypts with DPAPI using *different* optional entropy derived
from the scope identity. This means decrypting a "user" entry requires both
the same Windows login (DPAPI's own guarantee) *and* the matching scope
entropy - one Rex user cannot decrypt another Rex user's vault entries even
if both share a single Windows account.
Storage hardening (S4 correction pass):
- The on-disk JSON store is a versioned envelope (``__version__`` +
``entries``); an unrecognized/missing version fails closed with
``VaultCorruptedError`` instead of being silently accepted.
- Every mutating operation (and, for simplicity/correctness, every read)
holds an OS-level interprocess file lock (``msvcrt.locking``) around the
read-modify-write cycle, so concurrent writers cannot lose updates and
concurrent readers never observe a torn write.
- Writes go through a per-process temp file, ``fsync``, then an atomic
``os.replace`` onto the real path - a crash mid-write cannot corrupt the
store.
- The vault file (and lock file) get their ACL reset to grant access to the
current Windows user only. ACL hardening failure is fatal; packaged
production never silently downgrades to DPAPI with a broadly readable
ciphertext store.
- Every entry's stored ``scope``/``owner`` metadata is validated against the
vault instance performing the read before its ciphertext is decrypted (or,
for ``list_entries``, before it is surfaced at all) - metadata that has
been copied, tampered with, or otherwise ended up inconsistent with the
vault instance's own identity fails closed rather than returning
data for the wrong owner. Callers that know which ``integration``/
``account`` they expect (e.g. ``CredentialManager``) can pass those
through to ``get_secret`` for an additional cross-check.
This module never falls back to plaintext persistence. If the OS backend is
unavailable, callers get ``VaultUnavailableError`` and must fail closed -
see ``rex/credentials.py`` and ``rex/config.py`` for how read paths degrade
gracefully to their pre-vault behavior (only in an explicit, non-production
opt-in mode - see ``rex.credentials.legacy_plaintext_fallback_enabled``)
while write paths never do.
"""
from __future__ import annotations
import base64
import hashlib
import json
import logging
import os
import re
import secrets
import sys
import time
from collections.abc import Callable
from dataclasses import dataclass
from datetime import UTC, datetime
from pathlib import Path
from typing import BinaryIO, Protocol, TypeVar, runtime_checkable
logger = logging.getLogger(__name__)
try:
import win32crypt
import win32cryptcon
_PYWIN32_AVAILABLE = True
except ImportError: # pragma: no cover - exercised only when pywin32 is absent
win32crypt = None # optional Windows-only dependency
win32cryptcon = None # optional Windows-only dependency
_PYWIN32_AVAILABLE = False
try:
import msvcrt
_MSVCRT_AVAILABLE = True
except ImportError: # pragma: no cover - msvcrt is stdlib but Windows-only
msvcrt = None # type: ignore[assignment]
_MSVCRT_AVAILABLE = False
try:
import ntsecuritycon
import win32api
import win32security
_WIN32SECURITY_AVAILABLE = True
except ImportError: # pragma: no cover - exercised only when pywin32 is absent
ntsecuritycon = None
win32api = None
win32security = None
_WIN32SECURITY_AVAILABLE = False
_ENTROPY_PREFIX = b"askrex-credential-vault:v1:"
_VALID_SCOPES = {"household", "user"}
_SCHEMA_VERSION = 2
_LOCK_TIMEOUT_SECONDS = 10.0
_LOCK_POLL_SECONDS = 0.05
_REFERENCE_PATTERN = re.compile(r"^cred_[A-Za-z0-9_-]{32}$")
_CONTEXT_PATTERN = re.compile(r"^[A-Za-z0-9][A-Za-z0-9._:@-]{0,127}$")
_T = TypeVar("_T")
class VaultUnavailableError(Exception):
"""The OS-backed vault cannot be used in this process/platform.
Raised at construction/factory time, or when the vault is transiently
unusable (e.g. a lock acquisition timeout) - never a signal to silently
fall back to plaintext persistence. Write paths must surface this to the
caller; read paths may fail through to pre-vault behavior only under an
explicit, non-production opt-in (see
``rex.credentials.legacy_plaintext_fallback_enabled``).
"""
class VaultCorruptedError(Exception):
"""A vault entry could not be decrypted or trusted.
Covers a genuinely corrupted store, an unsupported/missing schema
version, and a scope/owner/integration/account metadata mismatch
(including a copied or swapped reference) - all of these must fail
closed rather than return wrong data.
"""
@dataclass(frozen=True)
class VaultEntryMetadata:
"""Non-secret metadata about one vault entry. Never carries the value."""
key: str
integration: str
account: str | None
slot: str
scope: str
owner: str
created_at: str
updated_at: str
@runtime_checkable
class CredentialVaultBackend(Protocol):
"""Contract shared by every vault backend implementation."""
def set_secret(
self,
key: str,
value: str,
*,
integration: str,
account: str | None,
slot: str,
) -> None: ...
def get_secret(
self, key: str, *, integration: str, account: str | None, slot: str
) -> str | None: ...
def delete_secret(
self, key: str, *, integration: str, account: str | None, slot: str
) -> bool: ...
def has_secret(self, key: str, *, integration: str, account: str | None, slot: str) -> bool: ...
def list_entries(self) -> list[VaultEntryMetadata]: ...
def _validate_scope(scope: str, user_id: str | None) -> tuple[str, str]:
"""Return (scope, owner) after validating. owner is 'household' or the user_id."""
if scope not in _VALID_SCOPES:
raise ValueError(f"Unknown vault scope: {scope!r}")
if scope == "user":
if not user_id:
raise ValueError("user_id is required when scope='user'")
from rex.identity import validate_user_id
return scope, validate_user_id(user_id)
return scope, "household"
def generate_credential_ref() -> str:
"""Return a random opaque reference suitable for non-secret config."""
return f"cred_{secrets.token_urlsafe(24)}"
def validate_credential_ref(key: str) -> str:
"""Validate and return an opaque credential reference."""
if not isinstance(key, str) or not _REFERENCE_PATTERN.fullmatch(key):
raise ValueError("Credential reference is not a valid opaque reference")
return key
def _validate_context(
key: str, integration: str, account: str | None, slot: str
) -> tuple[str, str, str | None, str]:
"""Validate the caller-owned authorization context for one credential."""
key = validate_credential_ref(key)
for label, value in (("integration", integration), ("slot", slot)):
if not isinstance(value, str) or not _CONTEXT_PATTERN.fullmatch(value):
raise ValueError(f"Credential {label} is invalid")
if account is not None and (
not isinstance(account, str) or not _CONTEXT_PATTERN.fullmatch(account)
):
raise ValueError("Credential account is invalid")
return key, integration, account, slot
def _entropy_for_entry(
scope: str,
owner: str,
key: str,
integration: str,
account: str | None,
slot: str,
) -> bytes:
"""Bind DPAPI ciphertext to every caller-validated context dimension."""
fields = (scope, owner, key, integration, account or "", slot)
return hashlib.sha256(_ENTROPY_PREFIX + "\0".join(fields).encode("utf-8")).digest()
class InMemoryCredentialVault:
"""In-memory vault backend.
Test/dev-only. Never selected implicitly by a production code path -
``get_credential_vault()`` only returns this when the caller explicitly
passes ``backend="memory"``. Honors the same metadata-validation
contract as ``WindowsDpapiCredentialVault`` (scope/owner/integration/
account checked on every read) so tests that use this backend for
speed/portability exercise real isolation behavior, not a stub that
merely looks similar.
"""
def __init__(
self,
*,
scope: str = "household",
user_id: str | None = None,
store: dict[str, tuple[str, VaultEntryMetadata]] | None = None,
) -> None:
self._scope, self._owner = _validate_scope(scope, user_id)
# `store` may be shared across instances (e.g. a test simulating two
# scopes' entries landing in one physical store) - defaults to a
# private dict, matching one-vault-per-scope production usage.
self._entries: dict[str, tuple[str, VaultEntryMetadata]] = (
store if store is not None else {}
)
def set_secret(
self,
key: str,
value: str,
*,
integration: str,
account: str | None,
slot: str,
) -> None:
key, integration, account, slot = _validate_context(key, integration, account, slot)
if not isinstance(value, str) or not value:
raise ValueError("Credential value must be a non-empty string")
now = datetime.now(UTC).isoformat()
existing = self._entries.get(key)
if existing is not None:
self._validated(
key,
existing[1],
integration=integration,
account=account,
slot=slot,
)
created_at = existing[1].created_at if existing else now
meta = VaultEntryMetadata(
key=key,
integration=integration,
account=account,
slot=slot,
scope=self._scope,
owner=self._owner,
created_at=created_at,
updated_at=now,
)
self._entries[key] = (value, meta)
def _validated(
self,
key: str,
meta: VaultEntryMetadata,
*,
integration: str,
account: str | None,
slot: str,
) -> None:
if meta.scope != self._scope or meta.owner != self._owner:
raise VaultCorruptedError(
f"Vault entry {key!r} scope/owner metadata does not match this vault; "
"refusing to use it."
)
if meta.integration != integration:
raise VaultCorruptedError(f"Vault entry {key!r} integration metadata mismatch.")
if meta.account != account:
raise VaultCorruptedError(f"Vault entry {key!r} account metadata mismatch.")
if meta.slot != slot:
raise VaultCorruptedError(f"Vault entry {key!r} credential-slot metadata mismatch.")
def get_secret(
self, key: str, *, integration: str, account: str | None, slot: str
) -> str | None:
key, integration, account, slot = _validate_context(key, integration, account, slot)
entry = self._entries.get(key)
if entry is None:
return None
value, meta = entry
self._validated(key, meta, integration=integration, account=account, slot=slot)
return value
def delete_secret(self, key: str, *, integration: str, account: str | None, slot: str) -> bool:
key, integration, account, slot = _validate_context(key, integration, account, slot)
entry = self._entries.get(key)
if entry is None:
return False
self._validated(key, entry[1], integration=integration, account=account, slot=slot)
del self._entries[key]
return True
def has_secret(self, key: str, *, integration: str, account: str | None, slot: str) -> bool:
return self.get_secret(key, integration=integration, account=account, slot=slot) is not None
def list_entries(self) -> list[VaultEntryMetadata]:
result = []
for key, (_, meta) in self._entries.items():
if meta.scope != self._scope or meta.owner != self._owner:
raise VaultCorruptedError("Vault entry scope/owner metadata is invalid")
_validate_context(key, meta.integration, meta.account, meta.slot)
result.append(meta)
return result
_ENTRY_FIELDS = {
"ciphertext",
"integration",
"account",
"slot",
"scope",
"owner",
"created_at",
"updated_at",
}
def _validate_schema(raw: object) -> dict:
"""Validate the on-disk envelope shape; fail closed on anything unexpected."""
if not isinstance(raw, dict):
raise VaultCorruptedError("Vault file has an invalid top-level structure.")
version = raw.get("__version__")
if version != _SCHEMA_VERSION:
raise VaultCorruptedError(
f"Vault file schema version {version!r} is not supported "
f"(expected {_SCHEMA_VERSION}); refusing to read it."
)
entries = raw.get("entries")
if not isinstance(entries, dict):
raise VaultCorruptedError("Vault file is missing a valid 'entries' section.")
if set(raw) != {"__version__", "entries"}:
raise VaultCorruptedError("Vault file contains unexpected top-level fields.")
for key, entry in entries.items():
if not isinstance(entry, dict) or set(entry) != _ENTRY_FIELDS:
raise VaultCorruptedError("Vault entry has an invalid schema.")
try:
_validate_context(
key,
entry["integration"],
entry["account"],
entry["slot"],
)
except (KeyError, TypeError, ValueError) as exc:
raise VaultCorruptedError("Vault entry has invalid context metadata.") from exc
if entry["scope"] not in _VALID_SCOPES or not isinstance(entry["owner"], str):
raise VaultCorruptedError("Vault entry has invalid ownership metadata.")
if entry["scope"] == "household" and entry["owner"] != "household":
raise VaultCorruptedError("Vault entry has invalid household ownership metadata.")
if entry["scope"] == "user":
try:
from rex.identity import validate_user_id
validate_user_id(entry["owner"])
except (TypeError, ValueError) as exc:
raise VaultCorruptedError(
"Vault entry has invalid user ownership metadata."
) from exc
if not isinstance(entry["ciphertext"], str):
raise VaultCorruptedError("Vault entry ciphertext is invalid.")
for timestamp_field in ("created_at", "updated_at"):
timestamp = entry[timestamp_field]
if not isinstance(timestamp, str):
raise VaultCorruptedError("Vault entry timestamp is invalid.")
try:
datetime.fromisoformat(timestamp)
except ValueError as exc:
raise VaultCorruptedError("Vault entry timestamp is invalid.") from exc
return raw
def _harden_file_acl(path: Path) -> None:
"""Restrict *path* to the current Windows user or fail closed."""
if not _WIN32SECURITY_AVAILABLE:
raise VaultUnavailableError("Windows ACL support is unavailable for the credential vault")
try:
username = win32api.GetUserName()
user_sid, _domain, _type = win32security.LookupAccountName("", username)
dacl = win32security.ACL()
dacl.AddAccessAllowedAce(
win32security.ACL_REVISION, ntsecuritycon.FILE_ALL_ACCESS, user_sid
)
security_descriptor = win32security.GetFileSecurity(
str(path), win32security.DACL_SECURITY_INFORMATION
)
security_descriptor.SetSecurityDescriptorDacl(1, dacl, 0)
win32security.SetFileSecurity(
str(path), win32security.DACL_SECURITY_INFORMATION, security_descriptor
)
except Exception as exc:
raise VaultUnavailableError("Could not secure the credential vault filesystem ACL") from exc
class WindowsDpapiCredentialVault:
"""Windows DPAPI-backed credential vault.
One JSON file per scope holds base64 DPAPI ciphertext plus non-secret
metadata per key, guarded by an interprocess file lock and written
atomically. ``CRYPTPROTECT_UI_FORBIDDEN`` is always set so a
background/bridge process fails instead of hanging on a Windows UI
prompt it can never show.
"""
_scope: str
_owner: str
_vault_path: Path
_lock_path: Path
def __init__(
self, *, scope: str = "household", user_id: str | None = None, vault_path: Path
) -> None:
if sys.platform != "win32" or not _PYWIN32_AVAILABLE:
raise VaultUnavailableError(
"The Windows credential vault requires pywin32 on Windows " "(pip install pywin32)."
)
self._scope, self._owner = _validate_scope(scope, user_id)
self._vault_path = vault_path
self._lock_path = vault_path.with_name(vault_path.name + ".lock")
# -- interprocess locking -------------------------------------------------
def _acquire_lock(self) -> BinaryIO:
self._vault_path.parent.mkdir(parents=True, exist_ok=True)
handle = open(self._lock_path, "a+b") # noqa: SIM115 - lifetime spans lock hold
handle.seek(0, os.SEEK_END)
if handle.tell() == 0:
handle.write(b"\0")
handle.flush()
os.fsync(handle.fileno())
_harden_file_acl(self._lock_path)
if not _MSVCRT_AVAILABLE: # pragma: no cover - Windows always has msvcrt
handle.close()
raise VaultUnavailableError("Windows interprocess locking is unavailable")
deadline = time.monotonic() + _LOCK_TIMEOUT_SECONDS
while True:
try:
handle.seek(0)
locking = msvcrt.locking # type: ignore[attr-defined, unused-ignore]
locking(handle.fileno(), msvcrt.LK_NBLCK, 1) # type: ignore[attr-defined, unused-ignore]
return handle
except OSError:
if time.monotonic() >= deadline:
handle.close()
raise VaultUnavailableError(
"Timed out waiting for the credential vault file lock; "
"another process may be holding it."
) from None
time.sleep(_LOCK_POLL_SECONDS)
def _release_lock(self, handle: BinaryIO) -> None:
try:
if _MSVCRT_AVAILABLE: # pragma: no branch - Windows always has msvcrt
handle.seek(0)
locking = msvcrt.locking # type: ignore[attr-defined, unused-ignore]
locking(handle.fileno(), msvcrt.LK_UNLCK, 1) # type: ignore[attr-defined, unused-ignore]
finally:
handle.close()
def _with_lock(self, fn: Callable[[], _T]) -> _T:
handle = self._acquire_lock()
try:
return fn()
finally:
self._release_lock(handle)
# -- store I/O --------------------------------------------------------------
def _read_store_unlocked(self) -> dict:
if not self._vault_path.exists():
return {"__version__": _SCHEMA_VERSION, "entries": {}}
try:
raw = json.loads(self._vault_path.read_text(encoding="utf-8"))
except (json.JSONDecodeError, OSError) as exc:
raise VaultCorruptedError(f"Vault file unreadable or corrupted: {exc}") from exc
return _validate_schema(raw)
def _write_store_unlocked(self, store: dict) -> None:
self._vault_path.parent.mkdir(parents=True, exist_ok=True)
tmp_path = self._vault_path.with_name(f"{self._vault_path.name}.tmp-{os.getpid()}")
payload = json.dumps(store, indent=2)
try:
with open(tmp_path, "w", encoding="utf-8") as handle:
handle.write(payload)
handle.flush()
os.fsync(handle.fileno())
_harden_file_acl(tmp_path)
os.replace(tmp_path, self._vault_path)
_harden_file_acl(self._vault_path)
finally:
try:
tmp_path.unlink()
except FileNotFoundError:
pass
def _encrypt(
self, value: str, key: str, integration: str, account: str | None, slot: str
) -> str:
assert win32crypt is not None and win32cryptcon is not None # __init__ already guarded this
blob = win32crypt.CryptProtectData(
value.encode("utf-8"),
"AskRex credential",
_entropy_for_entry(self._scope, self._owner, key, integration, account, slot),
None,
None,
win32cryptcon.CRYPTPROTECT_UI_FORBIDDEN,
)
return base64.b64encode(blob).decode("ascii")
def _decrypt(
self,
ciphertext_b64: str,
key: str,
integration: str,
account: str | None,
slot: str,
) -> str:
try:
assert win32crypt is not None and win32cryptcon is not None # __init__ guarded this
blob = base64.b64decode(ciphertext_b64, validate=True)
_description, plaintext = win32crypt.CryptUnprotectData(
blob,
_entropy_for_entry(self._scope, self._owner, key, integration, account, slot),
None,
None,
win32cryptcon.CRYPTPROTECT_UI_FORBIDDEN,
)
return plaintext.decode("utf-8") # type: ignore[no-any-return]
except Exception as exc:
raise VaultCorruptedError(
f"Could not decrypt vault entry {key!r}: wrong scope or corrupted data."
) from exc
def _validate_entry(
self,
key: str,
entry: dict,
*,
integration: str,
account: str | None,
slot: str,
) -> None:
if entry.get("scope") != self._scope or entry.get("owner") != self._owner:
raise VaultCorruptedError(
f"Vault entry {key!r} scope/owner metadata does not match this vault; "
"refusing to use it."
)
if entry.get("integration") != integration:
raise VaultCorruptedError(f"Vault entry {key!r} integration metadata mismatch.")
if entry.get("account") != account:
raise VaultCorruptedError(f"Vault entry {key!r} account metadata mismatch.")
if entry.get("slot") != slot:
raise VaultCorruptedError(f"Vault entry {key!r} credential-slot metadata mismatch.")
# -- public API ---------------------------------------------------------
def set_secret(
self,
key: str,
value: str,
*,
integration: str,
account: str | None,
slot: str,
) -> None:
key, integration, account, slot = _validate_context(key, integration, account, slot)
if not isinstance(value, str) or not value:
raise ValueError("Credential value must be a non-empty string")
def _do() -> None:
store = self._read_store_unlocked()
now = datetime.now(UTC).isoformat()
existing = store["entries"].get(key, {})
if existing:
self._validate_entry(
key,
existing,
integration=integration,
account=account,
slot=slot,
)
store["entries"][key] = {
"ciphertext": self._encrypt(value, key, integration, account, slot),
"integration": integration,
"account": account,
"slot": slot,
"scope": self._scope,
"owner": self._owner,
"created_at": existing.get("created_at", now),
"updated_at": now,
}
self._write_store_unlocked(store)
self._with_lock(_do)
def get_secret(
self, key: str, *, integration: str, account: str | None, slot: str
) -> str | None:
key, integration, account, slot = _validate_context(key, integration, account, slot)
def _do() -> str | None:
store = self._read_store_unlocked()
entry = store["entries"].get(key)
if entry is None:
return None
self._validate_entry(key, entry, integration=integration, account=account, slot=slot)
return self._decrypt(entry["ciphertext"], key, integration, account, slot)
return self._with_lock(_do)
def delete_secret(self, key: str, *, integration: str, account: str | None, slot: str) -> bool:
key, integration, account, slot = _validate_context(key, integration, account, slot)
def _do() -> bool:
store = self._read_store_unlocked()
if key not in store["entries"]:
return False
self._validate_entry(
key,
store["entries"][key],
integration=integration,
account=account,
slot=slot,
)
del store["entries"][key]
self._write_store_unlocked(store)
return True
return self._with_lock(_do)
def has_secret(self, key: str, *, integration: str, account: str | None, slot: str) -> bool:
return self.get_secret(key, integration=integration, account=account, slot=slot) is not None
def list_entries(self) -> list[VaultEntryMetadata]:
def _do() -> list[VaultEntryMetadata]:
store = self._read_store_unlocked()
result = []
for k, v in store["entries"].items():
if v.get("scope") != self._scope or v.get("owner") != self._owner:
raise VaultCorruptedError("Vault entry scope/owner metadata is invalid")
result.append(
VaultEntryMetadata(
key=k,
integration=v.get("integration", ""),
account=v.get("account"),
slot=v["slot"],
scope=v.get("scope", self._scope),
owner=v.get("owner", self._owner),
created_at=v.get("created_at", ""),
updated_at=v.get("updated_at", ""),
)
)
return result
return self._with_lock(_do)
def _default_vault_path(scope: str, user_id: str | None) -> Path:
from rex.runtime_paths import household_data_path, user_data_path
if scope == "user":
if not user_id:
raise ValueError("user_id is required when scope='user'")
return user_data_path(user_id, "credentials", "vault.json")
return household_data_path("credentials", "vault.json")
def get_credential_vault(
*,
scope: str = "household",
user_id: str | None = None,
backend: str | None = None,
vault_path_override: Path | None = None,
) -> CredentialVaultBackend:
"""Return the vault backend for *scope* (and *user_id* when scope='user').
``backend="memory"`` always returns the in-memory test/dev backend
regardless of platform. Otherwise, Windows gets the real DPAPI backend;
every other platform raises ``VaultUnavailableError`` - there is no
implicit non-Windows fallback in a normal code path.
"""
if backend == "memory":
return InMemoryCredentialVault(scope=scope, user_id=user_id)
if sys.platform == "win32":
vault_path = vault_path_override or _default_vault_path(scope, user_id)
return WindowsDpapiCredentialVault(scope=scope, user_id=user_id, vault_path=vault_path)
raise VaultUnavailableError(
f"No credential vault backend is available on platform {sys.platform!r}. "
"The desktop credential vault is Windows-only."
)
__all__ = [
"CredentialVaultBackend",
"InMemoryCredentialVault",
"VaultCorruptedError",
"VaultEntryMetadata",
"VaultUnavailableError",
"WindowsDpapiCredentialVault",
"generate_credential_ref",
"get_credential_vault",
"validate_credential_ref",
]