diff --git a/core/src/lan.rs b/core/src/lan.rs index 32dc218..9b559e8 100644 --- a/core/src/lan.rs +++ b/core/src/lan.rs @@ -1,13 +1,15 @@ -//! LAN transport — serverless same-network clipboard over mDNS + plain TCP. +//! LAN transport — serverless same-network sync + file transfer over mDNS + TCP. //! -//! Each engine advertises `_tether._tcp.local.` with TXT {room, from, source} -//! and browses for the same service. Peers in the *same room* connect directly -//! over TCP (lower `from` dials, to avoid double-connects) and exchange -//! newline-delimited JSON `Message`s. No server, no internet — just the LAN. +//! Each engine advertises `_tether._tcp.local.` (TXT room/from/source/name/ +//! account) and connects to peers in the same room — or sharing a non-empty +//! account — over a length-prefixed TCP framing: +//! [u32 BE len][u8 tag][payload] +//! tags: `H` hello · `J` JSON Message · `M`/`C`/`E` file frames (shared w/ RTC). +//! Lower `from` dials, to avoid double-connects. //! -//! Scope: room id is the only access control and the link is plaintext; fine -//! for a trusted LAN, a rustls layer keyed on a room secret is the follow-up. -//! iOS needs the multicast entitlement for mDNS — there this transport is inert. +//! Scope/caveats: room/account is the only access control and the link is +//! plaintext — fine for a trusted LAN; rustls keyed on a room secret is the +//! follow-up. iOS needs the multicast entitlement for mDNS (inert until then). use std::collections::HashMap; use std::net::SocketAddr; @@ -16,9 +18,9 @@ use std::sync::{Arc, Mutex as StdMutex}; use std::time::Instant; use mdns_sd::{ServiceDaemon, ServiceEvent, ServiceInfo}; -use serde_json::json; -use tokio::io::{AsyncBufReadExt, AsyncWriteExt, BufReader}; -use tokio::net::tcp::OwnedWriteHalf; +use serde_json::{json, Value}; +use tokio::io::{AsyncReadExt, AsyncWriteExt}; +use tokio::net::tcp::{OwnedReadHalf, OwnedWriteHalf}; use tokio::net::{TcpListener, TcpStream}; use tokio::runtime::Handle; use tokio::sync::Mutex; @@ -26,11 +28,14 @@ use tokio::sync::Mutex; use crate::{Config, DirectSet, Discovered, FileSink, Message, Peer, Sink, Status, Transport}; const SERVICE: &str = "_tether._tcp.local."; +type Writer = Arc>; // per-peer write half (lock per peer, not the map) pub(crate) struct LanTransport { cfg: Config, - peers: Arc>>, // peer `from` → its write half + peers: Arc>>, // peer `from` → its write half sink: StdMutex>, + files: StdMutex>, + incoming: Arc>>, // in-flight inbound files discovered: Discovered, // nearby devices (any room), for the engine's nearby() direct: DirectSet, // LAN-connected peers (a direct path; suppresses relay) } @@ -41,6 +46,8 @@ impl LanTransport { cfg, peers: Arc::new(Mutex::new(HashMap::new())), sink: StdMutex::new(None), + files: StdMutex::new(None), + incoming: Arc::new(Mutex::new(HashMap::new())), discovered, direct, } @@ -49,6 +56,31 @@ impl LanTransport { fn sink(&self) -> Option { self.sink.lock().unwrap().clone() } + + fn file_sink(&self) -> Option { + self.files.lock().unwrap().clone() + } +} + +async fn write_frame(w: &mut OwnedWriteHalf, tag: u8, payload: &[u8]) -> std::io::Result<()> { + let len = (1 + payload.len()) as u32; + w.write_all(&len.to_be_bytes()).await?; + w.write_all(&[tag]).await?; + w.write_all(payload).await +} + +async fn read_frame(r: &mut OwnedReadHalf) -> std::io::Result<(u8, Vec)> { + let mut lenb = [0u8; 4]; + r.read_exact(&mut lenb).await?; + let len = u32::from_be_bytes(lenb) as usize; + if len < 1 || len > 64 * 1024 * 1024 { + return Err(std::io::Error::new(std::io::ErrorKind::InvalidData, "bad frame len")); + } + let mut tag = [0u8; 1]; + r.read_exact(&mut tag).await?; + let mut payload = vec![0u8; len - 1]; + r.read_exact(&mut payload).await?; + Ok((tag[0], payload)) } impl Transport for LanTransport { @@ -62,9 +94,10 @@ impl Transport for LanTransport { running: Arc, sink: Sink, _status: Status, - _files: FileSink, + files: FileSink, ) { *self.sink.lock().unwrap() = Some(sink); + *self.files.lock().unwrap() = Some(files); let this = self; rt.spawn(async move { if let Err(e) = this.run(running).await { @@ -76,19 +109,33 @@ impl Transport for LanTransport { fn publish(&self, rt: &Handle, msg: Message) { let peers = self.peers.clone(); rt.spawn(async move { - let line = match serde_json::to_string(&msg) { - Ok(s) => format!("{s}\n"), + let body = match serde_json::to_vec(&msg) { + Ok(b) => b, Err(_) => return, }; - let mut map = peers.lock().await; - let mut dead = Vec::new(); - for (id, w) in map.iter_mut() { - if w.write_all(line.as_bytes()).await.is_err() { - dead.push(id.clone()); - } + let writers: Vec = peers.lock().await.values().cloned().collect(); + for w in writers { + let mut g = w.lock().await; + let _ = write_frame(&mut g, b'J', &body).await; } - for id in dead { - map.remove(&id); + }); + } + + fn send_file(&self, rt: &Handle, id: String, name: String, mime: String, data: Vec) { + let peers = self.peers.clone(); + rt.spawn(async move { + let writers: Vec = peers.lock().await.values().cloned().collect(); + if writers.is_empty() { + return; + } + let frames = crate::file_frames(&id, &name, &mime, &data); + for w in writers { + let mut g = w.lock().await; + for (tag, payload) in &frames { + if write_frame(&mut g, *tag, payload).await.is_err() { + break; + } + } } }); } @@ -101,7 +148,6 @@ impl LanTransport { .map_err(|e| e.to_string())?; let port = listener.local_addr().map_err(|e| e.to_string())?.port(); - // Advertise + browse over mDNS. let daemon = ServiceDaemon::new().map_err(|e| e.to_string())?; let instance = sanitize(&self.cfg.from); let props = [ @@ -147,8 +193,6 @@ impl LanTransport { if from.is_empty() || from == self.cfg.from { continue; } - // Record for the nearby list — every tether device on the - // network, regardless of room (the UI shows them all). let name = info .get_property_val_str("name") .filter(|s| !s.is_empty()) @@ -169,9 +213,7 @@ impl LanTransport { Instant::now(), ), ); - // Pair if same room OR same (non-empty) account — your own - // devices auto-connect on the LAN regardless of room. Lower - // id dials to avoid double-connects. + // Pair if same room OR same (non-empty) account; lower id dials. let same_account = !self.cfg.account.is_empty() && account == self.cfg.account; if (room != self.cfg.room && !same_account) @@ -182,7 +224,6 @@ impl LanTransport { if self.peers.lock().await.contains_key(&from) { continue; } - // Prefer IPv4; fall back to any resolved address. let ip = info .get_addresses_v4() .into_iter() @@ -204,41 +245,37 @@ impl LanTransport { Ok(()) } - /// Handle one connection (dialed or accepted): hello handshake, then read - /// `Message` lines into the sink. The write half is stored for `publish`. + /// Handle one connection: hello handshake, then read frames — JSON messages + /// to the sink, file frames reassembled to the file sink. async fn handle(self: &Arc, stream: TcpStream) { - let (rd, mut wr) = stream.into_split(); + let (mut rd, mut wr) = stream.into_split(); - let hello = format!( - "{}\n", - json!({ - "t": "hello", - "from": self.cfg.from, - "room": self.cfg.room, - "account": self.cfg.account, - }) - ); - if wr.write_all(hello.as_bytes()).await.is_err() { + let hello = json!({ + "from": self.cfg.from, "room": self.cfg.room, "account": self.cfg.account + }) + .to_string(); + if write_frame(&mut wr, b'H', hello.as_bytes()).await.is_err() { return; } - // Accept the connection if we share a room OR a non-empty account. - let mut lines = BufReader::new(rd).lines(); - let peer_from = match lines.next_line().await { - Ok(Some(l)) => match serde_json::from_str::(&l) { - Ok(v) if v["t"] == "hello" => { - let same_room = v["room"].as_str() == Some(self.cfg.room.as_str()); - let same_account = !self.cfg.account.is_empty() - && v["account"].as_str() == Some(self.cfg.account.as_str()); - if !same_room && !same_account { - return; - } - v["from"].as_str().unwrap_or("").to_string() - } - _ => return, - }, - _ => return, + // Accept if we share a room OR a non-empty account. + let (tag, payload) = match read_frame(&mut rd).await { + Ok(f) => f, + Err(_) => return, }; + if tag != b'H' { + return; + } + let Ok(v) = serde_json::from_slice::(&payload) else { + return; + }; + let same_room = v["room"].as_str() == Some(self.cfg.room.as_str()); + let same_account = + !self.cfg.account.is_empty() && v["account"].as_str() == Some(self.cfg.account.as_str()); + if !same_room && !same_account { + return; + } + let peer_from = v["from"].as_str().unwrap_or("").to_string(); if peer_from.is_empty() || peer_from == self.cfg.from { return; } @@ -247,18 +284,39 @@ impl LanTransport { if map.contains_key(&peer_from) { return; // already connected to this peer } - map.insert(peer_from.clone(), wr); + map.insert(peer_from.clone(), Arc::new(Mutex::new(wr))); } self.direct.lock().unwrap().insert(peer_from.clone()); // direct path up eprintln!("tethercore[lan]: peer {peer_from} connected"); - while let Ok(Some(line)) = lines.next_line().await { - if let Ok(m) = serde_json::from_str::(&line) { - if m.from != self.cfg.from { - if let Some(sink) = self.sink() { - sink(m); + loop { + let (tag, payload) = match read_frame(&mut rd).await { + Ok(f) => f, + Err(_) => break, + }; + match tag { + b'J' => { + if let Ok(m) = serde_json::from_slice::(&payload) { + if m.from != self.cfg.from { + if let Some(sink) = self.sink() { + sink(m); // engine routes + de-dups + } + } } } + b'M' | b'C' | b'E' => { + let done = { + let mut map = self.incoming.lock().await; + crate::apply_file_frame(&mut map, tag, &payload) + }; + if let Some((name, mime, bytes)) = done { + eprintln!("tethercore[lan]: received file {name:?} ({} bytes)", bytes.len()); + if let Some(fs) = self.file_sink() { + fs(name, mime, bytes); + } + } + } + _ => {} } } diff --git a/core/src/lib.rs b/core/src/lib.rs index aec3efa..ec58498 100644 --- a/core/src/lib.rs +++ b/core/src/lib.rs @@ -147,6 +147,71 @@ type Status = Arc; /// Called when a transport reassembles a complete inbound file (name, mime, bytes). type FileSink = Arc) + Send + Sync>; +/// A file being reassembled from inbound chunks (keyed by its id). Shared by the +/// RTC and LAN transports, which use the same M/C/E frame protocol. +pub(crate) struct IncomingFile { + pub name: String, + pub mime: String, + pub data: Vec, +} + +/// Build the frames for a file: M, then C<16-byte id>…, then +/// E<16-byte id>. Each is (tag, payload); the transport adds its own envelope +/// (RTC: tag-prefixed binary message; LAN: length-prefixed). +pub(crate) fn file_frames(id: &str, name: &str, mime: &str, data: &[u8]) -> Vec<(u8, Vec)> { + let mut out = Vec::new(); + let meta = serde_json::json!({ "id": id, "name": name, "mime": mime, "size": data.len() }).to_string(); + out.push((b'M', meta.into_bytes())); + let idb = id.as_bytes(); + for chunk in data.chunks(16 * 1024) { + let mut p = Vec::with_capacity(16 + chunk.len()); + p.extend_from_slice(idb); + p.extend_from_slice(chunk); + out.push((b'C', p)); + } + out.push((b'E', idb.to_vec())); + out +} + +/// Apply one inbound file frame to the in-progress map; returns the completed +/// (name, mime, data) on the terminating 'E' frame. +pub(crate) fn apply_file_frame( + map: &mut HashMap, + tag: u8, + rest: &[u8], +) -> Option<(String, String, Vec)> { + match tag { + b'M' => { + if let Ok(v) = serde_json::from_slice::(rest) { + let id = v["id"].as_str().unwrap_or("").to_string(); + if !id.is_empty() { + map.insert( + id, + IncomingFile { + name: v["name"].as_str().unwrap_or("file").to_string(), + mime: v["mime"].as_str().unwrap_or("application/octet-stream").to_string(), + data: Vec::new(), + }, + ); + } + } + None + } + b'C' if rest.len() >= 16 => { + let id = String::from_utf8_lossy(&rest[..16]).to_string(); + if let Some(f) = map.get_mut(&id) { + f.data.extend_from_slice(&rest[16..]); + } + None + } + b'E' if rest.len() >= 16 => { + let id = String::from_utf8_lossy(&rest[..16]).to_string(); + map.remove(&id).map(|f| (f.name, f.mime, f.data)) + } + _ => None, + } +} + /// A pluggable delivery channel. The engine owns one or more; today just SSE. /// RTC (WebRTC data channel) and BT (BLE GATT) implement the same trait. trait Transport: Send + Sync { @@ -238,7 +303,8 @@ impl Engine { let id = format!("{:016x}", n); let rt = self.rt.handle(); for t in &self.transports { - if t.name() == "rtc" { + // Direct transports only — files are never relayed over SSE. + if t.name() == "rtc" || t.name() == "lan" { t.send_file(rt, id.clone(), name.clone(), mime.clone(), data.clone()); } } @@ -337,8 +403,26 @@ impl Engine { }); let h_status = handler.clone(); let status: Status = Arc::new(move |c| h_status.on_status(c)); + // A file can arrive over both RTC and LAN — de-dup so on_file fires once. let h_file = handler.clone(); - let files: FileSink = Arc::new(move |name, mime, data| h_file.on_file(name, mime, data)); + let file_seen: Arc>> = Arc::new(Mutex::new(VecDeque::new())); + let files: FileSink = Arc::new(move |name: String, mime: String, data: Vec| { + use sha2::{Digest, Sha256}; + let mut h = Sha256::new(); + h.update(name.as_bytes()); + h.update(&data); + let key: String = h.finalize().iter().map(|b| format!("{b:02x}")).collect(); + { + let now = Instant::now(); + let mut seen = file_seen.lock().unwrap(); + seen.retain(|(_, t)| now.duration_since(*t) < Duration::from_secs(15)); + if seen.iter().any(|(k, _)| k == &key) { + return; + } + seen.push_back((key, now)); + } + h_file.on_file(name, mime, data); + }); for t in &self.transports { t.clone().start( diff --git a/core/src/rtc.rs b/core/src/rtc.rs index d57ea26..3df8307 100644 --- a/core/src/rtc.rs +++ b/core/src/rtc.rs @@ -40,13 +40,6 @@ struct Peer { offerer: bool, } -/// A file being reassembled from inbound chunks (keyed by its id). -struct IncomingFile { - name: String, - mime: String, - data: Vec, -} - pub(crate) struct RtcTransport { cfg: Config, http: reqwest::Client, @@ -54,7 +47,7 @@ pub(crate) struct RtcTransport { peers: Arc>>>, sink: StdMutex>, files: StdMutex>, - incoming: Arc>>, // in-flight inbound files + incoming: Arc>>, // in-flight inbound files ice: StdMutex>, // refreshed from /api/turn-cred at start direct: DirectSet, // peers with an open data channel present: PresentSet, // peers seen via presence chirps @@ -269,38 +262,15 @@ impl RtcTransport { let Some((&tag, rest)) = data.split_first() else { return; }; - match tag { - b'M' => { - if let Ok(v) = serde_json::from_slice::(rest) { - let id = v["id"].as_str().unwrap_or("").to_string(); - if !id.is_empty() { - self.incoming.lock().await.insert( - id, - IncomingFile { - name: v["name"].as_str().unwrap_or("file").to_string(), - mime: v["mime"].as_str().unwrap_or("application/octet-stream").to_string(), - data: Vec::new(), - }, - ); - } - } + let done = { + let mut map = self.incoming.lock().await; + crate::apply_file_frame(&mut map, tag, rest) + }; + if let Some((name, mime, bytes)) = done { + eprintln!("tethercore[rtc]: received file {name:?} ({} bytes)", bytes.len()); + if let Some(fs) = self.file_sink() { + fs(name, mime, bytes); } - b'C' if rest.len() >= 16 => { - let id = String::from_utf8_lossy(&rest[..16]).to_string(); - if let Some(f) = self.incoming.lock().await.get_mut(&id) { - f.data.extend_from_slice(&rest[16..]); - } - } - b'E' if rest.len() >= 16 => { - let id = String::from_utf8_lossy(&rest[..16]).to_string(); - if let Some(f) = self.incoming.lock().await.remove(&id) { - eprintln!("tethercore[rtc]: received file {:?} ({} bytes)", f.name, f.data.len()); - if let Some(fs) = self.file_sink() { - fs(f.name, f.mime, f.data); - } - } - } - _ => {} } } @@ -627,28 +597,17 @@ impl Transport for RtcTransport { if dcs.is_empty() { return; } - let meta = json!({ "id": id, "name": name, "mime": mime, "size": data.len() }).to_string(); - let mut mframe = Vec::with_capacity(1 + meta.len()); - mframe.push(b'M'); - mframe.extend_from_slice(meta.as_bytes()); - let idb = id.as_bytes(); // 16 ascii hex chars + // Each frame is sent as one binary message: tag byte + payload. + let frames = crate::file_frames(&id, &name, &mime, &data); for dc in dcs { - if dc.send(&Bytes::from(mframe.clone())).await.is_err() { - continue; - } - for chunk in data.chunks(16 * 1024) { - let mut c = Vec::with_capacity(1 + 16 + chunk.len()); - c.push(b'C'); - c.extend_from_slice(idb); - c.extend_from_slice(chunk); - if dc.send(&Bytes::from(c)).await.is_err() { + for (tag, payload) in &frames { + let mut f = Vec::with_capacity(1 + payload.len()); + f.push(*tag); + f.extend_from_slice(payload); + if dc.send(&Bytes::from(f)).await.is_err() { break; } } - let mut e = Vec::with_capacity(17); - e.push(b'E'); - e.extend_from_slice(idb); - let _ = dc.send(&Bytes::from(e)).await; } }); }