AnyBot/relay/src/main.rs

// AnyBot 帳號伺服器 (the thin backend).
//
// What it keeps: accounts (name + PBKDF2 password hash), signed-in phones (session tokens, stored
// hashed), the user's AnyBots ("hosts", tokens stored hashed), and one encrypted backup per account
// (sealed on the user's computer with a key made from the password; this server can't open it).
//
// What it does: each AnyBot keeps asking `GET /relay/host/next` (an outgoing connection, so the
// computer needs no open port). A phone signed in to the account calls `/h/<host>/api/...`; the
// request is handed to that AnyBot and its answer comes back. AnyBots of the same account can
// also reach each other (裝置連結) through `/relay/hosts/<host>/api/device/run`.
//
// Run: anybot-relay [--port 8080] [--data ./relay-data]   (put HTTPS in front, e.g. Caddy)
// Sign-ups: the first account is free; after that ANYBOT_RELAY_INVITE must be set and given.

#[path = "../../src/httpd.rs"]
mod httpd;
#[path = "../../src/wire.rs"]
mod wire;
mod push;

use httpd::{Req, Resp};
use serde::{Deserialize, Serialize};
use serde_json::{json, Value};
use std::collections::{HashMap, VecDeque};
use std::path::PathBuf;
use std::sync::mpsc::{channel, Sender};
use std::sync::atomic::{AtomicU64, Ordering};
use std::sync::{Arc, Condvar, Mutex};
use std::time::{Duration, Instant};

const PAGE: &str = include_str!("../../mobile/index.html");
/// The bots' computer, watched and used in a browser (noVNC).
const SCREEN_PAGE: &str = include_str!("../../mobile/screen.html");
const NOVNC: &str = include_str!("../../mobile/novnc.js");
const MANIFEST: &str = r##"{"name":"AnyBot","short_name":"AnyBot","start_url":"./","display":"standalone","background_color":"#121212","theme_color":"#121212","icons":[{"src":"icon.svg","sizes":"any","type":"image/svg+xml"},{"src":"icon-192.png","sizes":"192x192","type":"image/png","purpose":"any maskable"},{"src":"icon-512.png","sizes":"512x512","type":"image/png","purpose":"any maskable"}]}"##;
/// AnyBot's icon (the same drawing as the app's), and PNGs for the iPhone home screen and Android.
const ICON: &str = include_str!("../../mobile/icon.svg");
const ICON_PNGS: [(&str, &[u8]); 3] = [("icon-180.png", include_bytes!("../../mobile/icon-180.png")), ("icon-192.png", include_bytes!("../../mobile/icon-192.png")), ("icon-512.png", include_bytes!("../../mobile/icon-512.png"))];

/// How long a host's "next request?" waits before answering "nothing".
const HOST_POLL_SECS: u64 = 25;
/// How long a phone's request waits for the AnyBot to answer (its own long-poll is 20 s).
const FORWARD_SECS: u64 = 35;
/// Model requests relayed to a person's computer (model_fetch) are ordinary LLM latency:
/// the asking side waits minutes, so the relay must not answer first with a timeout.
const LONG_FORWARD_SECS: u64 = 610;
/// How long a 配對碼 lasts.
const PAIR_SECS: u64 = 600;
/// How long a request may wait to be picked up before the host counts as gone.
const PICKUP_SECS: u64 = 8;
/// The lasting connection: a ping this often, and given up after this long without a word.
const WS_PING_SECS: u64 = 20;
const WS_DEAD_SECS: u64 = 60;
const GONE: &str = "那台 AnyBot 斷線了（網路中斷，或 AnyBot 關掉了）";
const OFFLINE: &str = "那台 AnyBot 目前離線（電腦可能關機、睡眠、斷網，或 AnyBot 沒開）";
/// A host counts as online when it asked within this many seconds.
const ONLINE_SECS: u64 = 60;
/// Sessions last this long without use.
const SESSION_DAYS: i64 = 90;
const MAX_FAILS: u32 = 8;

#[derive(Clone, Serialize, Deserialize, Default)]
struct User {
    name: String,
    pass: String,
    created: i64,
    /// 2FA: the TOTP shared secret once enabled, and the one waiting for its first code.
    #[serde(default)]
    totp: String,
    #[serde(default)]
    totp_pending: String,
    /// sha256 of each unused recovery code (8 are handed out once, at enabling).
    #[serde(default)]
    recover: Vec<String>,
}

#[derive(Clone, Serialize, Deserialize, Default)]
struct Session {
    hash: String,
    user: String,
    device: String,
    created: i64,
    last: i64,
}

#[derive(Clone, Serialize, Deserialize, Default)]
struct Host {
    id: String,
    name: String,
    user: String,
    token_hash: String,
    created: i64,
    #[serde(default)]
    last_seen: i64,
    /// 團隊: other accounts this AnyBot is shared with, and what they may do.
    #[serde(default)]
    members: Vec<Member>,
}

#[derive(Clone, Serialize, Deserialize, Default)]
struct Member {
    user: String,
    /// "admin": everything the owner can do on the phone (approve, screens);
    /// "member": chat with the bots and attach files only.
    role: String,
}

impl Host {
    /// The account's role on this AnyBot (None = no access).
    fn role_of(&self, user: &str) -> Option<String> {
        if self.user == user {
            return Some("owner".into());
        }
        self.members.iter().find(|m| m.user == user).map(|m| m.role.clone())
    }
}

#[derive(Clone, Serialize, Deserialize, Default)]
struct Sub {
    user: String,
    sub: Value,
    created: i64,
}

#[derive(Serialize, Deserialize, Default)]
struct Data {
    users: Vec<User>,
    sessions: Vec<Session>,
    hosts: Vec<Host>,
    /// Phones that turned notifications on.
    #[serde(default)]
    subs: Vec<Sub>,
}

const SW: &str = r#"self.addEventListener("install", () => self.skipWaiting());
self.addEventListener("activate", (e) => e.waitUntil(self.clients.claim()));
self.addEventListener("push", (e) => {
  let d = {};
  try { d = e.data ? e.data.json() : {}; } catch (_) { d = {body: e.data ? e.data.text() : ""}; }
  e.waitUntil(self.registration.showNotification(d.title || "AnyBot", {body: d.body || "", tag: "bot-" + (d.bot || 0), renotify: true, data: d, icon: "icon-192.png", badge: "icon-192.png"}));
});
self.addEventListener("notificationclick", (e) => {
  e.notification.close();
  const d = e.notification.data || {};
  const url = "./#host=" + encodeURIComponent(d.host || "") + "&bot=" + (d.bot || "");
  e.waitUntil(self.clients.matchAll({type: "window", includeUncontrolled: true}).then((cs) => {
    for (const c of cs) { c.postMessage({open: d}); return c.focus(); }
    return self.clients.openWindow(url);
  }));
});
"#;

/// A forwarded request's answer.
struct Answer {
    status: u16,
    ctype: String,
    headers: Vec<(String, String)>,
    body: Vec<u8>,
    /// The body is deflate-compressed (only when the asker said it can read that).
    z: bool,
}

impl Answer {
    fn error(status: u16, msg: &str) -> Answer {
        Answer { status, ctype: "application/json; charset=utf-8".into(), headers: vec![], body: json!({"error": msg}).to_string().into_bytes(), z: false }
    }
    fn from_json(v: &Value) -> Answer {
        use base64::Engine;
        Answer {
            status: v["status"].as_u64().unwrap_or(502) as u16,
            ctype: v["ctype"].as_str().unwrap_or("application/octet-stream").to_string(),
            headers: v["headers"].as_array().into_iter().flatten().filter_map(|h| Some((h[0].as_str()?.to_string(), h[1].as_str()?.to_string()))).collect(),
            body: base64::engine::general_purpose::STANDARD.decode(v["body"].as_str().unwrap_or("")).unwrap_or_default(),
            z: v["z"].as_bool().unwrap_or(false),
        }
    }
    fn into_resp(self) -> Resp {
        let mut resp = Resp::bytes(&self.ctype, self.body);
        resp.status = self.status;
        for (k, v) in self.headers {
            // Only harmless headers pass through.
            if k.eq_ignore_ascii_case("x-size") {
                resp.headers.push((k, v));
            }
        }
        resp
    }
}

/// An AnyBot's lasting connection: frames to write to it.
struct Sock {
    tx: Sender<Vec<u8>>,
    serial: u64,
}

static SERIAL: AtomicU64 = AtomicU64::new(1);

/// A screen being watched: where the host's side goes, and that host.
type Stream = (Sender<Vec<u8>>, String);

struct Relay {
    dir: PathBuf,
    data: Mutex<Data>,
    /// Requests waiting for each host.
    queues: Mutex<HashMap<String, VecDeque<Value>>>,
    arrived: Condvar,
    /// Where each forwarded request's answer goes.
    /// Requests waiting for an answer: job id → (its answer, the host it was sent to —
    /// so only that host may answer it).
    waiting: Mutex<HashMap<String, (Sender<Answer>, String)>>,
    /// AnyBots connected the lasting way (WebSocket), and which requests went over which.
    sockets: Mutex<HashMap<String, Sock>>,
    flights: Mutex<HashMap<String, (String, u64)>>,
    /// The brain each AnyBot shows (when it is a screen of another), and which lend their hands.
    uses: Mutex<HashMap<String, String>>,
    hands: Mutex<std::collections::HashSet<String>>,
    /// People watching a bots' computer: stream id → (where the host's side goes, that host).
    streams: Mutex<HashMap<String, Stream>>,
    /// One-time links for a browser to watch a host's computer: ticket → (account, host, made).
    tickets: Mutex<HashMap<String, (String, String, Instant)>>,
    /// 配對碼: code → (account, the AnyBot it is for, when made). Ten minutes, once.
    pairs: Mutex<HashMap<String, (String, String, Instant)>>,
    /// When each host last asked (in memory; saved now and then).
    seen: Mutex<HashMap<String, Instant>>,
    fails: Mutex<HashMap<String, (u32, Instant)>>,
    /// 2FA: a right password opens a five-minute challenge (user, device) awaiting the code.
    challenges: Mutex<HashMap<String, (String, String, Instant)>>,
    invite: String,
    vapid: push::Vapid,
    contact: String,
}

/// How long a link the app opens for a browser works (reconnects included).
const TICKET_LIFE: Duration = Duration::from_secs(600);

fn now() -> i64 {
    chrono::Utc::now().timestamp()
}

