295 lines
13 KiB
Python
295 lines
13 KiB
Python
"""Gemeinsames für die HTML-Routen: Anmeldung, CSRF, Templates, Fehlerseiten."""
|
||
|
||
import logging
|
||
from pathlib import Path
|
||
from typing import Annotated, Any, NamedTuple
|
||
from urllib.parse import quote
|
||
|
||
from fastapi import Depends, FastAPI, Request
|
||
from fastapi.responses import HTMLResponse, RedirectResponse, Response
|
||
from fastapi.templating import Jinja2Templates
|
||
from jinja2 import pass_context
|
||
|
||
from app.core import charts, logs, markdown
|
||
from app.core.config import Settings
|
||
from app.core.crypto import Vault
|
||
from app.core.database import Database
|
||
from app.core.errors import Forbidden, Invalid, NotFound
|
||
from app.core.security import same
|
||
from app.help.service import link_for as help_link
|
||
from app.identity import Principal
|
||
from app.identity import service as identity
|
||
from app.workspaces.models import ROLE_LABELS
|
||
|
||
COOKIE = "aiw_session"
|
||
log = logging.getLogger(__name__)
|
||
templates = Jinja2Templates(directory=Path(__file__).parent / "templates")
|
||
templates.env.filters["markdown"] = markdown.render
|
||
templates.env.globals["chart"] = charts.render
|
||
templates.env.globals["role_labels"] = ROLE_LABELS
|
||
templates.env.globals["help_link"] = help_link
|
||
STATIC = Path(__file__).parent / "static"
|
||
|
||
|
||
@pass_context
|
||
def asset(context: dict, path: str) -> str:
|
||
"""URL einer statischen Datei mit Version (Änderungszeit und Größe): nach einem Update lädt der Browser
|
||
die neue Datei, statt eine zwischengespeicherte alte zu nehmen."""
|
||
url = context["request"].url_for("static", path=path)
|
||
try:
|
||
stat = (STATIC / path).stat()
|
||
except OSError:
|
||
return str(url)
|
||
return f"{url}?v={int(stat.st_mtime):x}{stat.st_size:x}"
|
||
|
||
|
||
templates.env.globals["asset"] = asset
|
||
|
||
|
||
def money(value: Any, places: int = 2) -> str:
|
||
"""Betrag deutsch formatiert; kleine Beträge mit mehr Stellen, damit sie nicht als 0,00 erscheinen."""
|
||
if value is None:
|
||
return "–"
|
||
value = float(value)
|
||
if 0 < abs(value) < 0.01:
|
||
places = 4
|
||
return f"{value:,.{places}f}".replace(",", "X").replace(".", ",").replace("X", ".")
|
||
|
||
|
||
def number(value: Any) -> str:
|
||
return f"{int(value or 0):,}".replace(",", ".")
|
||
|
||
|
||
templates.env.filters["money"] = money
|
||
templates.env.filters["number"] = number
|
||
templates.env.filters["combine_before"] = lambda query, before: {**query, "before": before}
|
||
|
||
|
||
def db(request: Request) -> Database:
|
||
return request.app.state.db
|
||
|
||
|
||
def settings(request: Request) -> Settings:
|
||
return request.app.state.settings
|
||
|
||
|
||
def vault(request: Request) -> Vault:
|
||
return request.app.state.vault
|
||
|
||
|
||
class LoginRequired(Exception):
|
||
pass
|
||
|
||
|
||
class CsrfFailed(Exception):
|
||
pass
|
||
|
||
|
||
class PasswordChangeRequired(Exception):
|
||
pass
|
||
|
||
|
||
# Erreichbar, solange ein neues Passwort gesetzt werden muss
|
||
PASSWORD_PATHS = ("/password", "/logout")
|
||
|
||
|
||
def current(request: Request) -> Principal | None:
|
||
if not hasattr(request.state, "principal"):
|
||
token = request.cookies.get(COOKIE, "")
|
||
with db(request).session() as session:
|
||
request.state.principal = identity.resolve(session, token)
|
||
if request.state.principal is not None:
|
||
logs.bind(user_id=request.state.principal.user_id)
|
||
return request.state.principal
|
||
|
||
|
||
def user(request: Request) -> Principal:
|
||
principal = current(request)
|
||
if principal is None:
|
||
raise LoginRequired
|
||
if principal.must_change_password and request.url.path not in PASSWORD_PATHS:
|
||
raise PasswordChangeRequired
|
||
return principal
|
||
|
||
|
||
async def checked(request: Request, principal: Annotated[Principal, Depends(user)]) -> Principal:
|
||
"""Für schreibende Anfragen: CSRF-Token aus Kopfzeile (htmx) oder Formularfeld."""
|
||
sent = request.headers.get("X-CSRF-Token")
|
||
if sent is None:
|
||
form = await request.form()
|
||
sent = str(form.get("_csrf", ""))
|
||
if not sent or not same(sent, principal.csrf):
|
||
raise CsrfFailed
|
||
return principal
|
||
|
||
|
||
User = Annotated[Principal, Depends(user)]
|
||
Checked = Annotated[Principal, Depends(checked)]
|
||
|
||
|
||
def render(request: Request, name: str, context: dict[str, Any] | None = None, status_code: int = 200) -> HTMLResponse:
|
||
context = {"p": current(request), "currency": settings(request).cost_currency,
|
||
"request_id": getattr(request.state, "request_id", ""), **(context or {})}
|
||
if context.get("current_workspace") is not None and context["p"] is not None:
|
||
context["ws_nav"] = workspace_nav(request, context["p"], context["current_workspace"].id)
|
||
context["trail"] = breadcrumbs(request.url.path, context)
|
||
return templates.TemplateResponse(request, name, context, status_code=status_code)
|
||
|
||
|
||
class Crumb(NamedTuple):
|
||
label: str
|
||
href: str | None = None
|
||
workspace: bool = False
|
||
|
||
|
||
# Wie in der Seitenleiste benannt
|
||
ADMIN_SECTIONS = {"teams": "Teams", "users": "Nutzer", "models": "Modelle und Zugänge", "tools": "Tools",
|
||
"mcp": "MCP-Server", "databases": "Datenbank-Treiber", "hugin": "Hugin",
|
||
"usage": "Nutzung und Kosten", "runs": "Runs", "audit": "Audit-Log",
|
||
"notifications": "Benachrichtigungen"}
|
||
WORKSPACE_SECTIONS = {"repo": "Repository", "data": "Daten", "memory": "Gedächtnis", "settings": "Einstellungen",
|
||
"usage": "Nutzung und Kosten", "runs": "Runs", "audit": "Audit-Log"}
|
||
HUGIN_PAGES = {"catalog": "Katalog", "import": "Aus .hugin.yaml"}
|
||
|
||
|
||
def breadcrumbs(path: str, context: dict[str, Any]) -> list[Crumb]:
|
||
"""Pfad in der Topbar (Workspaces › Workspace › Bereich › Detail); der letzte Eintrag ist die Seite selbst."""
|
||
parts = [part for part in path.split("/") if part]
|
||
head = parts[0] if parts else ""
|
||
if head == "approvals":
|
||
approval = context.get("a")
|
||
detail = [Crumb(approval.arguments.get("title") or approval.tool)] if approval is not None else []
|
||
return [Crumb("Freigaben", "/approvals"), *detail]
|
||
if head == "admin" and len(parts) > 1:
|
||
section, server = parts[1], context.get("server")
|
||
if section == "mcp" and server is not None and server.role == "database":
|
||
section = "databases"
|
||
if section not in ADMIN_SECTIONS:
|
||
return []
|
||
crumbs = [Crumb("Administration"), Crumb(ADMIN_SECTIONS[section], f"/admin/{section}")]
|
||
detail = context.get("team") if section == "teams" else server
|
||
if len(parts) > 2 and detail is not None:
|
||
crumbs.append(Crumb(detail.name if section == "teams" else detail.label))
|
||
return crumbs
|
||
if head == "hilfe":
|
||
page = context.get("page")
|
||
return [Crumb("Hilfe", "/hilfe"), *([Crumb(page.page.title)] if page else [])]
|
||
if head in ("profile", "password"):
|
||
return [Crumb("Profil", "/profile"), *([Crumb("Passwort ändern")] if head == "password" else [])]
|
||
|
||
workspace = context.get("current_workspace")
|
||
if workspace is None:
|
||
if parts == ["workspaces"]:
|
||
return [Crumb("Workspaces", "/workspaces")]
|
||
if parts == ["workspaces", "new"]:
|
||
return [Crumb("Workspaces", "/workspaces"), Crumb("Workspace anlegen")]
|
||
return []
|
||
wbase = f"/workspaces/{workspace.id}"
|
||
crumbs = [Crumb("Workspaces", "/workspaces"), Crumb(workspace.name, wbase, workspace=True)]
|
||
data = Crumb("Daten", f"{wbase}/data")
|
||
settings_ = Crumb("Einstellungen", f"{wbase}/settings")
|
||
mcp_tab = [settings_, Crumb("MCP-Server", f"{wbase}/settings?tab=mcp")]
|
||
rest = parts[2:] if head == "workspaces" else []
|
||
section = rest[0] if rest else ""
|
||
if head == "chats":
|
||
chat = context.get("chat")
|
||
crumbs += [Crumb("Chats", f"{wbase}#chats"), *([Crumb(chat.title)] if chat is not None else [])]
|
||
elif head == "sources" and context.get("source") is not None:
|
||
crumbs += [data, Crumb(f"{context['source'].name} bearbeiten")]
|
||
elif head == "folders" and context.get("folder") is not None:
|
||
crumbs += [Crumb("Daten", f"{wbase}/data#folders"), Crumb(context["folder"].name)]
|
||
elif head == "connections" and context.get("server") is not None:
|
||
crumbs += [*mcp_tab, Crumb(context["server"].label)]
|
||
elif rest == ["sources", "new"]:
|
||
crumbs += [data, Crumb("Git-Repository verbinden")]
|
||
elif rest == ["databases", "new"]:
|
||
crumbs += [*mcp_tab, Crumb("Datenbank hinzufügen")]
|
||
elif rest == ["connections", "new"]:
|
||
crumbs += [*mcp_tab, Crumb("MCP-Server hinzufügen")]
|
||
elif section == "hugin":
|
||
crumbs += [settings_, Crumb("Hugin", f"{wbase}/settings?tab=hugin"),
|
||
Crumb(HUGIN_PAGES.get(rest[1] if len(rest) > 1 else "", "Abgleichen"))]
|
||
elif section == "repo":
|
||
source, pr = context.get("source"), context.get("pr")
|
||
query = f"?source={quote(source.name)}" if source else ""
|
||
if rest[1:2] == ["compare"]:
|
||
query += ("&" if query else "?") + f"tab={'prs' if pr else 'branches'}"
|
||
crumbs.append(Crumb("Repository", f"{wbase}/repo{query}"))
|
||
if rest[1:2] == ["compare"]:
|
||
crumbs.append(Crumb(f"Pull Request #{pr}" if pr else "Vergleich"))
|
||
elif len(rest) > 1:
|
||
crumbs.append(Crumb(f"Commit {rest[-1][:7]}"))
|
||
elif section in WORKSPACE_SECTIONS:
|
||
crumbs.append(Crumb(WORKSPACE_SECTIONS[section], f"{wbase}/{section}"))
|
||
return crumbs
|
||
|
||
|
||
def workspace_nav(request: Request, principal: Principal, workspace_id: int) -> dict[str, Any]:
|
||
"""Navigation des aktiven Workspace in der Seitenleiste: Rolle, Gedächtnis, Repository, letzte Chats."""
|
||
from sqlalchemy import select
|
||
|
||
from app.chat.models import ChatSession
|
||
from app.mcp import service as mcp
|
||
from app.workspaces.models import WorkspaceMembership
|
||
with db(request).session() as session:
|
||
role = session.scalar(select(WorkspaceMembership.role).where(
|
||
WorkspaceMembership.workspace_id == workspace_id, WorkspaceMembership.user_id == principal.user_id))
|
||
chats = session.execute(select(ChatSession.id, ChatSession.title).where(
|
||
ChatSession.workspace_id == workspace_id).order_by(ChatSession.created.desc()).limit(8)).all()
|
||
memory = any(s.role == "memory" for s, _ in mcp.enabled(session, workspace_id))
|
||
from app.repo import service as repo
|
||
has_repo = repo.available(session, workspace_id)
|
||
return {"is_admin": role == "admin", "memory": memory, "chats": chats, "repo": has_repo}
|
||
|
||
|
||
def redirect(url: str) -> Response:
|
||
return RedirectResponse(url, status_code=303)
|
||
|
||
|
||
def is_htmx(request: Request) -> bool:
|
||
return request.headers.get("HX-Request") == "true"
|
||
|
||
|
||
def install_error_pages(app: FastAPI) -> None:
|
||
@app.exception_handler(LoginRequired)
|
||
async def login_required(request: Request, _: Exception) -> Response:
|
||
if is_htmx(request) or request.url.path.endswith("/events"):
|
||
return Response(status_code=401, headers={"HX-Redirect": "/login"})
|
||
target = request.url.path + (f"?{request.url.query}" if request.url.query else "")
|
||
return redirect(f"/login?next={quote(target)}")
|
||
|
||
@app.exception_handler(PasswordChangeRequired)
|
||
async def password_change_required(request: Request, _: Exception) -> Response:
|
||
if is_htmx(request) or request.url.path.endswith("/events"):
|
||
return Response(status_code=403, headers={"HX-Redirect": "/password"})
|
||
return redirect("/password")
|
||
|
||
@app.exception_handler(CsrfFailed)
|
||
async def csrf_failed(request: Request, _: Exception) -> Response:
|
||
return render(request, "error.html", {"title": "Sitzung abgelaufen",
|
||
"text": "Bitte die Seite neu laden und es erneut versuchen."}, 403)
|
||
|
||
@app.exception_handler(NotFound)
|
||
async def not_found(request: Request, _: Exception) -> Response:
|
||
return render(request, "error.html", {"title": "Nicht gefunden",
|
||
"text": "Das gibt es nicht, oder du hast keinen Zugriff."}, 404)
|
||
|
||
@app.exception_handler(Forbidden)
|
||
async def forbidden(request: Request, e: Exception) -> Response:
|
||
return render(request, "error.html", {"title": "Keine Berechtigung",
|
||
"text": str(e) or "Dafür brauchst du die Rolle Admin in diesem Workspace."},
|
||
403)
|
||
|
||
@app.exception_handler(Exception)
|
||
async def unexpected(request: Request, e: Exception) -> Response:
|
||
log.error("Unerwarteter Fehler", exc_info=e)
|
||
return templates.TemplateResponse(request, "error.html", {
|
||
"p": None, "title": "Unerwarteter Fehler", "request_id": getattr(request.state, "request_id", ""),
|
||
"text": "Das hat nicht geklappt. Bitte später erneut versuchen; wenn es bleibt, die Fehler-ID melden."},
|
||
status_code=500)
|
||
|
||
@app.exception_handler(Invalid)
|
||
async def invalid(request: Request, e: Exception) -> Response:
|
||
if is_htmx(request):
|
||
return render(request, "_alert.html", {"error": str(e)}, 422)
|
||
return render(request, "error.html", {"title": "Nicht möglich", "text": str(e)}, 400)
|