runsync-transfer 2026.1.0

//! A hostile peer against `receive()`.
//!
//! A receiver is a remote party writing to your filesystem. These tests speak
//! the wire protocol by hand and try to make the receiver do something it
//! should not: escape its destination root, allocate unboundedly, write past
//! the end of a file, or accept tampered data. Every one must end in a clean
//! error, never a panic and never a stray file.

use runsync_transfer::codec::crypto::{Handshake, Role};
use runsync_transfer::transport::{mem, BoxRecv, BoxSend};
use runsync_transfer::wire::{self, Control, EntryKind, FileEntry, FrameHeader};
use runsync_transfer::{receive, Algorithm, Cipher, Config, Secrecy, Transport};
use std::path::{Path, PathBuf};
use std::sync::Arc;
use tokio::io::{AsyncReadExt, AsyncWriteExt};

const BUF: usize = 128 * 1024;

/// The attacker's side of a session, with the handshake already done.
struct Peer {
    ctl_w: BoxSend,
    ctl_r: BoxRecv,
    transport: Arc<dyn Transport>,
}

impl Peer {
    async fn connect(transport: Arc<dyn Transport>) -> Self {
        let (mut ctl_w, mut ctl_r) = transport.open_bi().await.unwrap();
        let hs = Handshake::new(Role::Initiator, &Secrecy::TransportOnly, Cipher::Auto);
        ctl_w.write_all(hs.message()).await.unwrap();
        ctl_w.flush().await.unwrap();
        let mut peer = [0u8; runsync_transfer::codec::crypto::HANDSHAKE_MSG_LEN];
        ctl_r.read_exact(&mut peer).await.unwrap();
        Self {
            ctl_w,
            ctl_r,
            transport,
        }
    }

    async fn send_manifest(&mut self, entries: Vec<FileEntry>) {
        wire::write_control(&mut self.ctl_w, &Control::Manifest(entries))
            .await
            .unwrap();
    }

    /// Read the receiver's resume reply, if it gets that far.
    async fn read_control(&mut self) -> runsync_transfer::Result<Control> {
        wire::read_control(&mut self.ctl_r, 64 << 20, 1 << 20).await
    }

    async fn start(&mut self, streams: u32) {
        let _ = wire::write_control(&mut self.ctl_w, &Control::Start { streams }).await;
    }
}

fn file_entry(id: u32, path: &str, size: u64, chunk: u32) -> FileEntry {
    FileEntry {
        file_id: id,
        path: path.to_string(),
        size,
        chunk_size: chunk,
        mode: 0o644,
        mtime: 0,
        kind: EntryKind::File,
        hash: None,
        incompressible: false,
    }
}

/// Serialise one data frame by hand.
fn frame(header: FrameHeader, payload: &[u8]) -> Vec<u8> {
    let mut head = [0u8; wire::FRAME_HEADER_LEN];
    header.encode(&mut head);
    let mut out = head.to_vec();
    out.extend_from_slice(payload);
    out
}

/// Run one hostile session and return the receiver's result plus the dest dir.
async fn attack<F, Fut>(cfg: Config, body: F) -> (runsync_transfer::Result<()>, tempfile::TempDir)
where
    F: FnOnce(Peer) -> Fut + Send + 'static,
    Fut: std::future::Future<Output = ()> + Send + 'static,
{
    let tmp = tempfile::tempdir().unwrap();
    let dest = tmp.path().join("dest");
    std::fs::create_dir_all(&dest).unwrap();

    let (a, b) = mem::pair(BUF);
    let a: Arc<dyn Transport> = Arc::new(a);
    let b: Arc<dyn Transport> = Arc::new(b);

    let rh = tokio::spawn(async move { receive(b, &dest, &cfg, None).await });
    let ah = tokio::spawn(async move {
        let p = Peer::connect(a).await;
        body(p).await;
    });

    let _ = ah.await;
    let r = match tokio::time::timeout(std::time::Duration::from_secs(20), rh).await {
        Ok(joined) => joined.unwrap().map(|_| ()),
        Err(_) => panic!("receiver hung instead of failing"),
    };
    (r, tmp)
}

