server.py

#!/usr/bin/env python3
"""
server.py — Keybound: one self-contained Python server.

A complete, dependency-light demo of hardware-bound sessions on YubiKey 5.8:

  * SIGN UP  — register a passkey (WebAuthn create; native touch + PIN).
  * SIGN IN  — authenticate with the passkey (WebAuthn get; native touch + PIN).
               A hardware-bound session is established as part of sign-in — not a
               separate "enable security" step.
  * REMEMBER ME — issues a longer-lived session token. Returning to the app with
               it still requires ONE keypress (a passkey assertion) to re-bind the
               session to hardware before anything sensitive is shown.
  * SENSITIVE ACTIONS & READS — every authenticated route (including reading your
               profile PII) requires a live rolling-proof binding. A stolen cookie
               alone reveals nothing and can do nothing — it is inert.

The rolling-proof binding uses a per-epoch key (bk) held only in the browser's
memory and the server's session state, never in the cookie. Sensitive requests
carry a cheap HMAC over the canonical request; the epoch is (re)established with
a hardware signature. This is the "silent token" path — invisible after sign-in.

Modes:
  default   — WebAuthn is real (browser prompts); the rolling-proof epoch is
              signed server-side with a software stand-in key so the whole demo
              runs from ONE server with no external agent. Real WebAuthn, real
              request binding; the epoch signer is software.
  --hardware — the epoch proof is signed by the YubiKey via previewSign/ARKG
              (needs arkg_enrollment.json + a local signing path).

Run:  python server.py            # open http://localhost:8000
"""

from __future__ import annotations

import argparse
import hashlib
import json
import os
import sys
import threading
import time
from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
from urllib.parse import urlparse

from cryptography.hazmat.primitives import hashes, serialization
from cryptography.hazmat.primitives.asymmetric import ec
from cryptography.hazmat.primitives.asymmetric.utils import Prehashed

import webauthn_rp as wa
from rolling_proof import EpochProof, RequestBinding, RollingProofServer, _proof_bytes
from verifier import COSE_ESP256

HOST, PORT = "127.0.0.1", 8000
RP_ID = "localhost"
ORIGIN = f"http://localhost:{PORT}"
EPOCH_LIFETIME_MS = 30_000
REMEMBER_MS = 7 * 24 * 3600 * 1000
# The SESSION cookie's own TTL — separate from, and previously accidentally
# tied to, EPOCH_LIFETIME_MS (it was EPOCH_LIFETIME_MS*4 = 2 minutes, which is
# uncomfortably short for a live demo/bench-test session and has nothing to
# do with the hardware-binding security property; that's what bk/the epoch
# lifetime enforces, independent of this).
SESSION_TTL_MS = 30 * 60 * 1000
HERE = os.path.dirname(os.path.abspath(__file__))
USERS_FILE = os.path.join(HERE, "users.json")


# ------------------------------------------------------------------- stores ----

# Synthetic PII generated once at account creation — the kind of record an
# infostealer actually wants (identity + payment data), not a bank balance.
# All fake: no real-world SSN/card ranges, purely demo filler.
_STREETS = ["Maple", "Birch", "Cedar", "9th", "Elm", "Sunset", "River", "Oak"]
_CITIES = [("Austin", "TX"), ("Denver", "CO"), ("Portland", "OR"),
           ("Raleigh", "NC"), ("Boise", "ID")]
_CARD_BRANDS = ["Visa", "Mastercard", "Amex"]


def _rand_digits(n: int) -> str:
    return "".join(str(b % 10) for b in os.urandom(n))


def _generate_fake_pii() -> dict:
    street_num = int(_rand_digits(3)) + 1
    street = _STREETS[os.urandom(1)[0] % len(_STREETS)]
    city, state = _CITIES[os.urandom(1)[0] % len(_CITIES)]
    year = 1965 + (os.urandom(1)[0] % 45)
    month = 1 + (os.urandom(1)[0] % 12)
    day = 1 + (os.urandom(1)[0] % 28)
    return {
        "ssn": f"{_rand_digits(3)}-{_rand_digits(2)}-{_rand_digits(4)}",
        "dob": f"{year:04d}-{month:02d}-{day:02d}",
        "phone": f"({_rand_digits(3)}) {_rand_digits(3)}-{_rand_digits(4)}",
        "address": f"{street_num} {street} St, {city}, {state} {_rand_digits(5)}",
        "payment_card": f"{_CARD_BRANDS[os.urandom(1)[0] % len(_CARD_BRANDS)]} "
                        f"•••• {_rand_digits(4)}",
    }


