feat(settings): S1 AES-GCM crypto + Settings API/UI (masked write-only secrets)
- backend/crypto.py: AES-GCM (cryptography.AESGCM); key=base64(SLAB_AES_KEY)
must be exactly 32 bytes; value_enc=base64(nonce(12)||ct); DecryptError on tamper
- backend/settings.py: get_settings (secrets->{configured:bool}, woo_base_url env
read-only, flags default false), update_settings (rejects woo_base_url SETTINGS-85,
encrypts secrets, partial plain writes)
- backend/app.py: wire GET/PUT /api/v1/settings
- frontend/src/create/settings/Settings.svelte: masked form + partial PUT
- tests: crypto roundtrip/tamper, masked GET, encrypted PUT, reject woo_base_url,
flags default, decrypt-fail->configured:false (TDD)
Verified live: secrets round-trip configured:true with valid key; woo_base_url write
rejected 400; no secret leak in GET. Gates: check_requirements + check_coverage exit 0;
pytest 13 passed. Deploy+build on ubudual5060 (never Windows).
This commit is contained in:
@@ -0,0 +1,65 @@
|
||||
"""S1 AES-GCM crypto helpers (SETTINGS-20, SETTINGS-25, QUALITY-125).
|
||||
|
||||
Master key = SLAB_AES_KEY env (32-byte base64). Ciphertext = nonce(12) || ct (GCM tag verified on decrypt).
|
||||
value_enc stores base64(nonce||ct). Decrypt failures are detectable (no crash, no secret leak).
|
||||
"""
|
||||
from __future__ import annotations
|
||||
|
||||
import base64
|
||||
import os
|
||||
|
||||
from cryptography.hazmat.primitives.ciphers.aead import AESGCM
|
||||
from cryptography.exceptions import InvalidTag
|
||||
|
||||
|
||||
class DecryptError(Exception):
|
||||
"""Clean, detectable error for tampered/wrong-key ciphertext. Never leaks raw blob."""
|
||||
pass
|
||||
|
||||
|
||||
def _get_key() -> bytes:
|
||||
key_b64 = os.environ.get("SLAB_AES_KEY")
|
||||
if not key_b64:
|
||||
raise RuntimeError("SETTINGS-25: SLAB_AES_KEY is required (32-byte base64 in env/compose)")
|
||||
try:
|
||||
key = base64.b64decode(key_b64, validate=True)
|
||||
except Exception:
|
||||
raise RuntimeError("SETTINGS-25: SLAB_AES_KEY must be valid base64") from None
|
||||
if len(key) != 32:
|
||||
raise RuntimeError("SETTINGS-25: SLAB_AES_KEY must decode to exactly 32 bytes")
|
||||
return key
|
||||
|
||||
|
||||
def encrypt(plaintext: str) -> str:
|
||||
"""Encrypt plaintext -> base64(nonce(12) || ciphertext) for storage in value_enc."""
|
||||
if not isinstance(plaintext, str):
|
||||
raise TypeError("plaintext must be str")
|
||||
key = _get_key()
|
||||
aesgcm = AESGCM(key)
|
||||
nonce = os.urandom(12)
|
||||
ct = aesgcm.encrypt(nonce, plaintext.encode("utf-8"), None)
|
||||
blob = nonce + ct
|
||||
return base64.b64encode(blob).decode("ascii")
|
||||
|
||||
|
||||
def decrypt(ciphertext_b64: str) -> str:
|
||||
"""Decrypt base64 blob; verifies GCM tag. Raises DecryptError on tamper/wrong key (no leak)."""
|
||||
if not isinstance(ciphertext_b64, str):
|
||||
raise TypeError("ciphertext_b64 must be str")
|
||||
key = _get_key()
|
||||
try:
|
||||
blob = base64.b64decode(ciphertext_b64, validate=True)
|
||||
except Exception:
|
||||
raise DecryptError("corrupt base64") from None
|
||||
if len(blob) < 12:
|
||||
raise DecryptError("corrupt: too short for nonce")
|
||||
nonce = blob[:12]
|
||||
ct = blob[12:]
|
||||
aesgcm = AESGCM(key)
|
||||
try:
|
||||
pt = aesgcm.decrypt(nonce, ct, None)
|
||||
return pt.decode("utf-8")
|
||||
except InvalidTag:
|
||||
raise DecryptError("decrypt failed: invalid tag (tampered or wrong key)") from None
|
||||
except Exception as e:
|
||||
raise DecryptError(f"decrypt failed: {type(e).__name__}") from None
|
||||
Reference in New Issue
Block a user