M1.5 backend: admin role + DB-backed settings, DB-URL-only install
- users.is_admin; first registered user becomes admin; registration gated by the allow_registration setting (first account always allowed). is_admin in /api/auth/* responses; require_admin guard (live DB check). - settings table + code registry (site_name, allow_registration, session_ttl_days) with typed defaults — empty table = all defaults (rule 26). get/set/validate service; GET /api/config (public) + GET/PATCH /api/settings (admin), live session-TTL apply with no restart (rule 25). - Cookie-signing secret now persisted in the DB (before_serving load-or-create), so sessions survive restarts with no volume. Config: DATABASE_URL is the only required env; SECRET_KEY + DATA_DIR are optional break-glass items. - Migration 0003; DB-free tests for settings validation + admin guard. Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01FRgehjoz7Yv8LkUfADxACm
This commit is contained in:
@@ -0,0 +1,167 @@
|
||||
from __future__ import annotations
|
||||
|
||||
import json
|
||||
import secrets
|
||||
from dataclasses import dataclass
|
||||
from datetime import datetime, timezone
|
||||
from typing import Any, Literal
|
||||
|
||||
from .models.settings import Setting
|
||||
|
||||
SettingType = Literal["string", "bool", "int"]
|
||||
|
||||
|
||||
@dataclass(frozen=True)
|
||||
class SettingDef:
|
||||
key: str
|
||||
type: SettingType
|
||||
default: Any
|
||||
label: str
|
||||
description: str
|
||||
group: str
|
||||
|
||||
|
||||
# The source of truth for every user-facing setting. Add a row here and it appears
|
||||
# in the admin Settings UI with a working default — no migration, no env var.
|
||||
REGISTRY: list[SettingDef] = [
|
||||
SettingDef(
|
||||
"site_name", "string", "ThoughtSync", "Site name", "Shown in the header and the browser tab.", "General"
|
||||
),
|
||||
SettingDef(
|
||||
"allow_registration",
|
||||
"bool",
|
||||
True,
|
||||
"Allow new registrations",
|
||||
"When off, only existing users can sign in. The first account is always allowed.",
|
||||
"Access",
|
||||
),
|
||||
SettingDef(
|
||||
"session_ttl_days",
|
||||
"int",
|
||||
30,
|
||||
"Session length (days)",
|
||||
"How long a signed-in session stays valid before another login is required.",
|
||||
"Access",
|
||||
),
|
||||
]
|
||||
|
||||
_BY_KEY: dict[str, SettingDef] = {d.key: d for d in REGISTRY}
|
||||
|
||||
# Internal, non-UI reserved key: the persisted cookie-signing secret. Stored in the
|
||||
# same table but never listed in the registry, so it never shows in the Settings UI.
|
||||
SECRET_KEY_SETTING = "secret_key"
|
||||
|
||||
|
||||
def _coerce_bool(raw: Any) -> bool:
|
||||
if isinstance(raw, bool):
|
||||
return raw
|
||||
if isinstance(raw, str):
|
||||
return raw.strip().lower() in ("1", "true", "yes", "on")
|
||||
return bool(raw)
|
||||
|
||||
|
||||
def _coerce(defn: SettingDef, raw: Any) -> Any:
|
||||
if defn.type == "bool":
|
||||
return _coerce_bool(raw)
|
||||
if defn.type == "int":
|
||||
try:
|
||||
return int(raw)
|
||||
except (ValueError, TypeError):
|
||||
return defn.default
|
||||
return str(raw)
|
||||
|
||||
|
||||
async def _load_raw(db, key: str) -> Any:
|
||||
row = await db.get(Setting, key)
|
||||
if row is None:
|
||||
return None
|
||||
try:
|
||||
return json.loads(row.value)
|
||||
except (ValueError, TypeError):
|
||||
return None
|
||||
|
||||
|
||||
async def _upsert(db, key: str, value: Any) -> None:
|
||||
row = await db.get(Setting, key)
|
||||
payload = json.dumps(value)
|
||||
if row is None:
|
||||
db.add(Setting(key=key, value=payload))
|
||||
else:
|
||||
row.value = payload
|
||||
row.updated_at = datetime.now(timezone.utc)
|
||||
|
||||
|
||||
async def get_setting(db, key: str) -> Any:
|
||||
defn = _BY_KEY.get(key)
|
||||
if defn is None:
|
||||
raise KeyError(key)
|
||||
raw = await _load_raw(db, key)
|
||||
return defn.default if raw is None else _coerce(defn, raw)
|
||||
|
||||
|
||||
async def get_public_config(db) -> dict:
|
||||
"""Non-sensitive settings the unauthenticated login/register screen needs."""
|
||||
return {
|
||||
"site_name": await get_setting(db, "site_name"),
|
||||
"allow_registration": await get_setting(db, "allow_registration"),
|
||||
}
|
||||
|
||||
|
||||
async def get_admin_settings(db) -> list[dict]:
|
||||
"""Every registry setting with its current value + metadata, for the admin UI."""
|
||||
result: list[dict] = []
|
||||
for d in REGISTRY:
|
||||
result.append(
|
||||
{
|
||||
"key": d.key,
|
||||
"type": d.type,
|
||||
"value": await get_setting(db, d.key),
|
||||
"default": d.default,
|
||||
"label": d.label,
|
||||
"description": d.description,
|
||||
"group": d.group,
|
||||
}
|
||||
)
|
||||
return result
|
||||
|
||||
|
||||
def validate_updates(updates: dict) -> tuple[dict, str | None]:
|
||||
"""Coerce/validate a {key: value} dict against the registry. Returns
|
||||
(clean_values, error_message). An unknown key or a bad int is rejected."""
|
||||
clean: dict = {}
|
||||
for key, val in updates.items():
|
||||
defn = _BY_KEY.get(key)
|
||||
if defn is None:
|
||||
return {}, f"unknown setting: {key}"
|
||||
if defn.type == "int":
|
||||
try:
|
||||
clean[key] = int(val)
|
||||
except (ValueError, TypeError):
|
||||
return {}, f"{defn.label} must be a whole number"
|
||||
elif defn.type == "bool":
|
||||
clean[key] = _coerce_bool(val)
|
||||
else:
|
||||
clean[key] = str(val)
|
||||
return clean, None
|
||||
|
||||
|
||||
async def set_settings(db, updates: dict) -> None:
|
||||
for key, value in updates.items():
|
||||
await _upsert(db, key, value)
|
||||
|
||||
|
||||
async def load_or_create_secret_key(db) -> str:
|
||||
"""Return the persisted cookie-signing secret, generating + storing one on first
|
||||
run. Keeps sessions valid across restarts with no env var or volume required."""
|
||||
row = await db.get(Setting, SECRET_KEY_SETTING)
|
||||
if row is not None:
|
||||
try:
|
||||
val = json.loads(row.value)
|
||||
if isinstance(val, str) and val:
|
||||
return val
|
||||
except (ValueError, TypeError):
|
||||
pass
|
||||
key = secrets.token_urlsafe(48)
|
||||
await _upsert(db, SECRET_KEY_SETTING, key)
|
||||
await db.commit()
|
||||
return key
|
||||
Reference in New Issue
Block a user