impl Relay {
    fn load(dir: PathBuf) -> Relay {
        let _ = std::fs::create_dir_all(dir.join("backups"));
        let data: Data = std::fs::read(dir.join("data.json")).ok().and_then(|b| serde_json::from_slice(&b).ok()).unwrap_or_default();
        let vapid = push::Vapid::load(&dir);
        Relay {
            dir,
            data: Mutex::new(data),
            queues: Mutex::new(HashMap::new()),
            arrived: Condvar::new(),
            waiting: Mutex::new(HashMap::new()),
            sockets: Mutex::new(HashMap::new()),
            flights: Mutex::new(HashMap::new()),
            uses: Mutex::new(HashMap::new()),
            hands: Mutex::new(std::collections::HashSet::new()),
            streams: Mutex::new(HashMap::new()),
            tickets: Mutex::new(HashMap::new()),
            pairs: Mutex::new(HashMap::new()),
            seen: Mutex::new(HashMap::new()),
            fails: Mutex::new(HashMap::new()),
            challenges: Mutex::new(HashMap::new()),
            invite: std::env::var("ANYBOT_RELAY_INVITE").unwrap_or_default(),
            vapid,
            // Push services want a way to reach whoever runs the server.
            contact: std::env::var("ANYBOT_RELAY_CONTACT").ok().filter(|c| c.starts_with("mailto:") || c.starts_with("https://")).unwrap_or_else(|| "mailto:anybot-relay@example.com".into()),
        }
    }

    fn save(&self, d: &Data) {
        let tmp = self.dir.join("data.json.tmp");
        if std::fs::write(&tmp, serde_json::to_vec_pretty(d).unwrap_or_default()).is_ok() {
            let _ = std::fs::rename(&tmp, self.dir.join("data.json"));
        }
    }

    fn session(&self, req: &Req) -> Option<String> {
        let t = req.bearer();
        if t.is_empty() {
            return None;
        }
        let h = httpd::sha256_hex(&t);
        let mut d = self.data.lock().unwrap();
        let cutoff = now() - SESSION_DAYS * 86_400;
        let s = d.sessions.iter_mut().find(|s| s.hash == h && s.last > cutoff)?;
        // Count use at most once a minute (saves writing on every request).
        if now() - s.last > 60 {
            s.last = now();
            let user = s.user.clone();
            self.save(&d);
            return Some(user);
        }
        Some(s.user.clone())
    }

    fn host(&self, req: &Req) -> Option<Host> {
        let t = req.bearer();
        if t.is_empty() {
            return None;
        }
        let h = httpd::sha256_hex(&t);
        let d = self.data.lock().unwrap();
        d.hosts.iter().find(|x| x.token_hash == h).cloned()
    }

    fn online(&self, id: &str) -> bool {
        self.sockets.lock().unwrap().contains_key(id) || self.seen.lock().unwrap().get(id).map(|t| t.elapsed().as_secs() < ONLINE_SECS).unwrap_or(false)
    }

    fn too_many_fails(&self, key: &str) -> bool {
        let f = self.fails.lock().unwrap();
        matches!(f.get(key), Some((n, t)) if *n >= MAX_FAILS && t.elapsed() < Duration::from_secs(300))
    }

    fn failed(&self, key: &str) {
        let mut f = self.fails.lock().unwrap();
        let e = f.entry(key.to_string()).or_insert((0, Instant::now()));
        if e.1.elapsed() > Duration::from_secs(300) {
            *e = (0, Instant::now());
        }
        e.0 += 1;
        e.1 = Instant::now();
    }

