use std::collections::{HashMap, VecDeque};
use std::fs::{self, File};
use std::io::{BufWriter, Read, Write};
use std::path::PathBuf;
use std::time::{Duration, Instant};
use anyhow::{Result, bail};
use crate::transfer::{self, Common, Direction, Outcome, Progress};
const SOH: u8 = 0x01;
const CR: u8 = 0x0D;
pub const MAX_LONG: usize = 4000;
const TIMEOUT: Duration = Duration::from_secs(8);
const MAX_RETRIES: u32 = 10;
const START_TRIES: u32 = 15;
pub const WINDOW: usize = 16;
const CAP_LONG: u8 = 2;
const CAP_WINDOWS: u8 = 4;
const CAP_ATTRS: u8 = 8;
fn tochar(x: u8) -> u8 {
x + 32
}
fn unchar(c: u8) -> u8 {
c.wrapping_sub(32)
}
fn ctl(c: u8) -> u8 {
c ^ 64
}
fn block_check(kind: u8, data: &[u8]) -> Vec<u8> {
let sum: u32 = data.iter().map(|&b| b as u32).sum();
match kind {
2 => vec![tochar(((sum >> 6) & 63) as u8), tochar((sum & 63) as u8)],
3 => {
let crc = crc16_kermit(data);
vec![tochar(((crc >> 12) & 0x0F) as u8), tochar(((crc >> 6) & 0x3F) as u8), tochar((crc & 0x3F) as u8)]
}
_ => vec![tochar(((sum + ((sum & 192) >> 6)) & 63) as u8)],
}
}
pub fn crc16_kermit(data: &[u8]) -> u16 {
let mut crc: u16 = 0;
for &b in data {
crc ^= b as u16;
for _ in 0..8 {
crc = if crc & 1 != 0 { (crc >> 1) ^ 0x8408 } else { crc >> 1 };
}
}
crc
}
pub fn find_send_init(data: &[u8]) -> Option<usize> {
data.windows(4).position(|w| w[0] == SOH && (b' '..=b'~').contains(&w[1]) && w[2] == b' ' && w[3] == b'S')
}
#[derive(Clone, Debug)]
struct Params {
maxl: usize,
maxlx: Option<usize>,
time: u8,
npad: u8,
padc: u8,
eol: u8,
qctl: u8,
qbin: u8,
chkt: u8,
rept: u8,
capas: u8,
window: usize,
}
impl Params {
fn ours() -> Params {
Params {
maxl: 94,
maxlx: Some(MAX_LONG),
time: TIMEOUT.as_secs() as u8,
npad: 0,
padc: 0,
eol: CR,
qctl: b'#',
qbin: b'Y',
chkt: 3,
rept: b'~',
capas: CAP_LONG | CAP_WINDOWS | CAP_ATTRS,
window: WINDOW,
}
}
fn encode(&self) -> Vec<u8> {
let maxlx = self.maxlx.unwrap_or(0);
vec![
tochar(self.maxl as u8),
tochar(self.time),
tochar(self.npad),
ctl(self.padc),
tochar(self.eol),
self.qctl,
self.qbin,
b'0' + self.chkt,
self.rept,
tochar(self.capas),
tochar(self.window as u8),
tochar((maxlx / 95) as u8),
tochar((maxlx % 95) as u8),
]
}
fn decode(d: &[u8]) -> Params {
let at = |i: usize| d.get(i).copied();
let mut p = Params {
maxl: at(0).map_or(80, |c| unchar(c).clamp(10, 94) as usize),
maxlx: None,
time: at(1).map_or(5, unchar),
npad: at(2).map_or(0, unchar),
padc: at(3).map_or(0, ctl),
eol: at(4).map(unchar).filter(|&c| c > 0).unwrap_or(CR),
qctl: at(5).filter(|&c| c != b' ').unwrap_or(b'#'),
qbin: at(6).filter(|&c| c != b' ').unwrap_or(b'N'),
chkt: at(7).map(|c| c.wrapping_sub(b'0')).filter(|c| (1..=3).contains(c)).unwrap_or(1),
rept: at(8).unwrap_or(b' '),
capas: 0,
window: 1,
};
let mut i = 9;
while let Some(c) = at(i) {
if i == 9 {
p.capas = unchar(c);
}
i += 1;
if unchar(c) & 1 == 0 {
break;
}
}
if p.capas & CAP_WINDOWS != 0 {
p.window = at(i).map_or(1, |c| (unchar(c) as usize).clamp(1, 31));
}
if p.capas & CAP_LONG != 0 {
let len = match (at(i + 1), at(i + 2)) {
(Some(a), Some(b)) => unchar(a) as usize * 95 + unchar(b) as usize,
_ => 500,
};
p.maxlx = Some(if len == 0 { 500 } else { len });
}
p
}
}
#[derive(Clone, Debug)]
struct Link {
send_max: usize,
long: bool,
chkt: u8,
our_qctl: u8,
their_qctl: u8,
qbin: Option<u8>,
rept: Option<u8>,
attrs: bool,
window: usize,
eol: u8,
npad: u8,
padc: u8,
}
impl Link {
fn initial() -> Link {
Link {
send_max: 80,
long: false,
chkt: 1,
our_qctl: b'#',
their_qctl: b'#',
qbin: None,
rept: None,
attrs: false,
window: 1,
eol: CR,
npad: 0,
padc: 0,
}
}
fn agree(ours: &Params, theirs: &Params) -> Link {
let long = ours.capas & theirs.capas & CAP_LONG != 0;
let qbin = match (ours.qbin, theirs.qbin) {
(b'Y', c) | (c, b'Y') if c != b'Y' && c != b'N' && is_prefix(c) => Some(c),
(a, b) if a == b && is_prefix(a) => Some(a),
_ => None,
};
let rept = (ours.rept == theirs.rept && is_prefix(ours.rept)).then_some(ours.rept);
Link {
send_max: if long { theirs.maxlx.unwrap_or(500).min(MAX_LONG) } else { theirs.maxl },
long,
chkt: if ours.chkt == theirs.chkt { ours.chkt } else { 1 },
our_qctl: ours.qctl,
their_qctl: theirs.qctl,
qbin,
rept,
attrs: ours.capas & theirs.capas & CAP_ATTRS != 0,
window: if ours.capas & theirs.capas & CAP_WINDOWS != 0 { ours.window.min(theirs.window) } else { 1 },
eol: theirs.eol,
npad: theirs.npad,
padc: theirs.padc,
}
}
fn packet(&self, seq: u8, kind: u8, data: &[u8], chkt: u8) -> Vec<u8> {
let chk_len = chkt as usize;
let mut body = Vec::with_capacity(data.len() + 10);
let short_len = data.len() + 2 + chk_len;
if short_len <= 94 {
body.push(tochar(short_len as u8));
body.push(tochar(seq));
body.push(kind);
} else {
let ext = data.len() + chk_len;
body.extend_from_slice(&[tochar(0), tochar(seq), kind, tochar((ext / 95) as u8), tochar((ext % 95) as u8)]);
body.extend(block_check(1, &body));
}
body.extend_from_slice(data);
let chk = block_check(chkt, &body);
let mut out = vec![self.padc; self.npad as usize];
out.push(SOH);
out.extend(body);
out.extend(chk);
out.push(self.eol);
out
}
fn room(&self) -> usize {
if self.long { self.send_max - self.chkt as usize - 7 } else { self.send_max - 2 - self.chkt as usize }
}
fn encode(&self, src: &[u8], room: usize) -> (Vec<u8>, usize) {
let mut out = Vec::with_capacity(room);
let mut i = 0;
while i < src.len() {
let c = src[i];
let mut run = 1;
if self.rept.is_some() {
while i + run < src.len() && src[i + run] == c && run < 94 {
run += 1;
}
}
let one = self.encode_byte(c);
let unit = match self.rept {
Some(r) if run >= 3 => {
let mut u = vec![r, tochar(run as u8)];
u.extend_from_slice(&one);
u
}
_ => {
run = 1;
one
}
};
if out.len() + unit.len() > room {
break;
}
out.extend(unit);
i += run;
}
(out, i)
}
fn encode_byte(&self, mut c: u8) -> Vec<u8> {
let mut out = Vec::with_capacity(3);
if let Some(q) = self.qbin
&& c & 0x80 != 0
{
out.push(q);
c &= 0x7F;
}
let low = c & 0x7F;
if low < 32 || low == 127 {
out.extend_from_slice(&[self.our_qctl, ctl(c)]);
} else if low == self.our_qctl || Some(low) == self.qbin || Some(low) == self.rept {
out.extend_from_slice(&[self.our_qctl, c]);
} else {
out.push(c);
}
out
}
fn decode(&self, data: &[u8]) -> Vec<u8> {
let mut out = Vec::with_capacity(data.len());
let mut i = 0;
while i < data.len() {
let mut count = 1;
if let Some(r) = self.rept
&& data[i] == r
&& i + 1 < data.len()
{
count = unchar(data[i + 1]) as usize;
i += 2;
}
let mut high = 0;
if let Some(q) = self.qbin
&& data.get(i) == Some(&q)
{
high = 0x80;
i += 1;
}
let Some(&b) = data.get(i) else { break };
i += 1;
let c = if b == self.their_qctl {
let Some(&q) = data.get(i) else { break };
i += 1;
if (0x3F..=0x5F).contains(&(q & 0x7F)) { ctl(q) } else { q }
} else {
b
};
out.extend(std::iter::repeat_n(c | high, count));
}
out
}
}
fn is_prefix(c: u8) -> bool {
(33..=62).contains(&c) || (96..=126).contains(&c)
}
#[derive(Debug, PartialEq, Eq)]
struct Packet {
seq: u8,
kind: u8,
data: Vec<u8>,
}
fn read_packet(buf: &mut Vec<u8>, chkt: u8) -> std::result::Result<Option<Packet>, Option<u8>> {
let Some(start) = buf.iter().position(|&b| b == SOH) else {
buf.clear();
return Ok(None);
};
buf.drain(..start);
if buf.len() < 4 {
return Ok(None);
}
let len = unchar(buf[1]) as usize;
let (seq, kind) = (unchar(buf[2]), buf[3]);
let chk_len = if kind == b'S' { 1 } else { chkt as usize };
let (header, total) = if len == 0 {
if buf.len() < 7 {
return Ok(None);
}
if block_check(1, &buf[1..6]) != buf[6..7] {
buf.drain(..1);
return Err(Some(seq & 63));
}
(7, 7 + unchar(buf[4]) as usize * 95 + unchar(buf[5]) as usize)
} else if len < 3 {
buf.drain(..1);
return Err(None);
} else {
(4, 2 + len)
};
if total > MAX_LONG + 16 || total < header + chk_len {
buf.drain(..1);
return Err(None);
}
if buf.len() < total {
return Ok(None);
}
let (body, chk) = buf[1..total].split_at(total - 1 - chk_len);
if block_check(chk_len as u8, body) != chk {
buf.drain(..1);
return Err(Some(seq & 63));
}
let raw: Vec<u8> = buf.drain(..total).collect();
Ok(Some(Packet { seq: seq & 63, kind, data: raw[header..total - chk_len].to_vec() }))
}
#[derive(Clone, Copy, Debug, PartialEq, Eq)]
enum State {
Init,
File,
Data,
Header,
Attrs,
Eof,
Break,
Done,
}
struct Flight {
seq: u8,
packet: Vec<u8>,
acked: bool,
retries: u32,
}
fn seq_after(a: u8, b: u8) -> usize {
(b.wrapping_sub(a) & 63) as usize
}
enum Role {
Receive { dir: PathBuf, file: Option<(BufWriter<File>, PathBuf)>, saved: Vec<PathBuf> },
Send { queue: VecDeque<PathBuf>, file: Option<(File, PathBuf)>, sent: Vec<PathBuf>, eof: bool, skip: bool },
}
pub struct Kermit {
role: Role,
state: State,
link: Link,
seq: u8,
last: Vec<u8>,
pending_data: Vec<u8>,
flight: VecDeque<Flight>,
early: HashMap<u8, Packet>,
acks: Vec<Vec<u8>>,
retries: u32,
last_heard: Instant,
pub(crate) c: Common,
}
impl Kermit {
pub fn receive(dir: PathBuf) -> Kermit {
Kermit::new(Role::Receive { dir, file: None, saved: Vec::new() }, Direction::Download)
}
pub fn send(paths: Vec<PathBuf>) -> Result<Kermit> {
if paths.is_empty() {
bail!("no files to upload");
}
for p in &paths {
if !p.is_file() {
bail!("not a file: {}", p.display());
}
}
let mut k = Kermit::new(
Role::Send { queue: paths.into(), file: None, sent: Vec::new(), eof: false, skip: false },
Direction::Upload,
);
k.c.progress.files_queued = k.queued();
Ok(k)
}
fn new(role: Role, direction: Direction) -> Kermit {
Kermit {
role,
state: State::Init,
link: Link::initial(),
seq: 0,
last: Vec::new(),
pending_data: Vec::new(),
flight: VecDeque::new(),
early: HashMap::new(),
acks: vec![Vec::new(); 64],
retries: 0,
last_heard: Instant::now(),
c: Common::new(direction),
}
}
pub(crate) fn progress(&self) -> &Progress {
&self.c.progress
}
pub fn enqueue(&mut self, path: PathBuf) -> Result<()> {
match &mut self.role {
Role::Send { queue, .. } => {
queue.push_back(path);
self.c.progress.files_queued = self.queued();
Ok(())
}
Role::Receive { .. } => bail!("a download is in progress"),
}
}
fn queued(&self) -> usize {
match &self.role {
Role::Send { queue, .. } => queue.len(),
Role::Receive { .. } => 0,
}
}
fn sending(&self) -> bool {
matches!(self.role, Role::Send { .. })
}
fn chkt(&self) -> u8 {
if self.state == State::Init { 1 } else { self.link.chkt }
}
fn put(&mut self, kind: u8, data: &[u8], out: &mut Vec<u8>) {
let chkt = if kind == b'S' || self.state == State::Init { 1 } else { self.link.chkt };
self.last = self.link.packet(self.seq, kind, data, chkt);
out.extend_from_slice(&self.last);
self.retries = 0;
self.last_heard = Instant::now();
}
fn ack(&mut self, seq: u8, data: &[u8], out: &mut Vec<u8>) {
self.seq = seq;
if self.acks[seq as usize].is_empty() {
self.put(b'Y', data, out);
self.acks[seq as usize] = self.last.clone();
} else {
self.last = self.acks[seq as usize].clone();
self.retries = 0;
self.last_heard = Instant::now();
}
}
fn windowed(&self) -> bool {
self.sending() && self.state == State::Data && self.link.window > 1
}
fn resend(&mut self, out: &mut Vec<u8>) {
if self.windowed() {
if let Some(f) = self.flight.iter_mut().find(|f| !f.acked) {
f.retries += 1;
if f.retries > MAX_RETRIES {
return self.fail("too many retries", out);
}
out.extend_from_slice(&f.packet);
}
} else {
out.extend_from_slice(&self.last);
}
}
fn fail(&mut self, msg: impl Into<String>, out: &mut Vec<u8>) {
if self.c.outcome.is_some() {
return;
}
let msg = msg.into();
let data = self.link.encode(msg.as_bytes(), 80).0;
let p = self.link.packet(self.seq, b'E', &data, self.chkt());
out.extend_from_slice(&p);
self.drop_partial();
self.c.outcome = Some(Outcome::Failed(msg));
}
fn drop_partial(&mut self) {
if let Role::Receive { file, .. } = &mut self.role
&& let Some((_, path)) = file.take()
{
let _ = fs::remove_file(path);
}
}
pub(crate) fn cancel(&mut self, out: &mut Vec<u8>) {
self.c.cancelled = true;
if self.c.outcome.is_none() {
let p = self.link.packet(self.seq, b'E', b"Cancelled by the user", self.chkt());
out.extend_from_slice(&p);
}
self.drop_partial();
self.c.outcome.get_or_insert(Outcome::Cancelled);
}
pub fn start(&mut self, out: &mut Vec<u8>) {
if self.sending() && self.state == State::Init && self.last.is_empty() {
let data = Params::ours().encode();
self.put(b'S', &data, out);
}
}
pub(crate) fn tick(&mut self, now: Instant, out: &mut Vec<u8>) {
if self.c.outcome.is_some() || self.state == State::Done {
return;
}
if now.duration_since(self.last_heard) < TIMEOUT {
return;
}
self.last_heard = now;
self.retries += 1;
let limit = if !self.sending() && self.state == State::Init { START_TRIES } else { MAX_RETRIES };
if self.retries > limit {
return self.fail("the other side stopped answering", out);
}
if self.sending() {
self.resend(out);
} else if self.link.window > 1 && self.state != State::Init {
let p = self.link.packet((self.seq + 1) & 63, b'N', &[], self.chkt());
out.extend_from_slice(&p);
} else if !self.last.is_empty() {
out.extend_from_slice(&self.last);
} else {
let p = self.link.packet(0, b'N', &[], 1);
out.extend_from_slice(&p);
}
}
pub(crate) fn pump(&mut self, out: &mut Vec<u8>) {
while self.c.outcome.is_none() && !self.c.pending.is_empty() {
let chkt = self.chkt();
match read_packet(&mut self.c.pending, chkt) {
Ok(None) => break,
Ok(Some(p)) => {
self.last_heard = Instant::now();
if p.kind == b'E' {
let msg = String::from_utf8_lossy(&self.link.decode(&p.data)).into_owned();
self.drop_partial();
self.c.outcome = Some(Outcome::Failed(format!("the other side stopped: {msg}")));
break;
}
if self.sending() { self.sender_got(p, out) } else { self.receiver_got(p, out) }
}
Err(seq) => {
if self.sending() {
self.resend(out);
} else {
let p = self.link.packet(seq.unwrap_or(self.seq.wrapping_add(1) & 63), b'N', &[], chkt);
out.extend_from_slice(&p);
}
}
}
}
}
fn receiver_got(&mut self, p: Packet, out: &mut Vec<u8>) {
if self.state == State::Init {
if p.seq == 0 {
self.take_in_order(p, out);
} else if p.seq == self.seq && !self.last.is_empty() {
out.extend_from_slice(&self.last);
}
return;
}
let window = self.link.window;
let next = (self.seq + 1) & 63;
let ahead = seq_after(next, p.seq);
if ahead >= window {
if seq_after(p.seq, self.seq) < window {
out.extend_from_slice(&self.acks[p.seq as usize]);
}
return;
}
if self.early.contains_key(&p.seq) {
out.extend_from_slice(&self.acks[p.seq as usize]);
return;
}
self.acks[p.seq as usize].clear();
if ahead > 0 {
let asked = self.early.keys().map(|&k| seq_after(next, k) + 1).max().unwrap_or(0);
for n in asked..ahead {
let seq = (next as usize + n) as u8 & 63;
if !self.early.contains_key(&seq) {
let nak = self.link.packet(seq, b'N', &[], self.link.chkt);
out.extend_from_slice(&nak);
}
}
if p.kind == b'D' {
let ack = self.link.packet(p.seq, b'Y', &[], self.link.chkt);
out.extend_from_slice(&ack);
self.acks[p.seq as usize] = ack;
}
self.early.insert(p.seq, p);
return;
}
self.take_in_order(p, out);
while self.c.outcome.is_none()
&& let Some(p) = self.early.remove(&((self.seq + 1) & 63))
{
self.take_in_order(p, out);
}
}
fn take_in_order(&mut self, p: Packet, out: &mut Vec<u8>) {
match (self.state, p.kind) {
(State::Init, b'S') => {
let theirs = Params::decode(&p.data);
let ours = Params::ours();
self.ack(p.seq, &ours.encode(), out);
self.link = Link::agree(&ours, &theirs);
log::info!("Kermit: window {}, packets of {}", self.link.window, self.link.send_max);
self.state = State::File;
}
(State::File, b'F') => {
let wire = self.link.decode(&p.data);
let name = transfer::sanitize_file_name(&wire);
match self.open(&name) {
Ok(path) => {
let used = path.file_name().map(|n| n.to_string_lossy().into_owned()).unwrap_or(name);
let data = self.link.encode(used.as_bytes(), self.link.room()).0;
self.ack(p.seq, &data, out);
self.state = State::Data;
}
Err(e) => self.fail(format!("saving {name}: {e}"), out),
}
}
(State::File, b'X') => self.ack(p.seq, &[], out),
(State::File, b'B') => {
self.ack(p.seq, &[], out);
self.state = State::Done;
if let Role::Receive { saved, .. } = &self.role {
self.c.outcome = Some(Outcome::Completed(saved.clone()));
}
}
(State::Data, b'A') => {
self.c.progress.file_size = attribute_size(&p.data);
self.ack(p.seq, b"Y", out);
}
(State::Data, b'D') => {
let data = self.link.decode(&p.data);
if let Role::Receive { file: Some((f, path)), .. } = &mut self.role
&& let Err(e) = f.write_all(&data)
{
let msg = format!("writing {}: {e}", path.display());
return self.fail(msg, out);
}
self.c.progress.file_bytes += data.len() as u64;
self.ack(p.seq, &[], out);
}
(State::Data, b'Z') => {
let discard = self.link.decode(&p.data).first() == Some(&b'D');
if let Role::Receive { file, saved, .. } = &mut self.role
&& let Some((mut f, path)) = file.take()
{
if discard {
drop(f);
let _ = fs::remove_file(&path);
} else if let Err(e) = f.flush() {
let msg = format!("writing {}: {e}", path.display());
return self.fail(msg, out);
} else {
saved.push(path);
self.c.progress.files_done += 1;
}
}
self.ack(p.seq, &[], out);
self.state = State::File;
}
(_, kind) => self.fail(format!("unexpected Kermit packet '{}'", kind as char), out),
}
}
fn open(&mut self, name: &str) -> std::io::Result<PathBuf> {
let Role::Receive { dir, file, .. } = &mut self.role else { unreachable!("receiving") };
fs::create_dir_all(&*dir)?;
let path = transfer::unique_path(dir, name)?;
*file = Some((BufWriter::new(File::create(&path)?), path.clone()));
let p = &mut self.c.progress;
p.file_name = name.to_string();
p.file_size = None;
p.file_bytes = 0;
p.file_started = Instant::now();
Ok(path)
}
fn sender_got(&mut self, p: Packet, out: &mut Vec<u8>) {
if self.windowed() {
return self.window_got(p, out);
}
let acked = match p.kind {
b'Y' => p.seq == self.seq,
b'N' => p.seq == (self.seq + 1) & 63,
_ => false,
};
if !acked {
if p.kind == b'N' && p.seq == self.seq {
self.retries += 1;
if self.retries > MAX_RETRIES {
return self.fail("too many retries", out);
}
out.extend_from_slice(&self.last);
}
return;
}
let reply = if p.kind == b'Y' { p.data } else { Vec::new() };
self.seq = (self.seq + 1) & 63;
match self.state {
State::Init => {
self.link = Link::agree(&Params::ours(), &Params::decode(&reply));
log::info!("Kermit: window {}, packets of {}", self.link.window, self.link.send_max);
self.next_file(out);
}
State::Header if self.link.attrs => {
let size = self.file_size();
let digits = size.to_string();
let mut data = vec![b'1', tochar(digits.len() as u8)];
data.extend_from_slice(digits.as_bytes());
self.state = State::Attrs;
self.put(b'A', &data, out);
}
State::Header => self.start_data(out),
State::Attrs => {
if reply.first() == Some(&b'N') {
self.skip_file(out);
} else {
self.start_data(out);
}
}
State::Data => {
match reply.first() {
Some(b'X') => self.skip_file(out),
Some(b'Z') => {
if let Role::Send { queue, .. } = &mut self.role {
queue.clear();
}
self.skip_file(out);
}
_ => self.send_data(out),
}
}
State::Eof => {
if let Role::Send { file, sent, skip, .. } = &mut self.role
&& let Some((_, path)) = file.take()
{
if !*skip {
sent.push(path);
self.c.progress.files_done += 1;
}
*skip = false;
}
self.next_file(out);
}
State::Break => {
self.state = State::Done;
if let Role::Send { sent, .. } = &self.role {
self.c.outcome = Some(Outcome::Completed(sent.clone()));
}
}
State::File | State::Done => {}
}
}
fn file_size(&self) -> u64 {
match &self.role {
Role::Send { file: Some((f, _)), .. } => f.metadata().map(|m| m.len()).unwrap_or(0),
_ => 0,
}
}
fn next_file(&mut self, out: &mut Vec<u8>) {
let Role::Send { queue, file, eof, .. } = &mut self.role else { return };
let Some(path) = queue.pop_front() else {
self.state = State::Break;
return self.put(b'B', &[], out);
};
let f = match File::open(&path) {
Ok(f) => f,
Err(e) => return self.fail(format!("reading {}: {e}", path.display()), out),
};
let name = path.file_name().map(|n| n.to_string_lossy().into_owned()).unwrap_or_else(|| "file".into());
let size = f.metadata().map(|m| m.len()).ok();
*file = Some((f, path));
*eof = false;
self.pending_data.clear();
let p = &mut self.c.progress;
p.file_name = name.clone();
p.file_size = size;
p.file_bytes = 0;
p.file_started = Instant::now();
p.files_queued = queue.len();
let data = self.link.encode(name.as_bytes(), self.link.room()).0;
self.state = State::Header;
self.put(b'F', &data, out);
}
fn start_data(&mut self, out: &mut Vec<u8>) {
self.state = State::Data;
if self.link.window > 1 {
self.seq = self.seq.wrapping_sub(1) & 63;
self.flight.clear();
self.fill_window(out);
} else {
self.send_data(out);
}
}
fn window_got(&mut self, p: Packet, out: &mut Vec<u8>) {
let i = self.flight.iter().position(|f| f.seq == p.seq);
match (p.kind, i) {
(b'Y', Some(i)) => {
self.flight[i].acked = true;
if let Some(&c @ (b'X' | b'Z')) = p.data.first() {
self.stop_file(c == b'Z');
}
}
(b'N', Some(i)) if !self.flight[i].acked => {
let f = &mut self.flight[i];
f.retries += 1;
if f.retries > MAX_RETRIES {
return self.fail("too many retries", out);
}
out.extend_from_slice(&f.packet);
}
(b'N', None) if p.seq == (self.seq + 1) & 63 => self.flight.iter_mut().for_each(|f| f.acked = true),
_ => return,
}
self.last_heard = Instant::now();
self.retries = 0;
while self.flight.front().is_some_and(|f| f.acked) {
self.flight.pop_front();
}
self.fill_window(out);
}
fn stop_file(&mut self, whole_batch: bool) {
if let Role::Send { queue, eof, skip, .. } = &mut self.role {
*eof = true;
*skip = true;
if whole_batch {
queue.clear();
}
}
self.pending_data.clear();
}
fn fill_window(&mut self, out: &mut Vec<u8>) {
while self.flight.len() < self.link.window {
let Some(data) = self.next_chunk(out) else { return };
if data.is_empty() {
break;
}
self.seq = (self.seq + 1) & 63;
let packet = self.link.packet(self.seq, b'D', &data, self.link.chkt);
out.extend_from_slice(&packet);
self.flight.push_back(Flight { seq: self.seq, packet, acked: false, retries: 0 });
self.last_heard = Instant::now();
}
let at_end = matches!(self.role, Role::Send { eof: true, .. }) && self.pending_data.is_empty();
if self.flight.is_empty() && at_end {
let skip = matches!(self.role, Role::Send { skip: true, .. });
self.seq = (self.seq + 1) & 63;
self.state = State::Eof;
self.put(b'Z', if skip { b"D" } else { b"" }, out);
}
}
fn next_chunk(&mut self, out: &mut Vec<u8>) -> Option<Vec<u8>> {
let room = self.link.room();
loop {
if let Role::Send { file: Some((f, path)), eof, .. } = &mut self.role
&& !*eof
&& self.pending_data.len() < room * 2
{
let mut buf = vec![0u8; room * 4];
match f.read(&mut buf) {
Ok(0) => *eof = true,
Ok(n) => self.pending_data.extend_from_slice(&buf[..n]),
Err(e) => {
let msg = format!("reading {}: {e}", path.display());
self.fail(msg, out);
return None;
}
}
}
if !self.pending_data.is_empty() {
break;
}
if matches!(self.role, Role::Send { eof: true, .. } | Role::Send { file: None, .. }) {
return Some(Vec::new());
}
}
let (data, used) = self.link.encode(&self.pending_data, room);
self.pending_data.drain(..used);
self.c.progress.file_bytes += used as u64;
Some(data)
}
fn send_data(&mut self, out: &mut Vec<u8>) {
match self.next_chunk(out) {
None => {}
Some(data) if data.is_empty() => {
self.state = State::Eof;
self.put(b'Z', &[], out);
}
Some(data) => self.put(b'D', &data, out),
}
}
fn skip_file(&mut self, out: &mut Vec<u8>) {
if let Role::Send { skip, .. } = &mut self.role {
*skip = true;
}
self.pending_data.clear();
self.state = State::Eof;
self.put(b'Z', b"D", out);
}
}
fn attribute_size(data: &[u8]) -> Option<u64> {
let mut i = 0;
let mut size = None;
while i + 1 < data.len() {
let (tag, len) = (data[i], unchar(data[i + 1]) as usize);
let value = data.get(i + 2..i + 2 + len)?;
let n = std::str::from_utf8(value).ok().and_then(|s| s.trim().parse::<u64>().ok());
match tag {
b'1' => size = n.or(size),
b'!' if size.is_none() => size = n.map(|k| k * 1024),
_ => {}
}
i += 2 + len;
}
size
}
#[cfg(test)]
mod tests {
use super::*;
fn temp_dir(name: &str) -> PathBuf {
let d = std::env::temp_dir().join(format!("badterm-kermit-{name}-{}", std::process::id()));
let _ = fs::remove_dir_all(&d);
fs::create_dir_all(&d).unwrap();
d
}
fn run(tx: &mut Kermit, rx: &mut Kermit, damage: Option<usize>) {
run_lossy(tx, rx, damage, None);
}
fn drop_packets(data: Vec<u8>, nth: usize, count: &mut usize) -> Vec<u8> {
let mut out = Vec::new();
for packet in data.split_inclusive(|&b| b == CR) {
*count += 1;
if !count.is_multiple_of(nth) {
out.extend_from_slice(packet);
}
}
out
}
fn run_lossy(tx: &mut Kermit, rx: &mut Kermit, damage: Option<usize>, lose: Option<usize>) {
let (mut to_rx, mut to_tx) = (Vec::new(), Vec::new());
let (mut sent, mut answered) = (0, 0);
tx.start(&mut to_rx);
for n in 0..200_000 {
if tx.c.outcome.is_some() && rx.c.outcome.is_some() {
return;
}
let mut data = std::mem::take(&mut to_rx);
if damage.is_some_and(|d| n % d == d - 1) && data.len() > 8 {
let i = data.len() / 2;
data[i] ^= 0x01;
}
if let Some(nth) = lose {
data = drop_packets(data, nth, &mut sent);
}
rx.c.input(&data);
rx.pump(&mut to_tx);
let mut data = std::mem::take(&mut to_tx);
if let Some(nth) = lose {
data = drop_packets(data, nth + 2, &mut answered);
}
tx.c.input(&data);
tx.pump(&mut to_rx);
if to_rx.is_empty() && to_tx.is_empty() {
tx.last_heard = Instant::now() - TIMEOUT;
tx.tick(Instant::now(), &mut to_rx);
if n % 2 == 1 {
rx.last_heard = Instant::now() - TIMEOUT;
rx.tick(Instant::now(), &mut to_tx);
}
}
}
panic!("transfer did not finish");
}
#[test]
fn checks_and_prefixes() {
assert_eq!(crc16_kermit(b"123456789"), 0x2189);
let link = Link { qbin: Some(b'&'), rept: Some(b'~'), ..Link::agree(&Params::ours(), &Params::ours()) };
let src = b"ab\x01#\x80&~cccccc\xff\r\n".to_vec();
let (enc, used) = link.encode(&src, 1000);
assert_eq!(used, src.len());
assert!(enc.iter().all(|&b| (32..127).contains(&b)), "{enc:?}");
assert_eq!(link.decode(&enc), src);
let (part, n) = link.encode(&src, 5);
assert_eq!(link.decode(&part), src[..n]);
assert_eq!(attribute_size(b"#(20251031!\"12"), Some(12 * 1024));
assert_eq!(attribute_size(b"1#300!\"12"), Some(300));
assert_eq!(find_send_init(b"junk\x01) S~"), Some(4));
}
#[test]
fn round_trip_two_files() {
let (src, dst) = (temp_dir("src"), temp_dir("dst"));
let mut big = Vec::new();
for i in 0..150_000u32 {
big.push((i.wrapping_mul(2_654_435_761) >> 13) as u8);
}
big.extend_from_slice(&[0; 500]);
fs::write(src.join("big.bin"), &big).unwrap();
fs::write(src.join("empty.txt"), b"").unwrap();
let mut tx = Kermit::send(vec![src.join("big.bin"), src.join("empty.txt")]).unwrap();
let mut rx = Kermit::receive(dst.clone());
run(&mut tx, &mut rx, None);
assert_eq!(tx.c.outcome, Some(Outcome::Completed(vec![src.join("big.bin"), src.join("empty.txt")])));
assert_eq!(rx.c.outcome, Some(Outcome::Completed(vec![dst.join("big.bin"), dst.join("empty.txt")])));
assert_eq!(fs::read(dst.join("big.bin")).unwrap(), big);
assert_eq!(fs::read(dst.join("empty.txt")).unwrap(), b"");
assert!(tx.link.long && tx.link.chkt == 3 && tx.link.rept == Some(b'~'));
}
#[test]
fn damaged_packets_are_resent() {
let (src, dst) = (temp_dir("dsrc"), temp_dir("ddst"));
let data: Vec<u8> = (0..40_000u32).map(|i| (i % 251) as u8).collect();
fs::write(src.join("a.dat"), &data).unwrap();
let mut tx = Kermit::send(vec![src.join("a.dat")]).unwrap();
let mut rx = Kermit::receive(dst.clone());
run(&mut tx, &mut rx, Some(7));
assert!(matches!(rx.c.outcome, Some(Outcome::Completed(_))), "{:?}", rx.c.outcome);
assert_eq!(fs::read(dst.join("a.dat")).unwrap(), data);
}
#[test]
fn sliding_windows_survive_lost_packets() {
let (src, dst) = (temp_dir("wsrc"), temp_dir("wdst"));
let data: Vec<u8> = (0..300_000u32).map(|i| (i.wrapping_mul(2_654_435_761) >> 9) as u8).collect();
fs::write(src.join("w.dat"), &data).unwrap();
fs::write(src.join("x.dat"), &data[..5000]).unwrap();
let mut tx = Kermit::send(vec![src.join("w.dat")]).unwrap();
let mut rx = Kermit::receive(dst.clone());
let mut out = Vec::new();
tx.start(&mut out);
rx.c.input(&out);
let mut back = Vec::new();
rx.pump(&mut back);
tx.c.input(&back);
let mut out = Vec::new();
tx.pump(&mut out);
while tx.state != State::Data {
let mut back = Vec::new();
rx.c.input(&std::mem::take(&mut out));
rx.pump(&mut back);
tx.c.input(&back);
tx.pump(&mut out);
}
assert_eq!((tx.link.window, tx.flight.len()), (WINDOW, WINDOW));
assert_eq!(out.iter().filter(|&&b| b == SOH).count(), WINDOW);
for (lose, damage) in [(Some(5), None), (Some(13), Some(9)), (Some(3), None)] {
let dst = temp_dir("wdst2");
let mut tx = Kermit::send(vec![src.join("w.dat"), src.join("x.dat")]).unwrap();
let mut rx = Kermit::receive(dst.clone());
run_lossy(&mut tx, &mut rx, damage, lose);
assert!(matches!(rx.c.outcome, Some(Outcome::Completed(_))), "{lose:?}: {:?}", rx.c.outcome);
assert!(matches!(tx.c.outcome, Some(Outcome::Completed(_))), "{lose:?}: {:?}", tx.c.outcome);
assert_eq!(fs::read(dst.join("w.dat")).unwrap(), data, "{lose:?}");
assert_eq!(fs::read(dst.join("x.dat")).unwrap(), &data[..5000]);
}
let mut old = Params::ours();
old.capas &= !CAP_WINDOWS;
let theirs = Params::decode(&old.encode());
assert_eq!(Link::agree(&Params::ours(), &theirs).window, 1);
assert_eq!(Params::decode(&Params::ours().encode()).window, WINDOW);
}
#[test]
fn cancel_sends_an_error_packet() {
let mut rx = Kermit::receive(temp_dir("cancel"));
let mut out = Vec::new();
rx.cancel(&mut out);
assert_eq!(out[0], SOH);
assert_eq!(out[3], b'E');
assert_eq!(rx.c.outcome, Some(Outcome::Cancelled));
}
}