type-two
e1fcf94f93
- app/vault.py: Fernet-encrypted credential store (app_secret table); key lives only in RECORDGOD_SECRET_KEY (.env, gitignored) — refuses to store anything if the key is unset. - app/settings_routes.py: admin-gated GET/POST /settings/secrets; values encrypted at rest, NEVER returned. Known fields: Woo / Discogs / Cloudflare / DealGod credentials. - site/index.html: recordgod.com landing. site/dash.html: control room (token gate + paste-your-credentials form). main.py loads .env, mounts site at /, store at /store. - Proven: 401 on bad token; DB holds ciphertext (gAAAA…), not plaintext. Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
49 lines
1.7 KiB
Python
49 lines
1.7 KiB
Python
"""Encrypted secrets vault. Credentials are stored as Fernet ciphertext in app_secret;
|
|
the key lives only in the environment (RECORDGOD_SECRET_KEY), never in the DB or the repo.
|
|
If the key isn't set we REFUSE to store secrets — never silently store them in plaintext."""
|
|
import os
|
|
|
|
from cryptography.fernet import Fernet, InvalidToken
|
|
from sqlalchemy import text
|
|
|
|
_key = os.getenv("RECORDGOD_SECRET_KEY")
|
|
_fernet = Fernet(_key.encode()) if _key else None
|
|
|
|
|
|
class SecretsUnconfigured(RuntimeError):
|
|
pass
|
|
|
|
|
|
def _f():
|
|
if _fernet is None:
|
|
raise SecretsUnconfigured(
|
|
"RECORDGOD_SECRET_KEY not set — refusing to store secrets unencrypted")
|
|
return _fernet
|
|
|
|
|
|
async def set_secret(db, name, plaintext):
|
|
enc = _f().encrypt(plaintext.encode())
|
|
await db.execute(text("""
|
|
INSERT INTO app_secret (name, value_enc, updated_at) VALUES (:n, :v, now())
|
|
ON CONFLICT (name) DO UPDATE SET value_enc = EXCLUDED.value_enc, updated_at = now()
|
|
"""), {"n": name, "v": enc})
|
|
await db.commit()
|
|
|
|
|
|
async def get_secret(db, name):
|
|
"""Decrypt one secret for server-side use (e.g. the Woo client). Never exposed over HTTP."""
|
|
row = (await db.execute(text("SELECT value_enc FROM app_secret WHERE name = :n"),
|
|
{"n": name})).first()
|
|
if not row:
|
|
return None
|
|
try:
|
|
return _f().decrypt(bytes(row[0])).decode()
|
|
except InvalidToken:
|
|
return None # key rotated / corrupt — treat as not-set rather than crash
|
|
|
|
|
|
async def list_secret_names(db):
|
|
rows = await db.execute(text("SELECT name, updated_at FROM app_secret ORDER BY name"))
|
|
return [{"name": r[0], "updated_at": r[1].isoformat() if r[1] else None}
|
|
for r in rows]
|