crates/gunnar-core/tests/hostile_ipc.rs
//! Hostile Arrow IPC handed to gunnar-core's Arrow readers must be an `Err` (or
//! the clean, shorter read a log with a torn tail gives), never a panic, an
//! abort or a giant allocation. The inputs and the child-process harness are in
//! `tests/hostile/sweep.rs`.
#[path = "hostile/sweep.rs"]
mod hostile;
use std::sync::Arc;
use arrow_array::builder::{BinaryBuilder, FixedSizeBinaryBuilder, Int64Builder, StringBuilder};
use arrow_array::{ArrayRef, RecordBatch};
use arrow_ipc::writer::StreamWriter;
use arrow_schema::{DataType, Field, Schema};
use gunnar_core::arrow_log::{ArrowLogStore, REFS_LOG, SECRETS_LOG};
use gunnar_core::history::{live_mark, read_live_since, read_sealed_bytes, History};
use gunnar_store::SecretStore as _;
use hostile::{child_case, run_in_child, sweep};
const CHILD: &str = "hostile_child";
const KIND: gix_hash::Kind = gix_hash::Kind::Sha1;
/// A log as the store writes one: a schema message, then one batch per push,
/// no end-of-stream marker.
fn log(schema: Arc<Schema>, batches: &[RecordBatch]) -> Vec<u8> {
let mut out = Vec::new();
let mut w = StreamWriter::try_new(&mut out, &schema).unwrap();
for b in batches {
w.write(b).unwrap();
}
drop(w);
out
}
fn ref_schema() -> Arc<Schema> {
Arc::new(Schema::new(vec![
Field::new("refname", DataType::Utf8, false),
Field::new("old_oid", DataType::FixedSizeBinary(20), true),
Field::new("new_oid", DataType::FixedSizeBinary(20), true),
Field::new("old_symref", DataType::Utf8, true),
Field::new("new_symref", DataType::Utf8, true),
Field::new("actor", DataType::Utf8, false),
Field::new("at_unix_ms", DataType::Int64, false),
]))
}
fn ref_push(name: &str, old: Option<u8>, new: Option<u8>, at: i64) -> RecordBatch {
let mut n = StringBuilder::new();
let mut o = FixedSizeBinaryBuilder::new(20);
let mut w = FixedSizeBinaryBuilder::new(20);
let mut os = StringBuilder::new();
let mut ns = StringBuilder::new();
let mut a = StringBuilder::new();
let mut t = Int64Builder::new();
n.append_value(name);
match old {
Some(b) => o.append_value([b; 20]).unwrap(),
None => o.append_null(),
}
match new {
Some(b) => w.append_value([b; 20]).unwrap(),
None => w.append_null(),
}
os.append_null();
ns.append_null();
a.append_value("ann");
t.append_value(at);
let cols: Vec<ArrayRef> = vec![
Arc::new(n.finish()),
Arc::new(o.finish()),
Arc::new(w.finish()),
Arc::new(os.finish()),
Arc::new(ns.finish()),
Arc::new(a.finish()),
Arc::new(t.finish()),
];
RecordBatch::try_new(ref_schema(), cols).unwrap()
}
fn refs_log() -> Vec<u8> {
log(
ref_schema(),
&[
ref_push("refs/heads/main", None, Some(1), 1),
ref_push("refs/heads/main", Some(1), Some(2), 2),
],
)
}
fn secrets_log() -> Vec<u8> {
let schema = Arc::new(Schema::new(vec![
Field::new("key", DataType::Utf8, false),
Field::new("value", DataType::Binary, true),
Field::new("actor", DataType::Utf8, false),
Field::new("at_unix_ms", DataType::Int64, false),
]));
let mut k = StringBuilder::new();
let mut v = BinaryBuilder::new();
let mut a = StringBuilder::new();
let mut t = Int64Builder::new();
k.append_value("deploy-key");
v.append_value(b"secret-bytes");
a.append_value("ann");
t.append_value(1);
let cols: Vec<ArrayRef> = vec![
Arc::new(k.finish()),
Arc::new(v.finish()),
Arc::new(a.finish()),
Arc::new(t.finish()),
];
let b = RecordBatch::try_new(schema.clone(), cols).unwrap();
log(schema, &[b])
}
#[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();
let d = dir.path();
match case.as_str() {
"history" => {
let valid = refs_log();
std::fs::write(d.join(REFS_LOG), &valid).unwrap();
assert_eq!(History::read(d, KIND).unwrap().pushes(), 2);
sweep(&valid, |_, b| {
std::fs::write(d.join(REFS_LOG), b).unwrap();
let _ = History::read(d, KIND);
});
}
"live" => {
// A mark taken on the schema alone, then every hostile tail read
// since it (written in place, so the inode and the marked prefix
// stay the same and the tail is what is decoded).
let valid = refs_log();
let schema_len = 8 + u32::from_le_bytes(valid[4..8].try_into().unwrap()) as usize;
std::fs::write(d.join(REFS_LOG), &valid[..schema_len]).unwrap();
let mark = live_mark(d).unwrap();
std::fs::write(d.join(REFS_LOG), &valid).unwrap();
assert_eq!(read_live_since(d, KIND, &mark).unwrap().unwrap().len(), 2);
sweep(&valid, |_, b| {
std::fs::write(d.join(REFS_LOG), b).unwrap();
let _ = live_mark(d);
let _ = read_live_since(d, KIND, &mark);
});
}
"sealed" => {
std::fs::write(d.join(REFS_LOG), refs_log()).unwrap();
let valid = History::read(d, KIND).unwrap().seal_to_vec().unwrap();
assert_eq!(read_sealed_bytes(&valid).unwrap().len(), 2);
sweep(&valid, |_, b| {
let _ = read_sealed_bytes(b);
});
}
"secrets" => {
let valid = secrets_log();
std::fs::write(d.join(SECRETS_LOG), &valid).unwrap();
{
let s = ArrowLogStore::open(d, KIND).unwrap();
assert_eq!(
s.secrets().get("deploy-key").unwrap().as_deref(),
Some(&b"secret-bytes"[..])
);
}
sweep(&valid, |_, b| {
std::fs::write(d.join(SECRETS_LOG), b).unwrap();
if let Ok(s) = ArrowLogStore::open(d, KIND) {
let _ = s.secrets().get("deploy-key");
}
});
}
other => panic!("unknown case {other}"),
}
}
#[test]
fn a_hostile_reference_log_is_an_error() {
run_in_child(CHILD, "history");
}
#[test]
fn a_hostile_tail_since_a_live_mark_is_an_error() {
run_in_child(CHILD, "live");
}
#[test]
fn a_hostile_sealed_history_is_an_error() {
run_in_child(CHILD, "sealed");
}
#[test]
fn a_hostile_secret_log_is_an_error() {
run_in_child(CHILD, "secrets");
}