sticknife-home/app/web.py

from __future__ import annotations

import base64
import hashlib
import html
import json
import mimetypes
import urllib.parse
from http import HTTPStatus
from http.cookies import SimpleCookie
from pathlib import Path
from string import Template
from urllib.parse import parse_qs, urlparse

from . import pantheon
from .config import Config
from .authentik import trigger_ldap_sync
from .db import Database, sql_literal
from .ipa import approve_user, email_matches, provision_user, set_password, verify_password as verify_ipa_password
from .mailer import send_password_reset_email, send_verification_email
from .oidc import authorization_url, exchange_code, groups_from_claims, logout_url, userinfo
from .security import hash_password, sign, token_urlsafe, unsign, verify_password


ROOT = Path(__file__).resolve().parent
TEMPLATES = ROOT / "templates"
STATIC = ROOT / "static"

#: The only routes a role="home" deployment answers. Everything else belongs to the deployment
#: that holds the database and the OIDC client, and is reached through config.auth_base_url.
#: Served by a home deployment itself. Everything else is forwarded to the deployment that owns
#: identity. The auth routes are here so a home site can hold its own Authentik session: it needs
#: to know who you are to show "My Account" and "Admin", even though it owns no user table.
HOME_ROUTES = {
    ("GET", "/"),
    ("GET", "/trusted-domains"),
    ("GET", "/auth/start"),
    ("GET", "/auth/callback"),
    ("POST", "/logout"),
}


def esc(value: object) -> str:
    return html.escape("" if value is None else str(value), quote=True)