class User:
    def __init__(self, username):
        self.username = username
        self.uid = os.urandom(4).hex()
        self.credentials = {}   # credential_id hex -> {public_key_hex, alg, sign_count}
        self.profile = _generate_fake_pii()   # SSN, DOB, phone, address, card — the payload
        self.recovery_email = f"{username}@example.com"   # the writable, account-takeover-relevant field
        self.trusted_devices = []
        self.activity = []   # log of sensitive writes (email changes, devices added)

    def log_activity(self, action: str, detail: str) -> None:
        """Record a sensitive write. Surfaced in the UI as 'Recent activity' —
        an account-takeover attempt (a changed recovery email, an unfamiliar
        trusted device) is exactly the kind of thing a real user only ever
        catches by noticing it in a list like this."""
        self.activity.append({"action": action, "detail": detail,
                              "ts": int(time.time() * 1000)})
        del self.activity[:-20]   # keep the tail bounded; the UI shows the last few

    # Identity + WebAuthn credentials are the only things worth surviving a
    # server restart (no private key material ever exists server-side —
    # WebAuthn only gives us public keys). Sessions, rolling-proof epochs and
    # bk are deliberately NEVER persisted; they must die with the process.
    def to_dict(self) -> dict:
        return {"uid": self.uid, "credentials": self.credentials,
                "profile": self.profile, "recovery_email": self.recovery_email,
                "trusted_devices": self.trusted_devices, "activity": self.activity}

    @classmethod
    def from_dict(cls, username: str, d: dict) -> "User":
        u = cls(username)
        u.uid = d.get("uid", u.uid)
        u.credentials = d.get("credentials", {})
        u.profile = d.get("profile", u.profile)
        u.recovery_email = d.get("recovery_email", u.recovery_email)
        u.trusted_devices = d.get("trusted_devices", [])
        u.activity = d.get("activity", [])
        return u


class Session:
    def __init__(self, sid, username):
        self.sid = sid
        self.username = username
        self.remember = False
        self.expires_ms = 0
        self.epoch_id = 0
        self.bound = False            # is a live rolling-proof epoch active?
        self.pending_epoch = {}       # hardware route: epoch_id -> {pub,alg,start}