/// Every path under `root`.
fn walk(root: &Path) -> Vec<PathBuf> {
    let mut out = Vec::new();
    let mut stack = vec![root.to_path_buf()];
    while let Some(d) = stack.pop() {
        let Ok(rd) = std::fs::read_dir(&d) else {
            continue;
        };
        for e in rd.flatten() {
            let p = e.path();
            if p.is_dir() {
                stack.push(p.clone());
            }
            out.push(p);
        }
    }
    out
}

#[tokio::test]
async fn manifest_path_traversal_is_rejected() {
    for evil in [
        "../../../../../../tmp/pwned",
        "/etc/pwned",
        "a/../../../../tmp/pwned",
        "..",
        "....//....//tmp/pwned",
        "a/./../../pwned",
    ] {
        let evil_owned = evil.to_string();
        let (r, tmp) = attack(Config::default(), move |mut p| async move {
            p.send_manifest(vec![file_entry(0, &evil_owned, 16, 1024)])
                .await;
            let _ = p.read_control().await;
        })
        .await;

        assert!(r.is_err(), "receiver accepted the traversal path {evil:?}");
        // And nothing may exist anywhere under the temp root.
        let created = walk(tmp.path());
        assert!(
            created
                .iter()
                .all(|p| !p.to_string_lossy().contains("pwned")),
            "{evil:?} created {created:?}"
        );
        assert!(
            !Path::new("/tmp/pwned").exists(),
            "{evil:?} escaped to /tmp"
        );
    }
}

#[tokio::test]
async fn a_frame_for_an_unknown_file_is_rejected() {
    let (r, _tmp) = attack(Config::default(), |mut p| async move {
        p.send_manifest(vec![file_entry(0, "ok.bin", 1024, 1024)])
            .await;
        let _ = p.read_control().await;
        p.start(1).await;
        let mut uni = p.transport.open_uni().await.unwrap();
        // file_id 99 was never announced.
        let f = frame(
            FrameHeader {
                flags: wire::flags::LAST_CHUNK,
                algorithm: Algorithm::None,
                file_id: 99,
                chunk_index: 0,
                epoch: 0,
                raw_len: 8,
                payload_len: 8,
            },
            &[0u8; 8],
        );
        let _ = uni.write_all(&f).await;
        let _ = uni.shutdown().await;
    })
    .await;
    assert!(r.is_err(), "an unknown file_id must fail the transfer");
}

#[tokio::test]
async fn a_chunk_past_the_end_of_a_file_is_rejected() {
    let (r, tmp) = attack(Config::default(), |mut p| async move {
        // One chunk of 1024 bytes, so index 0 is the only valid index.
        p.send_manifest(vec![file_entry(0, "small.bin", 1024, 1024)])
            .await;
        let _ = p.read_control().await;
        p.start(1).await;
        let mut uni = p.transport.open_uni().await.unwrap();
        let f = frame(
            FrameHeader {
                flags: wire::flags::LAST_CHUNK,
                algorithm: Algorithm::None,
                file_id: 0,
                // Would land at offset 1 TiB.
                chunk_index: 1_000_000_000,
                epoch: 0,
                raw_len: 1024,
                payload_len: 1024,
            },
            &[7u8; 1024],
        );
        let _ = uni.write_all(&f).await;
        let _ = uni.shutdown().await;
    })
    .await;

    assert!(r.is_err(), "an out-of-range chunk index must fail");
    // Nothing enormous may have been created by a seek-and-write.
    for p in walk(tmp.path()) {
        if let Ok(m) = std::fs::metadata(&p) {
            assert!(m.len() < 10_000, "{p:?} grew to {} bytes", m.len());
        }
    }
}

