Skip to main content

badterm_core/
transfer.rs

1//! File transfers: ZMODEM here, and XMODEM and YMODEM in [`crate::xmodem`].
2//!
3//! ZMODEM transfers drive the caller-owned state machines of the
4//! `zmodem2` crate with file I/O, progress tracking and timeouts.
5//!
6//! A [`Transfer`] is fed raw (post-telnet) bytes from the remote and appends
7//! the bytes it wants sent to an output buffer, so it works the same over
8//! telnet, raw TCP and SSH.
9
10use std::collections::VecDeque;
11use std::fs::{self, File};
12use std::io::{BufWriter, Read, Seek, SeekFrom, Write};
13use std::path::{Path, PathBuf};
14use std::time::{Duration, Instant, UNIX_EPOCH};
15
16use anyhow::{Context, Result, bail};
17use zmodem2::{Action, Event, FileInfo, Position, Receiver, Sender};
18
19pub use crate::xmodem::{XyProtocol, XyReceive, XySend};
20
21/// `ZPAD ZPAD ZDLE 'B' "00"`: the hex ZRQINIT header a sender (`sz`) opens with.
22pub const ZRQINIT: &[u8] = b"**\x18B00";
23/// `ZPAD ZPAD ZDLE 'B' "01"`: the hex ZRINIT header a waiting receiver (`rz`) sends.
24pub const ZRINIT: &[u8] = b"**\x18B01";
25/// Ten CANs then ten backspaces: makes the other end abandon a session.
26pub const CANCEL: &[u8] = b"\x18\x18\x18\x18\x18\x18\x18\x18\x18\x18\x08\x08\x08\x08\x08\x08\x08\x08\x08\x08";
27
28/// Wait this long for input before nudging the other side.
29pub(crate) const TIMEOUT: Duration = Duration::from_secs(10);
30/// Give up after this many consecutive timeouts.
31pub(crate) const MAX_TIMEOUTS: u32 = 6;
32pub(crate) const MAX_FILE_SIZE: u64 = u32::MAX as u64;
33
34#[derive(Clone, Copy, Debug, PartialEq, Eq)]
35pub enum Direction {
36    Download,
37    Upload,
38}
39
40#[derive(Clone, Debug, PartialEq, Eq)]
41pub enum Outcome {
42    /// Finished; lists the files saved (download) or sent (upload).
43    Completed(Vec<PathBuf>),
44    Cancelled,
45    Failed(String),
46}
47
48#[derive(Clone, Debug)]
49pub struct Progress {
50    pub direction: Direction,
51    pub file_name: String,
52    pub file_bytes: u64,
53    pub file_size: Option<u64>,
54    pub files_done: usize,
55    /// Files still waiting to be sent (uploads only).
56    pub files_queued: usize,
57    pub file_started: Instant,
58    /// Where a resumed download picked up, if it was resumed.
59    pub resumed_at: Option<u64>,
60}
61
62impl Progress {
63    pub(crate) fn new(direction: Direction) -> Progress {
64        Progress {
65            direction,
66            file_name: String::new(),
67            file_bytes: 0,
68            file_size: None,
69            files_done: 0,
70            files_queued: 0,
71            file_started: Instant::now(),
72            resumed_at: None,
73        }
74    }
75
76    /// Average speed of the current file in bytes per second (of what
77    /// came this time, for a resumed download).
78    pub fn rate(&self) -> f64 {
79        let secs = self.file_started.elapsed().as_secs_f64();
80        let bytes = self.file_bytes.saturating_sub(self.resumed_at.unwrap_or(0));
81        if secs > 0.0 { bytes as f64 / secs } else { 0.0 }
82    }
83}
84
85/// State shared by both directions.
86pub(crate) struct Common {
87    pub(crate) pending: Vec<u8>,
88    pub(crate) progress: Progress,
89    pub(crate) outcome: Option<Outcome>,
90    pub(crate) last_input: Instant,
91    pub(crate) timeouts: u32,
92    pub(crate) cancelled: bool,
93}
94
95impl Common {
96    pub(crate) fn new(direction: Direction) -> Common {
97        Common {
98            pending: Vec::new(),
99            progress: Progress::new(direction),
100            outcome: None,
101            last_input: Instant::now(),
102            timeouts: 0,
103            cancelled: false,
104        }
105    }
106
107    pub(crate) fn input(&mut self, data: &[u8]) {
108        self.pending.extend_from_slice(data);
109        self.last_input = Instant::now();
110        self.timeouts = 0;
111    }
112
113    /// True when the timeout elapsed; sets a failure once too many have.
114    fn timed_out(&mut self, now: Instant) -> bool {
115        if self.outcome.is_some() || now.duration_since(self.last_input) < TIMEOUT {
116            return false;
117        }
118        self.last_input = now;
119        self.timeouts += 1;
120        if self.timeouts >= MAX_TIMEOUTS {
121            self.outcome = Some(Outcome::Failed("the other side stopped responding".into()));
122        }
123        true
124    }
125
126    pub(crate) fn aborted(&mut self) {
127        let outcome = if self.cancelled { Outcome::Cancelled } else { Outcome::Failed("cancelled by the BBS".into()) };
128        self.outcome.get_or_insert(outcome);
129    }
130
131    /// Bytes left after the session ended, to hand back to the terminal.
132    fn leftover(&mut self) -> Vec<u8> {
133        std::mem::take(&mut self.pending)
134    }
135}
136
137/// Downloads are written under this suffix until they are complete, so an
138/// interrupted one can be resumed when the BBS sends the file again.
139pub const PARTIAL_SUFFIX: &str = ".part";
140
141pub struct Download {
142    rx: Receiver,
143    dir: PathBuf,
144    /// The file being written (`name.part`), and its name once complete.
145    file: Option<(BufWriter<File>, PathBuf, String)>,
146    /// Where to ask the sender to start the file just announced, once the
147    /// receiver has room to say so.
148    accept: Option<u32>,
149    saved: Vec<PathBuf>,
150    c: Common,
151}
152
153impl Download {
154    pub fn new(dir: PathBuf) -> Result<Download> {
155        // TCP and SSH are reliable and flow-controlled, so let the sender
156        // stream without per-buffer acknowledgements.
157        let mut rx = Receiver::with_flow_control(0, true).map_err(|e| anyhow::anyhow!("{e}"))?;
158        // Decide where each file starts: a partial one is resumed.
159        rx.set_manual_file_accept(true);
160        Ok(Download { rx, dir, file: None, accept: None, saved: Vec::new(), c: Common::new(Direction::Download) })
161    }
162
163    fn fail(&mut self, msg: String) {
164        self.c.outcome.get_or_insert(Outcome::Failed(msg));
165        let _ = self.rx.abort();
166    }
167
168    /// Open the file the sender announced, as `name.part`: appending to
169    /// one left by an interrupted download of it (when its size says there
170    /// is more to come), or new. Returns where the sender should start.
171    fn open(&mut self, wire_name: &[u8], size: Option<u64>) -> Option<u32> {
172        self.close();
173        let name = sanitize_file_name(wire_name);
174        let part = self.dir.join(format!("{name}{PARTIAL_SUFFIX}"));
175        let have = fs::metadata(&part).map(|m| m.len()).unwrap_or(0);
176        let resume = size.is_some_and(|s| have > 0 && have < s && have <= u32::MAX as u64);
177        let result = fs::create_dir_all(&self.dir)
178            .and_then(|()| if resume { fs::OpenOptions::new().append(true).open(&part) } else { File::create(&part) });
179        match result {
180            Ok(f) => {
181                let offset = if resume { have } else { 0 };
182                self.file = Some((BufWriter::new(f), part, name.clone()));
183                let p = &mut self.c.progress;
184                p.file_name = name;
185                p.file_size = size;
186                p.file_bytes = offset;
187                p.resumed_at = resume.then_some(offset);
188                p.file_started = Instant::now();
189                Some(offset as u32)
190            }
191            Err(e) => {
192                self.fail(format!("creating {} in {}: {e}", part.display(), self.dir.display()));
193                None
194            }
195        }
196    }
197
198    /// Finish writing the current file. A complete one gets its own name
199    /// (never replacing a file); an incomplete one stays as `name.part`.
200    fn close_file(&mut self, complete: bool) -> Option<PathBuf> {
201        let (mut f, part, name) = self.file.take()?;
202        if let Err(e) = f.flush() {
203            self.fail(format!("writing {}: {e}", part.display()));
204            return None;
205        }
206        drop(f);
207        if !complete {
208            return None;
209        }
210        match unique_path(&self.dir, &name).and_then(|dest| fs::rename(&part, &dest).map(|()| dest)) {
211            Ok(dest) => Some(dest),
212            Err(e) => {
213                self.fail(format!("saving {name}: {e}"));
214                None
215            }
216        }
217    }
218
219    fn close(&mut self) {
220        self.close_file(false);
221    }
222
223    fn pump(&mut self, out: &mut Vec<u8>) {
224        loop {
225            match self.rx.poll() {
226                Action::WriteWire(b) => {
227                    out.extend_from_slice(b);
228                    let n = b.len();
229                    self.rx.wire_written(n);
230                }
231                Action::WriteFile(b) => {
232                    let n = b.len();
233                    let written = match &mut self.file {
234                        Some((f, _, _)) => f.write_all(b),
235                        None => Ok(()),
236                    };
237                    if let Err(e) = written {
238                        self.fail(format!("writing {}: {e}", self.c.progress.file_name));
239                        continue;
240                    }
241                    self.c.progress.file_bytes += n as u64;
242                    if let Err(e) = self.rx.file_written(n) {
243                        log::debug!("zmodem file_written: {e}");
244                    }
245                }
246                Action::Event(Event::FileStarted(info)) => {
247                    let name = info.name.to_vec();
248                    let size = info.size.map(|s| u64::from(s.get()));
249                    self.accept = self.open(&name, size);
250                }
251                Action::Event(Event::FileCompleted) => {
252                    let path = self.close_file(true);
253                    self.saved.extend(path);
254                    self.c.progress.files_done += 1;
255                }
256                Action::Event(Event::SessionCompleted) => {
257                    self.close();
258                    self.c.outcome.get_or_insert(Outcome::Completed(self.saved.clone()));
259                }
260                Action::Event(Event::Aborted) => {
261                    self.close();
262                    self.c.aborted();
263                }
264                Action::Idle => {
265                    if let Some(offset) = self.accept {
266                        match self.rx.accept_file_at(offset) {
267                            Ok(()) => {
268                                self.accept = None;
269                                continue;
270                            }
271                            // Outgoing bytes first.
272                            Err(zmodem2::Error::Backpressure) => {}
273                            Err(e) => {
274                                self.accept = None;
275                                self.fail(format!("starting {}: {e}", self.c.progress.file_name));
276                            }
277                        }
278                    }
279                    if self.c.outcome.is_some() || self.c.pending.is_empty() {
280                        break;
281                    }
282                    match self.rx.submit_wire(&self.c.pending) {
283                        Ok(0) => break,
284                        Ok(n) => {
285                            self.c.pending.drain(..n);
286                        }
287                        Err(e) => {
288                            // The receiver has asked for a resend; drop the damaged input.
289                            log::debug!("zmodem receive error: {e}");
290                            self.c.pending.clear();
291                        }
292                    }
293                }
294                _ => break,
295            }
296        }
297    }
298}
299
300pub struct Upload {
301    tx: Sender,
302    queue: VecDeque<PathBuf>,
303    file: Option<(File, PathBuf)>,
304    sent: Vec<PathBuf>,
305    /// Start the next file (or finish) once the outgoing buffer drains.
306    advance: bool,
307    finishing: bool,
308    buf: Vec<u8>,
309    c: Common,
310}
311
312impl Upload {
313    pub fn new(paths: Vec<PathBuf>) -> Result<Upload> {
314        if paths.is_empty() {
315            bail!("no files to upload");
316        }
317        for p in &paths {
318            check_upload(p)?;
319        }
320        let mut tx = Sender::new().map_err(|e| anyhow::anyhow!("{e}"))?;
321        tx.set_streaming_window(usize::MAX);
322        let mut up = Upload {
323            tx,
324            queue: paths.into(),
325            file: None,
326            sent: Vec::new(),
327            advance: false,
328            finishing: false,
329            buf: Vec::new(),
330            c: Common::new(Direction::Upload),
331        };
332        up.start_next();
333        Ok(up)
334    }
335
336    /// Add a file to the batch. Fails once the last file has been sent.
337    pub fn enqueue(&mut self, path: PathBuf) -> Result<()> {
338        if self.finishing || self.c.outcome.is_some() {
339            bail!("the upload is already finishing");
340        }
341        check_upload(&path)?;
342        self.queue.push_back(path);
343        self.c.progress.files_queued = self.queue.len();
344        Ok(())
345    }
346
347    fn fail(&mut self, msg: String) {
348        self.c.outcome.get_or_insert(Outcome::Failed(msg));
349        self.tx.abort();
350    }
351
352    fn start_next(&mut self) {
353        let Some(path) = self.queue.pop_front() else {
354            self.finishing = true;
355            if let Err(e) = self.tx.finish() {
356                self.fail(format!("finishing upload: {e}"));
357            }
358            return;
359        };
360        self.c.progress.files_queued = self.queue.len();
361        let opened = File::open(&path).and_then(|f| f.metadata().map(|m| (f, m)));
362        let (f, meta) = match opened {
363            Ok(v) => v,
364            Err(e) => return self.fail(format!("opening {}: {e}", path.display())),
365        };
366        let name = path.file_name().map(|n| n.to_string_lossy().into_owned()).unwrap_or_else(|| "file".into());
367        let size = meta.len();
368        let mut info = FileInfo::new(name.as_bytes(), Some(Position::new(size as u32)));
369        if let Some(t) = meta.modified().ok().and_then(|t| t.duration_since(UNIX_EPOCH).ok()) {
370            info = info.with_modified(t.as_secs() as u32);
371        }
372        if let Err(e) = self.tx.start_file(info) {
373            return self.fail(format!("starting {name}: {e}"));
374        }
375        let p = &mut self.c.progress;
376        p.file_name = name;
377        p.file_size = Some(size);
378        p.file_bytes = 0;
379        p.file_started = Instant::now();
380        self.file = Some((f, path));
381    }
382
383    fn read_chunk(&mut self, offset: u64, max_len: usize) -> std::io::Result<usize> {
384        let Some((f, _)) = &mut self.file else { return Ok(0) };
385        self.buf.resize(max_len, 0);
386        f.seek(SeekFrom::Start(offset))?;
387        f.read(&mut self.buf)
388    }
389
390    fn pump(&mut self, out: &mut Vec<u8>) {
391        loop {
392            match self.tx.poll() {
393                Action::WriteWire(b) => {
394                    out.extend_from_slice(b);
395                    let n = b.len();
396                    self.tx.wire_written(n);
397                }
398                Action::ReadFile { offset, max_len } => {
399                    let offset = u64::from(offset.get());
400                    match self.read_chunk(offset, max_len) {
401                        Ok(0) => self.fail(format!("{} got shorter while sending", self.c.progress.file_name)),
402                        Ok(n) => {
403                            if let Err(e) = self.tx.submit_file(&self.buf[..n]) {
404                                log::debug!("zmodem submit_file: {e}");
405                            }
406                            self.c.progress.file_bytes = offset + n as u64;
407                        }
408                        Err(e) => self.fail(format!("reading {}: {e}", self.c.progress.file_name)),
409                    }
410                }
411                Action::Event(Event::FileCompleted) => {
412                    if let Some((_, path)) = self.file.take() {
413                        self.sent.push(path);
414                    }
415                    self.c.progress.files_done += 1;
416                    self.advance = true;
417                }
418                Action::Event(Event::SessionCompleted) => {
419                    self.c.outcome.get_or_insert(Outcome::Completed(self.sent.clone()));
420                }
421                Action::Event(Event::Aborted) => self.c.aborted(),
422                Action::Event(_) => {}
423                Action::Idle => {
424                    if self.advance && self.c.outcome.is_none() {
425                        self.advance = false;
426                        self.start_next();
427                        continue;
428                    }
429                    if self.c.outcome.is_some() || self.c.pending.is_empty() {
430                        break;
431                    }
432                    match self.tx.submit_wire(&self.c.pending) {
433                        Ok(0) => break,
434                        Ok(n) => {
435                            self.c.pending.drain(..n);
436                        }
437                        Err(e) => {
438                            log::debug!("zmodem send error: {e}");
439                            self.c.pending.clear();
440                        }
441                    }
442                }
443                _ => break,
444            }
445        }
446    }
447}
448
449pub(crate) fn check_upload(p: &Path) -> Result<()> {
450    let meta = fs::metadata(p).with_context(|| format!("{}", p.display()))?;
451    if !meta.is_file() {
452        bail!("{} is not a file", p.display());
453    }
454    if meta.len() > MAX_FILE_SIZE {
455        bail!("{} is larger than the 4 GiB limit", p.display());
456    }
457    Ok(())
458}
459
460pub enum Transfer {
461    Download(Download),
462    Upload(Upload),
463    /// XMODEM or YMODEM.
464    XyDownload(XyReceive),
465    XyUpload(XySend),
466}
467
468impl Transfer {
469    fn parts(&mut self) -> &mut Common {
470        match self {
471            Transfer::Download(d) => &mut d.c,
472            Transfer::Upload(u) => &mut u.c,
473            Transfer::XyDownload(d) => &mut d.c,
474            Transfer::XyUpload(u) => &mut u.c,
475        }
476    }
477
478    fn pump(&mut self, out: &mut Vec<u8>) {
479        match self {
480            Transfer::Download(d) => d.pump(out),
481            Transfer::Upload(u) => u.pump(out),
482            Transfer::XyDownload(d) => d.pump(out),
483            Transfer::XyUpload(u) => u.pump(out),
484        }
485    }
486
487    /// Feed bytes received from the remote; bytes to send are appended to `out`.
488    pub fn feed(&mut self, data: &[u8], out: &mut Vec<u8>) {
489        self.parts().input(data);
490        self.pump(out);
491    }
492
493    /// Call periodically to drive timeouts and retries.
494    pub fn tick(&mut self, now: Instant, out: &mut Vec<u8>) {
495        // X/YMODEM keep their own time: a receiver asks every few seconds.
496        match self {
497            Transfer::XyDownload(d) => {
498                d.tick(now, out);
499                return self.pump(out);
500            }
501            Transfer::XyUpload(u) => {
502                u.tick(now, out);
503                return self.pump(out);
504            }
505            _ => {}
506        }
507        if !self.parts().timed_out(now) {
508            return;
509        }
510        let gave_up = self.parts().outcome.is_some();
511        match self {
512            Transfer::Download(d) if gave_up => {
513                let _ = d.rx.abort();
514            }
515            Transfer::Download(d) => {
516                let _ = d.rx.timeout();
517            }
518            Transfer::Upload(u) if gave_up => u.tx.abort(),
519            Transfer::Upload(u) => {
520                let _ = u.tx.timeout();
521            }
522            Transfer::XyDownload(_) | Transfer::XyUpload(_) => {}
523        }
524        self.pump(out);
525    }
526
527    /// Abort the transfer, telling the other side to stop.
528    pub fn cancel(&mut self, out: &mut Vec<u8>) {
529        self.parts().cancelled = true;
530        match self {
531            Transfer::Download(d) => {
532                let _ = d.rx.abort();
533            }
534            Transfer::Upload(u) => u.tx.abort(),
535            Transfer::XyDownload(d) => return d.cancel(out),
536            Transfer::XyUpload(u) => return u.cancel(out),
537        }
538        self.pump(out);
539        // The abort is final whether or not the other side acknowledges it.
540        self.parts().outcome.get_or_insert(Outcome::Cancelled);
541    }
542
543    pub fn progress(&self) -> &Progress {
544        match self {
545            Transfer::Download(d) => &d.c.progress,
546            Transfer::Upload(u) => &u.c.progress,
547            Transfer::XyDownload(d) => d.progress(),
548            Transfer::XyUpload(u) => u.progress(),
549        }
550    }
551
552    pub fn is_finished(&self) -> bool {
553        match self {
554            Transfer::Download(d) => d.c.outcome.is_some(),
555            Transfer::Upload(u) => u.c.outcome.is_some(),
556            Transfer::XyDownload(d) => d.c.outcome.is_some(),
557            Transfer::XyUpload(u) => u.c.outcome.is_some(),
558        }
559    }
560
561    /// The result and any bytes received after the session ended, which
562    /// belong to the terminal.
563    pub fn finish(mut self) -> (Outcome, Vec<u8>) {
564        let c = self.parts();
565        let leftover = c.leftover();
566        let outcome = c.outcome.take().unwrap_or_else(|| Outcome::Failed("interrupted".into()));
567        if let (Transfer::Download(d), Outcome::Failed(_) | Outcome::Cancelled) = (&mut self, &outcome) {
568            d.close();
569        }
570        (outcome, leftover)
571    }
572
573    pub fn enqueue(&mut self, path: PathBuf) -> Result<()> {
574        match self {
575            Transfer::Upload(u) => u.enqueue(path),
576            Transfer::XyUpload(u) => u.enqueue(path),
577            Transfer::Download(_) | Transfer::XyDownload(_) => bail!("a download is in progress"),
578        }
579    }
580}
581
582/// Reduce a file name from the wire to a safe base name: no directories, no
583/// control characters, never empty, "." or "..".
584pub fn sanitize_file_name(wire: &[u8]) -> String {
585    let s = String::from_utf8_lossy(wire);
586    let base = s.rsplit(['/', '\\']).next().unwrap_or("");
587    let cleaned: String = base
588        .chars()
589        .map(|c| if c.is_control() || matches!(c, ':' | '*' | '?' | '"' | '<' | '>' | '|') { '_' } else { c })
590        .collect();
591    let cleaned = cleaned.trim().trim_start_matches('.').to_string();
592    if cleaned.is_empty() { "download".to_string() } else { cleaned }
593}
594
595/// `dir/name`, or `dir/name (1).ext`, ... if that already exists.
596pub(crate) fn unique_path(dir: &Path, name: &str) -> std::io::Result<PathBuf> {
597    let first = dir.join(name);
598    if !first.exists() {
599        return Ok(first);
600    }
601    let (stem, ext) = match name.rsplit_once('.') {
602        Some((s, e)) if !s.is_empty() => (s, format!(".{e}")),
603        _ => (name, String::new()),
604    };
605    for i in 1..10_000 {
606        let p = dir.join(format!("{stem} ({i}){ext}"));
607        if !p.exists() {
608            return Ok(p);
609        }
610    }
611    Err(std::io::Error::new(std::io::ErrorKind::AlreadyExists, format!("too many copies of {name}")))
612}
613
614/// Strips the tail end of a ZMODEM session from data that follows it: the
615/// CR LF (often with the high bit set) and XON closing the last hex header,
616/// and the "OO" ("over and out") that `sz` sends after its final ZFIN. These
617/// can arrive in later packets than the end of the session, so the filter
618/// stays armed across calls until real data shows up or it expires.
619#[derive(Debug)]
620pub struct TrailerFilter {
621    until: Instant,
622    o_left: u8,
623}
624
625impl TrailerFilter {
626    const WINDOW: Duration = Duration::from_secs(2);
627
628    pub fn new() -> TrailerFilter {
629        TrailerFilter { until: Instant::now() + Self::WINDOW, o_left: 2 }
630    }
631
632    /// Remove trailer bytes from the start of `data`. Returns false once the
633    /// filter is done and can be dropped.
634    pub fn strip(&mut self, data: &mut Vec<u8>) -> bool {
635        if Instant::now() > self.until {
636            return false;
637        }
638        // Where the trailer ends, if it ends within this chunk.
639        let mut end = None;
640        for (n, &b) in data.iter().enumerate() {
641            match b {
642                b'\r' | b'\n' | 0x8D | 0x8A | 0x11 | 0x13 => {}
643                b'O' if self.o_left > 0 => {
644                    self.o_left -= 1;
645                    if self.o_left == 0 {
646                        // "OO" is the very last thing a sender writes.
647                        end = Some(n + 1);
648                        break;
649                    }
650                }
651                _ => {
652                    end = Some(n);
653                    break;
654                }
655            }
656        }
657        match end {
658            Some(cut) => {
659                data.drain(..cut);
660                false
661            }
662            None => {
663                data.clear();
664                true
665            }
666        }
667    }
668}
669
670impl Default for TrailerFilter {
671    fn default() -> Self {
672        Self::new()
673    }
674}
675
676/// Find the first occurrence of `needle` in `hay`.
677pub fn find(hay: &[u8], needle: &[u8]) -> Option<usize> {
678    hay.windows(needle.len()).position(|w| w == needle)
679}
680
681#[cfg(test)]
682mod tests {
683    use super::*;
684
685    fn temp_dir(name: &str) -> PathBuf {
686        let d = std::env::temp_dir().join(format!("badterm-test-{name}-{}", std::process::id()));
687        let _ = fs::remove_dir_all(&d);
688        fs::create_dir_all(&d).unwrap();
689        d
690    }
691
692    /// Run an upload against a download in memory until both finish.
693    fn run(up: &mut Transfer, down: &mut Transfer) {
694        let (mut to_down, mut to_up) = (Vec::new(), Vec::new());
695        up.feed(&[], &mut to_down);
696        for _ in 0..100_000 {
697            if up.is_finished() && down.is_finished() {
698                return;
699            }
700            let data = std::mem::take(&mut to_down);
701            down.feed(&data, &mut to_up);
702            let data = std::mem::take(&mut to_up);
703            up.feed(&data, &mut to_down);
704        }
705        panic!("transfer did not finish");
706    }
707
708    #[test]
709    fn upload_to_download_round_trip() {
710        let src = temp_dir("src");
711        let dst = temp_dir("dst");
712        let mut contents = Vec::new();
713        for i in 0..300_000u32 {
714            contents.push((i.wrapping_mul(2_654_435_761) >> 13) as u8);
715        }
716        // Include bytes that need escaping on the wire.
717        contents.extend_from_slice(&[0x18, 0x11, 0x13, 0xFF, b'\r', 0]);
718        fs::write(src.join("big.bin"), &contents).unwrap();
719        fs::write(src.join("empty.txt"), b"").unwrap();
720        fs::write(dst.join("big.bin"), b"already here").unwrap();
721
722        let mut up = Transfer::Upload(Upload::new(vec![src.join("big.bin"), src.join("empty.txt")]).unwrap());
723        let mut down = Transfer::Download(Download::new(dst.clone()).unwrap());
724        run(&mut up, &mut down);
725
726        let (up_outcome, _) = up.finish();
727        let (down_outcome, _) = down.finish();
728        assert_eq!(up_outcome, Outcome::Completed(vec![src.join("big.bin"), src.join("empty.txt")]));
729        // The existing file was not overwritten.
730        assert_eq!(down_outcome, Outcome::Completed(vec![dst.join("big (1).bin"), dst.join("empty.txt")]));
731        assert_eq!(fs::read(dst.join("big (1).bin")).unwrap(), contents);
732        assert_eq!(fs::read(dst.join("big.bin")).unwrap(), b"already here");
733        assert_eq!(fs::read(dst.join("empty.txt")).unwrap(), b"");
734    }
735
736    #[test]
737    fn cancel_sends_abort() {
738        let dst = temp_dir("cancel");
739        let mut down = Transfer::Download(Download::new(dst).unwrap());
740        let mut out = Vec::new();
741        down.cancel(&mut out);
742        assert!(find(&out, &CANCEL[..10]).is_some());
743        assert_eq!(down.finish().0, Outcome::Cancelled);
744    }
745
746    #[test]
747    fn trailer_filter_spans_chunks() {
748        let mut f = TrailerFilter::new();
749        let mut a = b"\r\x8a".to_vec();
750        assert!(f.strip(&mut a));
751        assert!(a.is_empty());
752        // The BBS's own line break after "OO" is kept.
753        let mut b = b"OO\r\nMenu: ".to_vec();
754        assert!(!f.strip(&mut b));
755        assert_eq!(b, b"\r\nMenu: ");
756        // Only two Os are ever eaten.
757        let mut f = TrailerFilter::new();
758        let mut c = b"OOOps".to_vec();
759        f.strip(&mut c);
760        assert_eq!(c, b"Ops");
761    }
762
763    #[test]
764    fn sanitizes_names() {
765        assert_eq!(sanitize_file_name(b"../../etc/passwd"), "passwd");
766        assert_eq!(sanitize_file_name(b"C:\\DOS\\GAME.ZIP"), "GAME.ZIP");
767        assert_eq!(sanitize_file_name(b".."), "download");
768        assert_eq!(sanitize_file_name(b".hidden"), "hidden");
769        assert_eq!(sanitize_file_name(b"a\x1bb?.txt"), "a_b_.txt");
770    }
771}