    /// A person's browser watching (and maybe using) an AnyBot's computer: its VNC stream goes
    /// to that AnyBot over its lasting connection, and back. Signed in by the account session
    /// (?t=, the phone page) or a ticket the account's app asked for (?ticket=); owners and
    /// admins only.
    fn vnc_viewer(self: &Arc<Self>, req: Req, mut c: std::net::TcpStream, host: &str) {
        use tungstenite::{protocol::Role, Message, WebSocket};
        let user = match req.query.get("ticket").filter(|t| !t.is_empty()) {
            Some(t) => self.tickets.lock().unwrap().get(t).filter(|(_, h, made)| h == host && made.elapsed() < TICKET_LIFE).map(|(u, _, _)| u.clone()),
            None => self.session(&req),
        };
        let Some(user) = user else { return httpd::write_response(&mut c, &Resp::error(401, "請先登入")) };
        let allowed = self.data.lock().unwrap().hosts.iter().any(|x| x.id == host && matches!(x.role_of(&user).as_deref(), Some("owner") | Some("admin")));
        if !allowed {
            return httpd::write_response(&mut c, &Resp::error(404, "找不到這台 AnyBot"));
        }
        let Some(to_host) = self.sockets.lock().unwrap().get(host).map(|s| s.tx.clone()) else {
            return httpd::write_response(&mut c, &Resp::error(503, "這台 AnyBot 現在沒有連線"));
        };
        if !httpd::accept_websocket(&req, &mut c) {
            return;
        }
        let mut ws = WebSocket::from_raw_socket(c, Role::Server, None);
        let sid = httpd::random_hex(8);
        let (tx, rx) = channel::<Vec<u8>>();
        self.streams.lock().unwrap().insert(sid.clone(), (tx, host.to_string()));
        if to_host.send(wire::pack(&json!({"t": "vopen", "sid": sid}), &[])).is_err() {
            self.streams.lock().unwrap().remove(&sid);
            return;
        }
        let (mut heard, mut pinged) = (Instant::now(), Instant::now());
        let mut pace = wire::Pace::new();
        'live: loop {
            pace.tune(ws.get_ref());
            let up = match ws.read() {
                Ok(Message::Binary(b)) => Some(("vdata", b.to_vec())),
                Ok(Message::Text(t)) => Some(("vtext", t.as_bytes().to_vec())),
                Ok(Message::Close(_)) => break,
                Ok(_) => None,
                Err(tungstenite::Error::Io(e)) if matches!(e.kind(), std::io::ErrorKind::WouldBlock | std::io::ErrorKind::TimedOut) => None,
                Err(_) => break,
            };
            if let Some((t, b)) = up {
                heard = Instant::now();
                pace.moved();
                if to_host.send(wire::pack(&json!({"t": t, "sid": sid}), &b)).is_err() {
                    break;
                }
            }
            loop {
                match rx.try_recv() {
                    Ok(b) => {
                        pace.moved();
                        if ws.write(Message::Binary(b.into())).is_err() {
                            break 'live;
                        }
                    }
                    Err(std::sync::mpsc::TryRecvError::Empty) => break,
                    // The host closed it (or went away).
                    Err(std::sync::mpsc::TryRecvError::Disconnected) => break 'live,
                }
            }
            if pinged.elapsed().as_secs() >= WS_PING_SECS {
                pinged = Instant::now();
                if ws.write(Message::Ping(Vec::new().into())).is_err() {
                    break;
                }
            }
            match ws.flush() {
                Err(tungstenite::Error::Io(e)) if matches!(e.kind(), std::io::ErrorKind::WouldBlock | std::io::ErrorKind::TimedOut) => {}
                Err(_) => break,
                Ok(()) => {}
            }
            if heard.elapsed().as_secs() >= WS_DEAD_SECS {
                break;
            }
        }
        let _ = ws.close(None);
        let _ = ws.flush();
        self.streams.lock().unwrap().remove(&sid);
        let _ = to_host.send(wire::pack(&json!({"t": "vclose", "sid": sid}), &[]));
    }

    /// Hand a request to a host and wait for its answer.
    fn forward(&self, host_id: &str, method: &str, path: &str, query: &HashMap<String, String>, body: &[u8], who: (&str, &str)) -> Resp {
        self.forward_raw(host_id, method, path, query, body, false, who, false, "").into_resp()
    }

    /// `z`: the body is compressed; `z_ok`: the asker reads a compressed answer.
    #[allow(clippy::too_many_arguments)]
    /// `from`: the AnyBot of the account asking (empty for the phone).
    fn forward_raw(&self, host_id: &str, method: &str, path: &str, query: &HashMap<String, String>, body: &[u8], z: bool, who: (&str, &str), z_ok: bool, from: &str) -> Answer {
        use base64::Engine;
        if !self.online(host_id) {
            return Answer::error(503, OFFLINE);
        }
        let id = httpd::random_hex(12);
        let secs = if path == "/api/device/run" { LONG_FORWARD_SECS } else { FORWARD_SECS };
        let (tx, rx) = channel();
        self.waiting.lock().unwrap().insert(id.clone(), (tx, host_id.to_string()));
        let head = json!({"t": "job", "id": id, "method": method, "path": path, "query": query, "user": who.0, "role": who.1, "z": z, "z_ok": z_ok, "from": from});
        // Connected the lasting way: straight down that connection.
        let sock = self.sockets.lock().unwrap().get(host_id).map(|s| (s.tx.clone(), s.serial));
        if let Some((stx, serial)) = sock {
            self.flights.lock().unwrap().insert(id.clone(), (host_id.to_string(), serial));
            let sent = stx.send(wire::pack(&head, body)).is_ok();
            let r = if sent { rx.recv_timeout(Duration::from_secs(secs)).ok() } else { None };
            self.waiting.lock().unwrap().remove(&id);
            self.flights.lock().unwrap().remove(&id);
            return match r {
                Some(a) => a,
                None if !sent => Answer::error(503, GONE),
                None => Answer::error(504, "那台 AnyBot 收到了，但太久沒有回應（可能太忙，或網路很慢）"),
            };
        }
        let mut job = head;
        job["body"] = json!(base64::engine::general_purpose::STANDARD.encode(body));
        self.queues.lock().unwrap().entry(host_id.to_string()).or_default().push_back(job);
        self.arrived.notify_all();
        // A running AnyBot always has a worker waiting, so it picks a request up within a moment.
        // Still in the queue after a few seconds means it went away (closed, asleep, offline).
        let mut r = rx.recv_timeout(Duration::from_secs(PICKUP_SECS));
        if r.is_err() {
            let queued = self.queues.lock().unwrap().get(host_id).map(|q| q.iter().any(|j| j["id"].as_str() == Some(id.as_str()))).unwrap_or(false);
            if !queued {
                r = rx.recv_timeout(Duration::from_secs(secs.max(PICKUP_SECS + 1) - PICKUP_SECS));
            }
        }
        self.waiting.lock().unwrap().remove(&id);
        // Not picked up in time: drop it so it doesn't run late.
        let was_queued = self.queues.lock().unwrap().get_mut(host_id).map(|q| {
            let n = q.len();
            q.retain(|j| j["id"].as_str() != Some(id.as_str()));
            n != q.len()
        });
        match r {
            Ok(a) => a,
            Err(_) if was_queued == Some(true) => Answer::error(503, OFFLINE),
            Err(_) => Answer::error(504, "那台 AnyBot 收到了，但太久沒有回應（可能太忙，或網路很慢）"),
        }
    }

    /// 長連線: an AnyBot keeps one WebSocket open; requests for it come down it at once, and its
    /// own requests to the account's other AnyBots go up it.
    fn ws(self: &Arc<Self>, req: Req, mut c: std::net::TcpStream) {
        use tungstenite::{protocol::Role, Message, WebSocket};
        if let Some(host) = req.path.strip_prefix("/relay/vnc/") {
            let host = host.to_string();
            return self.vnc_viewer(req, c, &host);
        }
        if req.path != "/relay/host/ws" {
            return httpd::write_response(&mut c, &Resp::error(404, "沒有這個網址"));
        }
        let Some(h) = self.host(&req) else { return httpd::write_response(&mut c, &Resp::error(401, "這台電腦的金鑰無效")) };
        if !httpd::accept_websocket(&req, &mut c) {
            return;
        }
        let cfg = tungstenite::protocol::WebSocketConfig::default().max_message_size(Some(httpd::MAX_BODY * 2)).max_frame_size(Some(httpd::MAX_BODY * 2));
        let mut ws = WebSocket::from_raw_socket(c, Role::Server, Some(cfg));
        let serial = SERIAL.fetch_add(1, Ordering::SeqCst);
        let (tx, rx) = channel::<Vec<u8>>();
        self.sockets.lock().unwrap().insert(h.id.clone(), Sock { tx: tx.clone(), serial });
        self.seen.lock().unwrap().insert(h.id.clone(), Instant::now());
        let (mut heard, mut pinged, mut marked) = (Instant::now(), Instant::now(), Instant::now());
        let mut pace = wire::Pace::new();
        'live: loop {
            pace.tune(ws.get_ref());
            match ws.read() {
                Ok(Message::Binary(b)) => {
                    heard = Instant::now();
                    pace.moved();
                    self.on_frame(&h, &b, &tx);
                }
                Ok(Message::Close(_)) => break,
                Ok(_) => heard = Instant::now(),
                Err(tungstenite::Error::Io(e)) if matches!(e.kind(), std::io::ErrorKind::WouldBlock | std::io::ErrorKind::TimedOut) => {}
                Err(_) => break,
            }
            while let Ok(f) = rx.try_recv() {
                pace.moved();
                if ws.write(Message::Binary(f.into())).is_err() {
                    break 'live;
                }
            }
            if pinged.elapsed().as_secs() >= WS_PING_SECS {
                pinged = Instant::now();
                if ws.write(Message::Ping(Vec::new().into())).is_err() {
                    break;
                }
            }
            match ws.flush() {
                Err(tungstenite::Error::Io(e)) if matches!(e.kind(), std::io::ErrorKind::WouldBlock | std::io::ErrorKind::TimedOut) => {}
                Err(_) => break,
                Ok(()) => {}
            }
            if heard.elapsed().as_secs() >= WS_DEAD_SECS {
                break;
            }
            if marked.elapsed().as_secs() >= 10 {
                marked = Instant::now();
                self.seen.lock().unwrap().insert(h.id.clone(), Instant::now());
            }
        }
        // Gone: its viewers' streams end, and whoever waits on it hears so now, not after a timeout.
        self.streams.lock().unwrap().retain(|_, (_, host)| *host != h.id);
        {
            let mut socks = self.sockets.lock().unwrap();
            if socks.get(&h.id).map(|s| s.serial) == Some(serial) {
                socks.remove(&h.id);
                self.seen.lock().unwrap().remove(&h.id);
            }
        }
        let lost: Vec<String> = self.flights.lock().unwrap().iter().filter(|(_, (host, s))| *host == h.id && *s == serial).map(|(id, _)| id.clone()).collect();
        for id in lost {
            if let Some((w, _)) = self.waiting.lock().unwrap().remove(&id) {
                let _ = w.send(Answer::error(503, GONE));
            }
        }
    }

    fn on_frame(self: &Arc<Self>, h: &Host, b: &[u8], tx: &Sender<Vec<u8>>) {
        let Some((head, body)) = wire::unpack(b) else { return };
        let s = |k: &str| head[k].as_str().unwrap_or("").to_string();
        match head["t"].as_str().unwrap_or("") {
            // A viewer's stream: only the host it was opened on may write to it.
            "vdata" => {
                if let Some((v, host)) = self.streams.lock().unwrap().get(&s("sid")) {
                    if *host == h.id {
                        let _ = v.send(body);
                    }
                }
            }
            "vclose" => {
                let mut all = self.streams.lock().unwrap();
                if all.get(&s("sid")).map(|(_, host)| *host == h.id).unwrap_or(false) {
                    all.remove(&s("sid"));
                }
            }
            "reply" => {
                // Only the host the job went to may answer it.
                let w = {
                    let mut waiting = self.waiting.lock().unwrap();
                    match waiting.get(&s("id")) {
                        Some((_, target)) if *target == h.id => waiting.remove(&s("id")).map(|(w, _)| w),
                        _ => None,
                    }
                };
                let Some(w) = w else { return };
                let headers = head["headers"].as_array().into_iter().flatten().filter_map(|x| Some((x[0].as_str()?.to_string(), x[1].as_str()?.to_string()))).collect();
                let _ = w.send(Answer { status: head["status"].as_u64().unwrap_or(502) as u16, ctype: s("ctype"), headers, body, z: head["z"].as_bool().unwrap_or(false) });
            }
            // One AnyBot asks another of the same account (大腦 sync, 裝置連結).
            "call" => {
                let (id, to, path) = (s("id"), s("to"), s("path"));
                let role = if path == "/api/device/run" {
                    "device"
                } else if path == "/api/sync" || path.starts_with("/api/sync/") {
                    "mirror"
                } else {
                    let a = Answer::error(404, "沒有這個網址");
                    let _ = tx.send(wire::pack(&json!({"t": "ret", "id": id, "status": a.status, "ctype": a.ctype}), &a.body));
                    return;
                };
                let same = self.data.lock().unwrap().hosts.iter().any(|x| x.id == to && x.user == h.user);
                if !same || to == h.id {
                    let a = Answer::error(403, "不是同一個帳號的電腦");
                    let _ = tx.send(wire::pack(&json!({"t": "ret", "id": id, "status": a.status, "ctype": a.ctype}), &a.body));
                    return;
                }
                let query: HashMap<String, String> = head["query"].as_object().map(|o| o.iter().map(|(k, v)| (k.clone(), v.as_str().unwrap_or("").to_string())).collect()).unwrap_or_default();
                let (me, tx, user, method, z, from) = (self.clone(), tx.clone(), h.user.clone(), s("method"), head["z"].as_bool().unwrap_or(false), h.id.clone());
                std::thread::spawn(move || {
                    let a = me.forward_raw(&to, &method, &path, &query, &body, z, (&user, role), true, &from);
                    // The answer's headers too (a live picture's number, a screen's size).
                    let headers: Vec<[&str; 2]> = a.headers.iter().map(|(k, v)| [k.as_str(), v.as_str()]).collect();
                    let _ = tx.send(wire::pack(&json!({"t": "ret", "id": id, "status": a.status, "ctype": a.ctype, "z": a.z, "headers": headers}), &a.body));
                });
            }
            "hello" => {
                if head["hands"].as_bool().unwrap_or(false) {
                    self.hands.lock().unwrap().insert(h.id.clone());
                } else {
                    self.hands.lock().unwrap().remove(&h.id);
                }
                let brain = s("brain");
                let mut u = self.uses.lock().unwrap();
                if brain.is_empty() {
                    u.remove(&h.id);
                } else {
                    u.insert(h.id.clone(), brain);
                }
            }
            _ => {}
        }
    }

    fn handle(&self, req: Req) -> Resp {
        let parts: Vec<String> = req.parts().iter().map(|s| s.to_string()).collect();
        let p: Vec<&str> = parts.iter().map(|s| s.as_str()).collect();
        match (req.method.as_str(), p.as_slice()) {
            ("GET", []) | ("GET", ["index.html"]) => Resp::bytes("text/html; charset=utf-8", PAGE.as_bytes().to_vec())
                .with_header("Content-Security-Policy", "default-src 'self'; img-src 'self' blob: data:; style-src 'self' 'unsafe-inline'; script-src 'self' 'unsafe-inline'; connect-src 'self'")
                .with_header("Strict-Transport-Security", "max-age=31536000"),
            ("GET", ["screen"]) => Resp::bytes("text/html; charset=utf-8", SCREEN_PAGE.as_bytes().to_vec())
                .with_header("Content-Security-Policy", "default-src 'self'; img-src 'self' blob: data:; style-src 'self' 'unsafe-inline'; script-src 'self' 'unsafe-inline'; connect-src 'self' wss: ws:")
                .with_header("Strict-Transport-Security", "max-age=31536000"),
            ("GET", ["novnc.js"]) => Resp::bytes("text/javascript; charset=utf-8", NOVNC.as_bytes().to_vec()).with_header("Cache-Control", "public, max-age=86400"),
            // A link for this account's browser to watch one of its AnyBots' computer (the app
            // opens it; the browser has no session of its own).
            ("POST", ["relay", "screen-ticket"]) => {
                let Some(h) = self.host(&req) else { return Resp::error(401, "這台電腦的金鑰無效") };
                let want = req.json()["host"].as_str().unwrap_or("").to_string();
                let ok = self.data.lock().unwrap().hosts.iter().any(|x| x.id == want && matches!(x.role_of(&h.user).as_deref(), Some("owner") | Some("admin")));
                if !ok {
                    return Resp::error(404, "找不到這台 AnyBot");
                }
                let t = httpd::random_hex(24);
                let mut all = self.tickets.lock().unwrap();
                all.retain(|_, (_, _, made)| made.elapsed() < TICKET_LIFE);
                all.insert(t.clone(), (h.user.clone(), want, Instant::now()));
                Resp::json(&json!({"ticket": t}))
            }
            ("GET", ["manifest.webmanifest"]) => Resp::bytes("application/manifest+json", MANIFEST.as_bytes().to_vec()),
            ("GET", ["icon.svg"]) => Resp::bytes("image/svg+xml", ICON.as_bytes().to_vec()),
            ("GET", [name]) if ICON_PNGS.iter().any(|(n, _)| n == name) => Resp::bytes("image/png", ICON_PNGS.iter().find(|(n, _)| n == name).map(|(_, b)| b.to_vec()).unwrap_or_default()),
            ("GET", ["api", "hello"]) => Resp::json(&json!({"app": "anybot-relay"})),
            ("GET", ["sw.js"]) => Resp::bytes("text/javascript; charset=utf-8", SW.as_bytes().to_vec()).with_header("Service-Worker-Allowed", "/"),
            ("GET", ["relay", "push", "key"]) => Resp::json(&json!({"key": self.vapid.public_b64})),
            ("POST", ["relay", "push", "subscribe"]) => {
                let Some(user) = self.session(&req) else { return Resp::error(401, "請先登入") };
                let sub = req.json();
                let endpoint = sub["endpoint"].as_str().unwrap_or("").to_string();
                if !endpoint.starts_with("https://") || sub["keys"]["p256dh"].as_str().is_none() || sub["keys"]["auth"].as_str().is_none() {
                    return Resp::error(400, "通知訂閱資料不完整");
                }
                let mut d = self.data.lock().unwrap();
                d.subs.retain(|s| s.sub["endpoint"].as_str() != Some(endpoint.as_str()));
                d.subs.push(Sub { user, sub, created: now() });
                self.save(&d);
                Resp::json(&json!({"ok": true}))
            }
            ("POST", ["relay", "push", "unsubscribe"]) => {
                let endpoint = req.json()["endpoint"].as_str().unwrap_or("").to_string();
                let mut d = self.data.lock().unwrap();
                d.subs.retain(|s| s.sub["endpoint"].as_str() != Some(endpoint.as_str()));
                self.save(&d);
                Resp::json(&json!({"ok": true}))
            }
            // An AnyBot tells its person something (a bot needs them, or finished while they were away).
            ("POST", ["relay", "host", "notify"]) => {
                let Some(h) = self.host(&req) else { return Resp::error(401, "這台電腦的金鑰無效") };
                let v = req.json();
                let payload = json!({"title": v["title"].as_str().unwrap_or("AnyBot").chars().take(80).collect::<String>(), "body": v["body"].as_str().unwrap_or("").chars().take(200).collect::<String>(), "bot": v["bot"], "host": h.id});
                let subs: Vec<Sub> = self.data.lock().unwrap().subs.iter().filter(|s| matches!(h.role_of(&s.user).as_deref(), Some("owner") | Some("admin"))).cloned().collect();
                let mut gone = vec![];
                let mut sent = 0;
                for s in &subs {
                    match push::send(&self.vapid, &s.sub, &payload, &self.contact) {
                        Ok(()) => sent += 1,
                        Err(true) => gone.push(s.sub["endpoint"].as_str().unwrap_or("").to_string()),
                        Err(false) => {}
                    }
                }
                if !gone.is_empty() {
                    let mut d = self.data.lock().unwrap();
                    d.subs.retain(|s| !gone.iter().any(|g| s.sub["endpoint"].as_str() == Some(g.as_str())));
                    self.save(&d);
                }
                Resp::json(&json!({"sent": sent}))
            }
            ("POST", ["relay", "signup"]) => self.signup(&req),
            ("POST", ["relay", "login"]) => self.login(&req),
            ("POST", ["relay", "2fa", "enroll"]) => self.enroll_2fa(&req),
            ("POST", ["relay", "2fa", "verify"]) => self.verify_2fa(&req),
            ("POST", ["relay", "2fa", "disable"]) => self.disable_2fa(&req),
            ("POST", ["relay", "sessions", "logout"]) => self.logout_everywhere(&req),
            ("POST", ["relay", "logout"]) => {
                let h = httpd::sha256_hex(&req.bearer());
                let mut d = self.data.lock().unwrap();
                d.sessions.retain(|s| s.hash != h);
                self.save(&d);
                Resp::json(&json!({"ok": true}))
            }
            ("GET", ["relay", "me"]) => {
                let Some(user) = self.session(&req) else { return Resp::error(401, "請先登入") };
                let d = self.data.lock().unwrap();
                let hosts: Vec<Value> = d
                    .hosts
                    .iter()
                    .filter_map(|h| h.role_of(&user).map(|r| (h, r)))
                    .map(|(h, r)| {
                        let mut v = self.host_json(h);
                        v["role"] = json!(r);
                        if r != "owner" {
                            v["name"] = json!(format!("{}（{} 的）", h.name, h.user));
                        }
                        v
                    })
                    .collect();
                let sessions: Vec<Value> = d.sessions.iter().filter(|s| s.user == user).map(|s| json!({"device": s.device, "created": s.created, "last": s.last})).collect();
                // (`d` above is this handler's lock: read the users through it, never a second lock.)
                let totp_on = d.users.iter().find(|u| u.name == user).map(|u| !u.totp.is_empty()).unwrap_or(false);
                Resp::json(&json!({"user": user, "two_factor": totp_on, "hosts": hosts, "sessions": sessions, "backup": self.dir.join("backups").join(format!("{}.bin", httpd::sha256_hex(&user))).exists()}))
            }
            ("POST", ["relay", "hosts"]) => {
                let Some(user) = self.session(&req) else { return Resp::error(401, "請先登入") };
                let name: String = req.json()["name"].as_str().unwrap_or("AnyBot").chars().take(60).collect();
                let (h, token) = self.add_host(&user, &name);
                Resp::json(&json!({"id": h.id, "token": token}))
            }
            // 加一台電腦: a signed-in phone or AnyBot asks for a code; the new computer gives it back.
            ("POST", ["relay", "pair"]) => {
                let Some(user) = self.session(&req).or_else(|| self.host(&req).map(|h| h.user)) else { return Resp::error(401, "請先登入") };
                let want = req.json()["for"].as_str().unwrap_or("").to_string();
                let for_host = if self.data.lock().unwrap().hosts.iter().any(|h| h.id == want && h.user == user) { want } else { String::new() };
                let mut pairs = self.pairs.lock().unwrap();
                pairs.retain(|_, p| p.2.elapsed().as_secs() < PAIR_SECS);
                let code = loop {
                    let n = u32::from_str_radix(&httpd::random_hex(4), 16).unwrap_or(0) % 1_000_000;
                    let c = format!("{n:06}");
                    if !pairs.contains_key(&c) {
                        break c;
                    }
                };
                pairs.insert(code.clone(), (user, for_host, Instant::now()));
                Resp::json(&json!({"code": code, "expires_in": PAIR_SECS}))
            }
            ("POST", ["relay", "pair", "claim"]) => {
                let key = format!("pair:{}", req.peer);
                if self.too_many_fails(&key) {
                    return Resp::error(429, "配對碼錯太多次，請 5 分鐘後再試");
                }
                let v = req.json();
                let code: String = v["code"].as_str().unwrap_or("").chars().filter(|c| c.is_ascii_digit()).collect();
                let got = {
                    let mut pairs = self.pairs.lock().unwrap();
                    pairs.retain(|_, p| p.2.elapsed().as_secs() < PAIR_SECS);
                    pairs.remove(&code)
                };
                let Some((user, for_host, _)) = got else {
                    self.failed(&key);
                    return Resp::error(404, "配對碼不對，或已經過期（10 分鐘內有效，只能用一次）");
                };
                let name: String = v["name"].as_str().unwrap_or("AnyBot").chars().take(60).collect();
                let (h, token) = self.add_host(&user, &name);
                let for_name = self.data.lock().unwrap().hosts.iter().find(|x| x.id == for_host).map(|x| x.name.clone()).unwrap_or_default();
                Resp::json(&json!({"id": h.id, "token": token, "user": user, "for": for_host, "for_name": for_name}))
            }
            ("POST", ["relay", "hosts", id, "remove"]) => {
                let Some(user) = self.session(&req) else { return Resp::error(401, "請先登入") };
                let mut d = self.data.lock().unwrap();
                d.hosts.retain(|x| !(x.user == user && x.id == *id));
                self.save(&d);
                Resp::json(&json!({"ok": true}))
            }
            ("GET", ["relay", "host", "next"]) => {
                let Some(h) = self.host(&req) else { return Resp::error(401, "這台電腦的金鑰無效") };
                self.seen.lock().unwrap().insert(h.id.clone(), Instant::now());
                let q = self.queues.lock().unwrap();
                let (mut q, _) = self.arrived.wait_timeout_while(q, Duration::from_secs(HOST_POLL_SECS), |q| q.get(&h.id).map(|x| x.is_empty()).unwrap_or(true)).unwrap();
                self.seen.lock().unwrap().insert(h.id.clone(), Instant::now());
                match q.get_mut(&h.id).and_then(|x| x.pop_front()) {
                    Some(job) => Resp::json(&job),
                    None => Resp::empty(204),
                }
            }
            ("POST", ["relay", "host", "reply", id]) => {
                let Some(h) = self.host(&req) else { return Resp::error(401, "這台電腦的金鑰無效") };
                // Only the host the job was sent to may answer it.
                let tx = {
                    let mut waiting = self.waiting.lock().unwrap();
                    match waiting.get(*id) {
                        Some((_, target)) if *target == h.id => waiting.remove(*id).map(|(w, _)| w),
                        _ => None,
                    }
                };
                if let Some(tx) = tx {
                    let _ = tx.send(Answer::from_json(&req.json()));
                }
                Resp::json(&json!({"ok": true}))
            }
            ("GET", ["relay", "host", "members"]) => {
                let Some(h) = self.host(&req) else { return Resp::error(401, "這台電腦的金鑰無效") };
                Resp::json(&json!({"members": h.members.iter().map(|m| json!({"user": m.user, "role": m.role})).collect::<Vec<_>>()}))
            }
            ("POST", ["relay", "host", "members"]) => {
                let Some(h) = self.host(&req) else { return Resp::error(401, "這台電腦的金鑰無效") };
                let v = req.json();
                let user = v["user"].as_str().unwrap_or("").trim().to_lowercase();
                let role = v["role"].as_str().unwrap_or("member").to_string();
                let mut d = self.data.lock().unwrap();
                if role != "remove" && !d.users.iter().any(|u| u.name == user) {
                    return Resp::error(404, "這台伺服器上沒有這個帳號（請對方先建立帳號）");
                }
                if user == h.user {
                    return Resp::error(400, "這是你自己的帳號");
                }
                if !matches!(role.as_str(), "member" | "admin" | "remove") {
                    return Resp::error(400, "角色只能是 member 或 admin");
                }
                if let Some(x) = d.hosts.iter_mut().find(|x| x.id == h.id) {
                    x.members.retain(|m| m.user != user);
                    if role != "remove" {
                        x.members.push(Member { user, role });
                    }
                }
                self.save(&d);
                Resp::json(&json!({"ok": true}))
            }
            // 發布網站: an AnyBot publishes a folder; it is served at /s/<user>-<name>/.
            ("POST", ["relay", "host", "sites", name]) => {
                let Some(h) = self.host(&req) else { return Resp::error(401, "這台電腦的金鑰無效") };
                let name = name.to_lowercase();
                if name.is_empty() || name.len() > 40 || !name.chars().all(|c| c.is_ascii_alphanumeric() || c == '-') {
                    return Resp::error(400, "網站名稱只能用英文字母、數字和 -（最多 40 個字）");
                }
                let site = format!("{}-{name}", h.user.replace(|c: char| !c.is_ascii_alphanumeric(), "-"));
                let dir = self.dir.join("sites").join(&site);
                let v = req.json();
                let _ = std::fs::remove_dir_all(&dir);
                if v["remove"].as_bool() == Some(true) {
                    return Resp::json(&json!({"ok": true}));
                }
                let mut n = 0;
                for f in v["files"].as_array().into_iter().flatten() {
                    let rel = f["path"].as_str().unwrap_or("");
                    if rel.is_empty() || rel.split('/').any(|c| c.is_empty() || c == "." || c == "..") {
                        continue;
                    }
                    let Ok(bytes) = base64::Engine::decode(&base64::engine::general_purpose::STANDARD, f["data"].as_str().unwrap_or("")) else { continue };
                    let path = dir.join(rel);
                    let _ = std::fs::create_dir_all(path.parent().unwrap_or(&dir));
                    if std::fs::write(&path, bytes).is_ok() {
                        n += 1;
                    }
                }
                let public = std::env::var("ANYBOT_RELAY_PUBLIC_URL").unwrap_or_default();
                Resp::json(&json!({"files": n, "path": format!("/s/{site}/"), "url": if public.is_empty() { String::new() } else { format!("{}/s/{site}/", public.trim_end_matches('/')) }}))
            }
            ("GET", ["s", site, rest @ ..]) => {
                if site.contains("..") || rest.iter().any(|c| c.is_empty() || *c == "." || *c == "..") {
                    return Resp::error(404, "找不到");
                }
                let mut path = self.dir.join("sites").join(site).join(rest.join("/"));
                if path.is_dir() || rest.is_empty() {
                    path = path.join("index.html");
                }
                let ext = path.extension().map(|e| e.to_string_lossy().to_lowercase()).unwrap_or_default();
                let ctype = match ext.as_str() {
                    "html" | "htm" => "text/html; charset=utf-8",
                    "css" => "text/css; charset=utf-8",
                    "js" | "mjs" => "text/javascript; charset=utf-8",
                    "json" => "application/json",
                    "svg" => "image/svg+xml",
                    "png" => "image/png",
                    "jpg" | "jpeg" => "image/jpeg",
                    "gif" => "image/gif",
                    "webp" => "image/webp",
                    "ico" => "image/x-icon",
                    "txt" | "md" => "text/plain; charset=utf-8",
                    "pdf" => "application/pdf",
                    "woff2" => "font/woff2",
                    "mp4" => "video/mp4",
                    _ => "application/octet-stream",
                };
                match std::fs::read(&path) {
                    // Sandboxed: a published page runs in its own opaque origin, so it can never
                    // read the phone app's sign-in on this same address.
                    Ok(b) => Resp::bytes(ctype, b).with_header("Content-Security-Policy", "sandbox allow-scripts allow-forms allow-popups allow-modals allow-downloads"),
                    Err(_) => Resp::error(404, "找不到這個網頁"),
                }
            }
            ("GET", ["relay", "host", "peers"]) => {
                let Some(me) = self.host(&req) else { return Resp::error(401, "這台電腦的金鑰無效") };
                let d = self.data.lock().unwrap();
                Resp::json(&json!({"hosts": d.hosts.iter().filter(|h| h.user == me.user).map(|h| self.host_json(h)).collect::<Vec<_>>()}))
            }
            // Signing out of the app unregisters this machine: its own key removes its own
            // registration, so a key left on a borrowed computer dies with the sign-out.
            ("POST", ["relay", "host", "remove"]) => {
                let Some(h) = self.host(&req) else { return Resp::error(401, "這台電腦的金鑰無效") };
                {
                    let mut d = self.data.lock().unwrap();
                    d.hosts.retain(|x| x.id != h.id);
                    self.save(&d);
                }
                // Its open connection is closed, so it hears "the key is gone" at once.
                if let Some(s) = self.sockets.lock().unwrap().remove(&h.id) {
                    let _ = s.tx.send(wire::pack(&json!({"t": "bye"}), &[]));
                }
                Resp::json(&json!({"ok": true}))
            }
            ("PUT", ["relay", "backup"]) => {
                let Some(h) = self.host(&req) else { return Resp::error(401, "這台電腦的金鑰無效") };
                if req.body.len() > 200 * 1024 * 1024 {
                    return Resp::error(413, "備份太大了（最多 200 MB）");
                }
                let path = self.dir.join("backups").join(format!("{}.bin", httpd::sha256_hex(&h.user)));
                let tmp = path.with_extension("tmp");
                match std::fs::write(&tmp, &req.body).and_then(|_| std::fs::rename(&tmp, &path)) {
                    Ok(()) => Resp::json(&json!({"ok": true, "bytes": req.body.len()})),
                    Err(e) => Resp::error(500, &e.to_string()),
                }
            }
            ("GET", ["relay", "backup"]) => {
                let Some(user) = self.session(&req) else { return Resp::error(401, "請先登入") };
                match std::fs::read(self.dir.join("backups").join(format!("{}.bin", httpd::sha256_hex(&user)))) {
                    Ok(b) => Resp::bytes("application/octet-stream", b),
                    Err(_) => Resp::error(404, "還沒有備份"),
                }
            }
            // 裝置連結: one AnyBot asks another AnyBot of the same account.
            ("POST", ["relay", "hosts", target, "api", "device", "run"]) => {
                let Some(me) = self.host(&req) else { return Resp::error(401, "這台電腦的金鑰無效") };
                let same = self.data.lock().unwrap().hosts.iter().any(|h| h.id == *target && h.user == me.user);
                if !same || *target == me.id {
                    return Resp::error(403, "不是同一個帳號的電腦");
                }
                self.forward_raw(target, "POST", "/api/device/run", &HashMap::new(), &req.body, false, (&me.user, "device"), false, &me.id).into_resp()
            }
            // 大腦: one AnyBot of the account shows (and edits) another one's bots and chats.
            (m, ["relay", "hosts", target, "api", "sync", rest @ ..]) => {
                let Some(me) = self.host(&req) else { return Resp::error(401, "這台電腦的金鑰無效") };
                let same = self.data.lock().unwrap().hosts.iter().any(|h| h.id == *target && h.user == me.user);
                if !same || *target == me.id {
                    return Resp::error(403, "不是同一個帳號的電腦");
                }
                let path = if rest.is_empty() { "/api/sync".to_string() } else { format!("/api/sync/{}", rest.join("/")) };
                self.forward_raw(target, m, &path, &req.query, &req.body, false, (&me.user, "mirror"), false, &me.id).into_resp()
            }
            // The phone, through to one of the account's AnyBots.
            (m, ["h", host, rest @ ..]) => {
                let Some(user) = self.session(&req) else { return Resp::error(401, "請先登入") };
                let role = self.data.lock().unwrap().hosts.iter().find(|h| h.id == *host).and_then(|h| h.role_of(&user));
                let Some(role) = role else { return Resp::error(404, "找不到這台 AnyBot") };
                if rest.first() != Some(&"api") || rest.get(1) == Some(&"device") {
                    return Resp::error(404, "沒有這個網址");
                }
                let mut q = req.query.clone();
                q.remove("t");
                self.forward(host, m, &format!("/{}", rest.join("/")), &q, &req.body, (&user, &role))
            }
            _ => Resp::error(404, "沒有這個網址"),
        }
    }

    /// Register an AnyBot of `user`. The same name again replaces its old key (keeping who it's
    /// shared with).
    fn add_host(&self, user: &str, name: &str) -> (Host, String) {
        let token = httpd::random_hex(32);
        let mut h = Host { id: httpd::random_hex(8), name: name.to_string(), user: user.to_string(), token_hash: httpd::sha256_hex(&token), created: now(), last_seen: 0, members: vec![] };
        let mut d = self.data.lock().unwrap();
        if let Some(old) = d.hosts.iter().find(|x| x.user == user && x.name == name) {
            h.members = old.members.clone();
        }
        d.hosts.retain(|x| !(x.user == user && x.name == name));
        d.hosts.push(h.clone());
        self.save(&d);
        (h, token)
    }

    fn host_json(&self, h: &Host) -> Value {
        let online = self.online(&h.id);
        json!({"id": h.id, "name": h.name, "online": online, "uses": self.uses.lock().unwrap().get(&h.id).cloned().unwrap_or_default(), "hands": self.hands.lock().unwrap().contains(&h.id), "last_seen_text": if online { "在線上".to_string() } else if h.last_seen > 0 { chrono::DateTime::from_timestamp(h.last_seen, 0).map(|t| t.format("%m/%d %H:%M UTC").to_string()).unwrap_or_default() } else { "從未連線".into() }})
    }

    fn signup(&self, req: &Req) -> Resp {
        let v = req.json();
        let user = v["user"].as_str().unwrap_or("").trim().to_lowercase();
        let pw = v["password"].as_str().unwrap_or("");
        if user.len() < 2 || user.len() > 64 || !user.chars().all(|c| c.is_ascii_alphanumeric() || "._@-".contains(c)) {
            return Resp::error(400, "帳號只能用英文字母、數字和 . _ @ -（2–64 個字）");
        }
        if pw.chars().count() < 8 {
            return Resp::error(400, "密碼至少 8 個字");
        }
        let mut d = self.data.lock().unwrap();
        if !d.users.is_empty() && (self.invite.is_empty() || v["invite"].as_str() != Some(self.invite.as_str())) {
            return Resp::error(403, if self.invite.is_empty() { "這台伺服器不開放註冊" } else { "邀請碼不對" });
        }
        if d.users.iter().any(|u| u.name == user) {
            return Resp::error(409, "這個帳號已經有人用了");
        }
        d.users.push(User { name: user, pass: httpd::hash_password(pw), created: now(), ..Default::default() });
        self.save(&d);
        Resp::json(&json!({"ok": true}))
    }

    fn login(&self, req: &Req) -> Resp {
        let v = req.json();
        let user = v["user"].as_str().unwrap_or("").trim().to_lowercase();
        // Rate-limit by the real peer. A client writes its own first X-Forwarded-For entry, so
        // taking the first value lets anyone reset the counter by changing the header. Only when
        // RELAY_TRUST_PROXY=1 (behind a reverse proxy) use the proxy's appended (last) entry.
        let ip = match (std::env::var("RELAY_TRUST_PROXY").is_ok(), req.header("x-forwarded-for")) {
            (true, xff) if !xff.is_empty() => xff.rsplit(',').next().unwrap_or(xff).trim().to_string(),
            _ => req.peer.clone(),
        };
        let key = format!("{ip}|{user}");
        if self.too_many_fails(&key) {
            return Resp::error(429, "密碼錯太多次，請 5 分鐘後再試");
        }
        let stored = self.data.lock().unwrap().users.iter().find(|u| u.name == user).map(|u| u.pass.clone());
        let ok = stored.map(|p| httpd::check_password(v["password"].as_str().unwrap_or(""), &p)).unwrap_or(false);
        if !ok {
            self.failed(&key);
            std::thread::sleep(Duration::from_millis(500));
            return Resp::error(401, "帳號或密碼不對");
        }
        // The password is the first factor. With 2FA on, it buys only a five-minute
        // challenge: the session comes after the code (or one of the recovery codes).
        let totp_on = self.data.lock().unwrap().users.iter().find(|u| u.name == user).map(|u| !u.totp.is_empty()).unwrap_or(false);
        if totp_on {
            let ch = httpd::random_hex(16);
            let device = v["device"].as_str().unwrap_or("").chars().take(40).collect();
            let mut c = self.challenges.lock().unwrap();
            c.retain(|_, (_, _, made)| made.elapsed() < Duration::from_secs(600));
            c.insert(ch.clone(), (user, device, Instant::now()));
            return Resp::json(&json!({"need_2fa": true, "challenge": ch}));
        }
        Resp::json(&json!({"token": self.issue_session(&user, &v["device"].as_str().unwrap_or("").chars().take(40).collect::<String>())}))
    }

    /// A signed-in session (the last act of a good login, with the code or without 2FA).
    fn issue_session(&self, user: &str, device: &str) -> String {
        let token = httpd::random_hex(32);
        let mut d = self.data.lock().unwrap();
        d.sessions.push(Session { hash: httpd::sha256_hex(&token), user: user.to_string(), device: device.to_string(), created: now(), last: now() });
        let cutoff = now() - SESSION_DAYS * 86_400;
        d.sessions.retain(|s| s.last > cutoff);
        self.save(&d);
        token
    }

    fn peer_ip(req: &Req) -> String {
        match (std::env::var("RELAY_TRUST_PROXY").is_ok(), req.header("x-forwarded-for")) {
            (true, xff) if !xff.is_empty() => xff.rsplit(',').next().unwrap_or(xff).trim().to_string(),
            _ => req.peer.clone(),
        }
    }

    /// 2FA, one endpoint, two moments. With a session and a code: finish enrolling (the
    /// secret in the authenticator app proves the person has it). With a login challenge:
    /// the second step of signing in — a TOTP code, or one of the eight recovery codes,
    /// each usable once.
    fn verify_2fa(&self, req: &Req) -> Resp {
        let v = req.json();
        let code = v["code"].as_str().unwrap_or("").trim().to_string();
        let ch = v["challenge"].as_str().filter(|s| !s.is_empty()).map(|s| s.to_string());
        if let Some(ch) = ch {
            let Some((user, device, made)) = self.challenges.lock().unwrap().remove(&ch) else {
                return Resp::error(400, "這次登入已過期，請重新輸入密碼");
            };
            if made.elapsed() > Duration::from_secs(300) {
                return Resp::error(400, "這次登入已過期，請重新輸入密碼");
            }
            let key = format!("2fa|{}|{user}", Self::peer_ip(req));
            if self.too_many_fails(&key) {
                return Resp::error(429, "驗證碼錯太多次，請 5 分鐘後再試");
            }
            let mut d = self.data.lock().unwrap();
            let Some(u) = d.users.iter_mut().find(|u| u.name == user) else { return Resp::error(404, "帳號不存在") };
            if httpd::totp_check(&u.totp, &code) {
                drop(d);
                return Resp::json(&json!({"token": self.issue_session(&user, &device)}));
            }
            // A recovery code works once, then it's gone (the hash slot is emptied).
            let h = httpd::sha256_hex(&code);
            match u.recover.iter_mut().find(|rh| *rh == &h) {
                Some(slot) => {
                    *slot = String::new();
                    self.save(&d);
                    drop(d);
                    return Resp::json(&json!({"token": self.issue_session(&user, &device), "used_recovery": true}))
                }
                None => {
                    drop(d);
                    self.failed(&key);
                    std::thread::sleep(Duration::from_millis(300));
                    return Resp::error(401, "驗證碼不對");
                }
            }
        }
        // Finishing enrollment (needs the session the app is signed in with).
        let Some(user) = self.session(req) else { return Resp::error(401, "請先登入") };
        let mut d = self.data.lock().unwrap();
        let secret = d.users.iter().find(|u| u.name == user).map(|u| u.totp_pending.clone()).unwrap_or_default();
        if secret.is_empty() {
            return Resp::error(400, "沒有待確認的設定：先點「開始設定」取得QR Code");
        }
        if !httpd::totp_check(&secret, &code) {
            return Resp::error(401, "驗證碼不對：輸入 App 上現在的那個六位數");
        }
        if let Some(u) = d.users.iter_mut().find(|u| u.name == user) {
            u.totp = std::mem::take(&mut u.totp_pending);
        }
        self.save(&d);
        Resp::json(&json!({"ok": true}))
    }

    /// Start 2FA: a new secret (the app scans the QR) and eight recovery codes, shown once,
    /// stored hashed. 2FA is not on until verify_2fa has seen a real code.
    fn enroll_2fa(&self, req: &Req) -> Resp {
        let Some(user) = self.session(req) else { return Resp::error(401, "請先登入") };
        let secret = httpd::totp_new_secret();
        let codes: Vec<String> = (0..8).map(|_| httpd::random_hex(5)).collect();
        let mut d = self.data.lock().unwrap();
        let Some(u) = d.users.iter_mut().find(|u| u.name == user) else { return Resp::error(404, "帳號不存在") };
        u.totp_pending = secret.clone();
        u.recover = codes.iter().map(|c| httpd::sha256_hex(c)).collect();
        self.save(&d);
        Resp::json(&json!({"secret": secret, "uri": httpd::totp_uri(&secret, &user), "recovery_codes": codes}))
    }

    /// Turn 2FA off (proves the code with the second factor itself).
    fn disable_2fa(&self, req: &Req) -> Resp {
        let Some(user) = self.session(req) else { return Resp::error(401, "請先登入") };
        let code = req.json()["code"].as_str().unwrap_or("").to_string();
        let mut d = self.data.lock().unwrap();
        let Some(u) = d.users.iter_mut().find(|u| u.name == user) else { return Resp::error(404, "帳號不存在") };
        if u.totp.is_empty() {
            return Resp::json(&json!({"ok": true}));
        }
        if !httpd::totp_check(&u.totp, &code) {
            return Resp::error(401, "驗證碼不對");
        }
        u.totp.clear();
        u.totp_pending.clear();
        u.recover.clear();
        self.save(&d);
        Resp::json(&json!({"ok": true}))
    }

    /// One-page logout: every session of this account is gone, including this one.
    fn logout_everywhere(&self, req: &Req) -> Resp {
        let Some(user) = self.session(req) else { return Resp::error(401, "請先登入") };
        let mut d = self.data.lock().unwrap();
        d.sessions.retain(|s| s.user != user);
        self.save(&d);
        Resp::json(&json!({"ok": true}))
    }

    /// Remember when hosts were last seen (for "離線（上次…）").
    fn save_seen(&self) {
        let seen: Vec<(String, u64)> = self.seen.lock().unwrap().iter().map(|(k, v)| (k.clone(), v.elapsed().as_secs())).collect();
        let mut d = self.data.lock().unwrap();
        let mut changed = false;
        for (id, ago) in seen {
            if let Some(h) = d.hosts.iter_mut().find(|h| h.id == id) {
                let t = now() - ago as i64;
                if t - h.last_seen > 60 {
                    h.last_seen = t;
                    changed = true;
                }
            }
        }
        if changed {
            self.save(&d);
        }
    }
}

