1395 lines
76 KiB
Rust
1395 lines
76 KiB
Rust
// 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,
|
||
}
|
||
|
||
#[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)>>,
|
||
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()),
|
||
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", "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();
|
||
Resp::json(&json!({"user": user, "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() });
|
||
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, "帳號或密碼不對");
|
||
}
|
||
let token = httpd::random_hex(32);
|
||
let mut d = self.data.lock().unwrap();
|
||
d.sessions.push(Session { hash: httpd::sha256_hex(&token), user, device: v["device"].as_str().unwrap_or("").chars().take(40).collect(), created: now(), last: now() });
|
||
let cutoff = now() - SESSION_DAYS * 86_400;
|
||
d.sessions.retain(|s| s.last > cutoff);
|
||
self.save(&d);
|
||
Resp::json(&json!({"token": token}))
|
||
}
|
||
|
||
/// 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())
|
||
}
|
||
|
||
#[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);
|
||
}
|
||
}
|