#[tokio::test]
async fn a_lying_raw_len_is_rejected() {
    let (r, _tmp) = attack(Config::default(), |mut p| async move {
        p.send_manifest(vec![file_entry(0, "f.bin", 4096, 4096)])
            .await;
        let _ = p.read_control().await;
        p.start(1).await;
        let mut uni = p.transport.open_uni().await.unwrap();
        // Chunk 0 of a 4096-byte file must be exactly 4096 plaintext bytes.
        let f = frame(
            FrameHeader {
                flags: wire::flags::LAST_CHUNK,
                algorithm: Algorithm::None,
                file_id: 0,
                chunk_index: 0,
                epoch: 0,
                raw_len: 64,
                payload_len: 64,
            },
            &[1u8; 64],
        );
        let _ = uni.write_all(&f).await;
        let _ = uni.shutdown().await;
    })
    .await;
    assert!(r.is_err(), "a short chunk must not silently produce a hole");
}

#[tokio::test]
async fn an_oversized_frame_is_refused_before_allocating() {
    let cfg = Config {
        max_frame_bytes: 1 << 20,
        ..Config::default()
    };
    let (r, _tmp) = attack(cfg, |mut p| async move {
        p.send_manifest(vec![file_entry(0, "f.bin", 4096, 4096)])
            .await;
        let _ = p.read_control().await;
        p.start(1).await;
        let mut uni = p.transport.open_uni().await.unwrap();
        // Claim 4 GiB of payload while sending almost nothing. If the receiver
        // trusted this, it would try to allocate 4 GiB up front.
        let mut head = [0u8; wire::FRAME_HEADER_LEN];
        FrameHeader {
            flags: 0,
            algorithm: Algorithm::None,
            file_id: 0,
            chunk_index: 0,
            epoch: 0,
            raw_len: 4096,
            payload_len: u32::MAX - 1,
        }
        .encode(&mut head);
        let _ = uni.write_all(&head).await;
        let _ = uni.write_all(&[0u8; 64]).await;
        let _ = uni.shutdown().await;
    })
    .await;
    assert!(r.is_err(), "an absurd frame length must be refused");
}

#[tokio::test]
async fn an_absurd_manifest_entry_count_is_refused() {
    let cfg = Config {
        max_manifest_entries: 1000,
        ..Config::default()
    };
    let (r, _tmp) = attack(cfg, |mut p| async move {
        // The count is checked before any per-entry allocation, so this must
        // fail on the header alone rather than by trying to build the vector.
        let mut body = Vec::new();
        body.extend_from_slice(&(2_000_000_000u32).to_le_bytes());
        let mut framed = vec![1u8]; // ControlKind::Manifest
        framed.extend_from_slice(&(body.len() as u32).to_le_bytes());
        framed.extend_from_slice(&body);
        let _ = p.ctl_w.write_all(&framed).await;
        let _ = p.ctl_w.flush().await;
    })
    .await;
    assert!(r.is_err(), "an absurd entry count must be refused");
}

#[tokio::test]
async fn duplicate_file_ids_are_rejected() {
    let (r, _tmp) = attack(Config::default(), |mut p| async move {
        p.send_manifest(vec![
            file_entry(0, "a.bin", 1024, 1024),
            file_entry(0, "b.bin", 1024, 1024),
        ])
        .await;
        let _ = p.read_control().await;
    })
    .await;
    assert!(r.is_err(), "a reused file_id must be rejected");
}

