"""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", "crop_fill", } ALL_KEYS = SECRET_KEYS | PLAIN_KEYS # Plain keys surfaced as numbers in the API (stored as TEXT in `settings`). NUMERIC_KEYS: set[str] = {"crop_fill"} DEFAULTS: dict[str, Any] = { "inference_enabled": False, "content_llm_enabled": False, "woo_create_status": "draft", # #99: global crop fill %, one source of truth for the ref line + PNG crop. "crop_fill": 90, } 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) if k in NUMERIC_KEYS: try: f = float(result[k]) result[k] = int(f) if f.is_integer() else f except (TypeError, ValueError): 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"}, } }, ) # #99: rules live here in the 4xx layer, not in the DB (structural only). if "crop_fill" in payload: raw = payload["crop_fill"] try: pct = float(raw) except (TypeError, ValueError): pct = None if pct is None or not (1 <= pct <= 100): raise HTTPException( status_code=400, detail={ "error": { "code": "validation", "message": "crop_fill must be a number between 1 and 100 (percent)", "detail": {"field": "crop_fill"}, } }, ) 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()