fn main() {
    let args: Vec<String> = std::env::args().collect();
    let arg = |name: &str| args.iter().position(|a| a == name).and_then(|i| args.get(i + 1)).cloned();
    let port: u16 = arg("--port").and_then(|p| p.parse().ok()).or_else(|| std::env::var("PORT").ok().and_then(|p| p.parse().ok())).unwrap_or(8080);
    let bind = arg("--bind").unwrap_or_else(|| "0.0.0.0".into());
    let dir = PathBuf::from(arg("--data").unwrap_or_else(|| "relay-data".into()));
    // Forgot the password: only someone on this computer can set a new one.
    if let Some(user) = arg("--reset-password") {
        std::process::exit(reset_password(&dir, &user.trim().to_lowercase(), port));
    }
    let relay = Arc::new(Relay::load(dir.clone()));
    let listener = std::net::TcpListener::bind((bind.as_str(), port)).unwrap_or_else(|e| {
        eprintln!("無法使用連接埠 {port}：{e}");
        std::process::exit(1)
    });
    eprintln!("AnyBot 帳號伺服器：http://{bind}:{port}（資料在 {}）", dir.display());
    if relay.data.lock().unwrap().users.is_empty() {
        eprintln!("還沒有帳號：第一個註冊的人會成為這台伺服器的使用者。");
    }
    let r2 = relay.clone();
    std::thread::spawn(move || loop {
        std::thread::sleep(Duration::from_secs(60));
        r2.save_seen();
    });
    let (r, w) = (relay.clone(), relay.clone());
    httpd::serve_upgrade(listener, Arc::new(move |req| r.handle(req)), Some(Arc::new(move |req, c| w.ws(req, c))), Arc::new(|| true));
}