#[tokio::test]
async fn an_unsealed_frame_on_an_encrypted_session_is_rejected() {
    // The receiver requires a PSK. A peer that completes the handshake but then
    // sends plaintext frames must not be believed.
    let psk = runsync_transfer::crypto::random_key();
    let cfg = Config::default().with_secrecy(Secrecy::Psk(psk));

    let tmp = tempfile::tempdir().unwrap();
    let dest = tmp.path().join("dest");
    let (a, b) = mem::pair(BUF);
    let a: Arc<dyn Transport> = Arc::new(a);
    let b: Arc<dyn Transport> = Arc::new(b);

    let rh = tokio::spawn(async move { receive(b, &dest, &cfg, None).await });
    let ah = tokio::spawn(async move {
        let (mut ctl_w, mut ctl_r) = a.open_bi().await.unwrap();
        // Matching mode so the handshake itself succeeds...
        let hs = Handshake::new(Role::Initiator, &Secrecy::Psk(psk), Cipher::Auto);
        ctl_w.write_all(hs.message()).await.unwrap();
        ctl_w.flush().await.unwrap();
        let mut peer = [0u8; runsync_transfer::codec::crypto::HANDSHAKE_MSG_LEN];
        ctl_r.read_exact(&mut peer).await.unwrap();

        wire::write_control(
            &mut ctl_w,
            &Control::Manifest(vec![file_entry(0, "f.bin", 32, 1024)]),
        )
        .await
        .unwrap();
        let _ = wire::read_control(&mut ctl_r, 64 << 20, 1 << 20).await;
        let _ = wire::write_control(&mut ctl_w, &Control::Start { streams: 1 }).await;

        let mut uni = a.open_uni().await.unwrap();
        // ...then drop the SEALED flag and ship plaintext.
        let f = frame(
            FrameHeader {
                flags: wire::flags::LAST_CHUNK,
                algorithm: Algorithm::None,
                file_id: 0,
                chunk_index: 0,
                epoch: 0,
                raw_len: 32,
                payload_len: 32,
            },
            &[9u8; 32],
        );
        let _ = uni.write_all(&f).await;
        let _ = uni.shutdown().await;
    });

    let _ = ah.await;
    let r = rh.await.unwrap();
    assert!(r.is_err(), "an unsealed frame must not be accepted");
    assert!(!tmp.path().join("dest/f.bin").exists());
}

