Verstack Local
f73167de85
Package Apple Silicon VM exports and cached runtime switching with local cabinet controls and conagent provisioning. Preserve experimental SPIKE 2 emulation. Improve preparation concurrency, Ghidra checkpoints, signature importing, sound indexing and client-side spectrograms. Include regression tests and validation notes.
392 lines
14 KiB
Rust
392 lines
14 KiB
Rust
use anyhow::Result;
|
|
use serde_json::json;
|
|
use std::{
|
|
fs,
|
|
path::Path,
|
|
sync::Arc,
|
|
time::{Duration, Instant},
|
|
};
|
|
use verstack::{
|
|
Archive, Config, Release,
|
|
imports::{ImportOptions, ImportRequest, ImportSource},
|
|
};
|
|
|
|
fn setup() -> Result<(tempfile::TempDir, Config)> {
|
|
let tmp = tempfile::tempdir()?;
|
|
let root = tmp.path();
|
|
fs::create_dir(root.join("input"))?;
|
|
fs::create_dir(root.join("signals"))?;
|
|
fs::write(
|
|
root.join("tool.py"),
|
|
r#"import json,os,pathlib,sys,time
|
|
r=json.loads(pathlib.Path(sys.argv[1]).read_text())
|
|
i=next(p for p in pathlib.Path(r['input_dir']).rglob('*') if p.is_file())
|
|
name=i.stem
|
|
s=pathlib.Path(r['settings']['signals'])
|
|
(s/(name+'.started')).write_text(str(r['workspace_bytes']))
|
|
pathlib.Path(os.environ['VERSTACK_PROGRESS_FILE']).write_text(json.dumps(dict(stage='Analyzing fixture',detail=name)))
|
|
deadline=time.monotonic()+20
|
|
while not (s/(name+'.release')).exists():
|
|
if time.monotonic()>deadline: raise RuntimeError('fixture release timed out')
|
|
time.sleep(.02)
|
|
(pathlib.Path(r['output_dir'])/'result.txt').write_text(name)
|
|
pathlib.Path(r['result_file']).write_text(json.dumps(dict(protocol=1,layer='derived',coverage='complete',warnings=[],files=['result.txt'])))
|
|
"#,
|
|
)?;
|
|
let config = serde_json::from_value(json!({
|
|
"archive":root.join("archive"), "workspace":root.join("work"),
|
|
"require_ram_workspace":false, "workspace_bytes":64*1024*1024,
|
|
"processing_enabled":false, "import_roots":[root.join("input")],
|
|
"plugins":{"ghidra":{"command":["python3",root.join("tool.py")],"version":"fixture","settings":{"signals":root.join("signals")},"timeout_seconds":25}}
|
|
}))?;
|
|
Ok((tmp, config))
|
|
}
|
|
fn request(root: &Path, name: &str, analyze: bool) -> Result<ImportRequest> {
|
|
let path = root.join("input").join(format!("{name}.exe"));
|
|
fs::write(&path, b"MZfixture")?;
|
|
Ok(ImportRequest {
|
|
source: ImportSource::Local { paths: vec![path] },
|
|
release: Release {
|
|
repository: name.into(),
|
|
version: "1".into(),
|
|
edition: String::new(),
|
|
generation: String::new(),
|
|
released_at: None,
|
|
},
|
|
options: ImportOptions {
|
|
extract: false,
|
|
media: false,
|
|
analyze,
|
|
},
|
|
})
|
|
}
|
|
fn wait(mut predicate: impl FnMut() -> bool) {
|
|
let start = Instant::now();
|
|
while !predicate() {
|
|
assert!(
|
|
start.elapsed() < Duration::from_secs(10),
|
|
"concurrent work did not make progress"
|
|
);
|
|
std::thread::sleep(Duration::from_millis(25));
|
|
}
|
|
}
|
|
|
|
#[test]
|
|
fn analysis_overlaps_imports_and_cancellation_is_isolated() -> Result<()> {
|
|
let (tmp, cfg) = setup()?;
|
|
let a = Arc::new(Archive::open(cfg)?);
|
|
let mut tasks = Vec::new();
|
|
for name in ["one", "two", "three"] {
|
|
tasks.push(a.submit_import(request(tmp.path(), name, true)?)?);
|
|
assert!(a.work_import_preparation_one()?);
|
|
}
|
|
let signals = tmp.path().join("signals");
|
|
let worker = a.clone();
|
|
let first = std::thread::spawn(move || worker.work_import_analysis_one());
|
|
wait(|| signals.join("one.started").exists());
|
|
let worker = a.clone();
|
|
let second = std::thread::spawn(move || worker.work_import_analysis_one());
|
|
wait(|| signals.join("two.started").exists());
|
|
assert!(
|
|
!a.work_import_analysis_one()?,
|
|
"third analysis must remain queued"
|
|
);
|
|
assert!(
|
|
a.cleanup_orphans().is_err(),
|
|
"maintenance cannot delete active inputs"
|
|
);
|
|
let plain = a.submit_import(request(tmp.path(), "plain", false)?)?;
|
|
assert!(a.work_import_preparation_one()?);
|
|
assert_eq!(
|
|
a.import_tasks()?
|
|
.iter()
|
|
.find(|t| t.id == plain.id)
|
|
.unwrap()
|
|
.state,
|
|
"completed"
|
|
);
|
|
wait(|| {
|
|
let view = a.activity_view().unwrap();
|
|
["one", "two"].iter().all(|name| {
|
|
view["items"]
|
|
.as_array()
|
|
.unwrap()
|
|
.iter()
|
|
.any(|t| t["release"]["repository"] == *name && t["progress"]["detail"] == *name)
|
|
})
|
|
});
|
|
a.control_import(&tasks[0].id, "cancel")?;
|
|
assert!(first.join().unwrap()?);
|
|
assert_eq!(
|
|
a.import_tasks()?
|
|
.iter()
|
|
.find(|t| t.id == tasks[0].id)
|
|
.unwrap()
|
|
.state,
|
|
"cancelled"
|
|
);
|
|
assert_eq!(
|
|
a.import_tasks()?
|
|
.iter()
|
|
.find(|t| t.id == tasks[1].id)
|
|
.unwrap()
|
|
.state,
|
|
"running"
|
|
);
|
|
let worker = a.clone();
|
|
let third = std::thread::spawn(move || worker.work_import_analysis_one());
|
|
wait(|| signals.join("three.started").exists());
|
|
fs::write(signals.join("two.release"), b"")?;
|
|
fs::write(signals.join("three.release"), b"")?;
|
|
assert!(second.join().unwrap()?);
|
|
assert!(third.join().unwrap()?);
|
|
for task in &tasks[1..] {
|
|
let saved = a
|
|
.import_tasks()?
|
|
.into_iter()
|
|
.find(|t| t.id == task.id)
|
|
.unwrap();
|
|
assert_eq!(saved.state, "completed", "{:?}", saved.error);
|
|
a.verify(&saved.outputs["ghidra"])?;
|
|
}
|
|
// Each analysis gets a quarter of this small test budget, less staged input.
|
|
let allowance: u64 = fs::read_to_string(signals.join("two.started"))?.parse()?;
|
|
assert!(allowance > 15 * 1024 * 1024 && allowance < 16 * 1024 * 1024);
|
|
Ok(())
|
|
}
|
|
|
|
#[test]
|
|
fn analysis_handoff_survives_restart_without_reimporting() -> Result<()> {
|
|
let (tmp, cfg) = setup()?;
|
|
let a = Archive::open(cfg.clone())?;
|
|
let task = a.submit_import(request(tmp.path(), "resume", true)?)?;
|
|
assert!(a.work_import_preparation_one()?);
|
|
let prepared = a.import_tasks()?.remove(0);
|
|
assert_eq!(prepared.state, "queued");
|
|
assert!(prepared.analysis_input.is_some());
|
|
assert!(prepared.finished.is_none());
|
|
drop(a);
|
|
fs::remove_file(tmp.path().join("input/resume.exe"))?;
|
|
fs::write(tmp.path().join("signals/resume.release"), b"")?;
|
|
let a = Archive::open(cfg)?;
|
|
assert!(!a.work_import_preparation_one()?);
|
|
assert!(a.work_import_analysis_one()?);
|
|
let done = a.import_tasks()?.remove(0);
|
|
assert_eq!(done.id, task.id);
|
|
assert_eq!(done.state, "completed", "{:?}", done.error);
|
|
assert_eq!(done.root, prepared.root);
|
|
assert_eq!(a.snapshots()?.len(), 2);
|
|
Ok(())
|
|
}
|
|
|
|
#[test]
|
|
fn shutdown_interrupts_both_analyses_and_releases_slots() -> Result<()> {
|
|
let (tmp, cfg) = setup()?;
|
|
let a = Arc::new(Archive::open(cfg.clone())?);
|
|
for name in ["one", "two"] {
|
|
a.submit_import(request(tmp.path(), name, true)?)?;
|
|
a.work_import_preparation_one()?;
|
|
}
|
|
let handles: Vec<_> = (0..2)
|
|
.map(|_| {
|
|
let a = a.clone();
|
|
std::thread::spawn(move || a.work_import_analysis_one())
|
|
})
|
|
.collect();
|
|
wait(|| {
|
|
["one", "two"]
|
|
.iter()
|
|
.all(|n| tmp.path().join(format!("signals/{n}.started")).exists())
|
|
});
|
|
a.request_shutdown();
|
|
for h in handles {
|
|
assert!(h.join().unwrap()?);
|
|
}
|
|
assert!(a.import_tasks()?.iter().all(|t| t.state == "interrupted"));
|
|
drop(a);
|
|
let a = Archive::open(cfg)?;
|
|
for task in a.import_tasks()? {
|
|
a.control_import(&task.id, "retry")?;
|
|
fs::write(
|
|
tmp.path().join(format!(
|
|
"signals/{}.release",
|
|
task.request.release.repository
|
|
)),
|
|
b"",
|
|
)?;
|
|
}
|
|
while a.work_import_one()? {}
|
|
assert!(a.import_tasks()?.iter().all(|t| t.state == "completed"));
|
|
Ok(())
|
|
}
|
|
|
|
#[test]
|
|
fn legacy_jobs_share_limits_and_leave_capacity_for_other_tools() -> Result<()> {
|
|
let (tmp, mut cfg) = setup()?;
|
|
cfg.processing_enabled = true;
|
|
cfg.plugins
|
|
.insert("quick".into(), cfg.plugins["ghidra"].clone());
|
|
let a = Arc::new(Archive::open(cfg)?);
|
|
let mut roots = Vec::new();
|
|
for name in ["one", "two", "three", "plain"] {
|
|
let req = request(tmp.path(), name, false)?;
|
|
let ImportSource::Local { paths } = req.source else {
|
|
unreachable!()
|
|
};
|
|
roots.push(a.import(&paths[0], req.release)?.id);
|
|
}
|
|
for root in &roots[..3] {
|
|
a.submit_job(root, "ghidra", false)?;
|
|
}
|
|
let signals = tmp.path().join("signals");
|
|
let mut handles = Vec::new();
|
|
for name in ["one", "two"] {
|
|
let worker = a.clone();
|
|
handles.push(std::thread::spawn(move || worker.work_one()));
|
|
wait(|| signals.join(format!("{name}.started")).exists());
|
|
}
|
|
assert!(a.work_one()?.is_none());
|
|
let jobs = a.jobs()?;
|
|
assert_eq!(jobs.iter().filter(|j| j.state == "running").count(), 2);
|
|
assert_eq!(jobs.iter().filter(|j| j.state == "queued").count(), 1);
|
|
a.submit_job(&roots[3], "quick", false)?;
|
|
fs::write(signals.join("plain.release"), b"")?;
|
|
assert_eq!(a.work_one()?.unwrap().tool, "quick");
|
|
for name in ["one", "two", "three"] {
|
|
fs::write(signals.join(format!("{name}.release")), b"")?;
|
|
}
|
|
for handle in handles {
|
|
assert_eq!(handle.join().unwrap()?.unwrap().state, "completed");
|
|
}
|
|
assert_eq!(a.work_one()?.unwrap().state, "completed");
|
|
assert!(a.work_one()?.is_none());
|
|
assert!(a.jobs()?.iter().all(|j| j.state == "completed"));
|
|
Ok(())
|
|
}
|
|
|
|
#[test]
|
|
fn analysis_cannot_consume_the_import_scratch_reservation() -> Result<()> {
|
|
let (tmp, mut cfg) = setup()?;
|
|
fs::write(
|
|
tmp.path().join("oversize.py"),
|
|
r#"import json,pathlib,sys
|
|
r=json.loads(pathlib.Path(sys.argv[1]).read_text())
|
|
(pathlib.Path(r['output_dir'])/'large').write_bytes(b'x'*(10*1024*1024))
|
|
pathlib.Path(r['result_file']).write_text(json.dumps(dict(protocol=1,layer='derived',coverage='complete',warnings=[],files=['large'])))
|
|
"#,
|
|
)?;
|
|
cfg.plugins.get_mut("ghidra").unwrap().command[1] = tmp
|
|
.path()
|
|
.join("oversize.py")
|
|
.to_string_lossy()
|
|
.into_owned();
|
|
let a = Archive::open(cfg)?;
|
|
a.submit_import(request(tmp.path(), "large", true)?)?;
|
|
a.work_import_one()?;
|
|
let task = a.import_tasks()?.remove(0);
|
|
assert_eq!(task.state, "failed");
|
|
assert!(task.error.unwrap().contains("workspace"));
|
|
assert_eq!(
|
|
a.snapshots()?.len(),
|
|
1,
|
|
"oversized analysis must not publish"
|
|
);
|
|
a.submit_import(request(tmp.path(), "plain", false)?)?;
|
|
assert!(a.work_import_preparation_one()?);
|
|
assert_eq!(a.snapshots()?.len(), 2);
|
|
Ok(())
|
|
}
|
|
|
|
#[test]
|
|
fn archived_analysis_bypasses_a_busy_import_worker() -> Result<()> {
|
|
let (tmp, mut cfg) = setup()?;
|
|
cfg.plugins
|
|
.insert("quick".into(), cfg.plugins["ghidra"].clone());
|
|
let a = Arc::new(Archive::open(cfg)?);
|
|
let mut roots = Vec::new();
|
|
for name in ["one", "plain"] {
|
|
let req = request(tmp.path(), name, false)?;
|
|
let ImportSource::Local { paths } = req.source else {
|
|
unreachable!()
|
|
};
|
|
roots.push(a.import(&paths[0], req.release)?);
|
|
}
|
|
let worker = a.clone();
|
|
let plain = roots[1].id.clone();
|
|
let busy = std::thread::spawn(move || worker.process(&plain, "quick"));
|
|
let signals = tmp.path().join("signals");
|
|
wait(|| signals.join("plain.started").exists());
|
|
let task = a.submit_import(ImportRequest {
|
|
source: ImportSource::Archived {
|
|
workspace: roots[0].id.clone(),
|
|
snapshot: roots[0].id.clone(),
|
|
paths: vec!["one.exe".into()],
|
|
},
|
|
release: roots[0].release.clone(),
|
|
options: ImportOptions {
|
|
extract: false,
|
|
media: false,
|
|
analyze: true,
|
|
},
|
|
})?;
|
|
assert!(task.analysis_input.is_some());
|
|
fs::write(signals.join("one.release"), b"")?;
|
|
assert!(a.work_import_analysis_one()?);
|
|
assert_eq!(a.import_tasks()?.remove(0).state, "completed");
|
|
fs::write(signals.join("plain.release"), b"")?;
|
|
busy.join().unwrap()?;
|
|
Ok(())
|
|
}
|
|
|
|
#[test]
|
|
fn preparation_slots_overlap_and_share_scratch() -> Result<()> {
|
|
let (tmp, mut cfg) = setup()?;
|
|
cfg.preparation_workers = 3;
|
|
let tool = cfg.plugins.remove("ghidra").unwrap();
|
|
cfg.plugins.insert("import-extract".into(), tool);
|
|
let a = Arc::new(Archive::open(cfg)?);
|
|
for name in ["prep1", "prep2", "prep3", "prep4"] {
|
|
let mut req = request(tmp.path(), name, false)?;
|
|
req.options.extract = true;
|
|
a.submit_import(req)?;
|
|
}
|
|
let signals = tmp.path().join("signals");
|
|
let mut workers = Vec::new();
|
|
for name in ["prep1", "prep2", "prep3"] {
|
|
let worker = a.clone();
|
|
workers.push(std::thread::spawn(move || {
|
|
worker.work_import_preparation_one()
|
|
}));
|
|
wait(|| signals.join(format!("{name}.started")).exists());
|
|
}
|
|
let view = a.activity_view()?;
|
|
assert_eq!(view["execution"]["preparation_limit"], 3);
|
|
assert_eq!(view["execution"]["preparation_used"], 3);
|
|
assert!(
|
|
!a.work_import_preparation_one()?,
|
|
"fourth preparation must wait"
|
|
);
|
|
assert!(a.cleanup_orphans().is_err());
|
|
for name in ["prep1", "prep2", "prep3"] {
|
|
let allowance: u64 =
|
|
fs::read_to_string(signals.join(format!("{name}.started")))?.parse()?;
|
|
// 64 MiB less two 16 MiB analysis reservations, divided three ways.
|
|
assert!(allowance > 10 * 1024 * 1024 && allowance <= 32 * 1024 * 1024 / 3);
|
|
fs::write(signals.join(format!("{name}.release")), b"")?;
|
|
}
|
|
for worker in workers {
|
|
assert!(worker.join().unwrap()?);
|
|
}
|
|
// With just one occupied slot, maintenance must still reserve all three.
|
|
let worker = a.clone();
|
|
let last = std::thread::spawn(move || worker.work_import_preparation_one());
|
|
wait(|| signals.join("prep4.started").exists());
|
|
assert!(a.cleanup_orphans().is_err());
|
|
fs::write(signals.join("prep4.release"), b"")?;
|
|
assert!(last.join().unwrap()?);
|
|
for task in a.import_tasks()? {
|
|
assert_eq!(task.state, "completed", "{:?}", task.error);
|
|
}
|
|
Ok(())
|
|
}
|