Relay locks, rate-limits and answers by the real peer (proxy trust opt-in); phone page follows

mobile/index.html

@@ -390,7 +390,9 @@ async function pickHost() {
   const live = hosts.filter((h) => h.online);
   // Several brains on: the one this account's computers use is the one the chats are on.
   const used = live.filter((h) => sharers(h.id).length);
-  if (used.length === 1 && !(S.host && sharers(target(S.host)).length)) S.host = used[0].id;
+  // First visit picks the brain this account's computers use; a choice the person made (or the
+  // host stored from last time) is kept as it is.
+  if (!S.host && used.length === 1) S.host = used[0].id;
   if (S.host) {
     const t = target(S.host);
     if (byId[t] && byId[t].online) { S.host = t; localStorage.setItem("anybot.host", t); return start(); }

relay/src/main.rs

@@ -36,6 +36,9 @@ const ICON: &str = r##"<svg xmlns="http://www.w3.org/2000/svg" viewBox="0 0 96 9
 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.
@@ -186,7 +189,9 @@ struct Relay {
     queues: Mutex<HashMap<String, VecDeque<Value>>>,
     arrived: Condvar,
     /// Where each forwarded request's answer goes.
-    waiting: Mutex<HashMap<String, Sender<Answer>>>,
+    /// 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)>>,
@@ -301,15 +306,16 @@ impl Relay {
             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);
+        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(FORWARD_SECS)).ok() } else { None };
+            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 {
@@ -328,7 +334,7 @@ impl Relay {
         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(FORWARD_SECS - PICKUP_SECS));
+                r = rx.recv_timeout(Duration::from_secs(secs.max(PICKUP_SECS + 1) - PICKUP_SECS));
             }
         }
         self.waiting.lock().unwrap().remove(&id);
@@ -416,7 +422,7 @@ impl Relay {
         }
         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) {
+            if let Some((w, _)) = self.waiting.lock().unwrap().remove(&id) {
                 let _ = w.send(Answer::error(503, GONE));
             }
         }
@@ -427,11 +433,18 @@ impl Relay {
         let s = |k: &str| head[k].as_str().unwrap_or("").to_string();
         match head["t"].as_str().unwrap_or("") {
             "reply" => {
-                if let Some(w) = self.waiting.lock().unwrap().remove(&s("id")) {
+                // 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"));
@@ -620,10 +633,16 @@ impl Relay {
                 }
             }
             ("POST", ["relay", "host", "reply", id]) => {
-                if self.host(&req).is_none() {
+                let Some(h) = self.host(&req) else { return Resp::error(401, "這台電腦的金鑰無效") };
-                    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) = self.waiting.lock().unwrap().remove(*id) {
+                };
+                if let Some(tx) = tx {
                     let _ = tx.send(Answer::from_json(&req.json()));
                 }
                 Resp::json(&json!({"ok": true}))
@@ -724,8 +743,26 @@ impl Relay {
                 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)) {
@@ -820,7 +857,14 @@ impl Relay {
     fn login(&self, req: &Req) -> Resp {
         let v = req.json();
         let user = v["user"].as_str().unwrap_or("").trim().to_lowercase();
-        let key = format!("{}|{user}", req.header("x-forwarded-for").split(',').next().unwrap_or(&req.peer).trim());
+        // 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 分鐘後再試");
         }
@@ -1121,4 +1165,94 @@ mod tests {
         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);
+    }
 }