/// `anybot-relay --data DIR --reset-password NAME`: a new password for an account (asked twice,
/// not shown). The server must be stopped first, or it would write its own copy back.
fn reset_password(dir: &std::path::Path, user: &str, port: u16) -> i32 {
    if std::net::TcpStream::connect_timeout(&std::net::SocketAddr::from(([127, 0, 0, 1], port)), Duration::from_millis(500)).is_ok() {
        eprintln!("帳號伺服器還在執行（埠 {port}）。先停掉它再重設密碼，之後再啟動。");
        return 1;
    }
    let relay = Relay::load(dir.to_path_buf());
    if !relay.data.lock().unwrap().users.iter().any(|u| u.name == user) {
        eprintln!("{} 裡沒有帳號「{user}」。", dir.display());
        return 1;
    }
    let ask = |prompt: &str| -> String {
        use std::io::Write;
        eprint!("{prompt}");
        let _ = std::io::stderr().flush();
        let hide = std::process::Command::new("stty").arg("-echo").stdin(std::process::Stdio::inherit()).stderr(std::process::Stdio::null()).status().map(|s| s.success()).unwrap_or(false);
        let mut s = String::new();
        let _ = std::io::stdin().read_line(&mut s);
        if hide {
            let _ = std::process::Command::new("stty").arg("echo").stdin(std::process::Stdio::inherit()).stderr(std::process::Stdio::null()).status();
        }
        eprintln!();
        s.trim_end_matches(['\n', '\r']).to_string()
    };
    let pw = ask(&format!("「{user}」的新密碼（至少 8 個字）："));
    if pw.chars().count() < 8 {
        eprintln!("密碼至少 8 個字，沒有改。");
        return 1;
    }
    if ask("再輸入一次：") != pw {
        eprintln!("兩次不一樣，沒有改。");
        return 1;
    }
    let mut d = relay.data.lock().unwrap();
    if let Some(u) = d.users.iter_mut().find(|u| u.name == user) {
        u.pass = httpd::hash_password(&pw);
    }
    // Phones signed in with the old password sign in again.
    d.sessions.retain(|s| s.user != user);
    relay.save(&d);
    eprintln!("已改好「{user}」的密碼。啟動帳號伺服器後，用新密碼登入。");
    eprintln!("（帳號備份是用舊密碼加密的，之後要先在電腦上按一次「立刻備份」。）");
    0
}

