znippy-plugin-git 0.2.2

//! Hostile Arrow IPC inputs for every read path of this crate, and the child
//! process the cases run in.
//!
//! Shared by the integration test `tests/hostile_ipc.rs` and by the unit tests
//! of the private decoders (`#[path]`-included from `src/lib.rs`), so every read
//! path faces exactly the same inputs:
//!
//! * the regression seeds of the Heimdall verify-bundle lane (znippy
//!   `tests/fixtures/hostile-ipc/`): every Arrow IPC stream found inside them —
//!   a schema with no field vector, a short validity bitmap, a buffer past its
//!   body — fed to the decoder as its own input;
//! * the valid input with every record batch's `bodyLength` set to 1 TiB, to a
//!   negative value and to `i64::MAX` (arrow-ipc allocates the body before it
//!   reads a byte of it: an abort no `catch_unwind` stops);
//! * every truncation and every single-byte flip (five values per byte) of the
//!   valid input.
//!
//! Each case runs in a CHILD process — the test binary re-run on one ignored
//! test — because the failure being guarded against is an abort that would take
//! the whole test binary down. The parent asserts the child exited cleanly and
//! printed no panic.

#![allow(dead_code)]

use std::path::{Path, PathBuf};
use std::process::Command;

use znippy_common::arrow::ipc::{MessageHeader, root_as_message};

/// The environment variable that names the case a child runs.
pub const CASE_ENV: &str = "GUNGUARD_HOSTILE_CASE";

/// The case this process is a child for, if it is one.
pub fn child_case() -> Option<String> {
    std::env::var(CASE_ENV).ok()
}

/// Re-run this test binary on the ignored test `child_test` (its full path, as
/// `--exact` wants it) for `case`, and fail unless the child ends cleanly.
pub fn run_in_child(child_test: &str, case: &str) {
    let exe = std::env::current_exe().unwrap();
    let out = Command::new(exe)
        .args([
            child_test,
            "--exact",
            "--ignored",
            "--nocapture",
            "--test-threads=1",
        ])
        .env(CASE_ENV, case)
        .output()
        .unwrap();
    let stdout = String::from_utf8_lossy(&out.stdout);
    let stderr = String::from_utf8_lossy(&out.stderr);
    assert!(
        out.status.success() && !stderr.contains("panicked") && stdout.contains("HOSTILE-DONE"),
        "hostile case {case}: the child did not end cleanly ({:?})\n--- stdout\n{}\n--- stderr\n{}",
        out.status,
        tail(&stdout),
        tail(&stderr)
    );
}

fn tail(s: &str) -> &str {
    let n = s.len();
    let mut at = n.saturating_sub(4000);
    while !s.is_char_boundary(at) {
        at += 1;
    }
    &s[at..]
}

/// Where the Heimdall seeds live: the znippy repository's shared fixtures.
pub fn fixture_dir() -> PathBuf {
    Path::new(env!("CARGO_MANIFEST_DIR")).join("../../../tests/fixtures/hostile-ipc")
}

/// Every Arrow IPC stream inside every seed: the bytes from each schema message
/// to the end of the file.
pub fn seed_streams() -> Vec<(String, Vec<u8>)> {
    let mut out = Vec::new();
    let mut names: Vec<_> = std::fs::read_dir(fixture_dir())
        .expect("znippy tests/fixtures/hostile-ipc")
        .map(|e| e.unwrap().path())
        .collect();
    names.sort();
    for p in names {
        let bytes = std::fs::read(&p).unwrap();
        let name = p.file_name().unwrap().to_string_lossy().into_owned();
        for at in schema_starts(&bytes) {
            out.push((format!("{name}@{at}"), bytes[at..].to_vec()));
        }
        out.push((format!("{name}:whole"), bytes));
    }
    assert!(out.len() > 6, "no IPC stream found in the seeds");
    out
}

