patrickandClaude Sonnet 5.5
347f55cf7c
Includes atomic migrations, per-rule reminder semantics, first-run helper script and the note on the undecided UI chat idea. Co-Authored-By: Claude Sonnet 5.5 <noreply@anthropic.com>
96 lines
3.1 KiB
Python
96 lines
3.1 KiB
Python
import secrets
|
|
from datetime import datetime
|
|
from pathlib import Path
|
|
|
|
from fastapi import Depends, HTTPException, Request
|
|
from fastapi.responses import RedirectResponse
|
|
from fastapi.templating import Jinja2Templates
|
|
from markupsafe import Markup, escape
|
|
|
|
from ..models import display_state, state_name
|
|
from ..users import User
|
|
|
|
BASE_DIR = Path(__file__).parent
|
|
SAFE_METHODS = frozenset({"GET", "HEAD", "OPTIONS"})
|
|
|
|
CSRF_FIELD = "csrf"
|
|
CSRF_HEADER = "x-csrf-token"
|
|
SESSION_UID = "uid"
|
|
SESSION_VERSION = "sv"
|
|
SESSION_CSRF = "csrf"
|
|
SESSION_FLASH = "flash"
|
|
|
|
FLASH_INFO = "info"
|
|
FLASH_OK = "ok"
|
|
FLASH_ERROR = "error"
|
|
|
|
TS_FORMAT = "%d.%m.%Y %H:%M:%S %Z"
|
|
REDIRECT_STATUS = 303
|
|
|
|
templates = Jinja2Templates(directory=str(BASE_DIR / "templates"))
|
|
templates.env.filters["fmt_ts"] = lambda ts: datetime.fromtimestamp(ts).astimezone().strftime(TS_FORMAT)
|
|
|
|
STATE_CLASSES = {0: "ok", 1: "warn", 2: "error", 3: "error"}
|
|
templates.env.filters["state_name"] = state_name
|
|
templates.env.filters["display_state"] = lambda state, service: display_state(service, state)
|
|
templates.env.filters["state_class"] = lambda state: STATE_CLASSES.get(state, "error")
|
|
|
|
|
|
class LoginRequired(Exception):
|
|
pass
|
|
|
|
|
|
def redirect(url: str) -> RedirectResponse:
|
|
return RedirectResponse(url, status_code=REDIRECT_STATUS)
|
|
|
|
|
|
def csrf_token(request: Request) -> str:
|
|
token = request.session.get(SESSION_CSRF)
|
|
if not token:
|
|
token = secrets.token_urlsafe(32)
|
|
request.session[SESSION_CSRF] = token
|
|
return token
|
|
|
|
|
|
async def verify_csrf(request: Request) -> None:
|
|
if request.method in SAFE_METHODS:
|
|
return
|
|
expected = request.session.get(SESSION_CSRF)
|
|
supplied = request.headers.get(CSRF_HEADER)
|
|
if supplied is None:
|
|
supplied = (await request.form()).get(CSRF_FIELD)
|
|
if not expected or not supplied or not secrets.compare_digest(str(supplied), expected):
|
|
raise HTTPException(403, "CSRF-Prüfung fehlgeschlagen")
|
|
|
|
|
|
def flash(request: Request, text: str, kind: str = FLASH_INFO) -> None:
|
|
request.session.setdefault(SESSION_FLASH, []).append({"kind": kind, "text": text})
|
|
|
|
|
|
def render(request: Request, name: str, user: User | None = None, status_code: int = 200, **context):
|
|
token = csrf_token(request)
|
|
page_context = {
|
|
"user": user,
|
|
"csrf": token,
|
|
"csrf_input": Markup('<input type="hidden" name="%s" value="%s">') % (CSRF_FIELD, escape(token)),
|
|
"flashes": request.session.pop(SESSION_FLASH, []),
|
|
**context,
|
|
}
|
|
return templates.TemplateResponse(request, name, page_context, status_code=status_code)
|
|
|
|
|
|
def current_user(request: Request) -> User:
|
|
uid = request.session.get(SESSION_UID)
|
|
user = request.app.state.users.get(uid) if uid else None
|
|
if user is None or user.disabled or user.session_version != request.session.get(SESSION_VERSION):
|
|
request.session.pop(SESSION_UID, None)
|
|
request.session.pop(SESSION_VERSION, None)
|
|
raise LoginRequired
|
|
return user
|
|
|
|
|
|
def require_admin(user: User = Depends(current_user)) -> User:
|
|
if not user.is_admin:
|
|
raise HTTPException(403, "Nur für Admins")
|
|
return user
|