stroblmeandClaude Opus 5
60d7ec81c0
A wheel whose top-level module is `app` collides with anything else in a user's venv, so the package that is about to be published takes the name it is published under. Only the Python package moves; the repo, the Docker WORKDIR and the compose project keep theirs. Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
119 lines
3.6 KiB
Python
119 lines
3.6 KiB
Python
"""Encrypted credential store for node integrations.
|
|
|
|
Nodes never hold credentials: a parameter written as ``{"$secret": "name"}``
|
|
is replaced with the stored value when the node is built. Secrets are kept
|
|
encrypted outside the flows repository, so what gets committed and shared
|
|
never contains a password.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import base64
|
|
import hashlib
|
|
import json
|
|
from pathlib import Path
|
|
from typing import Any
|
|
|
|
from cryptography.fernet import Fernet, InvalidToken
|
|
|
|
SECRET_REF = "$secret"
|
|
|
|
|
|
class SecretsUnreadable(RuntimeError):
|
|
"""The store exists but does not decrypt with the current SECRET_KEY."""
|
|
|
|
|
|
class SecretNotFound(KeyError):
|
|
"""A node asked for a secret that is not in the store."""
|
|
|
|
def __init__(self, name: str) -> None:
|
|
super().__init__(name)
|
|
self.name = name
|
|
|
|
def __str__(self) -> str:
|
|
return f"No secret named '{self.name}' — add it under Secrets."
|
|
|
|
|
|
class SecretsStore:
|
|
"""Named credentials, encrypted at rest with a key derived from the app key."""
|
|
|
|
def __init__(self, path: Path, key: str) -> None:
|
|
self._path = path
|
|
self._fernet = Fernet(
|
|
base64.urlsafe_b64encode(hashlib.sha256(key.encode()).digest())
|
|
)
|
|
|
|
def _read(self) -> dict[str, str]:
|
|
if not self._path.exists():
|
|
return {}
|
|
try:
|
|
data: dict[str, str] = json.loads(
|
|
self._fernet.decrypt(self._path.read_bytes())
|
|
)
|
|
return data
|
|
except (InvalidToken, ValueError) as exc:
|
|
# A changed app key makes existing secrets unreadable. Fail loudly:
|
|
# flows coming up with silently missing credentials are far harder
|
|
# to diagnose than a node that reports why it did not load.
|
|
raise SecretsUnreadable(
|
|
f"Cannot decrypt {self._path} — it was written with a "
|
|
"different SECRET_KEY. Re-encrypt it with "
|
|
"scripts/rotate_secret_key.py."
|
|
) from exc
|
|
|
|
def _write(self, data: dict[str, str]) -> None:
|
|
self._path.parent.mkdir(parents=True, exist_ok=True)
|
|
self._path.write_bytes(self._fernet.encrypt(json.dumps(data).encode()))
|
|
|
|
def list(self) -> list[str]:
|
|
return sorted(self._read())
|
|
|
|
def get(self, name: str) -> str:
|
|
data = self._read()
|
|
if name not in data:
|
|
raise SecretNotFound(name)
|
|
return data[name]
|
|
|
|
def set(self, name: str, value: str) -> None:
|
|
data = self._read()
|
|
data[name] = value
|
|
self._write(data)
|
|
|
|
def delete(self, name: str) -> None:
|
|
data = self._read()
|
|
if data.pop(name, None) is None:
|
|
raise SecretNotFound(name)
|
|
self._write(data)
|
|
|
|
|
|
_store: SecretsStore | None = None
|
|
|
|
|
|
def init_secrets(path: Path, key: str) -> SecretsStore:
|
|
"""Create the process-wide store (called once at startup)."""
|
|
global _store
|
|
_store = SecretsStore(path, key)
|
|
return _store
|
|
|
|
|
|
def get_secrets() -> SecretsStore:
|
|
if _store is None:
|
|
raise RuntimeError("Secrets store not initialised")
|
|
return _store
|
|
|
|
|
|
def get(name: str) -> str:
|
|
"""Look up a secret by name — the entry point for custom node code."""
|
|
return get_secrets().get(name)
|
|
|
|
|
|
def resolve_params(params: Any) -> Any:
|
|
"""Replace every ``{"$secret": "name"}`` reference with its value."""
|
|
if isinstance(params, dict):
|
|
if set(params) == {SECRET_REF} and isinstance(params[SECRET_REF], str):
|
|
return get(params[SECRET_REF])
|
|
return {k: resolve_params(v) for k, v in params.items()}
|
|
if isinstance(params, list):
|
|
return [resolve_params(v) for v in params]
|
|
return params
|