rightkit_send/
send.rs

1//! Sending: `prepare-upload` to a peer, then each accepted file as an upload.
2
3use super::error::{Error, peer_io};
4use super::net::{self, Wire};
5use super::proto::{
6    self, DeviceInfo, FileMeta, PrepareUploadRequest, PrepareUploadResponse, PulseExtra,
7};
8use std::collections::BTreeMap;
9use std::fs::File;
10use std::io::{self, Read, Write};
11use std::net::{IpAddr, TcpStream};
12use std::path::{Path, PathBuf};
13use std::sync::atomic::{AtomicBool, Ordering};
14use std::time::Duration;
15
16/// Where to send: a device found on the network.
17#[derive(Clone, Debug)]
18pub struct Peer {
19    pub ip: IpAddr,
20    pub port: u16,
21    pub https: bool,
22    pub fingerprint: String,
23    pub alias: String,
24}
25
26/// What the user chose to send.
27#[derive(Clone, Debug)]
28pub enum SendItem {
29    Path(PathBuf),
30    Text(String),
31}
32
33#[derive(Clone, Debug)]
34enum Source {
35    File(PathBuf),
36    Text(String),
37}
38
39#[derive(Clone, Debug)]
40pub struct Entry {
41    pub id: String,
42    /// Name as the receiver sees it; folders keep their structure ("Dir/a.txt").
43    pub name: String,
44    pub size: u64,
45    pub mime: String,
46    source: Source,
47}
48
49impl Entry {
50    /// Opens the entry's content for reading: the file from disk, or the message text.
51    ///
52    /// Each call returns a fresh reader positioned at the start. A file that cannot be opened
53    /// (moved, deleted, permission) returns the I/O error; nothing is cached, so a changed file is
54    /// read as it is now.
55    pub fn open_reader(&self) -> io::Result<Box<dyn Read + Send>> {
56        match &self.source {
57            Source::Text(text) => Ok(Box::new(io::Cursor::new(text.clone().into_bytes()))),
58            Source::File(path) => Ok(Box::new(File::open(path)?)),
59        }
60    }
61}
62
63#[derive(Clone, Copy, Debug, PartialEq, Eq)]
64pub enum Phase {
65    /// Waiting for the other device to accept.
66    Waiting,
67    Sending,
68    /// A message (text only) is on the other device's screen. No longer
69    /// reported early: a message waits for the other side's answer like a file.
70    Delivered,
71}
72
73#[derive(Clone, Debug)]
74pub struct Progress {
75    pub phase: Phase,
76    pub done: u64,
77    pub total: u64,
78    pub files_done: usize,
79    pub files_total: usize,
80    pub current: Option<String>,
81}
82
83#[derive(Clone, Copy, Debug, PartialEq, Eq)]
84pub enum Outcome {
85    Done,
86    Declined,
87    Cancelled,
88}
89
90const MAX_ENTRIES: usize = 20_000;
91
92/// The files behind the user's choice. Folders are walked (symbolic links
93/// inside them are skipped); a chosen path that is a link is followed.
94pub fn expand(items: &[SendItem]) -> Result<Vec<Entry>, Error> {
95    let mut entries: Vec<Entry> = Vec::new();
96    for item in items {
97        match item {
98            SendItem::Text(text) => {
99                if text.is_empty() {
100                    continue;
101                }
102                entries.push(Entry {
103                    id: format!("f{}", entries.len()),
104                    name: format!("{}.txt", proto::random_hex(4)),
105                    size: text.len() as u64,
106                    mime: "text/plain".to_string(),
107                    source: Source::Text(text.clone()),
108                });
109            }
110            SendItem::Path(path) => {
111                let metadata = std::fs::metadata(path).map_err(|e| Error::Unreadable {
112                    path: path.display().to_string(),
113                    detail: e.to_string(),
114                })?;
115                let name = path
116                    .file_name()
117                    .map(|n| n.to_string_lossy().into_owned())
118                    .unwrap_or_else(|| "file".to_string());
119                if metadata.is_dir() {
120                    walk(path, &name, &mut entries)?;
121                } else if metadata.is_file() {
122                    push_file(&mut entries, path, name, metadata.len());
123                } else {
124                    return Err(Error::NotFileOrFolder {
125                        path: path.display().to_string(),
126                    });
127                }
128            }
129        }
130        if entries.len() > MAX_ENTRIES {
131            return Err(Error::TooManyFiles { max: MAX_ENTRIES });
132        }
133    }
134    if entries.is_empty() {
135        return Err(Error::NothingToSend);
136    }
137    Ok(entries)
138}
139
140fn push_file(entries: &mut Vec<Entry>, path: &Path, name: String, size: u64) {
141    let mime = proto::mime_for(&name).to_string();
142    entries.push(Entry {
143        id: format!("f{}", entries.len()),
144        name,
145        size,
146        mime,
147        source: Source::File(path.to_path_buf()),
148    });
149}
150
151fn walk(directory: &Path, prefix: &str, entries: &mut Vec<Entry>) -> Result<(), Error> {
152    let mut stack = vec![(directory.to_path_buf(), prefix.to_string())];
153    while let Some((dir, relative)) = stack.pop() {
154        let listing = std::fs::read_dir(&dir).map_err(|e| Error::Unreadable {
155            path: dir.display().to_string(),
156            detail: e.to_string(),
157        })?;
158        for child in listing.flatten() {
159            let Ok(kind) = child.file_type() else {
160                continue;
161            };
162            let name = format!("{relative}/{}", child.file_name().to_string_lossy());
163            if kind.is_symlink() {
164                continue;
165            } else if kind.is_dir() {
166                stack.push((child.path(), name));
167            } else if kind.is_file() {
168                let size = child.metadata().map(|m| m.len()).unwrap_or(0);
169                push_file(entries, &child.path(), name, size);
170            }
171            if entries.len() > MAX_ENTRIES {
172                return Err(Error::TooManyFiles { max: MAX_ENTRIES });
173            }
174        }
175    }
176    Ok(())
177}
178
179fn exchange(
180    peer: &Peer,
181    target: &str,
182    json: Option<&[u8]>,
183    timeout: Duration,
184    register: &dyn Fn(Option<TcpStream>),
185    sent: &mut dyn FnMut(),
186) -> Result<net::Reply, Error> {
187    let mut wire = net::connect(peer.ip, peer.port, peer.https, &peer.fingerprint, timeout)
188        .map_err(|e| {
189            peer_io(&peer.alias, &e, |peer, detail| Error::PeerUnreachable {
190                peer,
191                detail,
192            })
193        })?;
194    register(wire.tcp().try_clone().ok());
195    let payload = json.unwrap_or(&[]);
196    let host = format!("{}:{}", peer.ip, peer.port);
197    let result = net::call(
198        &mut wire,
199        "POST",
200        &host,
201        target,
202        json.map(|_| "application/json"),
203        payload.len() as u64,
204        &mut |w| {
205            w.write_all(payload)?;
206            w.flush()?;
207            sent();
208            Ok(())
209        },
210    );
211    wire.finish();
212    register(None);
213    result.map_err(|e| {
214        peer_io(&peer.alias, &e, |peer, detail| Error::PeerLost {
215            peer,
216            detail,
217        })
218    })
219}
220
221fn query(session: &str, file: &str, token: &str) -> String {
222    format!(
223        "{}/upload?sessionId={}&fileId={}&token={}",
224        proto::API,
225        proto::url_encode(session),
226        proto::url_encode(file),
227        proto::url_encode(token)
228    )
229}
230
231/// Ask the peer to accept `entries`, and send them once it does. Blocks.
232/// `register` is told each connection's socket so another thread can break a
233/// wait by shutting it down; `on` hears progress after every chunk.
234pub fn deliver(
235    me: &DeviceInfo,
236    peer: &Peer,
237    entries: &[Entry],
238    clipboard: bool,
239    cancel: &AtomicBool,
240    register: &dyn Fn(Option<TcpStream>),
241    on: &mut dyn FnMut(Progress),
242) -> Result<Outcome, Error> {
243    // Sends go only over a verified TLS connection to a pinned identity.
244    if !peer.https {
245        return Err(Error::PeerPlaintext {
246            peer: peer.alias.clone(),
247        });
248    }
249    if !net::valid_pin(&peer.fingerprint) {
250        return Err(Error::InvalidIdentity {
251            peer: peer.alias.clone(),
252        });
253    }
254    let mut progress = Progress {
255        phase: Phase::Waiting,
256        done: 0,
257        total: entries.iter().map(|e| e.size).sum(),
258        files_done: 0,
259        files_total: entries.len(),
260        current: None,
261    };
262    on(progress.clone());
263
264    let mut files = BTreeMap::new();
265    for entry in entries {
266        files.insert(
267            entry.id.clone(),
268            FileMeta {
269                id: entry.id.clone(),
270                file_name: entry.name.clone(),
271                size: entry.size,
272                file_type: entry.mime.clone(),
273                sha256: None,
274                preview: match &entry.source {
275                    Source::Text(text) => Some(text.clone()),
276                    Source::File(_) => None,
277                },
278                metadata: None,
279            },
280        );
281    }
282    let mut info = me.clone();
283    info.announce = None;
284    let pulse = clipboard.then_some(PulseExtra { clipboard: true });
285    let body = serde_json::to_vec(&PrepareUploadRequest { info, files, pulse }).map_err(|e| {
286        Error::Internal {
287            detail: e.to_string(),
288        }
289    })?;
290    let prepare = format!("{}/prepare-upload", proto::API);
291    // The other person may take a while to say yes. A message counts as sent
292    // only once they have accepted it (or allow this device), like a file.
293    let reply = exchange(
294        peer,
295        &prepare,
296        Some(&body),
297        Duration::from_secs(190),
298        register,
299        &mut || {},
300    );
301    if cancel.load(Ordering::Relaxed) {
302        return Ok(Outcome::Cancelled);
303    }
304    let reply = reply?;
305    let session = match reply.status {
306        200 => serde_json::from_slice::<PrepareUploadResponse>(&reply.body).map_err(|_| {
307            Error::PeerReply {
308                peer: peer.alias.clone(),
309            }
310        })?,
311        204 => return Ok(Outcome::Done),
312        403 => return Ok(Outcome::Declined),
313        401 => {
314            return Err(Error::PeerNeedsPin {
315                peer: peer.alias.clone(),
316            });
317        }
318        409 => {
319            return Err(Error::PeerBusy {
320                peer: peer.alias.clone(),
321            });
322        }
323        413 => {
324            return Err(Error::PeerTooLarge {
325                peer: peer.alias.clone(),
326            });
327        }
328        429 => {
329            return Err(Error::PeerRateLimited {
330                peer: peer.alias.clone(),
331            });
332        }
333        other => {
334            return Err(Error::PeerAnswer {
335                peer: peer.alias.clone(),
336                status: other,
337            });
338        }
339    };
340    // A partial accept sends only the files it wants.
341    progress.files_total = entries
342        .iter()
343        .filter(|e| session.files.contains_key(&e.id))
344        .count();
345    progress.total = entries
346        .iter()
347        .filter(|e| session.files.contains_key(&e.id))
348        .map(|e| e.size)
349        .sum();
350    progress.phase = Phase::Sending;
351    on(progress.clone());
352
353    for entry in entries {
354        let Some(token) = session.files.get(&entry.id) else {
355            continue;
356        };
357        if cancel.load(Ordering::Relaxed) {
358            send_cancel(peer, &session.session_id);
359            return Ok(Outcome::Cancelled);
360        }
361        progress.current = Some(entry.name.clone());
362        on(progress.clone());
363        let target = query(&session.session_id, &entry.id, token);
364        let result = upload_one(peer, &target, entry, cancel, register, &mut progress, on);
365        match result {
366            Ok(()) => {
367                progress.files_done += 1;
368                on(progress.clone());
369            }
370            Err(error) => {
371                if cancel.load(Ordering::Relaxed) {
372                    send_cancel(peer, &session.session_id);
373                    return Ok(Outcome::Cancelled);
374                }
375                return Err(error);
376            }
377        }
378    }
379    Ok(Outcome::Done)
380}
381
382fn upload_one(
383    peer: &Peer,
384    target: &str,
385    entry: &Entry,
386    cancel: &AtomicBool,
387    register: &dyn Fn(Option<TcpStream>),
388    progress: &mut Progress,
389    on: &mut dyn FnMut(Progress),
390) -> Result<(), Error> {
391    let mut wire = net::connect(
392        peer.ip,
393        peer.port,
394        peer.https,
395        &peer.fingerprint,
396        Duration::from_secs(120),
397    )
398    .map_err(|e| {
399        peer_io(&peer.alias, &e, |peer, detail| Error::PeerUnreachable {
400            peer,
401            detail,
402        })
403    })?;
404    register(wire.tcp().try_clone().ok());
405    let host = format!("{}:{}", peer.ip, peer.port);
406    let result = net::call(
407        &mut wire,
408        "POST",
409        &host,
410        target,
411        Some("application/octet-stream"),
412        entry.size,
413        &mut |w: &mut Wire| match &entry.source {
414            Source::Text(text) => {
415                w.write_all(text.as_bytes())?;
416                progress.done += text.len() as u64;
417                on(progress.clone());
418                Ok(())
419            }
420            Source::File(path) => {
421                let mut file = File::open(path)?;
422                let mut buffer = vec![0u8; 64 * 1024];
423                let mut remaining = entry.size;
424                while remaining > 0 {
425                    if cancel.load(Ordering::Relaxed) {
426                        return Err(io::Error::new(io::ErrorKind::Interrupted, "cancelled"));
427                    }
428                    let want = remaining.min(buffer.len() as u64) as usize;
429                    let n = file.read(&mut buffer[..want])?;
430                    if n == 0 {
431                        return Err(io::Error::new(
432                            io::ErrorKind::UnexpectedEof,
433                            "the file got shorter",
434                        ));
435                    }
436                    w.write_all(&buffer[..n])?;
437                    remaining -= n as u64;
438                    progress.done += n as u64;
439                    on(progress.clone());
440                }
441                Ok(())
442            }
443        },
444    );
445    wire.finish();
446    register(None);
447    let reply = result.map_err(|e| {
448        peer_io(&peer.alias, &e, |_, detail| Error::FileStopped {
449            file: entry.name.clone(),
450            detail,
451        })
452    })?;
453    match reply.status {
454        200 | 204 => Ok(()),
455        403 => Err(Error::FileRefused {
456            peer: peer.alias.clone(),
457            file: entry.name.clone(),
458        }),
459        422 => Err(Error::FileDamaged {
460            peer: peer.alias.clone(),
461            file: entry.name.clone(),
462        }),
463        other => Err(Error::PeerAnswer {
464            peer: peer.alias.clone(),
465            status: other,
466        }),
467    }
468}
469
470/// Tell the peer the session is over. Best effort.
471fn send_cancel(peer: &Peer, session: &str) {
472    let target = format!(
473        "{}/cancel?sessionId={}",
474        proto::API,
475        proto::url_encode(session)
476    );
477    let _ = net::request_json(
478        peer.ip,
479        peer.port,
480        peer.https,
481        &peer.fingerprint,
482        "POST",
483        &target,
484        None,
485        Duration::from_secs(5),
486    );
487}
488
489#[cfg(test)]
490mod tests {
491    use super::*;
492
493    fn me() -> DeviceInfo {
494        DeviceInfo {
495            alias: "Me".into(),
496            version: "2.0".into(),
497            device_model: None,
498            device_type: None,
499            fingerprint: "f".repeat(64),
500            port: 53317,
501            protocol: "https".into(),
502            download: false,
503            announce: None,
504        }
505    }
506
507    #[test]
508    fn nothing_is_sent_over_plain_http() {
509        let peer = Peer {
510            ip: "127.0.0.1".parse().unwrap(),
511            port: 9,
512            https: false,
513            fingerprint: "a".repeat(64),
514            alias: "Plain".into(),
515        };
516        let entries = expand(&[SendItem::Text("hi".into())]).unwrap();
517        let cancel = AtomicBool::new(false);
518        let result = deliver(&me(), &peer, &entries, false, &cancel, &|_| {}, &mut |_| {});
519        assert_eq!(
520            result,
521            Err(Error::PeerPlaintext {
522                peer: "Plain".into()
523            })
524        );
525    }
526
527    #[test]
528    fn a_peer_without_a_valid_fingerprint_is_refused_before_connecting() {
529        let peer = Peer {
530            ip: "127.0.0.1".parse().unwrap(),
531            port: 9,
532            https: true,
533            fingerprint: "not-hex".into(),
534            alias: "Odd".into(),
535        };
536        let entries = expand(&[SendItem::Text("hi".into())]).unwrap();
537        let cancel = AtomicBool::new(false);
538        let result = deliver(&me(), &peer, &entries, false, &cancel, &|_| {}, &mut |_| {});
539        assert_eq!(result, Err(Error::InvalidIdentity { peer: "Odd".into() }));
540    }
541
542    #[test]
543    fn empty_selections_and_missing_paths_are_typed() {
544        assert!(matches!(expand(&[]), Err(Error::NothingToSend)));
545        assert!(matches!(
546            expand(&[SendItem::Path("/definitely/not/here/rightkit".into())]),
547            Err(Error::Unreadable { .. })
548        ));
549    }
550
551    #[test]
552    fn open_reader_reads_text_and_files_and_reports_missing_files() {
553        let entries = expand(&[SendItem::Text("hello".into())]).unwrap();
554        let mut body = String::new();
555        entries[0]
556            .open_reader()
557            .unwrap()
558            .read_to_string(&mut body)
559            .unwrap();
560        assert_eq!(body, "hello");
561
562        let dir = std::env::temp_dir().join(format!("rk-send-reader-{}", std::process::id()));
563        std::fs::create_dir_all(&dir).unwrap();
564        let path = dir.join("a.txt");
565        std::fs::write(&path, b"file body").unwrap();
566        let entries = expand(&[SendItem::Path(path.clone())]).unwrap();
567        let mut bytes = Vec::new();
568        entries[0]
569            .open_reader()
570            .unwrap()
571            .read_to_end(&mut bytes)
572            .unwrap();
573        assert_eq!(bytes, b"file body");
574
575        std::fs::remove_file(&path).unwrap();
576        assert!(entries[0].open_reader().is_err());
577        let _ = std::fs::remove_dir_all(&dir);
578    }
579}
rightkit_send/send.rs

rightkit_send/
send.rs