#[cfg(test)]
mod tests {
    use super::*;

    fn post(port: u16, path: &str, token: &str, body: Value) -> (u16, Value) {
        let agent: ureq::Agent = ureq::Agent::config_builder().http_status_as_error(false).build().into();
        let mut r = agent.post(&format!("http://127.0.0.1:{port}{path}")).header("Authorization", &format!("Bearer {token}")).send(body.to_string().as_bytes()).unwrap();
        let st = r.status().as_u16();
        (st, serde_json::from_str(&r.body_mut().read_to_string().unwrap_or_default()).unwrap_or(Value::Null))
    }

    fn get(port: u16, path: &str, token: &str) -> (u16, Vec<u8>) {
        let agent: ureq::Agent = ureq::Agent::config_builder().http_status_as_error(false).build().into();
        let mut r = agent.get(&format!("http://127.0.0.1:{port}{path}")).header("Authorization", &format!("Bearer {token}")).call().unwrap();
        (r.status().as_u16(), r.body_mut().read_to_vec().unwrap_or_default())
    }

    fn step_now() -> u64 {
        std::time::SystemTime::now().duration_since(std::time::UNIX_EPOCH).unwrap().as_secs() / 30
    }

    #[test]
    fn two_factor_codes_recovery_once_and_logout_everywhere() {
        let dir = std::env::temp_dir().join(format!("anybot-relay-2fa-{}", std::process::id()));
        let _ = std::fs::remove_dir_all(&dir);
        let relay = Arc::new(Relay::load(dir.clone()));
        let l = std::net::TcpListener::bind("127.0.0.1:0").unwrap();
        let port = l.local_addr().unwrap().port();
        let r = relay.clone();
        std::thread::spawn(move || httpd::serve(l, Arc::new(move |req| r.handle(req)), Arc::new(|| true)));

        post(port, "/relay/signup", "", json!({"user": "d2", "password": "12345678"}));
        let s = post(port, "/relay/login", "", json!({"user": "d2", "password": "12345678", "device": "phone"})).1["token"].as_str().unwrap().to_string();
        // Enroll: the app gets a secret (QR) and eight codes shown once; 2FA is on only
        // after a real code proves the app has it.
        let (st, v) = post(port, "/relay/2fa/enroll", &s, json!({}));
        assert_eq!(st, 200);
        let secret = v["secret"].as_str().unwrap().to_string();
        let recovery = v["recovery_codes"][0].as_str().unwrap().to_string();
        assert!(v["uri"].as_str().unwrap().starts_with("otpauth://totp/AnyBot:d2?secret="));
        assert_eq!(post(port, "/relay/2fa/verify", &s, json!({"code": "000000"})).0, 401);
        assert_eq!(post(port, "/relay/2fa/verify", &s, json!({"code": httpd::totp_code(&secret, step_now())})).0, 200);
        // Password alone no longer signs in: it buys a challenge, the code signs in.
        post(port, "/relay/logout", &s, json!({}));
        let (_, v) = post(port, "/relay/login", "", json!({"user": "d2", "password": "12345678", "device": "phone"}));
        assert_eq!(v["need_2fa"], true);
        let ch = v["challenge"].as_str().unwrap().to_string();
        let (st, v) = post(port, "/relay/2fa/verify", "", json!({"challenge": ch, "code": httpd::totp_code(&secret, step_now())}));
        assert_eq!(st, 200);
        let t1 = v["token"].as_str().unwrap().to_string();
        // Recovery code: in, once.
        let ch = post(port, "/relay/login", "", json!({"user": "d2", "password": "12345678"})).1["challenge"].as_str().unwrap().to_string();
        let (st, v) = post(port, "/relay/2fa/verify", "", json!({"challenge": ch, "code": recovery}));
        assert_eq!(st, 200);
        assert_eq!(v["used_recovery"], true);
        let t2 = v["token"].as_str().unwrap().to_string();
        let ch = post(port, "/relay/login", "", json!({"user": "d2", "password": "12345678"})).1["challenge"].as_str().unwrap().to_string();
        assert_eq!(post(port, "/relay/2fa/verify", "", json!({"challenge": ch, "code": recovery})).0, 401);
        // A stolen session, kicked from one page: every device is out.
        assert_eq!(get(port, "/relay/me", &t1).0, 200);
        post(port, "/relay/sessions/logout", &t2, json!({}));
        assert_eq!(get(port, "/relay/me", &t1).0, 401);
        assert_eq!(get(port, "/relay/me", &t2).0, 401);
        // Off again (proving the code — the login right after is still 2FA: 關掉也要驗證碼):
        // sign in with the code, disable with the code, and the password signs in by itself.
        let ch = post(port, "/relay/login", "", json!({"user": "d2", "password": "12345678"})).1["challenge"].as_str().unwrap().to_string();
        let s2 = post(port, "/relay/2fa/verify", "", json!({"challenge": ch, "code": httpd::totp_code(&secret, step_now())})).1["token"].as_str().unwrap().to_string();
        assert_eq!(post(port, "/relay/2fa/disable", &s2, json!({"code": httpd::totp_code(&secret, step_now())})).0, 200);
        let (_, v) = post(port, "/relay/login", "", json!({"user": "d2", "password": "12345678"}));
        assert!(v["token"].is_string());
    }

