1use 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
21pub const ZRQINIT: &[u8] = b"**\x18B00";
23pub const ZRINIT: &[u8] = b"**\x18B01";
25pub 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
28pub(crate) const TIMEOUT: Duration = Duration::from_secs(10);
30pub(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 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 pub files_queued: usize,
57 pub file_started: Instant,
58 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 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
85pub(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 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 fn leftover(&mut self) -> Vec<u8> {
133 std::mem::take(&mut self.pending)
134 }
135}
136
137pub const PARTIAL_SUFFIX: &str = ".part";
140
141pub struct Download {
142 rx: Receiver,
143 dir: PathBuf,
144 file: Option<(BufWriter<File>, PathBuf, String)>,
146 accept: Option<u32>,
149 saved: Vec<PathBuf>,
150 c: Common,
151}
152
153impl Download {
154 pub fn new(dir: PathBuf) -> Result<Download> {
155 let mut rx = Receiver::with_flow_control(0, true).map_err(|e| anyhow::anyhow!("{e}"))?;
158 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 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 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 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 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 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 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 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 pub fn feed(&mut self, data: &[u8], out: &mut Vec<u8>) {
489 self.parts().input(data);
490 self.pump(out);
491 }
492
493 pub fn tick(&mut self, now: Instant, out: &mut Vec<u8>) {
495 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 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 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 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
582pub 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
595pub(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#[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 pub fn strip(&mut self, data: &mut Vec<u8>) -> bool {
635 if Instant::now() > self.until {
636 return false;
637 }
638 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 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
676pub 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 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 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 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 let mut b = b"OO\r\nMenu: ".to_vec();
754 assert!(!f.strip(&mut b));
755 assert_eq!(b, b"\r\nMenu: ");
756 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}