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

rightkit_send/
send.rs
