diff --git a/Cargo.lock b/Cargo.lock index 369c0aa..e97e4da 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -2264,6 +2264,7 @@ name = "searchi-core" version = "0.1.0" dependencies = [ "anyhow", + "libc", "lopdf", "quick-xml", "rusqlite", @@ -2273,6 +2274,19 @@ dependencies = [ "sha2 0.10.9", "tempfile", "walkdir", + "windows-sys 0.61.2", + "zip", +] + +[[package]] +name = "searchi-extract-worker" +version = "0.1.0" +dependencies = [ + "anyhow", + "libc", + "searchi-core", + "serde_json", + "windows-sys 0.61.2", "zip", ] diff --git a/Cargo.toml b/Cargo.toml index 12c9794..8dcb843 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -1,5 +1,5 @@ [workspace] -members = ["crates/searchi-core", "crates/searchi-cli", "crates/searchi-semantic", "crates/searchi-service", "crates/searchi-assets", "crates/searchi-media", "crates/searchi-ai"] +members = ["crates/searchi-core", "crates/searchi-cli", "crates/searchi-semantic", "crates/searchi-service", "crates/searchi-assets", "crates/searchi-media", "crates/searchi-ai", "crates/searchi-extract-worker"] resolver = "2" [workspace.package] diff --git a/crates/searchi-core/Cargo.toml b/crates/searchi-core/Cargo.toml index c219931..0039319 100644 --- a/crates/searchi-core/Cargo.toml +++ b/crates/searchi-core/Cargo.toml @@ -16,6 +16,12 @@ lopdf = { version = "0.45", default-features = false } zip = { version = "8", default-features = false, features = ["deflate"] } quick-xml = "0.42" +[target.'cfg(windows)'.dependencies] +windows-sys = { version = "0.61", features = ["Win32_Foundation", "Win32_Security", "Win32_System_JobObjects", "Win32_System_Threading"] } + +[target.'cfg(unix)'.dependencies] +libc = "0.2" + [dev-dependencies] tempfile = "3" serde_json.workspace = true diff --git a/crates/searchi-core/src/extraction_process.rs b/crates/searchi-core/src/extraction_process.rs new file mode 100644 index 0000000..1fbea63 --- /dev/null +++ b/crates/searchi-core/src/extraction_process.rs @@ -0,0 +1,459 @@ +//! One bounded, path-free document request per resource-limited child process. +//! The configured executable is trusted application code, not a search result. +use crate::{ + extraction_protocol::{ExtractionResponse, RequestHeader}, + extraction_transport, +}; +use std::{ + io::{Read, Write}, + path::PathBuf, + process::{Child, Command, Stdio}, + sync::{ + atomic::{AtomicBool, Ordering}, + mpsc, + }, + thread, + time::{Duration, Instant}, +}; + +/// Parser limits are fixed; callers may shorten, but never extend, the deadline. +#[derive(Debug, Clone)] +pub struct ProcessConfig { + pub executable: PathBuf, + pub timeout: Duration, +} + +#[derive(Debug)] +pub enum ProcessError { + Cancelled, + TimedOut, + Failed(String), +} +impl std::fmt::Display for ProcessError { + fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { + match self { + Self::Cancelled => f.write_str("Document extraction cancelled"), + Self::TimedOut => f.write_str("Document extraction timed out"), + Self::Failed(s) => write!(f, "Document extraction failed: {s}"), + } + } +} +impl std::error::Error for ProcessError {} +fn failed(error: impl std::fmt::Display) -> ProcessError { + ProcessError::Failed(error.to_string()) +} +enum Event { + Written, + Response(ExtractionResponse), + Stderr, +} + +/// Includes pipe writes, response EOF and process exit in the same deadline. +/// No parser fallback occurs when limits, transport or worker execution fail. +pub fn run( + config: &ProcessConfig, + request: RequestHeader, + input: Vec, + cancelled: &AtomicBool, +) -> Result { + if config.timeout.is_zero() || config.timeout > Duration::from_secs(30) { + return Err(failed("Invalid extraction deadline")); + } + request.validate_payload(&input).map_err(failed)?; + if cancelled.load(Ordering::Relaxed) { + return Err(ProcessError::Cancelled); + } + let started = Instant::now(); + let mut command = Command::new(&config.executable); + command + .stdin(Stdio::piped()) + .stdout(Stdio::piped()) + .stderr(Stdio::piped()); + #[cfg(unix)] + { + use std::os::unix::process::CommandExt; + command.process_group(0); + } + #[cfg(windows)] + { + use std::os::windows::process::CommandExt; + command.creation_flags(0x08000000); + } + let child = command.spawn().map_err(failed)?; + let mut owned = OwnedChild { + child, + threads: Vec::new(), + completed: false, + #[cfg(windows)] + job: None, + }; + // The writer cannot transmit even the admission prefix until assignment succeeds. + #[cfg(windows)] + { + owned.job = Some(Job::assign(&owned.child).map_err(failed)?); + } + let mut stdin = owned + .child + .stdin + .take() + .ok_or_else(|| failed("Missing stdin"))?; + let mut stdout = owned + .child + .stdout + .take() + .ok_or_else(|| failed("Missing stdout"))?; + let mut stderr = owned + .child + .stderr + .take() + .ok_or_else(|| failed("Missing stderr"))?; + let (tx, rx) = mpsc::sync_channel(3); + let writer_tx = tx.clone(); + let writer_request = request.clone(); + owned.spawn("document-input", move || { + let result = (|| { + extraction_transport::write_request(&mut stdin, &writer_request, &input) + .map_err(failed)?; + stdin.flush().map_err(failed)?; + Ok(Event::Written) + })(); + drop(stdin); + let _ = writer_tx.send(result); + })?; + let reader_tx = tx.clone(); + owned.spawn("document-output", move || { + let result = (|| { + let response = + extraction_transport::read_response(&mut stdout, &request).map_err(failed)?; + let mut extra = [0]; + if stdout.read(&mut extra).map_err(failed)? != 0 { + return Err(failed("Trailing extraction output")); + } + Ok(Event::Response(response)) + })(); + let _ = reader_tx.send(result); + })?; + owned.spawn("document-errors", move || { + // Do not log document-derived stderr. Bound total bytes, not just retention. + let result = (|| { + let mut total = 0usize; + let mut buffer = [0; 1024]; + loop { + let count = stderr.read(&mut buffer).map_err(failed)?; + if count == 0 { + break; + } + total += count; + if total > 16 * 1024 { + return Err(failed("Worker stderr exceeds budget")); + } + } + Ok(Event::Stderr) + })(); + let _ = tx.send(result); + })?; + let mut received = 0; + let mut response = None; + loop { + if cancelled.load(Ordering::Relaxed) { + return Err(ProcessError::Cancelled); + } + if started.elapsed() >= config.timeout { + return Err(ProcessError::TimedOut); + } + match rx.recv_timeout( + Duration::from_millis(5).min(config.timeout.saturating_sub(started.elapsed())), + ) { + Ok(Ok(event)) => { + received += 1; + if let Event::Response(value) = event { + response = Some(value); + } + } + Ok(Err(error)) => return Err(error), + Err(mpsc::RecvTimeoutError::Timeout) => {} + Err(mpsc::RecvTimeoutError::Disconnected) if received != 3 => { + return Err(failed("Extraction I/O worker stopped")) + } + Err(mpsc::RecvTimeoutError::Disconnected) => thread::sleep(Duration::from_millis(1)), + } + // Keep the leader unreaped while pipes may still be inherited by descendants. + // Its PID then anchors the owned process-group identity during error cleanup. + if received == 3 { + if let Some(status) = owned.child.try_wait().map_err(failed)? { + owned.completed = true; + if !status.success() { + return Err(failed("Extraction worker exited unsuccessfully")); + } + let response = response.ok_or_else(|| failed("Missing extraction response"))?; + return Ok(response); + } + } + } +} + +struct OwnedChild { + child: Child, + threads: Vec>, + completed: bool, + #[cfg(windows)] + job: Option, +} +impl OwnedChild { + fn spawn( + &mut self, + name: &str, + work: impl FnOnce() + Send + 'static, + ) -> Result<(), ProcessError> { + self.threads.push( + thread::Builder::new() + .name(name.into()) + .spawn(work) + .map_err(failed)?, + ); + Ok(()) + } +} +impl Drop for OwnedChild { + fn drop(&mut self) { + #[cfg(windows)] + if let Some(job) = self.job.take() { + drop(job); + } + if !self.completed { + #[cfg(unix)] + unsafe { + libc::kill(-(self.child.id() as i32), libc::SIGKILL); + } + let _ = self.child.kill(); + let _ = self.child.wait(); + } + for handle in self.threads.drain(..) { + let _ = handle.join(); + } + } +} + +#[cfg(windows)] +struct Job(windows_sys::Win32::Foundation::HANDLE); +#[cfg(windows)] +impl Job { + fn assign(child: &Child) -> std::io::Result { + use std::os::windows::io::AsRawHandle; + use windows_sys::Win32::System::JobObjects::*; + unsafe { + let handle = CreateJobObjectW(std::ptr::null(), std::ptr::null()); + if handle.is_null() { + return Err(std::io::Error::last_os_error()); + } + let job = Self(handle); + let mut limits: JOBOBJECT_EXTENDED_LIMIT_INFORMATION = std::mem::zeroed(); + limits.BasicLimitInformation.LimitFlags = + JOB_OBJECT_LIMIT_KILL_ON_JOB_CLOSE | JOB_OBJECT_LIMIT_PROCESS_MEMORY; + limits.ProcessMemoryLimit = 512 * 1024 * 1024; + if SetInformationJobObject( + handle, + JobObjectExtendedLimitInformation, + (&limits as *const JOBOBJECT_EXTENDED_LIMIT_INFORMATION).cast(), + std::mem::size_of_val(&limits) as u32, + ) == 0 + || AssignProcessToJobObject(handle, child.as_raw_handle()) == 0 + { + return Err(std::io::Error::last_os_error()); + } + Ok(job) + } + } +} +#[cfg(windows)] +impl Drop for Job { + fn drop(&mut self) { + unsafe { + windows_sys::Win32::Foundation::CloseHandle(self.0); + } + } +} + +#[cfg(test)] +mod tests { + use super::*; + use crate::extraction_protocol::{DocumentFormat, ExtractionOutcome}; + + // These tests compile real executables. Avoid concurrent rustc/linker and + // executable-admission load contaminating the deliberately short deadlines. + static PROCESS_FIXTURE_SLOT: std::sync::Mutex<()> = std::sync::Mutex::new(()); + fn fixture_slot() -> std::sync::MutexGuard<'static, ()> { + PROCESS_FIXTURE_SLOT + .lock() + .unwrap_or_else(|poisoned| poisoned.into_inner()) + } + + fn fixtures() -> tempfile::TempDir { + let directory = tempfile::tempdir().unwrap(); + let source = directory.path().join("fake.rs"); + std::fs::write(&source, r#" +use std::{io::{Read, Write},time::Duration}; +fn main() { + let mode=std::env::current_exe().unwrap().file_stem().unwrap().to_string_lossy().into_owned(); + if mode=="blocked" { std::thread::sleep(Duration::from_secs(60)); return; } + if mode=="crash" { std::process::exit(7); } + let mut bytes=Vec::new(); std::io::stdin().read_to_end(&mut bytes).unwrap(); + if mode=="sleep" { std::thread::sleep(Duration::from_secs(60)); return; } + if mode=="oversize" { std::io::stdout().write_all(&u32::MAX.to_le_bytes()).unwrap(); return; } + let size=u32::from_le_bytes(bytes[..4].try_into().unwrap()) as usize; + let header=std::str::from_utf8(&bytes[4..4+size]).unwrap(); + let digest=header.split("\"input_sha256\":\"").nth(1).unwrap().split('"').next().unwrap(); + let json=format!("{{\"version\":1,\"input_sha256\":\"{}\",\"outcome\":{{\"status\":\"rejected\",\"reason\":\"controlled fixture\"}}}}",digest); + let mut out=std::io::stdout(); out.write_all(&(json.len() as u32).to_le_bytes()).unwrap();out.write_all(json.as_bytes()).unwrap(); + if mode=="stderr" { std::io::stderr().write_all(&vec![b'x';17000]).unwrap(); } + if mode=="extra" { out.write_all(b"extra").unwrap(); } +} +"#).unwrap(); + let executable = directory + .path() + .join(format!("valid{}", std::env::consts::EXE_SUFFIX)); + let output = Command::new("rustc") + .arg("--edition=2021") + .arg(&source) + .arg("-o") + .arg(&executable) + .output() + .unwrap(); + assert!( + output.status.success(), + "{}", + String::from_utf8_lossy(&output.stderr) + ); + for mode in ["blocked", "crash", "sleep", "oversize", "stderr", "extra"] { + std::fs::copy( + &executable, + directory + .path() + .join(format!("{mode}{}", std::env::consts::EXE_SUFFIX)), + ) + .unwrap(); + } + directory + } + fn config(directory: &tempfile::TempDir, mode: &str) -> ProcessConfig { + ProcessConfig { + executable: directory + .path() + .join(format!("{mode}{}", std::env::consts::EXE_SUFFIX)), + timeout: Duration::from_secs(2), + } + } + fn request(bytes: &[u8]) -> RequestHeader { + RequestHeader::from_bytes(DocumentFormat::Pdf, bytes).unwrap() + } + + #[test] + fn valid_response_requires_complete_input_eof_frame_and_exit() { + let _slot = fixture_slot(); + let directory = fixtures(); + let result = run( + &config(&directory, "valid"), + request(b"fixture"), + b"fixture".to_vec(), + &AtomicBool::new(false), + ) + .unwrap(); + assert!(matches!(result.outcome, ExtractionOutcome::Rejected { .. })); + } + #[test] + fn crashes_oversized_trailing_output_and_stderr_fail_closed() { + let _slot = fixture_slot(); + let directory = fixtures(); + for mode in ["crash", "oversize", "extra", "stderr"] { + let result = run( + &config(&directory, mode), + request(b"fixture"), + b"fixture".to_vec(), + &AtomicBool::new(false), + ); + assert!( + matches!(result, Err(ProcessError::Failed(_))), + "{mode}: {result:?}" + ); + } + } + #[test] + fn deadline_covers_blocked_stdin_and_missing_response() { + let _slot = fixture_slot(); + let directory = fixtures(); + for mode in ["blocked", "sleep"] { + let bytes = vec![0; 1024 * 1024]; + let started = Instant::now(); + assert!(matches!( + run( + &config(&directory, mode), + request(&bytes), + bytes, + &AtomicBool::new(false) + ), + Err(ProcessError::TimedOut) + )); + assert!(started.elapsed() < Duration::from_secs(5)); + } + } + #[test] + fn cancellation_reaps_a_blocked_writer_and_invalid_limits_do_not_spawn() { + let _slot = fixture_slot(); + let directory = fixtures(); + let cancelled = AtomicBool::new(false); + thread::scope(|scope| { + scope.spawn(|| { + thread::sleep(Duration::from_millis(50)); + cancelled.store(true, Ordering::Relaxed); + }); + let bytes = vec![0; 1024 * 1024]; + assert!(matches!( + run( + &config(&directory, "blocked"), + request(&bytes), + bytes, + &cancelled + ), + Err(ProcessError::Cancelled) + )); + }); + let mut invalid = config(&directory, "valid"); + invalid.timeout = Duration::from_secs(31); + assert!(matches!( + run(&invalid, request(b""), Vec::new(), &AtomicBool::new(false)), + Err(ProcessError::Failed(_)) + )); + } + + #[test] + #[ignore = "requires a separately built SEARCHI_EXTRACT_WORKER executable"] + fn real_worker_parses_exact_docx_bytes_in_parent_owned_limits() { + use std::io::Cursor; + let mut archive = zip::ZipWriter::new(Cursor::new(Vec::new())); + archive + .start_file( + "word/document.xml", + zip::write::SimpleFileOptions::default(), + ) + .unwrap(); + archive.write_all(b"Supplier cancellation notice").unwrap(); + let input = archive.finish().unwrap().into_inner(); + let header = RequestHeader::from_bytes(DocumentFormat::Docx, &input).unwrap(); + let config = ProcessConfig { + executable: std::env::var_os("SEARCHI_EXTRACT_WORKER") + .expect("Set actual worker executable") + .into(), + timeout: Duration::from_secs(5), + }; + let response = run(&config, header, input, &AtomicBool::new(false)).unwrap(); + match response.outcome { + ExtractionOutcome::Extracted { passages } => { + assert_eq!(passages.len(), 1); + assert_eq!(passages[0].text, "Supplier cancellation notice"); + assert_eq!(passages[0].source.kind, "paragraph"); + assert_eq!(passages[0].source.number, 1); + } + _ => panic!("Actual DOCX unexpectedly rejected"), + } + } +} diff --git a/crates/searchi-core/src/lib.rs b/crates/searchi-core/src/lib.rs index 9b37da2..639d6e3 100644 --- a/crates/searchi-core/src/lib.rs +++ b/crates/searchi-core/src/lib.rs @@ -5,6 +5,7 @@ //! Worker isolation remains a production milestone. mod code; mod extraction; +pub mod extraction_process; pub mod extraction_protocol; pub mod extraction_transport; pub use code::{code_file_predicate, is_code_path, CODE_EXTENSIONS}; diff --git a/crates/searchi-extract-worker/Cargo.toml b/crates/searchi-extract-worker/Cargo.toml new file mode 100644 index 0000000..78942dc --- /dev/null +++ b/crates/searchi-extract-worker/Cargo.toml @@ -0,0 +1,20 @@ +[package] +name = "searchi-extract-worker" +version.workspace = true +edition.workspace = true +license.workspace = true + +[dependencies] +anyhow.workspace = true +searchi-core = { path = "../searchi-core" } + +[target.'cfg(unix)'.dependencies] +libc = "0.2" + +[target.'cfg(windows)'.dependencies] +windows-sys = { version = "0.61", features = ["Win32_Foundation", "Win32_System_JobObjects", "Win32_System_Threading"] } + +[dev-dependencies] +serde_json.workspace = true +zip = { version = "8", default-features = false, features = ["deflate"] } + diff --git a/crates/searchi-extract-worker/src/main.rs b/crates/searchi-extract-worker/src/main.rs new file mode 100644 index 0000000..c03f81d --- /dev/null +++ b/crates/searchi-extract-worker/src/main.rs @@ -0,0 +1,40 @@ +//! Single-request, path-free document parser helper. Parent owns deadline, +//! cancellation and Windows Job lifetime; this helper never spawns children. +mod resources; +use anyhow::{ensure, Result}; +use searchi_core::{extraction_protocol::extract_document, extraction_transport}; +use std::io::{self, Read, Write}; + +fn run() -> Result<()> { + ensure!(std::env::args_os().len() == 1, "Arguments are unsupported"); + let mut input = io::stdin().lock(); + #[cfg(unix)] + resources::apply()?; + // Prefix arrival synchronizes with Windows parent job assignment. No + // attacker-declared allocation or parser runs before the policy check. + let mut prefix = [0u8; 4]; + input.read_exact(&mut prefix)?; + #[cfg(not(unix))] + resources::apply()?; + let mut input = io::Cursor::new(prefix).chain(&mut input); + let (request, bytes) = extraction_transport::read_request(&mut input)?; + // Parent closes stdin after the single frame. Extra data never reaches a parser. + let mut trailing = [0u8; 1]; + ensure!( + input.read(&mut trailing)? == 0, + "Trailing input is unsupported" + ); + let response = extract_document(&request, &bytes)?; + let mut output = io::stdout().lock(); + extraction_transport::write_response(&mut output, &response, &request)?; + output.flush()?; + Ok(()) +} + +fn main() { + if run().is_err() { + // Do not echo document contents, external paths or untrusted protocol values. + eprintln!("searchi extraction worker rejected request or resource policy"); + std::process::exit(1); + } +} diff --git a/crates/searchi-extract-worker/src/resources.rs b/crates/searchi-extract-worker/src/resources.rs new file mode 100644 index 0000000..3b89517 --- /dev/null +++ b/crates/searchi-extract-worker/src/resources.rs @@ -0,0 +1,61 @@ +use anyhow::{ensure, Result}; + +const MEMORY_BYTES: usize = 512 * 1024 * 1024; + +#[cfg(windows)] +pub fn apply() -> Result<()> { + use windows_sys::Win32::System::JobObjects::{ + JobObjectExtendedLimitInformation, QueryInformationJobObject, + JOBOBJECT_EXTENDED_LIMIT_INFORMATION, JOB_OBJECT_LIMIT_KILL_ON_JOB_CLOSE, + JOB_OBJECT_LIMIT_PROCESS_MEMORY, + }; + let mut information: JOBOBJECT_EXTENDED_LIMIT_INFORMATION = unsafe { std::mem::zeroed() }; + // NULL addresses the job associated with the current process. The parent + // must assign this process before writing its first input bytes. + let success = unsafe { + QueryInformationJobObject( + std::ptr::null_mut(), + JobObjectExtendedLimitInformation, + &mut information as *mut _ as *mut _, + std::mem::size_of_val(&information) as u32, + std::ptr::null_mut(), + ) + }; + ensure!( + success != 0, + "Extraction worker requires parent resource job" + ); + let required = JOB_OBJECT_LIMIT_PROCESS_MEMORY | JOB_OBJECT_LIMIT_KILL_ON_JOB_CLOSE; + ensure!( + information.BasicLimitInformation.LimitFlags & required == required + && information.ProcessMemoryLimit > 0 + && information.ProcessMemoryLimit <= MEMORY_BYTES, + "Extraction worker job resource policy is missing" + ); + Ok(()) +} + +#[cfg(unix)] +pub fn apply() -> Result<()> { + for (resource, maximum) in [ + (libc::RLIMIT_AS, MEMORY_BYTES as libc::rlim_t), + (libc::RLIMIT_CPU, 30 as libc::rlim_t), + ] { + let value = libc::rlimit { + rlim_cur: maximum, + rlim_max: maximum, + }; + // Applied before allocating input or invoking any parser. Platform + // enforcement semantics, especially macOS, require native evidence. + ensure!( + unsafe { libc::setrlimit(resource, &value) } == 0, + "Cannot install extraction resource limit" + ); + } + Ok(()) +} + +#[cfg(not(any(windows, unix)))] +pub fn apply() -> Result<()> { + anyhow::bail!("Extraction resource policy unsupported on this platform") +} diff --git a/crates/searchi-extract-worker/tests/parent_roundtrip.rs b/crates/searchi-extract-worker/tests/parent_roundtrip.rs new file mode 100644 index 0000000..aa3f090 --- /dev/null +++ b/crates/searchi-extract-worker/tests/parent_roundtrip.rs @@ -0,0 +1,63 @@ +//! Exercise the real helper through the shared engine's process owner. +use searchi_core::{ + extraction_process::{run, ProcessConfig, ProcessError}, + extraction_protocol::{DocumentFormat, ExtractionOutcome, RequestHeader}, +}; +use std::{ + io::{Cursor, Write}, + path::PathBuf, + sync::atomic::AtomicBool, + time::Duration, +}; + +fn config() -> ProcessConfig { + ProcessConfig { + executable: PathBuf::from(env!("CARGO_BIN_EXE_searchi-extract-worker")), + timeout: Duration::from_secs(10), + } +} + +#[test] +fn actual_parent_and_worker_return_original_docx_passage_and_digest() { + let mut document = zip::ZipWriter::new(Cursor::new(Vec::new())); + document + .start_file( + "word/document.xml", + zip::write::SimpleFileOptions::default(), + ) + .unwrap(); + document.write_all(br#"Cancel with sixty days written notice."#).unwrap(); + let input = document.finish().unwrap().into_inner(); + let request = RequestHeader::from_bytes(DocumentFormat::Docx, &input).unwrap(); + let expected_hash = request.input_sha256.clone(); + let response = run(&config(), request, input, &AtomicBool::new(false)).unwrap(); + assert_eq!(response.input_sha256, expected_hash); + let ExtractionOutcome::Extracted { passages } = response.outcome else { + panic!("Valid document rejected"); + }; + assert_eq!(passages.len(), 1); + assert_eq!(passages[0].text, "Cancel with sixty days written notice."); + assert_eq!(passages[0].source.kind, "paragraph"); + assert_eq!(passages[0].source.number, 1); +} + +#[test] +fn parser_rejection_and_cancellation_remain_distinct() { + let input = b"invalid PDF".to_vec(); + let request = RequestHeader::from_bytes(DocumentFormat::Pdf, &input).unwrap(); + let response = run( + &config(), + request.clone(), + input.clone(), + &AtomicBool::new(false), + ) + .unwrap(); + assert!(matches!( + response.outcome, + ExtractionOutcome::Rejected { .. } + )); + assert!(matches!( + run(&config(), request, input, &AtomicBool::new(true)), + Err(ProcessError::Cancelled) + )); +} diff --git a/crates/searchi-extract-worker/tests/protocol.rs b/crates/searchi-extract-worker/tests/protocol.rs new file mode 100644 index 0000000..16afaba --- /dev/null +++ b/crates/searchi-extract-worker/tests/protocol.rs @@ -0,0 +1,153 @@ +use searchi_core::{ + extraction_protocol::{DocumentFormat, ExtractionOutcome, RequestHeader, MAX_HEADER_BYTES}, + extraction_transport::{read_response, write_request}, +}; +use std::{ + io::{Cursor, Write}, + process::{Command, Stdio}, + time::{Duration, Instant}, +}; + +#[cfg(windows)] +struct Job(windows_sys::Win32::Foundation::HANDLE); +#[cfg(windows)] +impl Job { + fn attach(child: &std::process::Child) -> Self { + use std::os::windows::io::AsRawHandle; + use windows_sys::Win32::System::JobObjects::*; + let handle = unsafe { CreateJobObjectW(std::ptr::null(), std::ptr::null()) }; + assert!(!handle.is_null()); + let guard = Self(handle); + let mut info: JOBOBJECT_EXTENDED_LIMIT_INFORMATION = unsafe { std::mem::zeroed() }; + info.BasicLimitInformation.LimitFlags = + JOB_OBJECT_LIMIT_PROCESS_MEMORY | JOB_OBJECT_LIMIT_KILL_ON_JOB_CLOSE; + info.ProcessMemoryLimit = 512 * 1024 * 1024; + assert_ne!( + unsafe { + SetInformationJobObject( + handle, + JobObjectExtendedLimitInformation, + &info as *const _ as *const _, + std::mem::size_of_val(&info) as u32, + ) + }, + 0 + ); + assert_ne!( + unsafe { AssignProcessToJobObject(handle, child.as_raw_handle() as _) }, + 0 + ); + guard + } +} +#[cfg(windows)] +impl Drop for Job { + fn drop(&mut self) { + unsafe { + windows_sys::Win32::Foundation::CloseHandle(self.0); + } + } +} + +fn exchange(wire: &[u8], arguments: &[&str], attach: bool) -> std::process::Output { + let mut child = Command::new(env!("CARGO_BIN_EXE_searchi-extract-worker")) + .args(arguments) + .stdin(Stdio::piped()) + .stdout(Stdio::piped()) + .stderr(Stdio::piped()) + .spawn() + .unwrap(); + #[cfg(windows)] + let _job = attach.then(|| Job::attach(&child)); + #[cfg(not(windows))] + let _ = attach; + let mut input = child.stdin.take().unwrap(); + // Unknown arguments may terminate before the request write. + let _ = input.write_all(wire); + drop(input); + let deadline = Instant::now() + Duration::from_secs(30); + while child.try_wait().unwrap().is_none() { + if Instant::now() >= deadline { + child.kill().unwrap(); + child.wait().unwrap(); + panic!("Worker subprocess exceeded deadline"); + } + std::thread::sleep(Duration::from_millis(10)); + } + child.wait_with_output().unwrap() +} + +fn wire(format: DocumentFormat, input: &[u8]) -> (RequestHeader, Vec) { + let request = RequestHeader::from_bytes(format, input).unwrap(); + let mut bytes = Vec::new(); + write_request(&mut bytes, &request, input).unwrap(); + (request, bytes) +} + +#[test] +fn valid_docx_and_corrupt_pdf_are_real_worker_responses() { + let mut archive = zip::ZipWriter::new(Cursor::new(Vec::new())); + archive + .start_file( + "word/document.xml", + zip::write::SimpleFileOptions::default(), + ) + .unwrap(); + archive.write_all(br#"supplier cancellation"#).unwrap(); + let input = archive.finish().unwrap().into_inner(); + let (request, bytes) = wire(DocumentFormat::Docx, &input); + let output = exchange(&bytes, &[], true); + assert!( + output.status.success(), + "{}", + String::from_utf8_lossy(&output.stderr) + ); + let response = read_response(&mut Cursor::new(output.stdout), &request).unwrap(); + let ExtractionOutcome::Extracted { passages } = response.outcome else { + panic!("Valid DOCX rejected"); + }; + assert_eq!(passages[0].text, "supplier cancellation"); + assert_eq!(passages[0].source.number, 1); + let (request, bytes) = wire(DocumentFormat::Pdf, b"not a PDF"); + let output = exchange(&bytes, &[], true); + assert!(output.status.success()); + let response = read_response(&mut Cursor::new(output.stdout), &request).unwrap(); + assert!(matches!( + response.outcome, + ExtractionOutcome::Rejected { .. } + )); +} + +#[test] +fn rejects_unknown_arguments_trailing_input_hash_version_and_header_budget() { + let (request, mut bytes) = wire(DocumentFormat::Pdf, b"invalid PDF"); + bytes.push(0); + let output = exchange(&bytes, &[], true); + assert!(!output.status.success() && output.stdout.is_empty()); + for change_version in [false, true] { + let mut changed = request.clone(); + if change_version { + changed.version += 1; + } else { + changed.input_sha256 = "0".repeat(64); + } + let header = serde_json::to_vec(&changed).unwrap(); + let mut bytes = (header.len() as u32).to_le_bytes().to_vec(); + bytes.extend(header); + bytes.extend(b"invalid PDF"); + let output = exchange(&bytes, &[], true); + assert!(!output.status.success() && output.stdout.is_empty()); + } + let output = exchange(&((MAX_HEADER_BYTES + 1) as u32).to_le_bytes(), &[], true); + assert!(!output.status.success() && output.stdout.is_empty()); + let output = exchange(&[], &["--arbitrary-path"], true); + assert!(!output.status.success() && output.stdout.is_empty()); +} + +#[cfg(windows)] +#[test] +fn rejects_worker_without_parent_job_before_payload_parsing() { + let (_, bytes) = wire(DocumentFormat::Pdf, b"invalid PDF"); + let output = exchange(&bytes, &[], false); + assert!(!output.status.success() && output.stdout.is_empty()); +} diff --git a/docs/EXTRACTION-ISOLATION.md b/docs/EXTRACTION-ISOLATION.md index 8a40355..2d9c9ec 100644 --- a/docs/EXTRACTION-ISOLATION.md +++ b/docs/EXTRACTION-ISOLATION.md @@ -43,8 +43,10 @@ Source files and SQLite are not one atomic transaction. Preserve semantic and agent excerpt provenance checks and clear old passages on parser rejection. - [x] Buffer API and typed protocol with independent limits and genuine fixtures. -- [ ] Headless Rust helper with no GUI dependencies or arbitrary path access. -- [ ] Parent process deadlines, cancellation, resource setup and kill/reap tests. +- [x] Headless Rust helper with no GUI dependencies or arbitrary path arguments + (Windows execution tested; native Unix resource verification remains open). +- [x] Parent process deadlines, cancellation, resource setup and kill/reap tests + on Windows; native Unix execution remains open. - [ ] Shared-engine document adapter, explicit failures and post-parser guards. - [ ] Normal helper discovery and required release staging on all target systems. - [ ] Relocated Windows/WSL packages and actual parser fixtures through the helper. @@ -100,3 +102,30 @@ parsing or structured parser rejection must produce one bounded, flushed frame; the parent independently checks EOF and exit status. None of these limits is a filesystem/network sandbox or proof that every malicious parser input is safe. The helper does not add arbitrary file reads or writes to its application API. + +## Windows worker and process-owner evidence + +The new workspace binary and `searchi-core::extraction_process` now work +together. Three real worker subprocess tests cover DOCX parsing, corrupt PDF +rejection, malformed/trailing input, wrong version/hash, excessive header, +unknown arguments and missing Windows Job protection. Two further tests execute +the actual helper through the shared parent adapter, checking paragraph text, +source position, exact digest and parser-rejection/cancellation distinction. + +Four controlled child-process cases exercise crash, oversized/trailing output, +excess stderr, blocked input, missing response, timeout and cancellation. +The stderr case has a valid response, so an unrelated missing-frame failure +cannot mask the stderr budget. Per-test helper compiler activity is serialized +to avoid contaminating two-second test deadlines; product deadlines were not +relaxed. Owned temporary fixtures are cleaned after process/thread cleanup. +Unix cleanup defers reaping until all I/O events, avoiding use of a reaped +leader's process-group identifier. Native Unix execution still needs validation. + +The final combined Windows CPU run passed 59 tests (54 core and 5 worker); +the initially ignored actual-worker core test then passed explicitly, giving +60 executed checks. Strict all-target Clippy passed for both crates. No GPU or +model inference was used. Windows Job configuration/verification is exercised, +but adversarial memory-exhaustion behavior and native Unix limits remain open. +Default indexing still uses the earlier in-process parser; worker discovery, +post-parser source validation, production adapter activation and required +release staging are the next steps. No published installer contains this helper.