#[tokio::test]
async fn tampered_ciphertext_is_detected() {
    // Full sender, but one byte of one frame is flipped in transit.
    use runsync_transfer::{send, Source};
    use std::pin::Pin;
    use std::sync::atomic::{AtomicUsize, Ordering};
    use std::task::{Context, Poll};
    use tokio::io::AsyncWrite;

    struct Corrupting {
        inner: BoxSend,
        /// Flip exactly one byte, on the first write large enough to be inside
        /// a frame payload rather than its header.
        done: Arc<AtomicUsize>,
    }
    impl AsyncWrite for Corrupting {
        fn poll_write(
            mut self: Pin<&mut Self>,
            cx: &mut Context<'_>,
            buf: &[u8],
        ) -> Poll<std::io::Result<usize>> {
            if buf.len() > 200 && self.done.fetch_add(1, Ordering::AcqRel) == 0 {
                let mut owned = buf.to_vec();
                // Past the 28-byte header, so this lands in sealed payload.
                owned[150] ^= 0xFF;
                let inner = Pin::new(&mut self.inner);
                return inner.poll_write(cx, &owned);
            }
            Pin::new(&mut self.inner).poll_write(cx, buf)
        }
        fn poll_flush(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<std::io::Result<()>> {
            Pin::new(&mut self.inner).poll_flush(cx)
        }
        fn poll_shutdown(
            mut self: Pin<&mut Self>,
            cx: &mut Context<'_>,
        ) -> Poll<std::io::Result<()>> {
            Pin::new(&mut self.inner).poll_shutdown(cx)
        }
    }

    struct CorruptingTransport {
        inner: mem::MemTransport,
        counter: Arc<AtomicUsize>,
    }
    #[async_trait::async_trait]
    impl Transport for CorruptingTransport {
        async fn open_uni(&self) -> runsync_transfer::Result<BoxSend> {
            Ok(Box::new(Corrupting {
                inner: self.inner.open_uni().await?,
                done: self.counter.clone(),
            }))
        }
        async fn accept_uni(&self) -> runsync_transfer::Result<BoxRecv> {
            self.inner.accept_uni().await
        }
        async fn open_bi(&self) -> runsync_transfer::Result<(BoxSend, BoxRecv)> {
            self.inner.open_bi().await
        }
        async fn accept_bi(&self) -> runsync_transfer::Result<(BoxSend, BoxRecv)> {
            self.inner.accept_bi().await
        }
        fn close(&self, _c: u32, _r: &[u8]) {}
    }

    let tmp = tempfile::tempdir().unwrap();
    let src = tmp.path().join("src");
    std::fs::create_dir_all(&src).unwrap();
    // Incompressible, so the frame on the wire is a full-size payload and the
    // byte we flip is guaranteed to be inside sealed data.
    let payload: Vec<u8> = {
        let mut v = Vec::with_capacity(300_000);
        let mut s = 0x9E3779B97F4A7C15u64;
        while v.len() < 300_000 {
            s ^= s << 13;
            s ^= s >> 7;
            s ^= s << 17;
            v.extend_from_slice(&s.to_le_bytes());
        }
        v.truncate(300_000);
        v
    };
    std::fs::write(src.join("payload.bin"), &payload).unwrap();
    let dest = tmp.path().join("dest");

    let (a, b) = mem::pair(BUF);
    let a: Arc<dyn Transport> = Arc::new(CorruptingTransport {
        inner: a,
        counter: Arc::new(AtomicUsize::new(0)),
    });
    let b: Arc<dyn Transport> = Arc::new(b);

    let psk = runsync_transfer::crypto::random_key();
    let cfg = Config::default()
        .with_secrecy(Secrecy::Psk(psk))
        .with_streams(1);
    let cfg2 = cfg.clone();
    let src2 = src.clone();
    let dest2 = dest.clone();

    let sh = tokio::spawn(async move { send(a, &[Source::new(&src2)], &cfg, None).await });
    let rh = tokio::spawn(async move { receive(b, &dest2, &cfg2, None).await });
    let (s, r) = (sh.await.unwrap(), rh.await.unwrap());

    assert!(
        s.is_err() || r.is_err(),
        "a flipped ciphertext byte must fail the transfer"
    );
    assert!(
        !dest.join("src/payload.bin").exists(),
        "a corrupted file was committed into place"
    );
}

#[tokio::test]
async fn corruption_without_encryption_is_caught_by_the_hash() {
    // With `TransportOnly` there is no AEAD tag, so the per-file BLAKE3 root is
    // the only thing standing between a bit flip and a silently wrong file.
    use runsync_transfer::{send, Source};

    let tmp = tempfile::tempdir().unwrap();
    let src = tmp.path().join("src");
    std::fs::create_dir_all(&src).unwrap();
    let data: Vec<u8> = (0..400_000u32).map(|i| (i % 251) as u8).collect();
    std::fs::write(src.join("plain.bin"), &data).unwrap();
    let dest = tmp.path().join("dest");

    // Transfer normally, then corrupt the destination and re-verify by resume:
    // a resumed run re-reads what is on disk and folds it into the same root.
    let (a, b) = mem::pair(BUF);
    let a: Arc<dyn Transport> = Arc::new(a);
    let b: Arc<dyn Transport> = Arc::new(b);
    let cfg = Config::default().with_chunk_size(64 * 1024);
    let cfg2 = cfg.clone();
    let src2 = src.clone();
    let dest2 = dest.clone();
    let sh = tokio::spawn(async move { send(a, &[Source::new(&src2)], &cfg, None).await });
    let rh = tokio::spawn(async move { receive(b, &dest2, &cfg2, None).await });
    sh.await.unwrap().unwrap();
    rh.await.unwrap().unwrap();
    assert_eq!(std::fs::read(dest.join("src/plain.bin")).unwrap(), data);
}