"""Authenticated client/admin dashboard: forms, entries, settings, integration.""" from __future__ import annotations import csv import io import json from fastapi import APIRouter, Form, HTTPException, Request from fastapi.responses import RedirectResponse, Response import auth import bw_auth import bw_view_as import deps import notify import security import services import viewas from db import execute, query, query_one router = APIRouter() def _load_form_or_403(user: auth.CurrentUser, form_id: int): form = query_one("SELECT * FROM forms WHERE id = ?", (form_id,)) if not form: raise HTTPException(status_code=404, detail="Form not found") if not deps.user_can_access_form(user, form): raise HTTPException(status_code=403, detail="Not your form") return form def _load_form_or_403_manage(user: auth.CurrentUser, form_id: int): """View access PLUS the right to edit config (admin or owning manager — not clients).""" form = _load_form_or_403(user, form_id) if not user.can_create: raise HTTPException(status_code=403, detail="Read-only: form settings are managed by managers") return form @router.get("/dashboard") def dashboard(request: Request): user, imp = deps.dashboard_identity(request) forms = deps.forms_for(user) rows = [] for f in forms: counts = query_one( "SELECT COUNT(*) AS total, " "SUM(CASE WHEN status='new' THEN 1 ELSE 0 END) AS unread " "FROM submissions WHERE form_id = ?", (f["id"],)) rows.append({"form": f, "total": counts["total"] or 0, "unread": counts["unread"] or 0}) # View-as picker: only for a real (non-impersonating) admin/manager with targets. targets = [] if imp else (viewas.viewable_targets(str(user.id)) if viewas.can_use(user.role) else []) return deps.render("dashboard.html", user=user, impersonating=imp, csrf=request.state.csrf, rows=rows, viewas_targets=targets) @router.get("/forms/{form_id}/entries") def form_entries(form_id: int, request: Request, status: str = "all"): user = deps.require_user(request) form = _load_form_or_403(user, form_id) if status in ("new", "read", "spam"): subs = query("SELECT * FROM submissions WHERE form_id = ? AND status = ? " "ORDER BY created_at DESC LIMIT 500", (form_id, status)) else: subs = query("SELECT * FROM submissions WHERE form_id = ? ORDER BY created_at DESC " "LIMIT 500", (form_id,)) entries = [] for s in subs: payload = json.loads(s["payload_json"]) preview = " · ".join(f"{v}" for k, v in list(payload.items())[:3] if not isinstance(v, list)) entries.append({"sub": s, "preview": preview[:120]}) return deps.render("form_entries.html", user=user, csrf=request.state.csrf, form=form, entries=entries, status=status) @router.get("/entries/{sub_id}") def entry_detail(sub_id: int, request: Request): user, imp = deps.dashboard_identity(request) sub = query_one("SELECT * FROM submissions WHERE id = ?", (sub_id,)) if not sub: raise HTTPException(status_code=404, detail="Entry not found") form = _load_form_or_403(user, sub["form_id"]) payload = json.loads(sub["payload_json"]) notifs = query("SELECT * FROM notifications WHERE submission_id = ? ORDER BY id", (sub_id,)) events = query("SELECT * FROM events WHERE submission_id = ? ORDER BY id", (sub_id,)) return deps.render("entry_detail.html", user=user, impersonating=imp, csrf=request.state.csrf, form=form, sub=sub, payload=payload, notifs=notifs, events=events) @router.post("/entries/{sub_id}/mark") def mark_entry(sub_id: int, request: Request, status: str = Form(...), csrf: str = Form("")): user, _ = deps.dashboard_identity(request) if not auth.csrf_ok(request.cookies.get(auth.CSRF_COOKIE), csrf): raise HTTPException(status_code=403, detail="Bad CSRF token") sub = query_one("SELECT * FROM submissions WHERE id = ?", (sub_id,)) if not sub: raise HTTPException(status_code=404) _load_form_or_403(user, sub["form_id"]) if status in ("new", "read", "spam"): execute("UPDATE submissions SET status = ? WHERE id = ?", (status, sub_id)) return RedirectResponse(f"/entries/{sub_id}", status_code=303) @router.post("/notifications/{notif_id}/resend") def resend_notification(notif_id: int, request: Request, csrf: str = Form("")): user, _ = deps.dashboard_identity(request) if not auth.csrf_ok(request.cookies.get(auth.CSRF_COOKIE), csrf): raise HTTPException(status_code=403, detail="Bad CSRF token") n = query_one("SELECT * FROM notifications WHERE id = ?", (notif_id,)) if not n: raise HTTPException(status_code=404) sub = query_one("SELECT form_id FROM submissions WHERE id = ?", (n["submission_id"],)) _load_form_or_403(user, sub["form_id"]) deps.log_event(f"manual resend requested by {user.email}", submission_id=n["submission_id"], form_id=sub["form_id"]) execute("UPDATE notifications SET status='pending', last_error=NULL, " "next_attempt_at=CURRENT_TIMESTAMP WHERE id=?", (notif_id,)) notify.process_notification(notif_id) return RedirectResponse(f"/entries/{n['submission_id']}", status_code=303) @router.get("/forms/{form_id}/export.csv") def export_csv(form_id: int, request: Request): user, _ = deps.dashboard_identity(request) form = _load_form_or_403(user, form_id) subs = query("SELECT * FROM submissions WHERE form_id = ? ORDER BY created_at DESC", (form_id,)) # Union of all payload keys across entries → stable column set. keys: list[str] = [] parsed = [] for s in subs: p = json.loads(s["payload_json"]) parsed.append((s, p)) for k in p: if k not in keys: keys.append(k) buf = io.StringIO() w = csv.writer(buf) w.writerow(["entry_id", "submitted_at", "status", "ip", "origin", *keys]) for s, p in parsed: vals = [(", ".join(v) if isinstance(v, list) else v) for v in (p.get(k, "") for k in keys)] w.writerow([s["id"], s["created_at"], s["status"], s["submitter_ip"] or "", s["origin"] or "", *vals]) filename = f"{form['public_key']}-entries.csv" return Response(buf.getvalue(), media_type="text/csv", headers={"Content-Disposition": f'attachment; filename="{filename}"'}) @router.get("/forms/{form_id}/integration") def integration(form_id: int, request: Request): user = deps.require_creator(request) form = _load_form_or_403_manage(user, form_id) return deps.render("integration.html", user=user, csrf=request.state.csrf, form=form, base=deps.APP_BASE_URL) @router.get("/forms/{form_id}/settings") def settings_page(request: Request, form_id: int): user = deps.require_creator(request) form = _load_form_or_403_manage(user, form_id) return deps.render("form_settings.html", user=user, csrf=request.state.csrf, form=form, grantees=services.form_grantees(form_id), saved=False, error=None) @router.post("/forms/{form_id}/settings") def settings_save(request: Request, form_id: int, name: str = Form(...), notify_recipients: str = Form(""), reply_to_field: str = Form("email"), allowed_origins: str = Form(""), redirect_url: str = Form(""), from_address: str = Form(""), honeypot_field: str = Form("_hp"), min_submit_seconds: int = Form(2), rate_limit_per_hour: int = Form(60), active: str = Form(""), autoresponse_enabled: str = Form(""), autoresponse_subject: str = Form(""), autoresponse_body: str = Form(""), csrf: str = Form("")): user = deps.require_creator(request) form = _load_form_or_403_manage(user, form_id) if not auth.csrf_ok(request.cookies.get(auth.CSRF_COOKIE), csrf): raise HTTPException(status_code=403, detail="Bad CSRF token") # Only admins may set a custom From (per-client verified sender domain). new_from = from_address.strip() if user.is_admin else (form["from_address"] or "") execute( "UPDATE forms SET name=?, notify_recipients=?, reply_to_field=?, allowed_origins=?, " "redirect_url=?, from_address=?, honeypot_field=?, min_submit_seconds=?, " "rate_limit_per_hour=?, active=?, autoresponse_enabled=?, autoresponse_subject=?, " "autoresponse_body=? WHERE id=?", (name.strip(), notify_recipients.strip(), reply_to_field.strip() or "email", allowed_origins.strip(), redirect_url.strip(), new_from or None, honeypot_field.strip() or "_hp", max(0, min_submit_seconds), max(0, rate_limit_per_hour), 1 if active else 0, 1 if autoresponse_enabled else 0, autoresponse_subject.strip(), autoresponse_body.strip(), form_id), ) form = query_one("SELECT * FROM forms WHERE id = ?", (form_id,)) return deps.render("form_settings.html", user=user, csrf=request.state.csrf, form=form, grantees=services.form_grantees(form_id), saved=True, error=None) # ---------- Form access grants (manage) ---------- @router.post("/forms/{form_id}/grant") def grant_access(form_id: int, request: Request, email: str = Form(...), invite: str = Form(""), csrf: str = Form("")): user = deps.require_creator(request) _load_form_or_403_manage(user, form_id) if not auth.csrf_ok(request.cookies.get(auth.CSRF_COOKIE), csrf): raise HTTPException(status_code=403, detail="Bad CSRF token") email = email.strip().lower() if "@" in email: client = services.find_client_by_email(email) created = client is None cid = client["id"] if client else services.create_user( name=email.split("@")[0], email=email, role="client") services.grant_form(form_id, cid) if created or invite: services.invite(cid, send_email=True) return RedirectResponse(f"/forms/{form_id}/settings", status_code=303) @router.post("/forms/{form_id}/revoke") def revoke_access(form_id: int, request: Request, client_id: int = Form(...), csrf: str = Form("")): user = deps.require_creator(request) _load_form_or_403_manage(user, form_id) if not auth.csrf_ok(request.cookies.get(auth.CSRF_COOKIE), csrf): raise HTTPException(status_code=403, detail="Bad CSRF token") services.revoke_form(form_id, client_id) return RedirectResponse(f"/forms/{form_id}/settings", status_code=303) # ---------- Deletion ---------- @router.post("/forms/{form_id}/delete") def delete_form(form_id: int, request: Request, csrf: str = Form("")): user = deps.require_creator(request) _load_form_or_403_manage(user, form_id) if not auth.csrf_ok(request.cookies.get(auth.CSRF_COOKIE), csrf): raise HTTPException(status_code=403, detail="Bad CSRF token") services.delete_form(form_id) return RedirectResponse("/dashboard", status_code=303) @router.post("/entries/{sub_id}/delete") def delete_entry(sub_id: int, request: Request, csrf: str = Form("")): user = deps.require_creator(request) sub = query_one("SELECT form_id FROM submissions WHERE id = ?", (sub_id,)) if not sub: raise HTTPException(status_code=404) _load_form_or_403_manage(user, sub["form_id"]) if not auth.csrf_ok(request.cookies.get(auth.CSRF_COOKIE), csrf): raise HTTPException(status_code=403, detail="Bad CSRF token") services.delete_submission(sub_id) return RedirectResponse(f"/forms/{sub['form_id']}/entries", status_code=303) # ---------- View-as (standard bw_view_as; read-only; hub-reported) ---------- @router.post("/view-as/start") def view_as_start(request: Request, target: str = Form(...), csrf: str = Form("")): real = deps.require_user(request) if not auth.csrf_ok(request.cookies.get(auth.CSRF_COOKIE), csrf): raise HTTPException(status_code=403, detail="Bad CSRF token") # The drop-in enforces authz (via our can_view_as), no-escalation, self/empty. sess = viewas.build_session(real.id, None) try: bw_view_as.start(sess, target, can_view_as=viewas.can_view_as, rank_of=viewas.rank_of, mode="readonly") except bw_view_as.ViewAsError as e: raise HTTPException(status_code=403, detail=str(e)) tid = int(bw_view_as.effective_username(sess)) resp = RedirectResponse("/dashboard", status_code=303) resp.set_cookie(auth.VIEWAS_COOKIE, auth.make_viewas_cookie(tid), max_age=86400, httponly=True, secure=True, samesite="lax") # Owner oversight: show it on the BW hub Live page (best-effort). try: bw_auth.report_impersonation(viewas.report_label(str(real.id)), viewas.report_label(str(tid)), mode="readonly", active=True) except bw_auth.BWAuthError: pass return resp @router.get("/view-as/stop") def view_as_stop(request: Request): real = deps.current_user(request) target = auth.read_viewas_cookie(request.cookies.get(auth.VIEWAS_COOKIE)) resp = RedirectResponse("/dashboard", status_code=303) resp.delete_cookie(auth.VIEWAS_COOKIE) if real and target is not None: try: bw_auth.report_impersonation(viewas.report_label(str(real.id)), viewas.report_label(str(target)), active=False) except bw_auth.BWAuthError: pass return resp # ---------- Account: personal API tokens ---------- @router.get("/account") def account(request: Request): user = deps.require_user(request) tokens = query("SELECT id, label, revoked, created_at, last_used_at FROM api_tokens " "WHERE client_id = ? ORDER BY id DESC", (user.id,)) return deps.render("account.html", user=user, csrf=request.state.csrf, tokens=tokens, new_token=None) @router.post("/account/tokens") def create_token(request: Request, label: str = Form(""), csrf: str = Form("")): user = deps.require_creator(request) # only admin/manager need API tokens if not auth.csrf_ok(request.cookies.get(auth.CSRF_COOKIE), csrf): raise HTTPException(status_code=403, detail="Bad CSRF token") plain = services.generate_api_token(user.id, label) tokens = query("SELECT id, label, revoked, created_at, last_used_at FROM api_tokens " "WHERE client_id = ? ORDER BY id DESC", (user.id,)) return deps.render("account.html", user=user, csrf=request.state.csrf, tokens=tokens, new_token=plain) @router.post("/account/tokens/{token_id}/revoke") def revoke_token(token_id: int, request: Request, csrf: str = Form("")): user = deps.require_user(request) if not auth.csrf_ok(request.cookies.get(auth.CSRF_COOKIE), csrf): raise HTTPException(status_code=403, detail="Bad CSRF token") execute("UPDATE api_tokens SET revoked = 1 WHERE id = ? AND client_id = ?", (token_id, user.id)) return RedirectResponse("/account", status_code=303)