The auth service: public + admin aiohttp apps over unix sockets.

699 lines of Python Raw file Open on GitHub

#!/usr/bin/env python3.14
# -*- coding: utf-8 -*-
"""The auth service: public + admin aiohttp apps over unix sockets.

Two listeners, one process, one owner of all auth state:

- **Public** (`/run/euler/auth.sock`, group `euler-web`) — reached through
  Caddy: SRP login (challenge/verify), session resume/logout, and the
  `forward_auth` endpoint (`200 + X-User + X-Profile` or `401`). The
  shell-ticket endpoints also live here but are **not routed by Caddy**
  — only socket peers (the ws service, the PTY shells) can reach them.
- **Admin** (`/run/euler-adm/auth-admin.sock`, `0600` euler-auth-private)
  — the local admin plane, never routed through Caddy and **wheel-gated**:
  only root (the operator, via sudo) can connect, and `X-Admin-Token` (kept
  solely in root-readable `/etc/euler/auth.env`) is a second check. `users`
  add / list / enable / disable / remove.

Access logging is disabled on both listeners: registration/reset URLs carry
link tokens in the query string and must never reach the journal. App-level
events are logged without secrets instead.
"""
from __future__ import annotations

__all__ = ['AuthService', 'build_public_app', 'build_admin_app']

import asyncio
import hmac
import logging
import secrets
import time
from datetime import datetime, timezone
from pathlib import Path
from typing import Any

import aiohttp
import aiohttp_jinja2
import jinja2
from aiohttp import web

from solver.auth import Authorizations, Subject
from solver.auth.identity import system_slug
from euler_site.auth import policy
from euler_site.auth.returnto import return_path
from euler_site import assets
from euler_site.site import content
from euler_site.cache import cache_middleware
from euler_site.csp import csp_middleware
from euler_site.auth.config import AuthConfig
from euler_site.auth.mail import Mailer
from euler_site.auth.pending import PendingStore
from euler_site.auth.pow import ProofOfWork
from euler_site.auth.ratelimit import RateLimiter
from euler_site.auth.requests import RequestStore
from euler_site.auth.remember import RememberStore, load_or_create_secret
from euler_site.auth.sessions import SessionStore
from euler_site.auth.srp import KDF, SrpServer, SrpToken, decoy_token
from euler_site.auth.tickets import TicketStore
from euler_site.auth.users import UserStore, normalize_email

log = logging.getLogger('euler-auth')

_GENERIC_401 = 'authentication failed'


def _client_key(request: web.Request) -> str:
    """The rate-limit key: the first X-Forwarded-For hop (Caddy sets it), else 'local'."""
    forwarded = request.headers.get('X-Forwarded-For', '')
    return forwarded.split(',')[0].strip() or 'local'


def submitted_kdf(value: object) -> str | None:
    """The KDF a browser says it derived a new verifier with — the current :data:`KDF` — or None.

    A new verifier is always made at the current cost; anything else (a missing field from a page
    older than it, or other parameters) is refused rather than stored.
    """
    return KDF if value == KDF else None


async def _json_body(request: web.Request) -> dict[str, Any]:
    """The request's JSON object body; an empty dict for anything malformed."""
    try:
        data = await request.json()
    except Exception:
        return {}
    return data if isinstance(data, dict) else {}