/// Offsets of every `0xFFFFFFFF <len>` frame whose metadata is a schema message.
fn schema_starts(b: &[u8]) -> Vec<usize> {
    let mut v = Vec::new();
    let mut i = 0;
    while i + 8 <= b.len() {
        if b[i..i + 4] == [0xFF; 4] {
            let len = u32::from_le_bytes(b[i + 4..i + 8].try_into().unwrap()) as usize;
            if (8..=1 << 20).contains(&len)
                && i + 8 + len <= b.len()
                && root_as_message(&b[i + 8..i + 8 + len])
                    .is_ok_and(|m| m.header_type() == MessageHeader::Schema)
            {
                v.push(i);
            }
        }
        i += 1;
    }
    v
}

/// `(metadata start, metadata length, body length)` of every message in a run of
/// messages (one stream or several concatenated), stopping at the first that
/// does not frame.
pub fn messages(b: &[u8]) -> Vec<(usize, usize, i64)> {
    let mut out = Vec::new();
    let mut pos = 0usize;
    while pos + 8 <= b.len() {
        let mut len = u32::from_le_bytes(b[pos..pos + 4].try_into().unwrap()) as usize;
        pos += 4;
        if len == 0xFFFF_FFFF {
            len = u32::from_le_bytes(b[pos..pos + 4].try_into().unwrap()) as usize;
            pos += 4;
        }
        if len == 0 {
            continue; // end-of-stream marker; a log may carry another stream after it
        }
        if pos + len > b.len() {
            break;
        }
        let Ok(m) = root_as_message(&b[pos..pos + len]) else {
            break;
        };
        let body = m.bodyLength();
        out.push((pos, len, body));
        if body < 0 || pos + len + body as usize > b.len() {
            break;
        }
        pos += len + body as usize;
    }
    out
}

/// `valid` with the `bodyLength` of every record batch message set to `value`,
/// one variant per message.
pub fn body_length_variants(valid: &[u8], value: i64) -> Vec<Vec<u8>> {
    let mut out = Vec::new();
    for (at, len, body) in messages(valid) {
        let meta = &valid[at..at + len];
        if root_as_message(meta).unwrap().header_type() != MessageHeader::RecordBatch {
            continue;
        }
        let Some(p) = meta.windows(8).position(|w| w == body.to_le_bytes()) else {
            continue;
        };
        let mut m = valid.to_vec();
        m[at + p..at + p + 8].copy_from_slice(&value.to_le_bytes());
        out.push(m);
    }
    out
}

/// Every hostile variant of `valid`, plus every seed stream. `feed` is called on
/// each; it must return (an `Err` or an `Ok`) and never panic or abort.
pub fn sweep(valid: &[u8], mut feed: impl FnMut(&str, &[u8])) -> usize {
    let mut n = 0usize;
    let mut run = |what: &str, b: &[u8]| {
        feed(what, b);
        n += 1;
    };
    run("valid", valid);
    for v in [1i64 << 40, -8, i64::MAX] {
        let vs = body_length_variants(valid, v);
        assert!(!vs.is_empty(), "the valid input carries no record batch");
        for m in vs {
            run(&format!("bodyLength={v}"), &m);
        }
    }
    // Every 4-aligned 8-byte word of every record batch's metadata set to a
    // huge value: row counts, node lengths, buffer offsets and lengths that a
    // reader might size an allocation by (a fuzz find: an exploded table sized
    // its index by the declared row count).
    for (at, len, _) in messages(valid) {
        let meta = &valid[at..at + len];
        if root_as_message(meta).unwrap().header_type() != MessageHeader::RecordBatch {
            continue;
        }
        for w in (0..len.saturating_sub(7)).step_by(4) {
            for v in [(1i64 << 31) - 1, 1i64 << 40] {
                let mut m = valid.to_vec();
                m[at + w..at + w + 8].copy_from_slice(&v.to_le_bytes());
                run("huge word", &m);
            }
        }
    }
    for (name, s) in seed_streams() {
        run(&name, &s);
    }
    for cut in 0..valid.len() {
        run("truncated", &valid[..cut]);
    }
    for i in 0..valid.len() {
        for v in [0x00u8, 0xFF, 0x7F, 0x80, valid[i] ^ 0x01] {
            if v == valid[i] {
                continue;
            }
            let mut m = valid.to_vec();
            m[i] = v;
            run("flip", &m);
        }
    }
    println!("HOSTILE-DONE {n} inputs");
    n
}