class App:
    def __init__(self, hardware=False, derive=None, users_file=None):
        # derive() -> (pub_raw, alg, ticket_bytes); injectable so the hardware
        # epoch ROUTE can be exercised with software crypto in tests.
        self.derive = derive
        self.hardware = bool(hardware or derive is not None)
        # Coarse per-request lock: ThreadingHTTPServer runs one thread per
        # connection, and app state (users/sessions/challenges/rolling-proof
        # epochs) is plain, unlocked dicts. Serializing request handling
        # avoids races (e.g. a double-clicked action) at the cost of
        # concurrency — the right trade for a demo server.
        self.lock = threading.RLock()
        # users_file is overridable so tests (which construct their own App)
        # never read/write the real demo's persisted users.json.
        self.users_file = users_file or USERS_FILE
        self.users: dict[str, User] = {}
        self._load_users()
        self.sessions: dict[bytes, Session] = {}
        self.challenges: dict[bytes, bytes] = {}   # sid -> pending WebAuthn challenge
        self.rolling = RollingProofServer()
        self._seed = None
        if hardware and derive is None:
            self._load_seed()
            self.derive = self._arkg_derive

    # ---- identity persistence (users.json: public credentials only — no
    # secrets ever exist server-side for WebAuthn to begin with) ----
    def _load_users(self) -> None:
        if not os.path.exists(self.users_file):
            return
        with open(self.users_file) as f:
            raw = json.load(f)
        self.users = {name: User.from_dict(name, d) for name, d in raw.items()}

    def _save_users(self) -> None:
        raw = {name: u.to_dict() for name, u in self.users.items()}
        tmp = self.users_file + ".tmp"
        with open(tmp, "w") as f:
            json.dump(raw, f, indent=2)
        os.replace(tmp, self.users_file)   # atomic on both POSIX and Windows

    # ---- WebAuthn challenges ----
    def new_challenge(self, sid: bytes) -> str:
        ch = os.urandom(32)
        self.challenges[sid] = ch
        return wa.b64url_encode(ch)

    # ---- sessions ----
    def new_session(self, username: str, remember: bool) -> Session:
        sid = os.urandom(16)
        s = Session(sid, username)
        s.remember = remember
        s.expires_ms = int(time.time() * 1000) + (REMEMBER_MS if remember else SESSION_TTL_MS)
        self.sessions[sid] = s
        return s

    # ---- epoch establishment (software stand-in signer, single-server mode) ----
    def establish_epoch_software(self, s: Session) -> dict:
        priv = ec.generate_private_key(ec.SECP256R1())
        pub_raw = priv.public_key().public_bytes(
            serialization.Encoding.X962, serialization.PublicFormat.UncompressedPoint)
        bk = os.urandom(32)
        s.epoch_id += 1
        start = int(time.time() * 1000)
        commit = hashlib.sha256(bk).digest()
        payload = _proof_bytes(s.sid, ORIGIN, s.epoch_id, start, EPOCH_LIFETIME_MS, commit)
        sig = priv.sign(hashlib.sha256(payload).digest(),
                        ec.ECDSA(Prehashed(hashes.SHA256())))
        proof = EpochProof(s.sid, ORIGIN, s.epoch_id, start, EPOCH_LIFETIME_MS, commit, sig)
        reg = self.rolling.register_epoch(proof, bk, pub_raw, COSE_ESP256, now_ms=start)
        if not reg.ok:
            return {"ok": False, "reason": reg.reason.value}
        s.bound = True
        return {"ok": True, "epoch_id": s.epoch_id, "bk": bk.hex(),
                "lifetime_ms": EPOCH_LIFETIME_MS, "session_id": s.sid.hex(),
                "origin": ORIGIN, "mode": "software"}

    # ---- hardware route (ARKG derive; client signs via previewSign) ----
    def _load_seed(self):
        path = os.path.join(HERE, "arkg_enrollment.json")
        if not os.path.exists(path):
            sys.exit("--hardware needs arkg_enrollment.json in this folder.\n"
                     "Enroll your YubiKey's ARKG seed first (enroll_arkg2.py), "
                     "then copy arkg_enrollment.json here. Or run without --hardware "
                     "for the software demo.")
        from fido2 import cbor
        from fido2.cose import CoseKey
        from fido2.utils import websafe_decode
        with open(path) as f:
            rec = json.load(f)
        self._seed = CoseKey.parse(cbor.decode(websafe_decode(rec["seedPublicKey"])))

    def _arkg_derive(self):
        from fido2 import cbor
        dk, sign_args = self._seed.derive_public_key(os.urandom(32), b"inert-cookie/derive/v1")
        return b"\x04" + dk[-2] + dk[-3], dk[3], cbor.encode(sign_args)

    def epoch_prepare(self, s: Session) -> dict:
        pub_raw, alg, ticket = self.derive()
        s.epoch_id += 1
        start = int(time.time() * 1000)
        s.pending_epoch[s.epoch_id] = {"pub": pub_raw, "alg": alg, "start": start}
        return {"ok": True, "epoch_id": s.epoch_id, "start_ms": start,
                "lifetime_ms": EPOCH_LIFETIME_MS, "session_id": s.sid.hex(),
                "origin": ORIGIN, "ticket": ticket.hex(), "mode": "hardware"}

    def epoch_commit(self, s: Session, body: dict) -> dict:
        eid = int(body["epoch_id"])
        pend = s.pending_epoch.pop(eid, None)
        if not pend:
            return {"ok": False, "reason": "no pending epoch"}
        bk = bytes.fromhex(body["bk"])
        proof = EpochProof(s.sid, ORIGIN, eid, pend["start"], EPOCH_LIFETIME_MS,
                           hashlib.sha256(bk).digest(), bytes.fromhex(body["signature"]))
        reg = self.rolling.register_epoch(proof, bk, pend["pub"], pend["alg"])
        if reg.ok:
            s.bound = True
        return {"ok": reg.ok} if reg.ok else {"ok": False, "reason": reg.reason.value}

    # ---- rolling-proof request verification (protects ALL sensitive routes) ----
    def verify_binding(self, s: Session, method: str, path: str, body: bytes, headers):
        if not s.bound:
            return False, "session not hardware-bound"
        try:
            binding = RequestBinding(
                epoch_id=int(headers.get("X-IC-Epoch", "0")),
                timestamp_ms=int(headers.get("X-IC-Timestamp", "0")),
                counter=int(headers.get("X-IC-Counter", "0")),
                mac=bytes.fromhex(headers.get("X-IC-MAC", "")))
        except (ValueError, TypeError):
            return False, "malformed binding"
        res = self.rolling.verify_request(s.sid, method, path, body, binding)
        return res.ok, res.reason.value