class AuthService:
    """All auth state and behaviour; the route handlers are thin wrappers over this."""

    def __init__(self, config: AuthConfig) -> None:
        self.config = config
        state = config.state_dir
        self.users = UserStore(state / 'users.json')
        self.pending = PendingStore(state / 'pending.json')
        self.requests = RequestStore(state / 'requests.json',
                                     load_or_create_secret(state / 'requests-secret'))
        self.remember = RememberStore(state / 'remember.json',
                                      load_or_create_secret(state / 'session-secret'))
        self.sessions = SessionStore(policy.SESSION_TTL_SECONDS)
        #: The work a browser does before `/forgot` or `/request-invite` is accepted.
        self.pow = ProofOfWork(load_or_create_secret(state / 'pow-secret'),
                               policy.POW_BITS, policy.POW_TTL_SECONDS)
        self.tickets = TicketStore()
        self.mailer = Mailer(config.smtp_relay, config.base_url)
        self.rate = RateLimiter(policy.AUTH_RATE_MAX, policy.AUTH_RATE_WINDOW_SECONDS)
        #: The budget for mail an unauthenticated request can cause (:meth:`mail_allowed`).
        self._mail_per_recipient = RateLimiter(policy.MAIL_PER_RECIPIENT_MAX,
                                               policy.MAIL_PER_RECIPIENT_WINDOW_SECONDS)
        self._mail_daily = RateLimiter(policy.MAIL_DAILY_MAX, 86400)
        #: Mail sends in flight in the background (:meth:`send_later`), held so none is
        #: garbage-collected half-way.
        self._mail_tasks: set[asyncio.Task[None]] = set()
        #: In-flight SRP handshakes: challenge id → (email, server, the verifier it was built
        #: from, expiry). Keyed by a random id the browser carries back, not by the e-mail, so
        #: anyone starting a handshake for an address cannot cancel its owner's. The verifier
        #: is kept so a handshake begun before the credentials changed (a reset, a password
        #: change) cannot finish after it.
        self._challenges: dict[str, tuple[str, SrpServer, SrpToken, float]] = {}
        #: Secret for stable decoy salts (unknown-email challenges). Kept in the state dir:
        #: a real account's salt survives a restart, so a decoy's must too, or comparing the
        #: salt before and after one would tell the two apart.
        self._decoy_secret = load_or_create_secret(state / 'decoy-secret')

    # ── sessions & cookies ────────────────────────────────────────────────────────

    def session_identity(self, request: web.Request) -> tuple[str, str] | None:
        """`(email, profile)` for the request's session cookie, or None."""
        return self.sessions.get(request.cookies.get(policy.SESSION_COOKIE))

    @staticmethod
    def _set_cookie(response: web.Response, name: str, value: str, max_age: int) -> None:
        response.set_cookie(name, value, max_age=max_age, path='/',
                            secure=True, httponly=True, samesite='Lax')

    def open_session(self, response: web.Response, email: str, profile: str,
                     *, remember: bool) -> None:
        """Create the session (and optionally a remember-me token) as cookies on *response*."""
        token = self.sessions.create(email, profile)
        self._set_cookie(response, policy.SESSION_COOKIE, token, policy.SESSION_TTL_SECONDS)
        if remember:
            cookie = self.remember.issue(email)
            self._set_cookie(response, policy.REMEMBER_COOKIE, cookie, policy.REMEMBER_TTL_SECONDS)

    # ── SRP handshake ─────────────────────────────────────────────────────────────

    def start_challenge(self, email: str) -> tuple[str, str, str, str]:
        """Begin a handshake for *email*; return `(salt_hex, B_hex, challenge_id, kdf)`.

        Unknown or disabled accounts get a stable **decoy** challenge (same shape, an id
        included, the current KDF named, never stored), so the response does not reveal whether
        the account exists.
        """
        key = normalize_email(email)
        now = time.time()
        self._challenges = {k: v for k, v in self._challenges.items() if v[3] > now}
        cid = secrets.token_urlsafe(16)
        user = self.users.get(key)
        if user is None or user.disabled or not user.verifier:
            decoy = decoy_token(key, self._decoy_secret)
            server = SrpServer(key, decoy)
            return server.salt.hex(), f'{server.public:x}', cid, decoy.kdf
        token = user.srp_token
        server = SrpServer(key, token)
        while len(self._challenges) >= policy.MAX_PENDING_CHALLENGES:   # insertion order: oldest first
            del self._challenges[next(iter(self._challenges))]
        self._challenges[cid] = (key, server, token, now + policy.CHALLENGE_TTL_SECONDS)
        return server.salt.hex(), f'{server.public:x}', cid, token.kdf

    def finish_challenge(self, email: str, client_public_hex: str, client_proof_hex: str,
                         challenge_id: str) -> tuple[str, str] | None:
        """Verify the client proof; return `(M2_hex, profile)` or None (generic failure)."""
        key = normalize_email(email)
        entry = self._challenges.pop(challenge_id, None)
        if entry is None or entry[0] != key or entry[3] <= time.time():
            return None
        _email, server, token, _expiry = entry
        try:
            proof = server.verify(int(client_public_hex, 16), bytes.fromhex(client_proof_hex))
        except ValueError:
            return None
        user = self.users.get(key)
        # Re-check: state may have moved under us. A handshake proves the password the
        # verifier held when it began — if a reset or a change has replaced that verifier
        # since, the old password must not open a session now.
        if user is None or user.disabled or user.srp_token != token:
            return None
        return proof.hex(), self.profile_for(key)

    def forget_challenge(self, email: str) -> None:
        """Drop *email*'s in-flight handshakes (after its credentials change)."""
        key = normalize_email(email)
        self._challenges = {k: v for k, v in self._challenges.items() if v[0] != key}

    def profile_for(self, email: str) -> str:
        """The web profile for *email* from `authorizations.json`, defaulting to
        `reader` when unmapped. **Not capped** — in the per-user model an `admin`
        account is web-reachable, contained by its own uid + SRP, not the channel.

        Loaded **fresh** so a `users change` takes effect on the next login (sessions
        bake the profile in at login, and a change revokes them — the staleness rule).
        """
        return Authorizations.load().profile_for(email) or 'reader'

    # ── account lifecycle (admin plane) ───────────────────────────────────────────

    def invite(self, email: str, profile: str) -> str:
        """Mint a registration invite and email the link; return the invite URL."""
        token = self.pending.mint(email, profile, 'register')
        url = f'{self.config.base_url}/register?token={token}'
        self.mailer.send_invite(email, token, 'register')
        return url

    def mail_allowed(self, recipient: str) -> bool:
        """Spend one unit of the mail budget for *recipient*; False if it is exhausted.

        For mail an unauthenticated request triggers — reset links, OTPs, the invite-request
        notice. Per recipient so one address cannot be flooded, and per day for the whole
        service so the relay account's own quota is never reached and every other mail with
        it. A refusal is logged and otherwise silent: callers answer exactly as if it sent.
        """
        key = normalize_email(recipient)
        if not self._mail_per_recipient.allow(key) or not self._mail_daily.allow('*'):
            log.warning('mail budget exhausted — not sending to %s', key)
            return False
        return True

    def send_later(self, send: Any, *args: Any, what: str) -> None:
        """Run a blocking mail *send* in the background; failures are logged, never raised.

        So a response can go out at once whether or not a mail was sent — the time it takes
        to answer must not say which.
        """
        async def run() -> None:
            try:
                await asyncio.to_thread(send, *args)
            except OSError as exc:
                log.warning('%s failed: %s', what, exc)

        task = asyncio.get_running_loop().create_task(run())
        self._mail_tasks.add(task)
        task.add_done_callback(self._mail_tasks.discard)

    def notify_invite_request(self, name: str, email: str, remarks: str) -> None:
        """Best-effort: email the operator that an invite request landed. Never raises.

        The queue (`requests.json`) is the system of record; this is only a nudge.
        Skipped when no owner address is configured (`EULER_OWNER_EMAIL`). Runs the
        blocking submit under :func:`asyncio.to_thread` at the call site — never on the
        event loop.
        """
        owner = self.config.owner_email
        if not owner or not self.mail_allowed(owner):
            return
        try:
            self.mailer.send_invite_request(owner, name, email, remarks)
        except OSError as exc:
            log.warning('invite-request notice mail failed: %s', exc)

    def revoke_access(self, email: str) -> None:
        """Kill live access for *email* everywhere (disable/remove)."""
        self.sessions.revoke_email(email)
        self.remember.revoke_email(email)
        self.pending.revoke_email(email)

    async def push_shell_teardown(self, email: str) -> None:
        """Tear down *email*'s live web shell on their per-user instance.

        The auth service is the only party that sees a logout or a revocation, and
        a *running* shell baked its permissions in at startup — so a demoted or
        logged-out account would keep its old authority until the PTY happened to
        die. This closes that gap at the source: on those events we POST
        `/internal/logout` to **the user's own instance socket** (a single,
        deterministic `user-<slug>.sock` in the per-user model — no fan-out), which
        reaps that user's shell and its attached tabs.

        Best-effort: an absent socket (instance not running) or a refusal is skipped —
        the shell simply isn't there. Never raises into the caller (a teardown failure
        must not fail the logout / revocation itself). The endpoint is socket-peer only
        (Caddy never routes it), and `euler-auth` reaches it as a fellow `euler-web`
        member.
        """
        await self._push_user_instance(email, '/internal/logout')

    async def push_vault_reset(self, email: str) -> None:
        """Destroy *email*'s vault on their per-user instance after a password reset.

        An SRP reset re-mints the login verifier but shares nothing with the vault, so
        the old vault key is unrecoverable **by design** — the auth service (which never
        sees a password) could not preserve it if it wanted to, and that is exactly what
        keeps the vault operator-opaque. This push has the instance remove the stale
        blob and its ciphertext rather than leave secrets rotting under a dead key; the
        next sign-in initialises a fresh vault and the user re-provisions. A password
        *change* never comes here — the browser re-wraps and the vault survives.
        Best-effort and socket-peer only, exactly like :meth:`push_shell_teardown`.
        """
        await self._push_user_instance(email, '/internal/vault-reset')

    async def _push_user_instance(self, email: str, path: str) -> None:
        """POST to *email*'s own instance socket (`user-<slug>.sock`); never raises."""
        base = self.config.user_socket_dir
        if not base:
            return
        email = normalize_email(email)
        sock = Path(base) / f'user-{system_slug(email)}.sock'
        try:
            connector = aiohttp.UnixConnector(path=str(sock))
            timeout = aiohttp.ClientTimeout(total=3)
            async with aiohttp.ClientSession(connector=connector, timeout=timeout) as http:
                async with http.post(f'http://user{path}', json={'email': email}) as resp:
                    await resp.read()
        except (OSError, aiohttp.ClientError, asyncio.TimeoutError):
            pass                         # absent / down / slow — the instance isn't reachable


