Ty
34ee7eff60
Owner-directed, LAN/HTTP dev-test only. When the env flag is set,
GET /api/v1/settings returns decrypted secret values (string) instead of
{configured: bool}; the UI prefills and shows them in clear text. Default
off -> SETTINGS-75 masking and the existing tests stay intact. Revert
before any non-UAT deployment.
145 lines
4.5 KiB
Python
145 lines
4.5 KiB
Python
"""S1 Settings service (DATA-MODEL §6, API.md §10, SETTINGS-5/15/20/25/30/75/85/90).
|
|
|
|
Secrets (value_enc) are AES-GCM write-only; GET returns only {configured: bool}.
|
|
woo_base_url is env read-only. Flags default false. Partial PUT. Decrypt fail -> configured:false (no 500, no leak).
|
|
"""
|
|
from __future__ import annotations
|
|
|
|
import os
|
|
from typing import Any
|
|
|
|
from fastapi import HTTPException
|
|
|
|
from .crypto import decrypt, encrypt, DecryptError
|
|
from .db import get_connection
|
|
|
|
SECRET_KEYS: set[str] = {
|
|
"woo_app_password",
|
|
"inference_base_url",
|
|
"inference_api_key",
|
|
}
|
|
PLAIN_KEYS: set[str] = {
|
|
"woo_wp_username",
|
|
"inference_model",
|
|
"inference_enabled",
|
|
"content_llm_enabled",
|
|
"brand_voice",
|
|
"geo_context",
|
|
"woo_create_status",
|
|
"taxonomy_last_sync_at",
|
|
"taxonomy_anchor",
|
|
}
|
|
ALL_KEYS = SECRET_KEYS | PLAIN_KEYS
|
|
|
|
DEFAULTS: dict[str, Any] = {
|
|
"inference_enabled": False,
|
|
"content_llm_enabled": False,
|
|
"woo_create_status": "draft",
|
|
}
|
|
|
|
|
|
def _row_to_value(row: dict) -> Any:
|
|
if row is None:
|
|
return None
|
|
v = row.get("value")
|
|
if v is None:
|
|
return None
|
|
if isinstance(v, str):
|
|
if v.lower() in ("true", "false"):
|
|
return v.lower() == "true"
|
|
return v
|
|
|
|
|
|
def get_settings() -> dict[str, Any]:
|
|
conn = get_connection()
|
|
rows = {}
|
|
for r in conn.execute("SELECT key, value, value_enc FROM settings").fetchall():
|
|
rows[r["key"]] = {"value": r["value"], "value_enc": r["value_enc"]}
|
|
|
|
result: dict[str, Any] = {}
|
|
result["woo_base_url"] = os.environ.get("WOO_BASE_URL", "")
|
|
|
|
# TEMPORARY UAT override (owner-directed, LAN/HTTP only): when set, secrets are
|
|
# returned as their decrypted plaintext so they can be inspected on the dev-test
|
|
# box. Default OFF -> SETTINGS-75 masking intact. Remove before prod.
|
|
unmask = os.environ.get("SLAB_UAT_UNMASK_SECRETS") == "1"
|
|
|
|
for k in sorted(ALL_KEYS):
|
|
row = rows.get(k, {})
|
|
if k in SECRET_KEYS:
|
|
enc = row.get("value_enc")
|
|
if unmask:
|
|
if enc:
|
|
try:
|
|
result[k] = decrypt(enc)
|
|
except Exception:
|
|
result[k] = ""
|
|
else:
|
|
result[k] = ""
|
|
else:
|
|
if enc:
|
|
try:
|
|
decrypt(enc) # verify roundtrip possible
|
|
result[k] = {"configured": True}
|
|
except Exception:
|
|
result[k] = {"configured": False}
|
|
else:
|
|
result[k] = {"configured": False}
|
|
else:
|
|
if k in rows:
|
|
val = _row_to_value(row)
|
|
result[k] = val if val is not None else DEFAULTS.get(k)
|
|
else:
|
|
result[k] = DEFAULTS.get(k)
|
|
|
|
return result
|
|
|
|
|
|
def update_settings(payload: dict[str, Any]) -> dict[str, Any]:
|
|
if "woo_base_url" in payload:
|
|
raise HTTPException(
|
|
status_code=400,
|
|
detail={
|
|
"error": {
|
|
"code": "validation",
|
|
"message": "woo_base_url is read-only (SETTINGS-85: env only)",
|
|
"detail": {"field": "woo_base_url"},
|
|
}
|
|
},
|
|
)
|
|
|
|
conn = get_connection()
|
|
for k, v in payload.items():
|
|
if k not in ALL_KEYS:
|
|
continue # ignore unknown for partial
|
|
if k in SECRET_KEYS:
|
|
if v in (None, "", False):
|
|
conn.execute(
|
|
"INSERT OR REPLACE INTO settings(key, value, value_enc) VALUES (?,?,NULL)",
|
|
(k, None),
|
|
)
|
|
else:
|
|
try:
|
|
enc = encrypt(str(v))
|
|
except Exception:
|
|
enc = None
|
|
conn.execute(
|
|
"INSERT OR REPLACE INTO settings(key, value, value_enc) VALUES (?,?,?)",
|
|
(k, None, enc),
|
|
)
|
|
else:
|
|
# plain
|
|
if v in (None, ""):
|
|
conn.execute("DELETE FROM settings WHERE key=?", (k,))
|
|
else:
|
|
if k in ("inference_enabled", "content_llm_enabled"):
|
|
val_str = "true" if bool(v) else "false"
|
|
else:
|
|
val_str = str(v)
|
|
conn.execute(
|
|
"INSERT OR REPLACE INTO settings(key, value, value_enc) VALUES (?,?,NULL)",
|
|
(k, val_str),
|
|
)
|
|
conn.commit()
|
|
return get_settings()
|