plugins/native/znippy-plugin-git/tests/hostile_ipc.rs
//! Hostile Arrow IPC handed to this crate's public read paths must be an `Err`
//! (or a clean, shorter read where the format is a log with a torn tail), never
//! a panic, an abort or a giant allocation. The inputs and the child-process
//! harness are in `tests/common/hostile.rs`.
#[path = "common/hostile.rs"]
mod hostile;
use std::sync::Arc;
use hostile::{child_case, run_in_child, sweep};
use znippy_common::arrow::array::{BinaryBuilder, RecordBatch, StringBuilder, UInt64Builder};
use znippy_common::arrow::ipc::writer::StreamWriter;
use znippy_plugin_git::archive_write::{encode_journal_row, encode_journal_schema, read_journal};
use znippy_plugin_git::exploded_arrow::{ExplodePolicy, ExplodedArchive};
use znippy_plugin_git::{
CommitNode, GitObjectKind, decode_graph, decode_reach, reach_schema, scan_frames,
};
const CHILD: &str = "hostile_child";
/// An allocation bigger than this fails, as it would on a box with that much
/// memory free (the kernel overcommits a bare `malloc` here, so without the cap
/// a reader that sizes a buffer by a hostile count passes on a big test host).
const ALLOC_CAP: usize = 1 << 30;
struct Capped;
// SAFETY: forwards to the system allocator unchanged, except that an oversized
// request returns null, which `GlobalAlloc` allows.
unsafe impl std::alloc::GlobalAlloc for Capped {
unsafe fn alloc(&self, l: std::alloc::Layout) -> *mut u8 {
if l.size() > ALLOC_CAP {
return std::ptr::null_mut();
}
unsafe { std::alloc::System.alloc(l) }
}
unsafe fn dealloc(&self, p: *mut u8, l: std::alloc::Layout) {
unsafe { std::alloc::System.dealloc(p, l) }
}
unsafe fn alloc_zeroed(&self, l: std::alloc::Layout) -> *mut u8 {
if l.size() > ALLOC_CAP {
return std::ptr::null_mut();
}
unsafe { std::alloc::System.alloc_zeroed(l) }
}
unsafe fn realloc(&self, p: *mut u8, l: std::alloc::Layout, n: usize) -> *mut u8 {
if n > ALLOC_CAP {
return std::ptr::null_mut();
}
unsafe { std::alloc::System.realloc(p, l, n) }
}
}
#[global_allocator]
static CAPPED: Capped = Capped;
fn stream(batch: &RecordBatch) -> Vec<u8> {
let mut out = Vec::new();
let mut w = StreamWriter::try_new(&mut out, &batch.schema()).unwrap();
w.write(batch).unwrap();
w.finish().unwrap();
out
}
fn graph_bytes() -> Vec<u8> {
let nodes = vec![
CommitNode {
oid: "a".repeat(40),
parents: vec![],
tree: Some("t".repeat(40)),
committer_time: Some(1),
generation: 1,
},
CommitNode {
oid: "b".repeat(40),
parents: vec!["a".repeat(40)],
tree: None,
committer_time: None,
generation: 2,
},
];
znippy_plugin_git::graph::encode_graph(&nodes).unwrap()
}
fn reach_bytes() -> Vec<u8> {
let mut bm = Vec::new();
let mut r = roaring_bitmap();
r.extend([1u32, 5, 9]);
r.serialize_into(&mut bm).unwrap();
let mut o = StringBuilder::new();
let mut c = UInt64Builder::new();
let mut b = BinaryBuilder::new();
o.append_value("a".repeat(40));
c.append_value(3);
b.append_value(&bm);
let batch = RecordBatch::try_new(
reach_schema(),
vec![
Arc::new(o.finish()),
Arc::new(c.finish()),
Arc::new(b.finish()),
],
)
.unwrap();
stream(&batch)
}
fn roaring_bitmap() -> roaring::RoaringBitmap {
roaring::RoaringBitmap::new()
}
/// Two push-log frames: each a whole stream (schema, one batch, end marker).
fn pushlog_bytes() -> Vec<u8> {
let schema = Arc::new(znippy_common::arrow::datatypes::Schema::new(vec![
znippy_common::arrow::datatypes::Field::new(
"name",
znippy_common::arrow::datatypes::DataType::Utf8,
false,
),
]));
let mut out = Vec::new();
for n in ["refs/heads/main", "refs/heads/x"] {
let mut s = StringBuilder::new();
s.append_value(n);
let batch = RecordBatch::try_new(schema.clone(), vec![Arc::new(s.finish())]).unwrap();
out.extend(stream(&batch));
}
out
}
fn journal_bytes() -> Vec<u8> {
let mut out = encode_journal_schema().unwrap();
out.extend(encode_journal_row(0, 100).unwrap());
out.extend(encode_journal_row(100, 7).unwrap());
out
}
fn oid(n: u8) -> [u8; 20] {
[n; 20]
}
fn exploded_bytes(dir: &std::path::Path) -> Vec<u8> {
let p = dir.join("valid.arrow");
let t = ExplodedArchive::open_with_policy(&p, ExplodePolicy::Full).unwrap();
t.explode_at(
&oid(1),
GitObjectKind::Commit,
Some(0o100644),
Some("a/b"),
b"tree x\n",
)
.unwrap();
t.explode(&oid(2), GitObjectKind::Blob, &[b'z'; 3000])
.unwrap();
t.flush().unwrap();
drop(t);
std::fs::read(&p).unwrap()
}
/// Every read an exploded table offers, on a file holding `bytes`.
fn exercise_exploded(path: &std::path::Path, bytes: &[u8]) {
std::fs::write(path, bytes).unwrap();
let Ok(t) = ExplodedArchive::open_with_policy(path, ExplodePolicy::Full) else {
return;
};
let _ = t.rows();
for n in [1u8, 2, 9] {
let _ = t.has(&oid(n));
let _ = t.content(&oid(n));
let _ = t.name(&oid(n));
}
let _ = t.of_kind(GitObjectKind::Blob);
}
#[test]
#[ignore = "run as a child by the tests below"]
fn hostile_child() {
let Some(case) = child_case() else { return };
let dir = tempfile::tempdir().unwrap();
match case.as_str() {
"graph" => {
let valid = graph_bytes();
assert_eq!(decode_graph(&valid).unwrap().len(), 2);
sweep(&valid, |_, b| {
let _ = decode_graph(b);
});
}
"reach" => {
let valid = reach_bytes();
assert_eq!(decode_reach(&valid).unwrap().len(), 1);
sweep(&valid, |_, b| {
let _ = decode_reach(b);
});
}
"pushlog" => {
let valid = pushlog_bytes();
assert_eq!(scan_frames(&valid).pushes.len(), 2);
sweep(&valid, |_, b| {
let _ = scan_frames(b);
});
}
"journal" => {
let valid = journal_bytes();
let p = dir.path().join("journal");
std::fs::write(&p, &valid).unwrap();
assert_eq!(read_journal(&p).unwrap(), vec![(0, 100), (100, 7)]);
sweep(&valid, |_, b| {
std::fs::write(&p, b).unwrap();
let _ = read_journal(&p);
});
}
"exploded" => {
let valid = exploded_bytes(dir.path());
let p = dir.path().join("t.arrow");
exercise_exploded(&p, &valid);
{
let t = ExplodedArchive::open_with_policy(&p, ExplodePolicy::Full).unwrap();
assert_eq!(t.content(&oid(2)).unwrap().unwrap().1, vec![b'z'; 3000]);
}
sweep(&valid, |_, b| exercise_exploded(&p, b));
}
other => panic!("unknown case {other}"),
}
}
#[test]
fn hostile_graph_sections_are_errors() {
run_in_child(CHILD, "graph");
}
#[test]
fn hostile_reach_sections_are_errors() {
run_in_child(CHILD, "reach");
}
#[test]
fn hostile_push_logs_stop_at_the_first_bad_frame() {
run_in_child(CHILD, "pushlog");
}
#[test]
fn hostile_journals_keep_only_whole_rows() {
run_in_child(CHILD, "journal");
}
#[test]
fn hostile_exploded_tables_are_errors() {
run_in_child(CHILD, "exploded");
}