# ── public app (auth.sock — via Caddy, plus the socket-peer ticket endpoints) ──────

def build_public_app(service: AuthService) -> web.Application:
    """The public listener: login, forward_auth, session, and shell tickets."""

    async def healthz(_request: web.Request) -> web.Response:
        return web.Response(text='ok')

    async def check(request: web.Request) -> web.Response:
        """The Caddy `forward_auth` endpoint: 200 + identity headers, or 401.

        `X-User-Slug` (the e-mail's :func:`system_slug`) is what Caddy routes on in
        the per-user model — every request goes to that user's own instance socket
        `X-User`/`X-Profile` still ride along for the app tier.
        """
        identity = service.session_identity(request)
        if identity is None:
            return web.Response(status=401, text=_GENERIC_401)
        email, profile = identity
        return web.Response(status=200, headers={
            'X-User': email, 'X-Profile': profile, 'X-User-Slug': system_slug(email)})

    async def challenge(request: web.Request) -> web.Response:
        if not service.rate.allow(_client_key(request)):
            return web.Response(status=429, text='rate limited')
        body = await _json_body(request)
        email = str(body.get('email', ''))
        if '@' not in email:
            return web.Response(status=400, text='email required')
        salt_hex, public_hex, challenge_id, kdf = service.start_challenge(email)
        return web.json_response({'salt': salt_hex, 'B': public_hex, 'id': challenge_id, 'kdf': kdf})

    async def verify(request: web.Request) -> web.Response:
        if not service.rate.allow(_client_key(request)):
            return web.Response(status=429, text='rate limited')
        body = await _json_body(request)
        try:
            result = service.finish_challenge(str(body.get('email', '')), str(body.get('A', '')),
                                              str(body.get('M1', '')), str(body.get('id', '')))
        except ValueError:
            result = None
        if result is None:
            log.info('login failed for %s', normalize_email(str(body.get('email', '?'))))
            return web.Response(status=401, text=_GENERIC_401)
        server_proof_hex, profile = result
        email = normalize_email(str(body.get('email', '')))
        response = web.json_response({'M2': server_proof_hex, 'email': email, 'profile': profile})
        service.open_session(response, email, profile, remember=bool(body.get('remember', False)))
        log.info('login ok for %s (%s)', email, profile)
        return response

    async def resume(request: web.Request) -> web.Response:
        """Redeem a remember-me cookie for a fresh session (rotating the token)."""
        if not service.rate.allow(_client_key(request)):
            return web.Response(status=429, text='rate limited')
        cookie = request.cookies.get(policy.REMEMBER_COOKIE, '')
        redeemed = service.remember.redeem(cookie) if cookie else None
        user = service.users.get(redeemed[0]) if redeemed else None
        if redeemed is None or user is None or user.disabled:
            # A dead token is cleared with the session it could not renew: until both are
            # gone the edge keeps routing this browser to the login page instead of the
            # public site (frontend.sh, @stale_read).
            refused = web.Response(status=401, text=_GENERIC_401)
            refused.del_cookie(policy.SESSION_COOKIE, path='/')
            refused.del_cookie(policy.REMEMBER_COOKIE, path='/')
            return refused
        email, new_cookie = redeemed
        profile = service.profile_for(email)
        response = web.json_response({'email': email, 'profile': profile})
        token = service.sessions.create(email, profile)
        service._set_cookie(response, policy.SESSION_COOKIE, token, policy.SESSION_TTL_SECONDS)
        service._set_cookie(response, policy.REMEMBER_COOKIE, new_cookie, policy.REMEMBER_TTL_SECONDS)
        log.info('session resumed for %s', email)
        return response

    async def password_change(request: web.Request) -> web.Response:
        """Change the signed-in user's password (`POST /auth/password`).

        One atomic exchange, distinct from the unauthenticated forgot/reset
        flow: the browser proves the **current** password with an SRP handshake
        (`/auth/challenge` first, then `A`/`M1` here) and submits the new
        `{salt, verifier}` derived locally — no password ever reaches the
        server. The email comes from the live session, never the client. On
        success every *other* session and all remember-me tokens are revoked;
        the session that made the change stays signed in.
        """
        if not service.rate.allow(_client_key(request)):
            return web.Response(status=429, text='rate limited')
        identity = service.session_identity(request)
        if identity is None:
            return web.Response(status=401, text=_GENERIC_401)
        email = identity[0]
        body = await _json_body(request)
        try:
            result = service.finish_challenge(email, str(body.get('A', '')),
                                              str(body.get('M1', '')), str(body.get('id', '')))
        except ValueError:
            result = None
        if result is None:
            log.info('password change failed for %s (current password not proven)', email)
            return web.Response(status=401, text='current password incorrect')
        salt = str(body.get('salt', '')).lower()
        verifier = str(body.get('verifier', '')).lower()
        kdf = submitted_kdf(body.get('kdf'))
        try:                                    # sanity: well-formed hex of sane size
            if not (kdf and len(salt) == 32 and 0 < len(verifier) <= 512 and int(verifier, 16) > 0):
                raise ValueError
            bytes.fromhex(salt)
        except ValueError:
            return web.Response(status=400, text='malformed credentials')
        if not service.users.set_credentials(email, salt, verifier, kdf=kdf):
            return web.Response(status=401, text=_GENERIC_401)
        service.forget_challenge(email)
        current = request.cookies.get(policy.SESSION_COOKIE)
        service.sessions.revoke_email(email, keep=current)   # other devices out;
        service.remember.revoke_email(email)                 # persistent tokens die
        await service.push_shell_teardown(email)             # …and any live shell
        log.info('password changed for %s', email)
        return web.json_response({'M2': result[0]})

    async def logout(request: web.Request) -> web.Response:
        identity = service.session_identity(request)         # (email, profile) — before we drop it
        service.sessions.drop(request.cookies.get(policy.SESSION_COOKIE))
        remember_cookie = request.cookies.get(policy.REMEMBER_COOKIE)
        if remember_cookie:
            service.remember.revoke(remember_cookie)
        if identity is not None:
            await service.push_shell_teardown(identity[0])   # the user left — reap the shell
        # A browser form post (the site's user menu) lands back on the page it came from
        # (its `next` field — returnto), now served signed out by the public site; a page
        # only a session can see, or none, lands on home. 303 turns the POST into a GET.
        # A programmatic caller (no text/html Accept) keeps the JSON contract.
        if 'text/html' in request.headers.get('Accept', ''):
            form = await request.post() if request.content_type == 'application/x-www-form-urlencoded' else {}
            location = return_path(str(form.get('next', '')), signed_in=False)
            # A plain 303, not web.HTTPSeeOther: a handler *returning* an HTTPException
            # is deprecated in aiohttp (it is meant to be raised), and this one must be
            # returned so the cookie deletions below land on it.
            response: web.Response = web.Response(status=303, headers={'Location': location})
        else:
            response = web.json_response({'ok': True})
        response.del_cookie(policy.SESSION_COOKIE, path='/')
        response.del_cookie(policy.REMEMBER_COOKIE, path='/')
        return response

    async def ticket_mint(request: web.Request) -> web.Response:
        """Mint a one-time shell ticket against the caller's live session.

        Not routed by Caddy: only socket peers (the ws service, forwarding the
        user's cookie) reach this. A shell user cannot mint — no cookie.
        """
        identity = service.session_identity(request)
        if identity is None:
            return web.Response(status=401, text=_GENERIC_401)
        email, profile = identity
        ticket = service.tickets.mint(email, profile)
        log.info('shell ticket minted for %s', email)
        return web.json_response({'ticket': ticket})

    async def ticket_redeem(request: web.Request) -> web.Response:
        """Consume a shell ticket, returning the authoritative identity."""
        body = await _json_body(request)
        redeemed = service.tickets.redeem(str(body.get('ticket', '')))
        if redeemed is None:
            return web.Response(status=401, text='invalid ticket')
        email, profile = redeemed
        log.info('shell ticket redeemed for %s', email)
        return web.json_response({'email': email, 'profile': profile})

    async def shell_context(request: web.Request) -> dict[str, Any]:
        """The app shell's own context, for every page this service renders.

        The auth pages are standalone public documents, but they are not a separate
        place: they render the content service's `base.html` / `_nav.html` /
        `_home.html`, so this service must supply what those templates read.

        - `subject` — the header's switch. `None` on the signed-out pages, which
          is exactly what renders the header and the start page's cards inert. It is
          **presentational only**: this service's own routes gate on
          `session_identity` directly, never on this. It is built here so that the
          two pages a *signed-in* visitor can still land on (`/terms`,
          `/password`, both deep-link fallbacks) show them their own header rather
          than an invitation to sign in.
        - `readme_html` — the start page's README, rendered from the packaged copy
          (cached), since this uid has no clone to read one from.
        """
        identity = service.session_identity(request)
        subject = None
        if identity is not None:
            email, profile = identity
            subject = Subject(user=email, slug=system_slug(email), channel='web',
                              auth_method='session-cookie', profile=profile)
        # Signed out, the content is still browsable: the public site serves it, so the start
        # page's tiles and the header's nav lead there rather than wearing padlocks.
        # `page_path` stays empty: an auth page is never the place a sign-in returns to.
        # `auth_tier`: the right pane here is the page's own dialogue, not the public site's
        # minimizable window, so the header's terminal chip and the footer's hidden window
        # (which act on that window) stay inert — the content tiers never set it.
        return {'subject': subject, 'browsable': True, 'page_path': '', 'auth_tier': True,
                'readme_html': content.readme_html()}

    app = web.Application(
        middlewares=[cache_middleware, csp_middleware, aiohttp_jinja2.context_processors_middleware])
    # Templates ship as package data: packages/site/euler_site/templates, autoescape on.
    # The content service's templates come second, and that ordering is the design:
    # this service renders the *shell* (base.html, _nav.html, _home.html, the crumbs
    # and actions partials) from site/templates, and overrides nothing — there is one
    # source of truth for the chrome, not a copy here that silently drifts from it.
    # Its own dir holds only what is genuinely auth's: the dialogues and auth_base.html.
    # request_processor puts the live request in every template's context — the terms
    # page's footer needs its own path (?back=) to send its reader back here.
    web_dir = Path(__file__).resolve().parent.parent
    env = aiohttp_jinja2.setup(
        app,
        loader=jinja2.ChoiceLoader([
            jinja2.FileSystemLoader(str(web_dir / 'templates')),
            jinja2.FileSystemLoader(str(web_dir / 'site' / 'templates')),
        ]),
        context_processors=[aiohttp_jinja2.request_processor, shell_context])
    assets.install(env)
    app.add_routes([
        web.get('/healthz', healthz),
        web.get('/auth/check', check),
        web.post('/auth/challenge', challenge),
        web.post('/auth/verify', verify),
        web.post('/auth/resume', resume),
        web.post('/auth/password', password_change),
        web.post('/auth/logout', logout),
        web.post('/shell-ticket', ticket_mint),
        web.post('/shell-ticket/redeem', ticket_redeem),
    ])
    # Deferred import: pages.py imports AuthService from this module.
    from euler_site.auth.pages import add_page_routes
    add_page_routes(app, service)
    return app