# ------------------------------------------------------------------ routing ----

def make_handler(app: App):
    class H(BaseHTTPRequestHandler):
        server_version = "Keybound/2.0"

        def log_message(self, *a):
            pass

        # -- session cookie --
        def _session(self):
            for part in self.headers.get("Cookie", "").split(";"):
                part = part.strip()
                if part.startswith("sid="):
                    try:
                        s = app.sessions.get(bytes.fromhex(part[4:]))
                    except ValueError:
                        return None
                    if s and s.expires_ms > int(time.time() * 1000):
                        return s
            return None

        def _body(self):
            n = int(self.headers.get("Content-Length", "0") or "0")
            return self.rfile.read(n) if n else b""

        def _json(self, code, obj, cookie=None):
            data = json.dumps(obj).encode()
            self.send_response(code)
            self.send_header("Content-Type", "application/json")
            self.send_header("Content-Length", str(len(data)))
            if cookie:
                self.send_header("Set-Cookie", cookie)
            self.end_headers()
            self.wfile.write(data)

        def _file(self, name, ctype):
            try:
                with open(os.path.join(HERE, "web", name), "rb") as f:
                    data = f.read()
            except FileNotFoundError:
                # canonical.js lives at project root
                try:
                    with open(os.path.join(HERE, name), "rb") as f:
                        data = f.read()
                except FileNotFoundError:
                    return self.send_error(404)
            self.send_response(200)
            self.send_header("Content-Type", ctype)
            self.send_header("Content-Length", str(len(data)))
            self.end_headers()
            self.wfile.write(data)

        # ---------------------------------------------------------- GET ----
        def do_GET(self):
            with app.lock:
                self._handle_get()

        def _handle_get(self):
            r = urlparse(self.path).path
            if r == "/":
                return self._file("index.html", "text/html")
            if r == "/app.js":
                return self._file("app.js", "application/javascript")
            if r == "/canonical.js":
                return self._file("canonical.js", "application/javascript")
            if r == "/api/whoami":
                s = self._session()
                # `hardware` is reported even when signed out: it's a deployment
                # mode, not a secret, and having it unauthenticated means you can
                # always answer "is this server actually in --hardware mode?" with
                # a plain curl instead of having to sign in first. Restarting the
                # server without the flag looks exactly like a code regression
                # from the UI, so make the ground truth cheap to check.
                if not s:
                    return self._json(200, {"authenticated": False,
                                            "hardware": app.hardware})
                # bound status tells the client whether it must re-arm with a keypress
                return self._json(200, {"authenticated": True, "username": s.username,
                                        "bound": s.bound, "remember": s.remember,
                                        "hardware": app.hardware})
            if r == "/api/profile":
                s = self._session()
                if not s:
                    return self._json(401, {"error": "no session"})
                ok, reason = app.verify_binding(s, "GET", "/api/profile", b"", self.headers)
                if not ok:
                    return self._json(401, {"error": "inert", "reason": reason})
                u = app.users[s.username]
                # This is the payload an infostealer actually wants: identity +
                # payment data, not a bank balance. It's exactly as exposed to a
                # stolen cookie as a balance would be — the point is what's behind
                # the binding, not what kind of app it is.
                return self._json(200, {"profile": u.profile,
                                        "recovery_email": u.recovery_email,
                                        "trusted_devices": u.trusted_devices,
                                        "activity": u.activity[-5:], "user": u.username})
            return self.send_error(404)

        # --------------------------------------------------------- POST ----
        def do_POST(self):
            with app.lock:
                self._handle_post()

        def _handle_post(self):
            r = urlparse(self.path).path
            body = self._body()

            # ---- passkey registration (sign up) ----
            if r == "/auth/register/begin":
                data = json.loads(body or b"{}")
                username = (data.get("username") or "").strip()
                if not username:
                    return self._json(400, {"error": "username required"})
                # temp session to carry the challenge
                existing = app.users.get(username)
                if existing and existing.credentials:
                    return self._json(400, {"error": "username already registered"})
                s = app.new_session(username, remember=False)
                challenge = app.new_challenge(s.sid)
                app.users.setdefault(username, User(username))
                return self._json(200, {
                    "rp": {"id": RP_ID, "name": "Keybound"},
                    "user": {"id": wa.b64url_encode(username.encode()),
                             "name": username, "displayName": username},
                    "challenge": challenge,
                    "pubKeyCredParams": [{"type": "public-key", "alg": -7},
                                         {"type": "public-key", "alg": -8}],
                    "authenticatorSelection": {"userVerification": "required",
                                               "residentKey": "preferred",
                                               "authenticatorAttachment": "cross-platform"},
                    "attestation": "none",
                }, cookie=f"sid={s.sid.hex()}; Path=/; SameSite=Strict")

            if r == "/auth/register/finish":
                s = self._session()
                if not s:
                    return self._json(401, {"error": "no session"})
                data = json.loads(body)
                try:
                    reg = wa.verify_registration(
                        attestation_object_b64=data["attestationObject"],
                        client_data_json_b64=data["clientDataJSON"],
                        rp_id=RP_ID, expected_challenge=app.challenges.pop(s.sid, b""),
                        expected_origin=ORIGIN)
                except (ValueError, KeyError) as e:
                    return self._json(400, {"error": f"registration failed: {e}"})
                u = app.users[s.username]
                u.credentials[reg["credential_id"].hex()] = {
                    "public_key_hex": reg["public_key_hex"], "alg": reg["alg"],
                    "sign_count": reg["sign_count"]}
                app._save_users()
                # Auto-sign-in: the registration ceremony just verified above
                # already proved a live touch (verify_registration enforces
                # the UP flag) — making the user immediately repeat a whole
                # separate WebAuthn assertion just to sign in would be
                # redundant re-authentication, not additional security. Bind
                # the session right here, exactly like login/finish does.
                out = {"ok": True, "username": s.username,
                       "credential_id": reg["credential_id"].hex(), "hardware": app.hardware}
                if not app.hardware:
                    out["epoch"] = app.establish_epoch_software(s)
                return self._json(200, out)

            # ---- passkey authentication (sign in / re-arm) ----
            if r == "/auth/login/begin":
                data = json.loads(body or b"{}")
                username = (data.get("username") or "").strip()
                u = app.users.get(username)
                if not u or not u.credentials:
                    return self._json(400, {"error": "unknown user or no passkey"})
                s = app.new_session(username, remember=bool(data.get("remember")))
                challenge = app.new_challenge(s.sid)
                return self._json(200, {
                    "rpId": RP_ID, "challenge": challenge,
                    "userVerification": "required",
                    "allowCredentials": [{"type": "public-key", "id": cid}
                                         for cid in u.credentials],
                }, cookie=f"sid={s.sid.hex()}; Path=/; SameSite=Strict")

            if r == "/auth/login/finish":
                s = self._session()
                if not s:
                    return self._json(401, {"error": "no session"})
                data = json.loads(body)
                u = app.users.get(s.username)
                cred = u.credentials.get(data["credentialId"]) if u else None
                if not cred:
                    return self._json(400, {"error": "unknown credential"})
                try:
                    res = wa.verify_authentication(
                        authenticator_data_b64=data["authenticatorData"],
                        client_data_json_b64=data["clientDataJSON"],
                        signature_b64=data["signature"],
                        public_key_hex=cred["public_key_hex"], alg=cred["alg"],
                        rp_id=RP_ID, expected_challenge=app.challenges.pop(s.sid, b""),
                        expected_origin=ORIGIN, require_uv=True,
                        previous_sign_count=cred["sign_count"])
                except (ValueError, KeyError) as e:
                    return self._json(400, {"error": f"authentication failed: {e}"})
                cred["sign_count"] = res["sign_count"]
                # Authenticated. In software mode, bind the session immediately.
                out = {"ok": True, "username": s.username, "up": res["up"],
                       "uv": res["uv"], "remember": s.remember, "hardware": app.hardware}
                if not app.hardware:
                    out["epoch"] = app.establish_epoch_software(s)
                return self._json(200, out)

            # ---- re-arm a remembered session (keypress on return) ----
            if r == "/auth/rearm/begin":
                s = self._session()
                if not s:
                    return self._json(401, {"error": "no session"})
                u = app.users.get(s.username)
                if not u or not u.credentials:
                    return self._json(400, {"error": "no passkey"})
                challenge = app.new_challenge(s.sid)
                return self._json(200, {
                    "rpId": RP_ID, "challenge": challenge, "userVerification": "required",
                    "allowCredentials": [{"type": "public-key", "id": cid}
                                         for cid in u.credentials]})

            # rearm/finish reuses the same verification + binding as login/finish
            if r == "/auth/rearm/finish":
                s = self._session()
                if not s:
                    return self._json(401, {"error": "no session"})
                data = json.loads(body)
                u = app.users.get(s.username)
                cred = u.credentials.get(data["credentialId"]) if u else None
                if not cred:
                    return self._json(400, {"error": "unknown credential"})
                try:
                    res = wa.verify_authentication(
                        authenticator_data_b64=data["authenticatorData"],
                        client_data_json_b64=data["clientDataJSON"],
                        signature_b64=data["signature"],
                        public_key_hex=cred["public_key_hex"], alg=cred["alg"],
                        rp_id=RP_ID, expected_challenge=app.challenges.pop(s.sid, b""),
                        expected_origin=ORIGIN, require_uv=True,
                        previous_sign_count=cred["sign_count"])
                except (ValueError, KeyError) as e:
                    return self._json(400, {"error": f"re-arm failed: {e}"})
                cred["sign_count"] = res["sign_count"]
                out = {"ok": True}
                if not app.hardware:
                    out["epoch"] = app.establish_epoch_software(s)
                return self._json(200, out)

            # ---- hardware epoch route ----
            if r == "/epoch/prepare":
                s = self._session()
                if not s:
                    return self._json(401, {"error": "no session"})
                if not app.hardware:
                    return self._json(400, {"error": "server not in hardware mode"})
                return self._json(200, app.epoch_prepare(s))
            if r == "/epoch/commit":
                s = self._session()
                if not s:
                    return self._json(401, {"error": "no session"})
                return self._json(200, app.epoch_commit(s, json.loads(body)))
            # ---- software-mode silent refresh (stand-in for touchless ARKG) ----
            if r == "/epoch/refresh":
                s = self._session()
                if not s:
                    return self._json(401, {"error": "no session"})
                if app.hardware:
                    return self._json(400, {"error": "use /epoch/prepare in hardware mode"})
                if not s.bound:
                    return self._json(401, {"error": "session not bound"})
                return self._json(200, app.establish_epoch_software(s))

            # ---- add a trusted device (bound). Account-persistence-relevant:
            # this is the kind of action an infostealer uses to keep access
            # after the victim changes their password. ----
            if r == "/api/add-device":
                s = self._session()
                if not s:
                    return self._json(401, {"error": "no session"})
                ok, reason = app.verify_binding(s, "POST", "/api/add-device", body, self.headers)
                if not ok:
                    return self._json(401, {"error": "inert", "reason": reason})
                try:
                    p = json.loads(body or b"{}")
                    device = str(p.get("device", "")).strip()
                except (json.JSONDecodeError, ValueError):
                    return self._json(400, {"error": "bad body"})
                if not device:
                    return self._json(400, {"error": "device name required"})
                u2 = app.users.get(s.username)
                u2.trusted_devices.append(device)
                u2.log_activity("add_device", device)
                app._save_users()
                return self._json(200, {"ok": True, "trusted_devices": u2.trusted_devices,
                                        "activity": u2.activity[-5:]})

            # ---- logout ----
            if r == "/auth/logout":
                s = self._session()
                if s:
                    app.sessions.pop(s.sid, None)
                return self._json(200, {"ok": True},
                                  cookie="sid=; Path=/; Max-Age=0")

            # ---- sensitive action: update recovery email (bound). This is
            # deliberately the headline "sensitive write" — redirecting
            # password-reset email is exactly what an account-hijacker does
            # with a stolen session, not moving money. ----
            if r == "/api/update-email":
                s = self._session()
                if not s:
                    return self._json(401, {"error": "no session"})
                ok, reason = app.verify_binding(s, "POST", "/api/update-email", body, self.headers)
                if not ok:
                    return self._json(401, {"error": "inert", "reason": reason})
                try:
                    p = json.loads(body or b"{}")
                    email = str(p.get("email", "")).strip()
                except (json.JSONDecodeError, ValueError):
                    return self._json(400, {"error": "bad body"})
                # Deliberately light validation (not full RFC 5322) — this is a
                # demo, not an email-address parser; just reject the obviously blank.
                if not email or "@" not in email or email.startswith("@") or email.endswith("@"):
                    return self._json(400, {"error": "a valid-looking email is required"})
                u = app.users[s.username]
                u.recovery_email = email
                u.log_activity("update_email", email)
                app._save_users()
                return self._json(200, {"ok": True, "recovery_email": u.recovery_email,
                                        "activity": u.activity[-5:]})

            return self.send_error(404)

    return H