    #[test]
    fn account_host_and_phone_round_trip() {
        let dir = std::env::temp_dir().join(format!("anybot-relay-test-{}", std::process::id()));
        let _ = std::fs::remove_dir_all(&dir);
        let relay = Arc::new(Relay::load(dir.clone()));
        let l = std::net::TcpListener::bind("127.0.0.1:0").unwrap();
        let port = l.local_addr().unwrap().port();
        let r = relay.clone();
        std::thread::spawn(move || httpd::serve(l, Arc::new(move |req| r.handle(req)), Arc::new(|| true)));

        assert_eq!(post(port, "/relay/signup", "", json!({"user": "Daniel", "password": "12345678"})).0, 200);
        // Second sign-up without an invite: refused.
        assert_eq!(post(port, "/relay/signup", "", json!({"user": "eve", "password": "12345678"})).0, 403);
        assert_eq!(post(port, "/relay/login", "", json!({"user": "daniel", "password": "wrong-pw"})).0, 401);
        let (_, v) = post(port, "/relay/login", "", json!({"user": "daniel", "password": "12345678", "device": "Mac"}));
        let session = v["token"].as_str().unwrap().to_string();
        let (_, h) = post(port, "/relay/hosts", &session, json!({"name": "工作室 Mac"}));
        let (host_id, host_token) = (h["id"].as_str().unwrap().to_string(), h["token"].as_str().unwrap().to_string());

        // Offline host: the phone is told so.
        let (st, _) = get(port, &format!("/h/{host_id}/api/state"), &session);
        assert_eq!(st, 503);

        // The "AnyBot": answers whatever comes in with the path it got.
        let ht = host_token.clone();
        std::thread::spawn(move || loop {
            let (st, body) = get(port, "/relay/host/next", &ht);
            if st == 200 {
                let job: Value = serde_json::from_slice(&body).unwrap();
                let answer = json!({"status": 200, "ctype": "application/json", "body": base64::Engine::encode(&base64::engine::general_purpose::STANDARD, json!({"path": job["path"], "q": job["query"]}).to_string()), "headers": [["X-Size", "10x20"], ["Set-Cookie", "no"]]});
                post(port, &format!("/relay/host/reply/{}", job["id"].as_str().unwrap()), &ht, answer);
            }
        });
        std::thread::sleep(Duration::from_millis(300));
        let agent: ureq::Agent = ureq::Agent::config_builder().http_status_as_error(false).build().into();
        let mut resp = agent.get(&format!("http://127.0.0.1:{port}/h/{host_id}/api/state?since=3")).header("Authorization", &format!("Bearer {session}")).call().unwrap();
        assert_eq!(resp.status().as_u16(), 200);
        assert_eq!(resp.headers().get("x-size").unwrap(), "10x20");
        assert!(resp.headers().get("set-cookie").is_none());
        let v: Value = serde_json::from_str(&resp.body_mut().read_to_string().unwrap()).unwrap();
        assert_eq!(v["path"], "/api/state");
        assert_eq!(v["q"]["since"], "3");

        // The device endpoint is only for the account's other AnyBots, never the phone.
        assert_eq!(get(port, &format!("/h/{host_id}/api/device/run"), &session).0, 404);
        // Without a session: refused.
        assert_eq!(get(port, &format!("/h/{host_id}/api/state"), "nope").0, 401);

        // Backup: the host uploads, the person downloads after signing in.
        let agent2: ureq::Agent = ureq::Agent::config_builder().http_status_as_error(false).build().into();
        let st = agent2.put(&format!("http://127.0.0.1:{port}/relay/backup")).header("Authorization", &format!("Bearer {host_token}")).send(&[1u8, 2, 3][..]).unwrap().status().as_u16();
        assert_eq!(st, 200);
        assert_eq!(get(port, "/relay/backup", &session).1, vec![1, 2, 3]);
        let (_, me) = get(port, "/relay/me", &session);
        let me: Value = serde_json::from_slice(&me).unwrap();
        assert_eq!(me["hosts"][0]["online"], true);
        // Data survives a restart; tokens are stored hashed.
        let saved = std::fs::read_to_string(dir.join("data.json")).unwrap();
        assert!(!saved.contains(&session) && !saved.contains(&host_token) && !saved.contains("12345678"));
        let _ = std::fs::remove_dir_all(&dir);
    }

    #[test]
    fn team_members_get_their_role() {
        let dir = std::env::temp_dir().join(format!("anybot-relay-team-{}", std::process::id()));
        let _ = std::fs::remove_dir_all(&dir);
        let mut relay = Relay::load(dir.clone());
        relay.invite = "join-us".into();
        let relay = Arc::new(relay);
        let l = std::net::TcpListener::bind("127.0.0.1:0").unwrap();
        let port = l.local_addr().unwrap().port();
        let r = relay.clone();
        std::thread::spawn(move || httpd::serve(l, Arc::new(move |req| r.handle(req)), Arc::new(|| true)));
        post(port, "/relay/signup", "", json!({"user": "boss", "password": "12345678"}));
        assert_eq!(post(port, "/relay/signup", "", json!({"user": "amy", "password": "12345678", "invite": "nope"})).0, 403);
        assert_eq!(post(port, "/relay/signup", "", json!({"user": "amy", "password": "12345678", "invite": "join-us"})).0, 200);
        let boss = post(port, "/relay/login", "", json!({"user": "boss", "password": "12345678"})).1["token"].as_str().unwrap().to_string();
        let amy = post(port, "/relay/login", "", json!({"user": "amy", "password": "12345678"})).1["token"].as_str().unwrap().to_string();
        let h = post(port, "/relay/hosts", &boss, json!({"name": "辦公室"})).1;
        let (host_id, host_token) = (h["id"].as_str().unwrap().to_string(), h["token"].as_str().unwrap().to_string());
        // Not shared yet: Amy can't see it.
        assert_eq!(get(port, &format!("/h/{host_id}/api/state"), &amy).0, 404);
        assert_eq!(post(port, "/relay/host/members", &host_token, json!({"user": "ghost", "role": "member"})).0, 404);
        assert_eq!(post(port, "/relay/host/members", &host_token, json!({"user": "amy", "role": "member"})).0, 200);
        let me: Value = serde_json::from_slice(&get(port, "/relay/me", &amy).1).unwrap();
        assert_eq!(me["hosts"][0]["role"], "member");
        // The host hears who asked and in which role.
        let ht = host_token.clone();
        std::thread::spawn(move || loop {
            let (st, body) = get(port, "/relay/host/next", &ht);
            if st == 200 {
                let job: Value = serde_json::from_slice(&body).unwrap();
                let answer = json!({"status": 200, "ctype": "application/json", "body": base64::Engine::encode(&base64::engine::general_purpose::STANDARD, json!({"user": job["user"], "role": job["role"]}).to_string()), "headers": []});
                post(port, &format!("/relay/host/reply/{}", job["id"].as_str().unwrap()), &ht, answer);
            }
        });
        std::thread::sleep(Duration::from_millis(300));
        let v: Value = serde_json::from_slice(&get(port, &format!("/h/{host_id}/api/state"), &amy).1).unwrap();
        assert_eq!((v["user"].as_str(), v["role"].as_str()), (Some("amy"), Some("member")));
        let v: Value = serde_json::from_slice(&get(port, &format!("/h/{host_id}/api/state"), &boss).1).unwrap();
        assert_eq!(v["role"], "owner");
        // Removing ends access.
        post(port, "/relay/host/members", &host_token, json!({"user": "amy", "role": "remove"}));
        assert_eq!(get(port, &format!("/h/{host_id}/api/state"), &amy).0, 404);
        let _ = std::fs::remove_dir_all(&dir);
    }

