feat(crypto): name the failing setting in wrong-key decrypt log
The "could not decrypt a stored secret" warning was generic, so an operator couldn't tell which of the six secret settings was encrypted under an old key. Thread the setting key through _decode → decrypt_secret(context=...) so the log now reads e.g. "Could not decrypt stored secret smtp.password (wrong/rotated key — re-enter it)". Pure diagnostic; decrypt behaviour unchanged. Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
This commit is contained in:
@@ -81,8 +81,12 @@ def encrypt_secret(plaintext: str) -> str:
|
||||
return ENC_PREFIX + f.encrypt(plaintext.encode("utf-8")).decode("ascii")
|
||||
|
||||
|
||||
def decrypt_secret(value: str) -> str:
|
||||
"""Decrypt an ``enc:v1:`` token; pass through plaintext / undecryptable values."""
|
||||
def decrypt_secret(value: str, *, context: str = "") -> str:
|
||||
"""Decrypt an ``enc:v1:`` token; pass through plaintext / undecryptable values.
|
||||
|
||||
context (e.g. the setting key) is only used to name the value in the
|
||||
wrong-key log line, so an operator knows exactly which secret to re-enter.
|
||||
"""
|
||||
if not is_encrypted(value):
|
||||
return value
|
||||
f = _get()
|
||||
@@ -91,7 +95,8 @@ def decrypt_secret(value: str) -> str:
|
||||
try:
|
||||
return f.decrypt(value[len(ENC_PREFIX):].encode("ascii")).decode("utf-8")
|
||||
except InvalidToken:
|
||||
logger.error("Could not decrypt a stored secret (wrong/rotated key?)")
|
||||
logger.error("Could not decrypt stored secret %s (wrong/rotated key — re-enter it)",
|
||||
context or "(unknown setting)")
|
||||
return value
|
||||
|
||||
|
||||
|
||||
@@ -124,10 +124,14 @@ SECRET_KEYS: set[str] = {
|
||||
}
|
||||
|
||||
|
||||
def _decode(value: Any) -> Any:
|
||||
"""Decrypt a stored value if it's an encrypted token; else pass through."""
|
||||
def _decode(value: Any, key: str = "") -> Any:
|
||||
"""Decrypt a stored value if it's an encrypted token; else pass through.
|
||||
|
||||
key is passed through to the decrypt log so a wrong-key failure names the
|
||||
exact setting that needs re-entering.
|
||||
"""
|
||||
from steward.core.crypto import decrypt_secret, is_encrypted
|
||||
return decrypt_secret(value) if is_encrypted(value) else value
|
||||
return decrypt_secret(value, context=key) if is_encrypted(value) else value
|
||||
|
||||
|
||||
# ─────────────────────────────────────────────────────────────────────────────
|
||||
@@ -142,7 +146,7 @@ async def get_setting(session: AsyncSession, key: str) -> Any:
|
||||
row = result.scalar_one_or_none()
|
||||
if row is None:
|
||||
return DEFAULTS.get(key)
|
||||
return _decode(json.loads(row.value_json))
|
||||
return _decode(json.loads(row.value_json), key)
|
||||
|
||||
|
||||
async def set_setting(session: AsyncSession, key: str, value: Any) -> None:
|
||||
@@ -166,7 +170,7 @@ async def set_setting(session: AsyncSession, key: str, value: Any) -> None:
|
||||
async def get_all_settings(session: AsyncSession) -> dict[str, Any]:
|
||||
"""Return flat key→value dict with defaults filled in for missing keys."""
|
||||
result = await session.execute(select(AppSetting))
|
||||
stored = {row.key: _decode(json.loads(row.value_json)) for row in result.scalars()}
|
||||
stored = {row.key: _decode(json.loads(row.value_json), row.key) for row in result.scalars()}
|
||||
out: dict[str, Any] = {}
|
||||
for key, default in DEFAULTS.items():
|
||||
out[key] = stored.get(key, default)
|
||||
@@ -247,7 +251,7 @@ def load_settings_sync(db_url: str) -> dict[str, Any]:
|
||||
try:
|
||||
async with factory() as session:
|
||||
result = await session.execute(select(AppSetting))
|
||||
return {row.key: _decode(json.loads(row.value_json)) for row in result.scalars()}
|
||||
return {row.key: _decode(json.loads(row.value_json), row.key) for row in result.scalars()}
|
||||
finally:
|
||||
await engine.dispose()
|
||||
|
||||
|
||||
Reference in New Issue
Block a user