Ty
49c22c9385
Clean re-implementation of the full product (not a port of the old M3 tree): - P0 foundation: 13-table schema + per-call sqlite factory (thread-safe), FastAPI app (health, JSON logging + secret redaction, CORS, startup guard). - P1 M1 Capture: mask/sheetKeep, geometry/rect (min-area rect), measure, png crop/inventory/overlay, EXIF decodeOrientedBitmap. - P2 M2 Photos: stateful capture session (batch, pick-cal, apply mask/crop to batch) + photos-ready lineup. - P3 M3 Create: real backend (crypto, woo client, taxonomy sync, pricing, inference Call1/Call2, review, publish, settings, slabs, photos) + real FE review/publish/settings screens consuming the API (no hardcoded Woo ids). - P4 gate exit: check_requirements + check_coverage both exit 0 (137/155 IDs, 0 gaps); backend 130/130, frontend 150/150, build green. Cite: docs/REQUIREMENTS.md (155 rows), docs/MILESTONES.md M1-M3.
60 lines
2.2 KiB
Python
60 lines
2.2 KiB
Python
"""AES-GCM secret encryption for Woo app password + inference API key (SETTINGS-20/25, QUALITY-125).
|
|
|
|
Master key = SLAB_AES_KEY env (32-byte base64). Stored form value_enc =
|
|
base64(nonce(12) || ciphertext+GCM tag). Decrypt failures raise DecryptError —
|
|
detectable, no crash, no secret leak (SETTINGS-75 masking degrades to
|
|
configured:false on a bad blob).
|
|
"""
|
|
from __future__ import annotations
|
|
|
|
import base64
|
|
import os
|
|
|
|
from cryptography.exceptions import InvalidTag
|
|
from cryptography.hazmat.primitives.ciphers.aead import AESGCM
|
|
|
|
|
|
class DecryptError(Exception):
|
|
"""Tampered, corrupt, or wrong-key ciphertext. Message never carries the blob."""
|
|
|
|
|
|
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)")
|
|
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 a plaintext secret -> base64(nonce || ciphertext) for value_enc."""
|
|
if not isinstance(plaintext, str):
|
|
raise TypeError("plaintext must be str")
|
|
nonce = os.urandom(12)
|
|
ct = AESGCM(_get_key()).encrypt(nonce, plaintext.encode("utf-8"), None)
|
|
return base64.b64encode(nonce + ct).decode("ascii")
|
|
|
|
|
|
def decrypt(ciphertext_b64: str) -> str:
|
|
"""Decrypt a value_enc blob; verifies the GCM tag. Raises DecryptError (no leak)."""
|
|
if not isinstance(ciphertext_b64, str):
|
|
raise TypeError("ciphertext_b64 must be str")
|
|
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")
|
|
try:
|
|
pt = AESGCM(_get_key()).decrypt(blob[:12], blob[12:], None)
|
|
except InvalidTag:
|
|
raise DecryptError("decrypt failed: invalid tag (tampered or wrong key)") from None
|
|
except Exception:
|
|
raise DecryptError("decrypt failed") from None
|
|
return pt.decode("utf-8")
|