class WebApp:
    def __init__(self, config: Config):
        self.config = config
        # A home deployment has no database credentials at all, so there is nothing to point at.
        self.db = None if config.role == "home" else Database(config.database_url)
        #: Whether this deployment can run the OIDC flow itself. A home site without these falls
        #: back to handing sign-in off to auth_base_url.
        self.oidc_ready = bool(
            config.oidc_issuer and config.oidc_client_id and config.oidc_client_secret
        )

    def dispatch(self, handler) -> None:
        parsed = urlparse(handler.path)
        path = parsed.path
        method = handler.command
        if path.startswith("/static/"):
            return self.static(handler, path)
        routes = {
            ("GET", "/"): self.register_form,
            ("POST", "/register"): self.register_submit,
            ("GET", "/verify"): self.verify_email,
            ("GET", "/trusted-domains"): self.trusted_domains_page,
            ("GET", "/login"): self.login_form,
            ("POST", "/login"): self.login_submit,
            ("POST", "/logout"): self.logout,
            ("GET", "/auth/start"): self.auth_start,
            ("GET", "/auth/callback"): self.auth_callback,
            ("GET", "/profile"): self.profile,
            ("POST", "/profile"): self.profile_update,
            ("POST", "/profile/password"): self.profile_password_update,
            ("GET", "/password/request"): self.password_request_form,
            ("POST", "/password/request"): self.password_request_submit,
            ("GET", "/password/reset"): self.password_reset_form,
            ("POST", "/password/reset"): self.password_reset_submit,
            ("GET", "/admin"): self.admin,
            ("POST", "/admin/approve"): self.admin_approve,
            ("POST", "/admin/reject"): self.admin_reject,
            ("POST", "/admin/delete-unverified"): self.admin_delete_unverified,
            ("POST", "/admin/purge-unverified"): self.admin_purge_unverified,
        }
        if self.config.role == "home" and (method, path) not in HOME_ROUTES:
            # Not a 404: the page exists, it just lives on the deployment that owns identity. Only
            # GET is forwarded — a 303 would turn a POST into a GET and drop the form body, so any
            # form that submits off-box has to target auth_base_url directly.
            if method != "GET":
                return self.render(handler, "Not found", "<p>That page does not exist.</p>", HTTPStatus.NOT_FOUND)
            query = f"?{parsed.query}" if parsed.query else ""
            return self.redirect(handler, f"{self.config.auth_base_url}{path}{query}")
        view = routes.get((method, path))
        if view is None:
            return self.render(handler, "Not found", "<p>That page does not exist.</p>", HTTPStatus.NOT_FOUND)
        return view(handler)

    def form_data(self, handler) -> dict[str, str]:
        length = int(handler.headers.get("Content-Length", "0"))
        raw = handler.rfile.read(length).decode()
        return {key: values[0] for key, values in parse_qs(raw).items()}

    def current_user(self, handler) -> dict[str, str | None] | None:
        cookie = SimpleCookie(handler.headers.get("Cookie"))
        morsel = cookie.get("snreg_session")
        if not morsel:
            return None
        value = unsign(morsel.value, self.config.secret)
        if not value:
            return None
        if self.db is None:
            # No user table here, so the claims we care about travel in the cookie itself. It is
            # signed, so the contents are tamper-evident; nothing secret goes in.
            try:
                return json.loads(base64.urlsafe_b64decode(value.encode()).decode())
            except Exception:
                return None
        return self.db.one(f"SELECT * FROM snreg_users WHERE id = {sql_literal(value)}")

    def require_user(self, handler) -> dict[str, str | None] | None:
        user = self.current_user(handler)
        if user:
            return user
        self.redirect(handler, f"/auth/start?next={urllib.parse.quote(urlparse(handler.path).path)}")
        return None

    def require_admin(self, handler) -> dict[str, str | None] | None:
        user = self.require_user(handler)
        if not user:
            return None
        if user.get("is_admin") == "t":
            return user
        self.render(handler, "Forbidden", "<p>You do not have access to the admin dashboard.</p>", HTTPStatus.FORBIDDEN)
        return None

    def static(self, handler, path: str) -> None:
        rel = path.removeprefix("/static/").lstrip("/")
        target = (STATIC / rel).resolve()
        if not str(target).startswith(str(STATIC.resolve())) or not target.exists():
            handler.send_error(404)
            return
        content_type = mimetypes.guess_type(str(target))[0] or "application/octet-stream"
        data = target.read_bytes()
        # Nothing here is fingerprinted, so a cached copy is a stale copy: no-cache makes the
        # browser ask every time, and the ETag lets it keep the bytes it already has.
        tag = '"%s"' % hashlib.sha1(data).hexdigest()[:16]
        if handler.headers.get("If-None-Match") == tag:
            handler.send_response(HTTPStatus.NOT_MODIFIED.value)
            handler.send_header("ETag", tag)
            handler.send_header("Cache-Control", "no-cache")
            handler.end_headers()
            return
        handler.send_response(200)
        handler.send_header("Content-Type", content_type)
        handler.send_header("Content-Length", str(len(data)))
        handler.send_header("ETag", tag)
        handler.send_header("Cache-Control", "no-cache")
        handler.end_headers()
        handler.wfile.write(data)

    def render(
        self,
        handler,
        title: str,
        content: str,
        status: HTTPStatus = HTTPStatus.OK,
        wide: bool = False,
        head: str = "",
        body_class: str = "",
        shell_class: str = "",
    ) -> None:
        template = Template((TEMPLATES / "base.html").read_text())
        user = self.current_user(handler)
        html_out = template.substitute(
            title=esc(title),
            content=content,
            shell_class=shell_class or ("shell shell-wide" if wide else "shell"),
            nav=self.nav(user),
            head=head,
            body_class=body_class,
        ).encode()
        handler.send_response(status.value)
        handler.send_header("Content-Type", "text/html; charset=utf-8")
        handler.send_header("Content-Length", str(len(html_out)))
        handler.end_headers()
        handler.wfile.write(html_out)

    def nav(self, user: dict[str, str | None] | None) -> str:
        if not user:
            # Sign in here when this deployment can run the flow itself; otherwise hand off to the
            # deployment that owns identity.
            if self.oidc_ready:
                start, nxt = "/auth/start", "/" if self.config.role == "home" else "/profile"
            else:
                start, nxt = self.off_box("/auth/start"), "/profile"
            return f'<a class="nav-button" href="{start}?next={nxt}">Sign in</a>'
        # off_box leaves these relative on a full deployment and absolute on a home one, so the
        # same markup serves both.
        admin = (
            f'<a class="nav-button" href="{self.off_box("/admin")}">Admin</a>'
            if user.get("is_admin") == "t"
            else ""
        )
        account = f'<a class="nav-button" href="{self.off_box("/profile")}">My Account</a>'
        return f'<a class="nav-button" href="/">Home</a>{account}{admin}<form method="post" action="/logout"><button class="nav-button">Sign out</button></form>'

    def redirect(self, handler, location: str) -> None:
        handler.send_response(303)
        handler.send_header("Location", location)
        handler.end_headers()

    def set_session(self, handler, user_id: str) -> None:
        cookie = f"snreg_session={sign(user_id, self.config.secret)}; Path=/; HttpOnly; SameSite=Lax"
        if self.config.cookie_secure:
            cookie += "; Secure"
        handler.send_header("Set-Cookie", cookie)

    def set_home_session(self, handler, claims: dict) -> None:
        """Session for a deployment with no user table, built straight from the OIDC claims."""
        groups = groups_from_claims(claims)
        email = str(claims.get("email") or "").lower()
        payload = {
            "username": str(claims.get("preferred_username") or email.split("@", 1)[0] or ""),
            "full_name": str(claims.get("name") or ""),
            "email": email,
            # Same rule the full deployment applies in upsert_oidc_user.
            "is_admin": "t"
            if email in self.config.app_admin_emails or groups & self.config.admin_groups
            else "f",
        }
        encoded = base64.urlsafe_b64encode(json.dumps(payload).encode()).decode()
        self.set_session(handler, encoded)

    def off_box(self, path: str) -> str:
        """Absolute URL for a page this deployment does not serve; unchanged when it does.

        Only worth using where a relative link would not survive the forward in ``dispatch`` —
        form actions above all, since a POST cannot be redirected without losing its body.
        """
        return path if self.config.role != "home" else f"{self.config.auth_base_url}{path}"

    def trusted_domain(self, email: str) -> bool:
        domain = email.rsplit("@", 1)[-1].lower()
        return domain in self.config.trusted_domains

    def trusted_domains_page(self, handler) -> None:
        """Placeholder — explains the shortcut the registration fine print points at."""
        domains = sorted(self.config.trusted_domains)
        if domains:
            listing = (
                "<p>Right now these are trusted:</p><ul>"
                + "".join(f"<li><code>@{esc(d)}</code></li>" for d in domains)
                + "</ul>"
            )
        else:
            listing = "<p>No domains are trusted yet, so every request is read by a human.</p>"
        self.render(
            handler,
            "Trusted domains",
            "<h1>Trusted domains</h1>"
            "<p>Some email domains belong to people I already know &mdash; a workplace, or a "
            "shared project. An address at one of those is approved the moment you click the "
            "link in the verification email, with no wait for me to read the request.</p>"
            f"{listing}"
            "<p>Everything else still works exactly the same, it just waits for me. If you "
            "think your domain belongs on this list, ask.</p>"
            '<p><a class="primary" href="/">Back to the pantheon</a></p>',
        )

    def create_password_token(self, username: str, email: str) -> str:
        token = token_urlsafe()
        self.db.execute(
            f"""
            INSERT INTO snreg_password_tokens (token, username, email, expires_at)
            VALUES ({sql_literal(token)}, {sql_literal(username)}, {sql_literal(email)}, now() + interval '2 hours');
            """
        )
        return token

    def send_password_link(self, username: str, email: str) -> str:
        token = self.create_password_token(username, email)
        reset_url = f"{self.config.base_url}/password/reset?token={token}"
        send_password_reset_email(self.config, email, reset_url)
        return reset_url

    def auth_start(self, handler) -> None:
        if not self.config.oidc_client_id or not self.config.oidc_client_secret:
            return self.render(handler, "Sign in unavailable", "<p>Authentik OIDC is not configured for Charon yet.</p>", HTTPStatus.SERVICE_UNAVAILABLE)
        query = parse_qs(urlparse(handler.path).query)
        # A home deployment does not serve /profile, so land back on its own front page.
        default_next = "/" if self.config.role == "home" else "/profile"
        next_path = query.get("next", [default_next])[0]
        if not next_path.startswith("/") or next_path.startswith("//"):
            next_path = default_next
        state = sign(next_path, self.config.secret)
        self.redirect(handler, authorization_url(self.config, state, next_path))

    def auth_callback(self, handler) -> None:
        query = parse_qs(urlparse(handler.path).query)
        code = query.get("code", [""])[0]
        state = query.get("state", [""])[0]
        next_path = unsign(state, self.config.secret, max_age=600) or "/profile"
        if not code:
            return self.render(handler, "Sign in failed", "<p>Authentik did not return an authorization code.</p>", HTTPStatus.BAD_REQUEST)
        try:
            token = exchange_code(self.config, code)
            claims = userinfo(self.config, token["access_token"])
            user = None if self.db is None else self.upsert_oidc_user(claims)
        except Exception as exc:
            return self.render(handler, "Sign in failed", f"<p>{esc(exc)}</p>", HTTPStatus.BAD_GATEWAY)
        handler.send_response(303)
        handler.send_header("Location", next_path)
        if user is None:
            self.set_home_session(handler, claims)
        else:
            self.set_session(handler, user["id"] or "")
        handler.end_headers()

    def upsert_oidc_user(self, claims: dict) -> dict[str, str | None]:
        subject = str(claims.get("sub") or "")
        email = str(claims.get("email") or "").lower()
        username = str(claims.get("preferred_username") or email.split("@", 1)[0] or subject).lower()
        full_name = str(claims.get("name") or username)
        groups = groups_from_claims(claims)
        is_admin = bool(email in self.config.app_admin_emails or groups.intersection(self.config.admin_groups))
        group_text = ",".join(sorted(groups))
        existing = self.db.one(
            f"""
            SELECT * FROM snreg_users
            WHERE oidc_subject = {sql_literal(subject)}
               OR lower(email) = {sql_literal(email)}
               OR lower(username) = {sql_literal(username)}
            ORDER BY CASE WHEN oidc_subject = {sql_literal(subject)} THEN 0 ELSE 1 END
            LIMIT 1
            """
        )
        if existing:
            self.db.execute(
                f"""
                UPDATE snreg_users
                SET oidc_subject = {sql_literal(subject)},
                    email = {sql_literal(email)},
                    full_name = {sql_literal(full_name)},
                    status = CASE WHEN status IN ('pending_email', 'pending_approval', 'rejected') THEN status ELSE 'active' END,
                    is_admin = {sql_literal(is_admin)},
                    auth_groups = {sql_literal(group_text)},
                    updated_at = now()
                WHERE id = {sql_literal(existing['id'])};
                """
            )
            return self.db.one(f"SELECT * FROM snreg_users WHERE id = {sql_literal(existing['id'])}") or existing
        self.db.execute(
            f"""
            INSERT INTO snreg_users (username, email, full_name, password_hash, status, email_verified_at, trusted_domain, is_admin, oidc_subject, auth_groups)
            VALUES ({sql_literal(username)}, {sql_literal(email)}, {sql_literal(full_name)}, {sql_literal(hash_password(token_urlsafe()))}, 'active', now(), {sql_literal(self.trusted_domain(email))}, {sql_literal(is_admin)}, {sql_literal(subject)}, {sql_literal(group_text)});
            """
        )
        return self.db.one(f"SELECT * FROM snreg_users WHERE oidc_subject = {sql_literal(subject)}") or {}

    def registration_flyout(self, handler) -> str:
        if self.current_user(handler):
            return ""
        return """
  <details class="register-flyout">
    <summary>Create account</summary>
    <div class="register-card">
      <div class="identity">
        <h1>Create your Sticknife account</h1>
        <p>Use a verified email address to request access. You will set your password after verification.</p>
      </div>
      <form method="post" action="/register" class="stack">
        <label>Username <input name="username" autocomplete="username" required></label>
        <label>Full name <input name="full_name" autocomplete="name" required></label>
        <label>Email <input name="email" type="email" autocomplete="email" required></label>
        <button class="primary" type="submit">Create account</button>
      </form>
      <p class="fineprint">Trusted email domains are auto-approved after verification. Other requests go to admin review.</p>
    </div>
  </details>
"""

    def registration_card(self, handler) -> str:
        if self.current_user(handler):
            return (
                '<h2>You already have an account</h2>'
                '<p class="sn-fineprint">It opens everything below. '
                f'<a href="{self.off_box("/profile")}">Your account</a>.</p>'
            )
        return f"""
      <h2>Ask for an account</h2>
      <form method="post" action="{self.off_box("/register")}" class="stack">
        <label>Username <input name="username" autocomplete="username" required></label>
        <label>Full name <input name="full_name" autocomplete="name" required></label>
        <label>Email <input name="email" type="email" autocomplete="email" required></label>
        <button type="submit">Send request</button>
      </form>
      <ul class="sn-fineprint">
        <li>Registration requires validation* by the administrator, so please be patient or
          send me a message if you sign up.</li>
        <li>I don't collect any personal information except name and email, and I don't use
          any analytics or tracking.</li>
        <li>*emails at a <a href="/trusted-domains">trusted domain</a> automatically
          validate</li>
      </ul>
"""

    def register_form(self, handler, message: str = "") -> None:
        content = Template((TEMPLATES / "landing.html").read_text()).substitute(
            message=message,
            registration_form=self.registration_card(handler),
            payload=pantheon.payload(),
            fallback=pantheon.fallback(),
        )
        self.render(
            handler,
            "The Pantheon",     # base.html appends " · Sticknife"
            content,
            head='<link rel="stylesheet" href="/static/css/pantheon.css">',
            body_class="landing",
            shell_class="landing-shell",
        )

    def register_submit(self, handler) -> None:
        data = self.form_data(handler)
        username = data.get("username", "").strip().lower()
        email = data.get("email", "").strip().lower()
        full_name = data.get("full_name", "").strip()
        if not username or not email or not full_name or "@" not in email:
            return self.register_form(handler, '<div class="notice error">Fill every field and use a valid email address.</div>')
        placeholder_password_hash = hash_password(token_urlsafe())
        trusted = self.trusted_domain(email)
        token = token_urlsafe()
        try:
            self.db.execute(
                f"""
                WITH inserted AS (
                    INSERT INTO snreg_users (username, email, full_name, password_hash, trusted_domain, is_admin)
                    VALUES ({sql_literal(username)}, {sql_literal(email)}, {sql_literal(full_name)}, {sql_literal(placeholder_password_hash)}, {sql_literal(trusted)}, {sql_literal(email in self.config.app_admin_emails)})
                    RETURNING id
                )
                INSERT INTO snreg_email_tokens (token, user_id, expires_at)
                SELECT {sql_literal(token)}, id, now() + interval '24 hours' FROM inserted;
                """
            )
        except Exception:
            return self.register_form(handler, '<div class="notice error">That username or email is already registered.</div>')
        verify_url = f"{self.config.base_url}/verify?token={token}"
        try:
            send_verification_email(self.config, email, verify_url)
            message = '<div class="notice">Check your email for a verification link.</div>'
        except Exception as exc:
            message = f'<div class="notice error">Registration saved, but email could not be sent: {esc(exc)}</div>'
        self.register_form(handler, message)

    def verify_email(self, handler) -> None:
        token = parse_qs(urlparse(handler.path).query).get("token", [""])[0]
        row = self.db.one(
            f"""
            SELECT u.* FROM snreg_email_tokens t
            JOIN snreg_users u ON u.id = t.user_id
            WHERE t.token = {sql_literal(token)}
              AND t.used_at IS NULL
              AND t.expires_at > now()
            """
        )
        if not row:
            return self.render(handler, "Verification failed", '<p>The verification link is invalid or expired.</p>')

        trusted = row.get("trusted_domain") == "t"
        next_status = "approved" if trusted else "pending_approval"
        self.db.execute(
            f"""
            UPDATE snreg_email_tokens SET used_at = now() WHERE token = {sql_literal(token)};
            UPDATE snreg_users
            SET email_verified_at = now(), status = {sql_literal(next_status)}, updated_at = now()
            WHERE id = {sql_literal(row['id'])};
            """
        )

        fresh = self.db.one(f"SELECT * FROM snreg_users WHERE id = {sql_literal(row['id'])}")
        user = fresh or row
        result = provision_user(self.config, user, active=trusted)
        if not result.ok:
            failed_status = "ipa_error" if trusted else "pending_approval"
            self.db.execute(
                f"""
                UPDATE snreg_users
                SET status = {sql_literal(failed_status)},
                    ipa_message = {sql_literal(result.message)},
                    updated_at = now()
                WHERE id = {sql_literal(row['id'])};
                INSERT INTO snreg_audit_log (target_user_id, action, detail)
                VALUES ({sql_literal(row['id'])}, 'email_verify_provision', {sql_literal(result.message)});
                """
            )
            return self.render(handler, "Verified", '<p>Your email is verified, but your account could not be prepared automatically. An admin will review it.</p>')

        reset_url = ""
        if self.config.ipa_mode == "cli":
            reset_url = self.send_password_link(user["username"] or "", user["email"] or "")

        if trusted:
            sync_result = trigger_ldap_sync(self.config)
            self.db.execute(
                f"""
                INSERT INTO snreg_audit_log (target_user_id, action, detail)
                VALUES ({sql_literal(row['id'])}, 'auto_provision', {sql_literal(result.message)});
                INSERT INTO snreg_audit_log (target_user_id, action, detail)
                VALUES ({sql_literal(row['id'])}, 'authentik_sync', {sql_literal(sync_result.message)});
                DELETE FROM snreg_users WHERE id = {sql_literal(row['id'])};
                """
            )
            if reset_url:
                return self.redirect(handler, reset_url)
            return self.render(handler, "Verified", '<p>Your email is verified. Your account is ready for Sticknife services.</p><a class="primary" href="/auth/start?next=/profile">Continue</a>')

        self.db.execute(
            f"""
            UPDATE snreg_users
            SET status = 'pending_approval',
                ipa_provisioned_at = now(),
                ipa_message = {sql_literal(result.message)},
                updated_at = now()
            WHERE id = {sql_literal(row['id'])};
            INSERT INTO snreg_audit_log (target_user_id, action, detail)
            VALUES ({sql_literal(row['id'])}, 'pending_provision', {sql_literal(result.message)});
            """
        )
        if reset_url:
            return self.redirect(handler, reset_url)
        return self.render(handler, "Verified", '<p>Your email is verified. Your account exists, but an admin must approve it before it can access Sticknife services.</p>')

    def login_form(self, handler, message: str = "") -> None:
        content = Template((TEMPLATES / "login.html").read_text()).substitute(message=message)
        self.render(handler, "Sign in", content)

    def login_submit(self, handler) -> None:
        data = self.form_data(handler)
        identifier = data.get("identifier", "").strip().lower()
        password = data.get("password", "")
        user = self.db.one(
            f"SELECT * FROM snreg_users WHERE lower(username) = {sql_literal(identifier)} OR lower(email) = {sql_literal(identifier)}"
        )
        if not user or not verify_password(password, user.get("password_hash") or ""):
            return self.login_form(handler, '<div class="notice error">Invalid username or password.</div>')
        handler.send_response(303)
        handler.send_header("Location", "/profile")
        self.set_session(handler, user["id"] or "")
        handler.end_headers()

    def logout(self, handler) -> None:
        try:
            location = logout_url(self.config)
        except Exception:
            location = "/"
        handler.send_response(303)
        handler.send_header("Location", location)
        handler.send_header("Set-Cookie", "snreg_session=; Path=/; Max-Age=0; HttpOnly; SameSite=Lax")
        handler.end_headers()

    def profile(self, handler, message: str = "") -> None:
        user = self.require_user(handler)
        if not user:
            return
        content = Template((TEMPLATES / "profile.html").read_text()).substitute(
            message=message,
            username=esc(user["username"]),
            email=esc(user["email"]),
            full_name=esc(user["full_name"]),
            status=esc(user["status"]),
        )
        self.render(handler, "Profile", content, wide=True)

    def profile_update(self, handler) -> None:
        user = self.require_user(handler)
        if not user:
            return
        data = self.form_data(handler)
        full_name = data.get("full_name", "").strip()
        email = data.get("email", "").strip().lower()
        if not full_name or "@" not in email:
            return self.profile(handler, '<div class="notice error">Use a valid name and email.</div>')
        self.db.execute(
            f"""
            UPDATE snreg_users
            SET full_name = {sql_literal(full_name)}, email = {sql_literal(email)}, updated_at = now()
            WHERE id = {sql_literal(user['id'])};
            INSERT INTO snreg_audit_log (actor_user_id, target_user_id, action, detail)
            VALUES ({sql_literal(user['id'])}, {sql_literal(user['id'])}, 'profile_update', 'profile fields updated');
            """
        )
        self.profile(handler, '<div class="notice">Profile updated.</div>')

    def profile_password_update(self, handler) -> None:
        user = self.require_user(handler)
        if not user:
            return
        data = self.form_data(handler)
        old_password = data.get("old_password", "")
        new_password = data.get("new_password", "")
        confirm_password = data.get("confirm_password", "")
        if len(new_password) < 12 or new_password != confirm_password:
            return self.profile(handler, '<div class="notice error">Use matching new passwords of at least 12 characters.</div>')
        username = user["username"] or ""
        verified = verify_ipa_password(self.config, username, old_password)
        if not verified.ok:
            return self.profile(handler, '<div class="notice error">Old password did not match.</div>')
        result = set_password(self.config, username, new_password)
        if not result.ok:
            return self.profile(handler, f'<div class="notice error">Password could not be changed: {esc(result.message)}</div>')
        self.db.execute(
            f"""
            INSERT INTO snreg_audit_log (actor_user_id, target_user_id, action, detail)
            VALUES ({sql_literal(user['id'])}, {sql_literal(user['id'])}, 'profile_password_change', 'password changed');
            """
        )
        self.profile(handler, '<div class="notice">Password changed.</div>')

    def password_request_form(self, handler, message: str = "") -> None:
        content = Template((TEMPLATES / "password_request.html").read_text()).substitute(message=message)
        self.render(handler, "Password reset", content)

    def password_request_submit(self, handler) -> None:
        data = self.form_data(handler)
        username = data.get("username", "").strip().lower()
        email = data.get("email", "").strip().lower()
        detail = "password reset requested"
        if username and email and email_matches(self.config, username, email):
            try:
                self.send_password_link(username, email)
                detail = "password reset email sent"
            except Exception as exc:
                detail = f"password reset email failed: {exc}"
        self.db.execute(
            f"""
            INSERT INTO snreg_audit_log (action, detail)
            VALUES ('password_request', {sql_literal(username + ' ' + detail)});
            """
        )
        message = '<div class="notice">If that account exists, a password reset link has been sent.</div>'
        self.password_request_form(handler, message)

    def password_reset_form(self, handler, message: str = "") -> None:
        token = parse_qs(urlparse(handler.path).query).get("token", [""])[0]
        row = self.db.one(
            f"""
            SELECT * FROM snreg_password_tokens
            WHERE token = {sql_literal(token)}
              AND used_at IS NULL
              AND expires_at > now()
            """
        )
        if not row:
            return self.render(handler, "Password reset", '<p>The password reset link is invalid or expired.</p><a class="primary" href="/password/request">Request another link</a>')
        content = Template((TEMPLATES / "password_reset.html").read_text()).substitute(
            message=message,
            username=esc(row["username"]),
            token=esc(token),
        )
        self.render(handler, "Set password", content)

    def password_reset_submit(self, handler) -> None:
        data = self.form_data(handler)
        token = data.get("token", "")
        password = data.get("password", "")
        confirm_password = data.get("confirm_password", "")
        row = self.db.one(
            f"""
            SELECT * FROM snreg_password_tokens
            WHERE token = {sql_literal(token)}
              AND used_at IS NULL
              AND expires_at > now()
            """
        )
        if not row:
            return self.render(handler, "Password reset", '<p>The password reset link is invalid or expired.</p><a class="primary" href="/password/request">Request another link</a>')
        if len(password) < 12 or password != confirm_password:
            return self.password_reset_form(handler, '<div class="notice error">Use matching passwords of at least 12 characters.</div>')
        result = set_password(self.config, row["username"] or "", password)
        if not result.ok:
            return self.password_reset_form(handler, f'<div class="notice error">Password could not be set: {esc(result.message)}</div>')
        self.db.execute(
            f"""
            UPDATE snreg_password_tokens SET used_at = now() WHERE token = {sql_literal(token)};
            INSERT INTO snreg_audit_log (action, detail)
            VALUES ('password_reset', {sql_literal(result.message)});
            """
        )
        self.render(handler, "Password set", '<p>Your password is set. You can now sign in.</p><a class="primary" href="/auth/start?next=/profile">Sign in</a>')

    def admin(self, handler) -> None:
        user = self.require_admin(handler)
        if not user:
            return
        pending = self.db.rows("SELECT * FROM snreg_users WHERE status = 'pending_approval' ORDER BY created_at ASC")
        unverified = self.db.rows("SELECT * FROM snreg_users WHERE status = 'pending_email' ORDER BY created_at ASC")
        recent = self.db.rows("SELECT * FROM snreg_users ORDER BY created_at DESC LIMIT 25")
        pending_rows = "".join(self.user_row(row, actions=True) for row in pending) or '<tr><td colspan="5">No pending approvals.</td></tr>'
        unverified_rows = "".join(self.user_row(row, actions=False, delete_unverified=True) for row in unverified) or '<tr><td colspan="5">No unverified registrations.</td></tr>'
        recent_rows = "".join(self.user_row(row, actions=False) for row in recent)
        purge_disabled = "disabled" if not unverified else ""
        content = Template((TEMPLATES / "admin.html").read_text()).substitute(
            pending_rows=pending_rows,
            unverified_rows=unverified_rows,
            purge_disabled=purge_disabled,
            recent_rows=recent_rows,
        )
        self.render(handler, "Admin", content, wide=True)

    def user_row(self, row: dict[str, str | None], actions: bool, delete_unverified: bool = False) -> str:
        controls = ""
        if actions:
            controls = f"""
            <form method="post" action="/admin/approve"><input type="hidden" name="user_id" value="{esc(row['id'])}"><button class="small">Approve</button></form>
            <form method="post" action="/admin/reject"><input type="hidden" name="user_id" value="{esc(row['id'])}"><button class="small danger">Reject</button></form>
            """
        elif delete_unverified:
            controls = f"""
            <form method="post" action="/admin/delete-unverified"><input type="hidden" name="user_id" value="{esc(row['id'])}"><button class="small danger">Delete</button></form>
            """
        return f"""
        <tr>
            <td>{esc(row['username'])}</td>
            <td>{esc(row['email'])}</td>
            <td>{esc(row['full_name'])}</td>
            <td><span class="pill">{esc(row['status'])}</span></td>
            <td class="actions">{controls}</td>
        </tr>
        """

    def admin_approve(self, handler) -> None:
        admin = self.require_admin(handler)
        if not admin:
            return
        user_id = self.form_data(handler).get("user_id", "")
        user = self.db.one(f"SELECT * FROM snreg_users WHERE id = {sql_literal(user_id)}")
        if user:
            already_provisioned = bool(user.get("ipa_provisioned_at"))
            result = approve_user(self.config, user["username"] or "") if already_provisioned else provision_user(self.config, user, active=True)
            if result.ok and self.config.ipa_mode == "cli":
                if not already_provisioned:
                    self.send_password_link(user["username"] or "", user["email"] or "")
                sync_result = trigger_ldap_sync(self.config)
                self.db.execute(
                    f"""
                    INSERT INTO snreg_audit_log (actor_user_id, target_user_id, action, detail)
                    VALUES ({sql_literal(admin['id'])}, {sql_literal(user_id)}, 'admin_approve', {sql_literal(result.message)});
                    INSERT INTO snreg_audit_log (actor_user_id, target_user_id, action, detail)
                    VALUES ({sql_literal(admin['id'])}, {sql_literal(user_id)}, 'authentik_sync', {sql_literal(sync_result.message)});
                    DELETE FROM snreg_users WHERE id = {sql_literal(user_id)};
                    """
                )
            else:
                status = "active" if result.ok else "ipa_error"
                self.db.execute(
                    f"""
                    UPDATE snreg_users
                    SET status = {sql_literal(status)},
                        ipa_provisioned_at = CASE WHEN {sql_literal(result.ok)} THEN COALESCE(ipa_provisioned_at, now()) ELSE ipa_provisioned_at END,
                        ipa_message = {sql_literal(result.message)},
                        updated_at = now()
                    WHERE id = {sql_literal(user_id)};
                    INSERT INTO snreg_audit_log (actor_user_id, target_user_id, action, detail)
                    VALUES ({sql_literal(admin['id'])}, {sql_literal(user_id)}, 'admin_approve', {sql_literal(result.message)});
                    """
                )
        self.redirect(handler, "/admin")

    def admin_reject(self, handler) -> None:
        admin = self.require_admin(handler)
        if not admin:
            return
        user_id = self.form_data(handler).get("user_id", "")
        self.db.execute(
            f"""
            UPDATE snreg_users SET status = 'rejected', updated_at = now() WHERE id = {sql_literal(user_id)};
            INSERT INTO snreg_audit_log (actor_user_id, target_user_id, action, detail)
            VALUES ({sql_literal(admin['id'])}, {sql_literal(user_id)}, 'admin_reject', 'registration rejected');
            """
        )
        self.redirect(handler, "/admin")

    def admin_delete_unverified(self, handler) -> None:
        admin = self.require_admin(handler)
        if not admin:
            return
        user_id = self.form_data(handler).get("user_id", "")
        self.db.execute(
            f"""
            WITH deleted AS (
                DELETE FROM snreg_users
                WHERE id = {sql_literal(user_id)} AND status = 'pending_email'
                RETURNING id
            )
            INSERT INTO snreg_audit_log (actor_user_id, action, detail)
            SELECT {sql_literal(admin['id'])}, 'admin_delete_unverified', 'deleted pending_email user id ' || id
            FROM deleted;
            """
        )
        self.redirect(handler, "/admin")

    def admin_purge_unverified(self, handler) -> None:
        admin = self.require_admin(handler)
        if not admin:
            return
        self.db.execute(
            f"""
            WITH deleted AS (
                DELETE FROM snreg_users
                WHERE status = 'pending_email'
                RETURNING id
            ), count_deleted AS (
                SELECT count(*) AS total FROM deleted
            )
            INSERT INTO snreg_audit_log (actor_user_id, action, detail)
            SELECT {sql_literal(admin['id'])}, 'admin_purge_unverified', 'purged ' || total || ' pending_email users'
            FROM count_deleted;
            """
        )
        self.redirect(handler, "/admin")