use std::collections::VecDeque;
use std::fs::{self, File};
use std::io::{BufWriter, Read, Seek, SeekFrom, Write};
use std::path::{Path, PathBuf};
use std::time::{Duration, Instant, UNIX_EPOCH};
use anyhow::{Context, Result, bail};
use zmodem2::{Action, Event, FileInfo, Position, Receiver, Sender};
pub use crate::xmodem::{XyProtocol, XyReceive, XySend};
pub const ZRQINIT: &[u8] = b"**\x18B00";
pub const ZRINIT: &[u8] = b"**\x18B01";
pub const CANCEL: &[u8] = b"\x18\x18\x18\x18\x18\x18\x18\x18\x18\x18\x08\x08\x08\x08\x08\x08\x08\x08\x08\x08";
pub(crate) const TIMEOUT: Duration = Duration::from_secs(10);
pub(crate) const MAX_TIMEOUTS: u32 = 6;
pub(crate) const MAX_FILE_SIZE: u64 = u32::MAX as u64;
#[derive(Clone, Copy, Debug, PartialEq, Eq)]
pub enum Direction {
Download,
Upload,
}
#[derive(Clone, Debug, PartialEq, Eq)]
pub enum Outcome {
Completed(Vec<PathBuf>),
Cancelled,
Failed(String),
}
#[derive(Clone, Debug)]
pub struct Progress {
pub direction: Direction,
pub file_name: String,
pub file_bytes: u64,
pub file_size: Option<u64>,
pub files_done: usize,
pub files_queued: usize,
pub file_started: Instant,
pub resumed_at: Option<u64>,
}
impl Progress {
pub(crate) fn new(direction: Direction) -> Progress {
Progress {
direction,
file_name: String::new(),
file_bytes: 0,
file_size: None,
files_done: 0,
files_queued: 0,
file_started: Instant::now(),
resumed_at: None,
}
}
pub fn rate(&self) -> f64 {
let secs = self.file_started.elapsed().as_secs_f64();
let bytes = self.file_bytes.saturating_sub(self.resumed_at.unwrap_or(0));
if secs > 0.0 { bytes as f64 / secs } else { 0.0 }
}
}
pub(crate) struct Common {
pub(crate) pending: Vec<u8>,
pub(crate) progress: Progress,
pub(crate) outcome: Option<Outcome>,
pub(crate) last_input: Instant,
pub(crate) timeouts: u32,
pub(crate) cancelled: bool,
}
impl Common {
pub(crate) fn new(direction: Direction) -> Common {
Common {
pending: Vec::new(),
progress: Progress::new(direction),
outcome: None,
last_input: Instant::now(),
timeouts: 0,
cancelled: false,
}
}
pub(crate) fn input(&mut self, data: &[u8]) {
self.pending.extend_from_slice(data);
self.last_input = Instant::now();
self.timeouts = 0;
}
fn timed_out(&mut self, now: Instant) -> bool {
if self.outcome.is_some() || now.duration_since(self.last_input) < TIMEOUT {
return false;
}
self.last_input = now;
self.timeouts += 1;
if self.timeouts >= MAX_TIMEOUTS {
self.outcome = Some(Outcome::Failed("the other side stopped responding".into()));
}
true
}
pub(crate) fn aborted(&mut self) {
let outcome = if self.cancelled { Outcome::Cancelled } else { Outcome::Failed("cancelled by the BBS".into()) };
self.outcome.get_or_insert(outcome);
}
fn leftover(&mut self) -> Vec<u8> {
std::mem::take(&mut self.pending)
}
}
pub const PARTIAL_SUFFIX: &str = ".part";
pub struct Download {
rx: Receiver,
dir: PathBuf,
file: Option<(BufWriter<File>, PathBuf, String)>,
accept: Option<u32>,
saved: Vec<PathBuf>,
c: Common,
}
impl Download {
pub fn new(dir: PathBuf) -> Result<Download> {
let mut rx = Receiver::with_flow_control(0, true).map_err(|e| anyhow::anyhow!("{e}"))?;
rx.set_manual_file_accept(true);
Ok(Download { rx, dir, file: None, accept: None, saved: Vec::new(), c: Common::new(Direction::Download) })
}
fn fail(&mut self, msg: String) {
self.c.outcome.get_or_insert(Outcome::Failed(msg));
let _ = self.rx.abort();
}
fn open(&mut self, wire_name: &[u8], size: Option<u64>) -> Option<u32> {
self.close();
let name = sanitize_file_name(wire_name);
let part = self.dir.join(format!("{name}{PARTIAL_SUFFIX}"));
let have = fs::metadata(&part).map(|m| m.len()).unwrap_or(0);
let resume = size.is_some_and(|s| have > 0 && have < s && have <= u32::MAX as u64);
let result = fs::create_dir_all(&self.dir)
.and_then(|()| if resume { fs::OpenOptions::new().append(true).open(&part) } else { File::create(&part) });
match result {
Ok(f) => {
let offset = if resume { have } else { 0 };
self.file = Some((BufWriter::new(f), part, name.clone()));
let p = &mut self.c.progress;
p.file_name = name;
p.file_size = size;
p.file_bytes = offset;
p.resumed_at = resume.then_some(offset);
p.file_started = Instant::now();
Some(offset as u32)
}
Err(e) => {
self.fail(format!("creating {} in {}: {e}", part.display(), self.dir.display()));
None
}
}
}
fn close_file(&mut self, complete: bool) -> Option<PathBuf> {
let (mut f, part, name) = self.file.take()?;
if let Err(e) = f.flush() {
self.fail(format!("writing {}: {e}", part.display()));
return None;
}
drop(f);
if !complete {
return None;
}
match unique_path(&self.dir, &name).and_then(|dest| fs::rename(&part, &dest).map(|()| dest)) {
Ok(dest) => Some(dest),
Err(e) => {
self.fail(format!("saving {name}: {e}"));
None
}
}
}
fn close(&mut self) {
self.close_file(false);
}
fn pump(&mut self, out: &mut Vec<u8>) {
loop {
match self.rx.poll() {
Action::WriteWire(b) => {
out.extend_from_slice(b);
let n = b.len();
self.rx.wire_written(n);
}
Action::WriteFile(b) => {
let n = b.len();
let written = match &mut self.file {
Some((f, _, _)) => f.write_all(b),
None => Ok(()),
};
if let Err(e) = written {
self.fail(format!("writing {}: {e}", self.c.progress.file_name));
continue;
}
self.c.progress.file_bytes += n as u64;
if let Err(e) = self.rx.file_written(n) {
log::debug!("zmodem file_written: {e}");
}
}
Action::Event(Event::FileStarted(info)) => {
let name = info.name.to_vec();
let size = info.size.map(|s| u64::from(s.get()));
self.accept = self.open(&name, size);
}
Action::Event(Event::FileCompleted) => {
let path = self.close_file(true);
self.saved.extend(path);
self.c.progress.files_done += 1;
}
Action::Event(Event::SessionCompleted) => {
self.close();
self.c.outcome.get_or_insert(Outcome::Completed(self.saved.clone()));
}
Action::Event(Event::Aborted) => {
self.close();
self.c.aborted();
}
Action::Idle => {
if let Some(offset) = self.accept {
match self.rx.accept_file_at(offset) {
Ok(()) => {
self.accept = None;
continue;
}
Err(zmodem2::Error::Backpressure) => {}
Err(e) => {
self.accept = None;
self.fail(format!("starting {}: {e}", self.c.progress.file_name));
}
}
}
if self.c.outcome.is_some() || self.c.pending.is_empty() {
break;
}
match self.rx.submit_wire(&self.c.pending) {
Ok(0) => break,
Ok(n) => {
self.c.pending.drain(..n);
}
Err(e) => {
log::debug!("zmodem receive error: {e}");
self.c.pending.clear();
}
}
}
_ => break,
}
}
}
}
pub struct Upload {
tx: Sender,
queue: VecDeque<PathBuf>,
file: Option<(File, PathBuf)>,
sent: Vec<PathBuf>,
advance: bool,
finishing: bool,
buf: Vec<u8>,
c: Common,
}
impl Upload {
pub fn new(paths: Vec<PathBuf>) -> Result<Upload> {
if paths.is_empty() {
bail!("no files to upload");
}
for p in &paths {
check_upload(p)?;
}
let mut tx = Sender::new().map_err(|e| anyhow::anyhow!("{e}"))?;
tx.set_streaming_window(usize::MAX);
let mut up = Upload {
tx,
queue: paths.into(),
file: None,
sent: Vec::new(),
advance: false,
finishing: false,
buf: Vec::new(),
c: Common::new(Direction::Upload),
};
up.start_next();
Ok(up)
}
pub fn enqueue(&mut self, path: PathBuf) -> Result<()> {
if self.finishing || self.c.outcome.is_some() {
bail!("the upload is already finishing");
}
check_upload(&path)?;
self.queue.push_back(path);
self.c.progress.files_queued = self.queue.len();
Ok(())
}
fn fail(&mut self, msg: String) {
self.c.outcome.get_or_insert(Outcome::Failed(msg));
self.tx.abort();
}
fn start_next(&mut self) {
let Some(path) = self.queue.pop_front() else {
self.finishing = true;
if let Err(e) = self.tx.finish() {
self.fail(format!("finishing upload: {e}"));
}
return;
};
self.c.progress.files_queued = self.queue.len();
let opened = File::open(&path).and_then(|f| f.metadata().map(|m| (f, m)));
let (f, meta) = match opened {
Ok(v) => v,
Err(e) => return self.fail(format!("opening {}: {e}", path.display())),
};
let name = path.file_name().map(|n| n.to_string_lossy().into_owned()).unwrap_or_else(|| "file".into());
let size = meta.len();
let mut info = FileInfo::new(name.as_bytes(), Some(Position::new(size as u32)));
if let Some(t) = meta.modified().ok().and_then(|t| t.duration_since(UNIX_EPOCH).ok()) {
info = info.with_modified(t.as_secs() as u32);
}
if let Err(e) = self.tx.start_file(info) {
return self.fail(format!("starting {name}: {e}"));
}
let p = &mut self.c.progress;
p.file_name = name;
p.file_size = Some(size);
p.file_bytes = 0;
p.file_started = Instant::now();
self.file = Some((f, path));
}
fn read_chunk(&mut self, offset: u64, max_len: usize) -> std::io::Result<usize> {
let Some((f, _)) = &mut self.file else { return Ok(0) };
self.buf.resize(max_len, 0);
f.seek(SeekFrom::Start(offset))?;
f.read(&mut self.buf)
}
fn pump(&mut self, out: &mut Vec<u8>) {
loop {
match self.tx.poll() {
Action::WriteWire(b) => {
out.extend_from_slice(b);
let n = b.len();
self.tx.wire_written(n);
}
Action::ReadFile { offset, max_len } => {
let offset = u64::from(offset.get());
match self.read_chunk(offset, max_len) {
Ok(0) => self.fail(format!("{} got shorter while sending", self.c.progress.file_name)),
Ok(n) => {
if let Err(e) = self.tx.submit_file(&self.buf[..n]) {
log::debug!("zmodem submit_file: {e}");
}
self.c.progress.file_bytes = offset + n as u64;
}
Err(e) => self.fail(format!("reading {}: {e}", self.c.progress.file_name)),
}
}
Action::Event(Event::FileCompleted) => {
if let Some((_, path)) = self.file.take() {
self.sent.push(path);
}
self.c.progress.files_done += 1;
self.advance = true;
}
Action::Event(Event::SessionCompleted) => {
self.c.outcome.get_or_insert(Outcome::Completed(self.sent.clone()));
}
Action::Event(Event::Aborted) => self.c.aborted(),
Action::Event(_) => {}
Action::Idle => {
if self.advance && self.c.outcome.is_none() {
self.advance = false;
self.start_next();
continue;
}
if self.c.outcome.is_some() || self.c.pending.is_empty() {
break;
}
match self.tx.submit_wire(&self.c.pending) {
Ok(0) => break,
Ok(n) => {
self.c.pending.drain(..n);
}
Err(e) => {
log::debug!("zmodem send error: {e}");
self.c.pending.clear();
}
}
}
_ => break,
}
}
}
}
pub(crate) fn check_upload(p: &Path) -> Result<()> {
let meta = fs::metadata(p).with_context(|| format!("{}", p.display()))?;
if !meta.is_file() {
bail!("{} is not a file", p.display());
}
if meta.len() > MAX_FILE_SIZE {
bail!("{} is larger than the 4 GiB limit", p.display());
}
Ok(())
}
pub enum Transfer {
Download(Download),
Upload(Upload),
XyDownload(XyReceive),
XyUpload(XySend),
}
impl Transfer {
fn parts(&mut self) -> &mut Common {
match self {
Transfer::Download(d) => &mut d.c,
Transfer::Upload(u) => &mut u.c,
Transfer::XyDownload(d) => &mut d.c,
Transfer::XyUpload(u) => &mut u.c,
}
}
fn pump(&mut self, out: &mut Vec<u8>) {
match self {
Transfer::Download(d) => d.pump(out),
Transfer::Upload(u) => u.pump(out),
Transfer::XyDownload(d) => d.pump(out),
Transfer::XyUpload(u) => u.pump(out),
}
}
pub fn feed(&mut self, data: &[u8], out: &mut Vec<u8>) {
self.parts().input(data);
self.pump(out);
}
pub fn tick(&mut self, now: Instant, out: &mut Vec<u8>) {
match self {
Transfer::XyDownload(d) => {
d.tick(now, out);
return self.pump(out);
}
Transfer::XyUpload(u) => {
u.tick(now, out);
return self.pump(out);
}
_ => {}
}
if !self.parts().timed_out(now) {
return;
}
let gave_up = self.parts().outcome.is_some();
match self {
Transfer::Download(d) if gave_up => {
let _ = d.rx.abort();
}
Transfer::Download(d) => {
let _ = d.rx.timeout();
}
Transfer::Upload(u) if gave_up => u.tx.abort(),
Transfer::Upload(u) => {
let _ = u.tx.timeout();
}
Transfer::XyDownload(_) | Transfer::XyUpload(_) => {}
}
self.pump(out);
}
pub fn cancel(&mut self, out: &mut Vec<u8>) {
self.parts().cancelled = true;
match self {
Transfer::Download(d) => {
let _ = d.rx.abort();
}
Transfer::Upload(u) => u.tx.abort(),
Transfer::XyDownload(d) => return d.cancel(out),
Transfer::XyUpload(u) => return u.cancel(out),
}
self.pump(out);
self.parts().outcome.get_or_insert(Outcome::Cancelled);
}
pub fn progress(&self) -> &Progress {
match self {
Transfer::Download(d) => &d.c.progress,
Transfer::Upload(u) => &u.c.progress,
Transfer::XyDownload(d) => d.progress(),
Transfer::XyUpload(u) => u.progress(),
}
}
pub fn is_finished(&self) -> bool {
match self {
Transfer::Download(d) => d.c.outcome.is_some(),
Transfer::Upload(u) => u.c.outcome.is_some(),
Transfer::XyDownload(d) => d.c.outcome.is_some(),
Transfer::XyUpload(u) => u.c.outcome.is_some(),
}
}
pub fn finish(mut self) -> (Outcome, Vec<u8>) {
let c = self.parts();
let leftover = c.leftover();
let outcome = c.outcome.take().unwrap_or_else(|| Outcome::Failed("interrupted".into()));
if let (Transfer::Download(d), Outcome::Failed(_) | Outcome::Cancelled) = (&mut self, &outcome) {
d.close();
}
(outcome, leftover)
}
pub fn enqueue(&mut self, path: PathBuf) -> Result<()> {
match self {
Transfer::Upload(u) => u.enqueue(path),
Transfer::XyUpload(u) => u.enqueue(path),
Transfer::Download(_) | Transfer::XyDownload(_) => bail!("a download is in progress"),
}
}
}
pub fn sanitize_file_name(wire: &[u8]) -> String {
let s = String::from_utf8_lossy(wire);
let base = s.rsplit(['/', '\\']).next().unwrap_or("");
let cleaned: String = base
.chars()
.map(|c| if c.is_control() || matches!(c, ':' | '*' | '?' | '"' | '<' | '>' | '|') { '_' } else { c })
.collect();
let cleaned = cleaned.trim().trim_start_matches('.').to_string();
if cleaned.is_empty() { "download".to_string() } else { cleaned }
}
pub(crate) fn unique_path(dir: &Path, name: &str) -> std::io::Result<PathBuf> {
let first = dir.join(name);
if !first.exists() {
return Ok(first);
}
let (stem, ext) = match name.rsplit_once('.') {
Some((s, e)) if !s.is_empty() => (s, format!(".{e}")),
_ => (name, String::new()),
};
for i in 1..10_000 {
let p = dir.join(format!("{stem} ({i}){ext}"));
if !p.exists() {
return Ok(p);
}
}
Err(std::io::Error::new(std::io::ErrorKind::AlreadyExists, format!("too many copies of {name}")))
}
#[derive(Debug)]
pub struct TrailerFilter {
until: Instant,
o_left: u8,
}
impl TrailerFilter {
const WINDOW: Duration = Duration::from_secs(2);
pub fn new() -> TrailerFilter {
TrailerFilter { until: Instant::now() + Self::WINDOW, o_left: 2 }
}
pub fn strip(&mut self, data: &mut Vec<u8>) -> bool {
if Instant::now() > self.until {
return false;
}
let mut end = None;
for (n, &b) in data.iter().enumerate() {
match b {
b'\r' | b'\n' | 0x8D | 0x8A | 0x11 | 0x13 => {}
b'O' if self.o_left > 0 => {
self.o_left -= 1;
if self.o_left == 0 {
end = Some(n + 1);
break;
}
}
_ => {
end = Some(n);
break;
}
}
}
match end {
Some(cut) => {
data.drain(..cut);
false
}
None => {
data.clear();
true
}
}
}
}
impl Default for TrailerFilter {
fn default() -> Self {
Self::new()
}
}
pub fn find(hay: &[u8], needle: &[u8]) -> Option<usize> {
hay.windows(needle.len()).position(|w| w == needle)
}
#[cfg(test)]
mod tests {
use super::*;
fn temp_dir(name: &str) -> PathBuf {
let d = std::env::temp_dir().join(format!("badterm-test-{name}-{}", std::process::id()));
let _ = fs::remove_dir_all(&d);
fs::create_dir_all(&d).unwrap();
d
}
fn run(up: &mut Transfer, down: &mut Transfer) {
let (mut to_down, mut to_up) = (Vec::new(), Vec::new());
up.feed(&[], &mut to_down);
for _ in 0..100_000 {
if up.is_finished() && down.is_finished() {
return;
}
let data = std::mem::take(&mut to_down);
down.feed(&data, &mut to_up);
let data = std::mem::take(&mut to_up);
up.feed(&data, &mut to_down);
}
panic!("transfer did not finish");
}
#[test]
fn upload_to_download_round_trip() {
let src = temp_dir("src");
let dst = temp_dir("dst");
let mut contents = Vec::new();
for i in 0..300_000u32 {
contents.push((i.wrapping_mul(2_654_435_761) >> 13) as u8);
}
contents.extend_from_slice(&[0x18, 0x11, 0x13, 0xFF, b'\r', 0]);
fs::write(src.join("big.bin"), &contents).unwrap();
fs::write(src.join("empty.txt"), b"").unwrap();
fs::write(dst.join("big.bin"), b"already here").unwrap();
let mut up = Transfer::Upload(Upload::new(vec![src.join("big.bin"), src.join("empty.txt")]).unwrap());
let mut down = Transfer::Download(Download::new(dst.clone()).unwrap());
run(&mut up, &mut down);
let (up_outcome, _) = up.finish();
let (down_outcome, _) = down.finish();
assert_eq!(up_outcome, Outcome::Completed(vec![src.join("big.bin"), src.join("empty.txt")]));
assert_eq!(down_outcome, Outcome::Completed(vec![dst.join("big (1).bin"), dst.join("empty.txt")]));
assert_eq!(fs::read(dst.join("big (1).bin")).unwrap(), contents);
assert_eq!(fs::read(dst.join("big.bin")).unwrap(), b"already here");
assert_eq!(fs::read(dst.join("empty.txt")).unwrap(), b"");
}
#[test]
fn cancel_sends_abort() {
let dst = temp_dir("cancel");
let mut down = Transfer::Download(Download::new(dst).unwrap());
let mut out = Vec::new();
down.cancel(&mut out);
assert!(find(&out, &CANCEL[..10]).is_some());
assert_eq!(down.finish().0, Outcome::Cancelled);
}
#[test]
fn trailer_filter_spans_chunks() {
let mut f = TrailerFilter::new();
let mut a = b"\r\x8a".to_vec();
assert!(f.strip(&mut a));
assert!(a.is_empty());
let mut b = b"OO\r\nMenu: ".to_vec();
assert!(!f.strip(&mut b));
assert_eq!(b, b"\r\nMenu: ");
let mut f = TrailerFilter::new();
let mut c = b"OOOps".to_vec();
f.strip(&mut c);
assert_eq!(c, b"Ops");
}
#[test]
fn sanitizes_names() {
assert_eq!(sanitize_file_name(b"../../etc/passwd"), "passwd");
assert_eq!(sanitize_file_name(b"C:\\DOS\\GAME.ZIP"), "GAME.ZIP");
assert_eq!(sanitize_file_name(b".."), "download");
assert_eq!(sanitize_file_name(b".hidden"), "hidden");
assert_eq!(sanitize_file_name(b"a\x1bb?.txt"), "a_b_.txt");
}
}