# ── admin app (auth-admin.sock — local only, never via Caddy) ──────────────────────

def build_admin_app(service: AuthService) -> web.Application:
    """The admin plane: account lifecycle behind the euler-adm socket + token."""

    @web.middleware
    async def require_token(request: web.Request,
                            handler: Any) -> web.StreamResponse:
        if request.path != '/healthz':
            token = request.headers.get('X-Admin-Token', '')
            if not hmac.compare_digest(token, service.config.admin_token):
                return web.Response(status=401, text='bad admin token')
        return await handler(request)  # type: ignore[no-any-return]

    async def healthz(_request: web.Request) -> web.Response:
        return web.Response(text='ok')

    async def list_users(_request: web.Request) -> web.Response:
        """The roster: every identity in `authorizations.json` — web emails
        **and** OS logins — with its profile, joined with SRP registration state for the
        web ones (and any registered web account not yet in the map)."""
        authz_users = Authorizations.load().all_users()          # identity → profile (web + local)
        srp = {rec.email: rec for rec in service.users.all()}    # registered web accounts
        roster: list[dict[str, Any]] = []
        for name, profile in sorted(authz_users.items()):
            scope = 'web' if '@' in name else 'local'
            if scope == 'local':
                state = 'os-login'
            elif name in srp:
                state = 'disabled' if srp[name].disabled else 'registered'
            else:
                state = 'invited'
            # The per-user unix name (uid/home/socket/branch) for a web account; a local
            # OS login has no per-user instance, so it carries none.
            slug = system_slug(name) if scope == 'web' else ''
            # No public key here: the registry a rotation re-issues against is the tracked
            # roster (`/etc/euler/roster/users.json`), read by anyone with no `sudo`. This plane holds
            # the login secrets, and holding a *second* copy of the public keys only created a
            # pair that could disagree.
            roster.append({'user': name, 'profile': profile, 'scope': scope, 'state': state,
                           'slug': slug})
        for email in sorted(srp):                                # registered but unmapped → default reader
            if email not in authz_users:
                roster.append({'user': email, 'profile': 'reader (unmapped)', 'scope': 'web',
                               'state': 'disabled' if srp[email].disabled else 'registered',
                               'slug': system_slug(email)})
        return web.json_response({
            'roster': roster,
            'pending': [record.summary() for record in service.pending.all()],
        })

    async def add_user(request: web.Request) -> web.Response:
        """Mint + mail an invite (the first registration step). No user record exists until completion."""
        body = await _json_body(request)
        email = normalize_email(str(body.get('email', '')))
        profile = str(body.get('profile', 'reader'))
        if '@' not in email:
            return web.Response(status=400, text='valid email required')
        if profile not in policy.PROFILES:
            return web.Response(status=400, text=f'profile must be one of {policy.PROFILES}')
        if service.users.get(email) is not None:
            return web.Response(status=409, text='user already registered')
        try:
            # In a thread: the mail submission is blocking smtplib, and the
            # upstream (relay → Gmail) can take many seconds — the event loop
            # must keep serving forward_auth meanwhile.
            url = await asyncio.to_thread(service.invite, email, profile)
        except OSError as exc:
            log.warning('invite mail to %s failed: %s', email, exc)
            return web.Response(status=502, text=f'invite mail failed: {exc}')
        log.info('invited %s (%s)', email, profile)
        return web.json_response({'email': email, 'profile': profile, 'invite_url': url,
                                  'expires': f'{policy.INVITE_TTL_SECONDS // 86400}d'}, status=201)

    async def set_enabled(request: web.Request) -> web.Response:
        email = normalize_email(request.match_info['email'])
        enable = request.match_info['action'] == 'enable'
        if not service.users.set_enabled(email, enable):
            return web.Response(status=404, text='no such user')
        if not enable:
            service.revoke_access(email)
            await service.push_shell_teardown(email)         # disable ends the shell now
        log.info('%s %s', 'enabled' if enable else 'disabled', email)
        return web.json_response({'email': email, 'disabled': not enable})

    async def remove_user(request: web.Request) -> web.Response:
        email = normalize_email(request.match_info['email'])
        existed = service.users.remove(email)
        revoked = service.pending.revoke_email(email)
        service.revoke_access(email)
        await service.push_shell_teardown(email)             # remove ends the shell now
        if not existed and not revoked:
            return web.Response(status=404, text='no such user or invite')
        log.info('removed %s', email)
        return web.json_response({'email': email, 'removed': True})

    async def revoke_sessions(request: web.Request) -> web.Response:
        """Drop a web account's live sessions + remember tokens so a profile change
        (written to authorizations.json by the sudo CLI) takes effect on next login —
        and tear down the live shell so the *running* PTY's baked-in permissions die
        with them, not at next login (this is the `users change` path)."""
        email = normalize_email(request.match_info['email'])
        n = service.sessions.revoke_email(email) + service.remember.revoke_email(email)
        await service.push_shell_teardown(email)
        log.info('revoked %d session/remember token(s) for %s', n, email)
        return web.json_response({'email': email, 'revoked': n})

    async def revoke_everyone(_request: web.Request) -> web.Response:
        """Drop every session and remember token (`logout-all`).

        No teardown push: `logout-all` has closed the shells by now, and a push would start
        every stopped instance, since dialling a socket is what activates one.
        """
        n = service.sessions.revoke_all() + service.remember.revoke_all()
        log.info('revoked %d session/remember token(s) for everyone', n)
        return web.json_response({'revoked': n})

    async def list_requests(_request: web.Request) -> web.Response:
        """The invite-request queue — prospective collaborators from the public form."""
        return web.json_response({'requests': [r.summary() for r in service.requests.all()]})

    async def dismiss_request(request: web.Request) -> web.Response:
        email = normalize_email(request.match_info['email'])
        if not service.requests.dismiss(email):
            return web.Response(status=404, text='no such request')
        log.info('dismissed invite request from %s', email)
        return web.json_response({'email': email, 'dismissed': True})

    app = web.Application(middlewares=[require_token])
    app.add_routes([
        web.get('/healthz', healthz),
        web.get('/admin/users', list_users),
        web.post('/admin/users', add_user),
        web.post('/admin/users/{email}/{action:enable|disable}', set_enabled),
        web.post('/admin/users/{email}/revoke', revoke_sessions),
        web.post('/admin/revoke-all', revoke_everyone),
        web.delete('/admin/users/{email}', remove_user),
        web.get('/admin/requests', list_requests),
        web.delete('/admin/requests/{email}', dismiss_request),
    ])
    return app


def utcnow_iso() -> str:
    """Current UTC time in ISO-8601 (shared by the registration/reset flow handlers)."""
    return datetime.now(timezone.utc).isoformat(timespec='seconds')