import hashlib from datetime import timedelta from pathlib import Path from fastapi import APIRouter, Form, HTTPException, Request, status from fastapi.responses import HTMLResponse, RedirectResponse from fastapi.templating import Jinja2Templates from app.auth.admin import password_matches, require_admin from app.auth.csrf import ensure_csrf_token, validate_csrf from app.db.models import SyncUser, utcnow from app.db.repositories import ActivityRepository, SyncRunRepository, UserRepository from app.security.credentials import CredentialCipher from app.web.forms import UserFormData DASHBOARD_SUMMARY_WINDOW = timedelta(days=7) CACHE_BUSTED_STATIC_FILES = ("style.css", "app.js", "htmx.min.js") router = APIRouter() templates = Jinja2Templates(directory=str(Path(__file__).resolve().parent / "templates")) def _compute_static_version(static_dir: Path) -> str: hasher = hashlib.sha256() for name in CACHE_BUSTED_STATIC_FILES: try: hasher.update((static_dir / name).read_bytes()) except FileNotFoundError: continue return hasher.hexdigest()[:10] templates.env.globals["static_version"] = _compute_static_version( Path(__file__).resolve().parent / "static" ) def _next_sync_tick(request: Request): scheduler = getattr(request.app.state, "scheduler", None) return getattr(scheduler, "next_tick", None) templates.env.globals["next_sync_tick"] = _next_sync_tick def _get_user_or_404(repository: UserRepository, user_id: int) -> SyncUser: user = repository.get(user_id) if user is None: raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="User not found") return user def _cipher(request: Request) -> CredentialCipher: return CredentialCipher(request.app.state.settings.credential_encryption_key) def _require_non_empty(value: str, field_name: str) -> None: if not value.strip(): raise HTTPException( status_code=status.HTTP_400_BAD_REQUEST, detail=f"{field_name} must not be empty", ) @router.get("/login", response_class=HTMLResponse) def login_page(request: Request): return templates.TemplateResponse(request, "login.html", {"csrf_token": ensure_csrf_token(request)}) @router.post("/login") def login( request: Request, password: str = Form(...), csrf_token: str = Form(...), ): validate_csrf(request, csrf_token) settings = request.app.state.settings if not password_matches(password, settings.admin_password): return templates.TemplateResponse( request, "login.html", {"csrf_token": ensure_csrf_token(request), "error": "Invalid password"}, status_code=401, ) request.session["admin_authenticated"] = True return RedirectResponse("/", status_code=303) @router.get("/", response_class=HTMLResponse) def dashboard(request: Request): require_admin(request) with request.app.state.session_factory() as session: repository = UserRepository(session) rows = repository.dashboard_rows() summary = repository.dashboard_summary(since=utcnow() - DASHBOARD_SUMMARY_WINDOW) return templates.TemplateResponse( request, "dashboard.html", {"rows": rows, "summary": summary, "csrf_token": ensure_csrf_token(request)}, ) @router.get("/users/new", response_class=HTMLResponse) def new_user_page(request: Request): require_admin(request) return templates.TemplateResponse( request, "users/form.html", { "csrf_token": ensure_csrf_token(request), "user": None, "form_action": "/users", "mywhoosh_email": "", "garmin_email": "", }, ) @router.post("/users") def create_user( request: Request, csrf_token: str = Form(...), name: str = Form(...), mywhoosh_email: str = Form(""), mywhoosh_password: str = Form(""), garmin_email: str = Form(""), garmin_password: str = Form(""), enabled: str | None = Form(None), notify_email_enabled: str | None = Form(None), notification_email: str = Form(""), ): require_admin(request) validate_csrf(request, csrf_token) _require_non_empty(mywhoosh_email, "mywhoosh_email") _require_non_empty(mywhoosh_password, "mywhoosh_password") _require_non_empty(garmin_email, "garmin_email") _require_non_empty(garmin_password, "garmin_password") form = UserFormData( name=name, mywhoosh_email=mywhoosh_email, mywhoosh_password=mywhoosh_password, garmin_email=garmin_email, garmin_password=garmin_password, enabled=enabled is not None, notify_email_enabled=notify_email_enabled is not None, notification_email=notification_email, ) if form.notify_email_enabled: _require_non_empty(form.notification_email, "notification_email") cipher = _cipher(request) with request.app.state.session_factory() as session: repository = UserRepository(session) user = repository.create( name=form.name.strip(), enabled=form.enabled, mywhoosh_email_enc=cipher.encrypt(form.mywhoosh_email.strip()), mywhoosh_password_enc=cipher.encrypt(form.mywhoosh_password), garmin_email_enc=cipher.encrypt(form.garmin_email.strip()), garmin_password_enc=cipher.encrypt(form.garmin_password), notify_email_enabled=form.notify_email_enabled, notification_email=form.notification_email.strip() or None, ) user_id = user.id return RedirectResponse(f"/users/{user_id}", status_code=303) @router.get("/users/{user_id}", response_class=HTMLResponse) def user_detail(request: Request, user_id: int): require_admin(request) with request.app.state.session_factory() as session: user = _get_user_or_404(UserRepository(session), user_id) activities = ActivityRepository(session).list_pending_for_user(user_id) recent_runs = SyncRunRepository(session).list_recent_for_user(user_id, limit=10) return templates.TemplateResponse( request, "users/detail.html", { "csrf_token": ensure_csrf_token(request), "user": user, "activities": activities, "recent_runs": recent_runs, }, ) @router.get("/users/{user_id}/edit", response_class=HTMLResponse) def edit_user_page(request: Request, user_id: int): require_admin(request) cipher = _cipher(request) with request.app.state.session_factory() as session: user = _get_user_or_404(UserRepository(session), user_id) return templates.TemplateResponse( request, "users/form.html", { "csrf_token": ensure_csrf_token(request), "user": user, "form_action": f"/users/{user_id}", "mywhoosh_email": cipher.decrypt(user.mywhoosh_email_enc), "garmin_email": cipher.decrypt(user.garmin_email_enc), }, ) @router.post("/users/{user_id}") def update_user( request: Request, user_id: int, csrf_token: str = Form(...), name: str = Form(...), mywhoosh_email: str = Form(""), mywhoosh_password: str = Form(""), garmin_email: str = Form(""), garmin_password: str = Form(""), enabled: str | None = Form(None), notify_email_enabled: str | None = Form(None), notification_email: str = Form(""), ): require_admin(request) validate_csrf(request, csrf_token) _require_non_empty(mywhoosh_email, "mywhoosh_email") _require_non_empty(garmin_email, "garmin_email") form = UserFormData( name=name, mywhoosh_email=mywhoosh_email, mywhoosh_password=mywhoosh_password, garmin_email=garmin_email, garmin_password=garmin_password, enabled=enabled is not None, notify_email_enabled=notify_email_enabled is not None, notification_email=notification_email, ) if form.notify_email_enabled: _require_non_empty(form.notification_email, "notification_email") cipher = _cipher(request) with request.app.state.session_factory() as session: repository = UserRepository(session) user = _get_user_or_404(repository, user_id) values = { "name": form.name.strip(), "enabled": form.enabled, "mywhoosh_email_enc": cipher.encrypt(form.mywhoosh_email.strip()), "garmin_email_enc": cipher.encrypt(form.garmin_email.strip()), "notify_email_enabled": form.notify_email_enabled, "notification_email": form.notification_email.strip() or None, } if form.mywhoosh_password: values["mywhoosh_password_enc"] = cipher.encrypt(form.mywhoosh_password) if form.garmin_password: values["garmin_password_enc"] = cipher.encrypt(form.garmin_password) repository.update(user, **values) return RedirectResponse(f"/users/{user_id}", status_code=303)