Files
siftlode/backend/app/routes/config.py
T
peter 99df44d7fb fix(api): mutations answer ok; route helpers flush and leave the commit to the handler
Two rows of the canonical-patterns table had drifted, found while closing the roadmap audit:

- 63 mutation responses carried no `ok` — `{"unsubscribed": true}`, `{"status": "ok"}`,
  `{"added": 3, …}` and so on. Each now leads with `"ok": true` (additive: every field a client
  reads is still there).
- Twelve route helpers committed on their own. Six were just the end of their handler's work and now
  flush: the playlist item add/remove and Watch later get-or-create, the Plex series/season bulk
  mark, and the invite decision + account activation — which an approval used to land in separate
  transactions. The other six commit on purpose (a token durable before its link is mailed, a
  single-use token burned as its own unit, the Google callback's tail, a background registration
  with its own session, the staged search ingest around YouTube calls), each listed with its reason.

`tests/test_route_conventions.py` holds both rules in code; CLAUDE.md's two rows point at it.
2026-09-19 18:21:06 +02:00

134 lines
4.7 KiB
Python

"""Admin Configuration page API: read/edit the DB-overridable operational settings defined in
app.sysconfig. Secret values (e.g. SMTP password) are write-only — never returned."""
import re
from typing import Any
from fastapi import APIRouter, Depends
from pydantic import BaseModel
from sqlalchemy.orm import Session
from app import audit
from app import email as email_mod
from app import logs
from app import mailtext
from app import sysconfig
from app.audit import AuditAction
from app.auth import admin_user
from app.db import get_db
from app.models import User
from app.refusal import refused
from app.routes._common import not_found
from app.refusal import setting_refused
router = APIRouter(prefix="/api/admin/config", tags=["admin"])
# Strip URL userinfo (user:pass@) so a non-secret setting that happens to embed credentials —
# e.g. an authenticated egress proxy http://user:pass@host:port — never lands plaintext in the
# audit trail. Non-secret values are shown live in the UI, but the log shouldn't retain creds.
_URL_USERINFO = re.compile(r"(://)[^/@\s]+@")
def _redact(v):
return _URL_USERINFO.sub(r"\1***@", v) if isinstance(v, str) else v
def _apply_runtime(db: Session, key: str) -> None:
"""Some knobs change how THIS process behaves, not just what a later read returns — they have to
take effect now, without a restart. One dispatch for both call sites (set and reset), so the next
such key can't be wired into one of them and forgotten in the other."""
if key == "log_level":
logs.set_level(sysconfig.get_str(db, "log_level"))
@router.get("")
def get_config(_: User = Depends(admin_user), db: Session = Depends(get_db)) -> dict:
return sysconfig.describe(db)
class ConfigSetIn(BaseModel):
# A config value is arbitrarily typed (str/int/bool/…), validated downstream by
# sysconfig.set_value. `value` defaults to a sentinel so a MISSING value stays a 400
# ("Missing 'value'") while an explicit `null` is a real, settable value.
value: Any = None
@router.patch("/{key}")
def set_config(
key: str,
body: ConfigSetIn,
admin: User = Depends(admin_user),
db: Session = Depends(get_db),
) -> dict:
s = sysconfig.spec(key)
if s is None:
raise not_found("config_key")
if "value" not in body.model_fields_set:
raise refused(400, "value_missing")
if s.secret and not sysconfig.secrets_manageable():
raise refused(400, "no_encryption_key")
old = None if s.secret else sysconfig.get(db, key)
try:
sysconfig.set_value(db, key, body.value)
except sysconfig.InvalidSetting as exc:
raise setting_refused(exc) from None
# Never log a secret's value — only that it was changed, by whom. Non-secret values are logged
# but with URL credentials redacted (a proxy/URL setting can embed user:pass). Skip the row when
# a non-secret value didn't actually change (a no-op re-save shouldn't add trail noise); secrets
# are write-only so we can't compare — always log.
new = None if s.secret else sysconfig.get(db, key)
if s.secret or new != old:
audit.record(
db,
admin.id,
AuditAction.CONFIG_SET,
target_type="config",
target_id=key,
summary=f"{key} = (secret updated)" if s.secret else f"{key} = {_redact(new)}",
before=None if s.secret else {"value": _redact(old)},
after=None if s.secret else {"value": _redact(new)},
)
db.commit()
_apply_runtime(db, key)
return sysconfig.describe(db)
@router.delete("/{key}")
def reset_config(
key: str,
admin: User = Depends(admin_user),
db: Session = Depends(get_db),
) -> dict:
s = sysconfig.spec(key)
if s is None:
raise not_found("config_key")
old = None if s.secret else sysconfig.get(db, key)
sysconfig.reset(db, key)
audit.record(
db,
admin.id,
AuditAction.CONFIG_RESET,
target_type="config",
target_id=key,
summary=f"{key} → default",
before=None if s.secret else {"value": _redact(old)},
)
db.commit()
_apply_runtime(db, key)
return sysconfig.describe(db)
@router.post("/test-email")
def send_test_email(
user: User = Depends(admin_user),
db: Session = Depends(get_db),
) -> dict:
"""Send a test email to the requesting admin using the current (DB or env) SMTP config —
lets the admin verify the settings actually work."""
if not email_mod.email_enabled():
raise refused(400, "smtp_off")
sent = email_mod.send_test(user.email, mailtext.lang_for_user(user))
if not sent:
raise refused(422, "smtp_failed")
return {"ok": True, "sent": True, "to": user.email}