class DemoServer(ThreadingHTTPServer):
    """ThreadingHTTPServer that refuses to share its port on Windows.

    http.server sets allow_reuse_address = 1 (SO_REUSEADDR). On POSIX that
    only skips the TIME_WAIT wait when restarting, which is what you want.
    On Windows SO_REUSEADDR means something else entirely: it lets a SECOND
    process bind a port another process is already serving, and new
    connections are then split between them non-deterministically.

    That caused a genuinely confusing failure during development — a freshly
    started `--hardware` server sat there Listening while a stale
    software-mode process from an earlier run kept answering the browser, so
    the UI reported "software epoch signer" no matter how many times the
    server was restarted with the right flag. It looks exactly like a code
    regression and isn't one. Refusing the second bind makes the real
    problem obvious: startup fails with "address already in use", and you go
    kill the old process instead of debugging phantom behaviour.
    """
    allow_reuse_address = (os.name != "nt")


def main():
    global PORT, ORIGIN
    ap = argparse.ArgumentParser()
    ap.add_argument("--hardware", action="store_true")
    ap.add_argument("--port", type=int, default=PORT)
    args = ap.parse_args()
    PORT = args.port
    ORIGIN = f"http://localhost:{PORT}"

    app = App(hardware=args.hardware)
    try:
        httpd = DemoServer((HOST, PORT), make_handler(app))
    except OSError as e:
        sys.exit(f"Could not bind {HOST}:{PORT} — {e}\n"
                 f"Something is already serving that port (most likely an earlier\n"
                 f"server.py you forgot to stop). Kill it and retry, or pass --port.")
    print("Keybound — single-server demo")
    if args.hardware:
        print("  MODE: HARDWARE — epoch signed on the YubiKey (previewSign/ARKG)")
        print("        needs signing_agent.py running in another terminal")
    else:
        print("  MODE: SOFTWARE epoch signer — real WebAuthn, but the epoch proof")
        print("        is signed in-process, NOT on the key. Pass --hardware for the")
        print("        previewSign/ARKG path.")
    print(f"  Open http://localhost:{PORT}   (WebAuthn passkeys are REAL — browser prompts)")
    try:
        httpd.serve_forever()
    except KeyboardInterrupt:
        print("\nbye")


if __name__ == "__main__":
    main()