    #[test]
    fn published_sites_are_sandboxed() {
        let dir = std::env::temp_dir().join(format!("anybot-relay-sites-{}", std::process::id()));
        let _ = std::fs::remove_dir_all(&dir);
        let relay = Arc::new(Relay::load(dir.clone()));
        let l = std::net::TcpListener::bind("127.0.0.1:0").unwrap();
        let port = l.local_addr().unwrap().port();
        let r = relay.clone();
        std::thread::spawn(move || httpd::serve(l, Arc::new(move |req| r.handle(req)), Arc::new(|| true)));
        post(port, "/relay/signup", "", json!({"user": "boss", "password": "12345678"}));
        let s = post(port, "/relay/login", "", json!({"user": "boss", "password": "12345678"})).1["token"].as_str().unwrap().to_string();
        let ht = post(port, "/relay/hosts", &s, json!({"name": "m"})).1["token"].as_str().unwrap().to_string();
        let b = |t: &str| base64::Engine::encode(&base64::engine::general_purpose::STANDARD, t);
        let (st, v) = post(port, "/relay/host/sites/trip", &ht, json!({"files": [{"path": "index.html", "data": b("<h1>台南</h1>")}, {"path": "css/a.css", "data": b("h1{}")}, {"path": "../evil", "data": b("x")}]}));
        assert_eq!((st, v["files"].as_u64()), (200, Some(2)));
        assert_eq!(v["path"], "/s/boss-trip/");
        let agent: ureq::Agent = ureq::Agent::config_builder().http_status_as_error(false).build().into();
        let mut r = agent.get(&format!("http://127.0.0.1:{port}/s/boss-trip/")).call().unwrap();
        assert!(r.headers().get("content-security-policy").unwrap().to_str().unwrap().starts_with("sandbox"));
        assert_eq!(r.body_mut().read_to_string().unwrap(), "<h1>台南</h1>");
        assert_eq!(get(port, "/s/boss-trip/css/a.css", "").0, 200);
        assert_eq!(get(port, "/s/boss-trip/../../data.json", "").0, 404);
        assert!(!dir.join("evil").exists() && !dir.join("sites").join("evil").exists());
        assert_eq!(post(port, "/relay/host/sites/Bad_Name", &ht, json!({})).0, 400);
        let _ = std::fs::remove_dir_all(&dir);
    }

    #[test]
    fn a_pairing_code_adds_a_computer_once() {
        let dir = std::env::temp_dir().join(format!("anybot-relay-pair-{}", std::process::id()));
        let _ = std::fs::remove_dir_all(&dir);
        let relay = Arc::new(Relay::load(dir.clone()));
        let l = std::net::TcpListener::bind("127.0.0.1:0").unwrap();
        let port = l.local_addr().unwrap().port();
        let r = relay.clone();
        std::thread::spawn(move || httpd::serve(l, Arc::new(move |req| r.handle(req)), Arc::new(|| true)));
        post(port, "/relay/signup", "", json!({"user": "amy", "password": "12345678"}));
        let (_, v) = post(port, "/relay/login", "", json!({"user": "amy", "password": "12345678", "device": "phone"}));
        let session = v["token"].as_str().unwrap().to_string();
        let (_, brain) = post(port, "/relay/hosts", &session, json!({"name": "雲端大腦"}));
        // Asked from the phone, for the brain.
        let (st, p) = post(port, "/relay/pair", &session, json!({"for": brain["id"]}));
        assert_eq!(st, 200);
        let code = p["code"].as_str().unwrap().to_string();
        assert_eq!(code.len(), 6);
        // Without signing in: no code.
        assert_eq!(post(port, "/relay/pair", "", json!({})).0, 401);
        let (st, h) = post(port, "/relay/pair/claim", "", json!({"code": format!("{} {}", &code[..3], &code[3..]), "name": "書房電腦"}));
        assert_eq!(st, 200, "{h}");
        assert_eq!(h["user"], "amy");
        assert_eq!(h["for_name"], "雲端大腦");
        // The new computer's key works.
        let (st, peers) = get(port, "/relay/host/peers", h["token"].as_str().unwrap());
        assert_eq!(st, 200);
        assert!(String::from_utf8_lossy(&peers).contains("雲端大腦"));
        // Once only; a wrong code is refused, and guessing is stopped.
        assert_eq!(post(port, "/relay/pair/claim", "", json!({"code": code, "name": "x"})).0, 404);
        for _ in 0..10 {
            post(port, "/relay/pair/claim", "", json!({"code": "000000", "name": "x"}));
        }
        assert_eq!(post(port, "/relay/pair/claim", "", json!({"code": "000000", "name": "x"})).0, 429);
        let _ = std::fs::remove_dir_all(&dir);
    }

    #[test]
    fn brain_mirror_link_answers_only_from_the_right_machine() {
        // One account with a brain, a hands machine, and a nosy third machine: the sync
        // channel (brain → relay → hands → relay → brain) must carry the right roles, must
        // take answers only from the machine a job went to, and a signing-out machine's key
        // must die with its registration.
        let dir = std::env::temp_dir().join(format!("anybot-relay-brain-{}", std::process::id()));
        let _ = std::fs::remove_dir_all(&dir);
        let relay = Arc::new(Relay::load(dir.clone()));
        let l = std::net::TcpListener::bind("127.0.0.1:0").unwrap();
        let port = l.local_addr().unwrap().port();
        let r = relay.clone();
        std::thread::spawn(move || httpd::serve(l, Arc::new(move |req| r.handle(req)), Arc::new(|| true)));
        post(port, "/relay/signup", "", json!({"user": "boss", "password": "12345678"}));
        let s = post(port, "/relay/login", "", json!({"user": "boss", "password": "12345678", "device": "phone"})).1["token"].as_str().unwrap().to_string();
        let hb = post(port, "/relay/hosts", &s, json!({"name": "大腦"})).1;
        let btoken = hb["token"].as_str().unwrap().to_string();
        let hh = post(port, "/relay/hosts", &s, json!({"name": "手腳"})).1;
        let (hid, htoken) = (hh["id"].as_str().unwrap().to_string(), hh["token"].as_str().unwrap().to_string());
        let hm = post(port, "/relay/hosts", &s, json!({"name": "旁聽"})).1;
        let mtoken = hm["token"].as_str().unwrap().to_string();

        // The hands machine: takes its jobs, lets the snooper see the id, answers a beat later.
        let (tx, rx) = channel::<String>();
        let ht = htoken.clone();
        std::thread::spawn(move || loop {
            let (st, body) = get(port, "/relay/host/next", &ht);
            if st == 200 {
                let job: Value = serde_json::from_slice(&body).unwrap();
                let id = job["id"].as_str().unwrap().to_string();
                let _ = tx.send(id.clone());
                std::thread::sleep(Duration::from_millis(400));
                let answer = json!({"status": 200, "ctype": "application/json", "body": base64::Engine::encode(&base64::engine::general_purpose::STANDARD, json!({"from": "hands", "role": job["role"], "path": job["path"]}).to_string()), "headers": []});
                post(port, &format!("/relay/host/reply/{id}"), &ht, answer);
            }
        });
        // The snooper answers with a lie as soon as it learns a job id.
        let sn = mtoken.clone();
        std::thread::spawn(move || {
            let id = rx.recv().unwrap();
            let lie = json!({"status": 200, "ctype": "application/json", "body": base64::Engine::encode(&base64::engine::general_purpose::STANDARD, json!({"from": "snooper"}).to_string()), "headers": []});
            post(port, &format!("/relay/host/reply/{id}"), &sn, lie);
        });
        std::thread::sleep(Duration::from_millis(200));

        // 大腦 syncs 手腳's chats: the answer must come from the hands, with the mirror role.
        let v: Value = serde_json::from_slice(&get(port, &format!("/relay/hosts/{hid}/api/sync?since=0"), &btoken).1).unwrap();
        assert_eq!((v["from"].as_str(), v["role"].as_str(), v["path"].as_str()), (Some("hands"), Some("mirror"), Some("/api/sync")), "the brain's sync answer must be the hands' own");
        // 裝置連結: running a tool on the other machine carries the device role.
        let v: Value = post(port, &format!("/relay/hosts/{hid}/api/device/run"), &btoken, json!({"tool": "shell", "args": "{}", "who": "大腦"})).1;
        assert_eq!((v["from"].as_str(), v["role"].as_str()), (Some("hands"), Some("device")));

        // Signing out: the machine's own key removes its registration; others can't reach it.
        assert_eq!(post(port, "/relay/host/remove", &mtoken, json!({})).0, 200);
        let hosts: Value = serde_json::from_slice(&get(port, "/relay/host/peers", &btoken).1).unwrap();
        assert_eq!(hosts["hosts"].as_array().map(|a| a.len()), Some(2));
        let _ = std::fs::remove_dir_all(&dir);
    }

    #[test]
    fn forged_forwarded_for_cannot_outrun_the_login_lock() {
        // The login throttle follows the real peer: rewriting the first X-Forwarded-For
        // entry on every try used to reset the counter, so passwords could be guessed forever.
        let dir = std::env::temp_dir().join(format!("anybot-relay-throttle-{}", std::process::id()));
        let _ = std::fs::remove_dir_all(&dir);
        let relay = Arc::new(Relay::load(dir.clone()));
        let l = std::net::TcpListener::bind("127.0.0.1:0").unwrap();
        let port = l.local_addr().unwrap().port();
        let r = relay.clone();
        std::thread::spawn(move || httpd::serve(l, Arc::new(move |req| r.handle(req)), Arc::new(|| true)));
        post(port, "/relay/signup", "", json!({"user": "boss", "password": "12345678"}));
        let agent: ureq::Agent = ureq::Agent::config_builder().http_status_as_error(false).build().into();
        for i in 0..8 {
            let mut rr = agent
                .post(&format!("http://127.0.0.1:{port}/relay/login"))
                .header("X-Forwarded-For", &format!("10.9.{i}.{i}"))
                .send(json!({"user": "boss", "password": "wrong-password"}).to_string().as_bytes())
                .unwrap();
            let _ = rr.body_mut().read_to_string();
        }
        // Every fake IP was new; the one real peer is locked out — even for the right password.
        let rr = agent
            .post(&format!("http://127.0.0.1:{port}/relay/login"))
            .header("X-Forwarded-For", "10.9.99.99")
            .send(json!({"user": "boss", "password": "12345678"}).to_string().as_bytes())
            .unwrap();
        assert_eq!(rr.status().as_u16(), 429);
        let _ = std::fs::remove_dir_all(&dir);
    }
}