diff --git a/crates/searchi-core/Cargo.toml b/crates/searchi-core/Cargo.toml index 6d0360f..c219931 100644 --- a/crates/searchi-core/Cargo.toml +++ b/crates/searchi-core/Cargo.toml @@ -8,6 +8,7 @@ license.workspace = true searchi-media = { path = "../searchi-media" } anyhow.workspace = true serde.workspace = true +serde_json.workspace = true sha2 = "0.10" rusqlite = { version = "0.32", features = ["bundled"] } walkdir = "2" diff --git a/crates/searchi-core/src/extraction.rs b/crates/searchi-core/src/extraction.rs index 6000014..9e53a74 100644 --- a/crates/searchi-core/src/extraction.rs +++ b/crates/searchi-core/src/extraction.rs @@ -25,6 +25,20 @@ pub(crate) struct SourceInput { pub hash: String, } +/// Parse a bounded document snapshot without accessing its source path. +pub(crate) fn document_bytes( + format: crate::extraction_protocol::DocumentFormat, + bytes: &[u8], +) -> Result> { + if bytes.len() > MAX_DOCUMENT { + bail!("document input exceeds {MAX_DOCUMENT} bytes"); + } + match format { + crate::extraction_protocol::DocumentFormat::Pdf => pdf(bytes), + crate::extraction_protocol::DocumentFormat::Docx => docx(bytes), + } +} + /// Read once: the fingerprint and parser consume precisely the same bytes. pub(crate) fn read_source(path: &Path) -> Result> { let extension = path diff --git a/crates/searchi-core/src/extraction_protocol.rs b/crates/searchi-core/src/extraction_protocol.rs new file mode 100644 index 0000000..7186939 --- /dev/null +++ b/crates/searchi-core/src/extraction_protocol.rs @@ -0,0 +1,301 @@ +//! Path-free document extraction protocol building blocks. +//! +//! These types do not launch or isolate a parser. The caller must enforce +//! process deadlines/resource limits and bounded transport framing separately. +use crate::SourceLocation; +use anyhow::{ensure, Result}; +use serde::{Deserialize, Serialize}; +use sha2::{Digest, Sha256}; + +pub const VERSION: u32 = 1; +pub const MAX_INPUT_BYTES: usize = 32 * 1024 * 1024; +pub const MAX_HEADER_BYTES: usize = 1024; +pub const MAX_RESPONSE_BYTES: usize = 13 * 1024 * 1024; +const MAX_TEXT_BYTES: usize = 2 * 1024 * 1024; +const MAX_PASSAGES: usize = 10_000; +const MAX_REJECTION_BYTES: usize = 2048; + +#[derive(Debug, Clone, Copy, Serialize, Deserialize, PartialEq, Eq)] +#[serde(rename_all = "snake_case")] +pub enum DocumentFormat { + Pdf, + Docx, +} + +#[derive(Debug, Clone, Serialize, Deserialize)] +#[serde(deny_unknown_fields)] +pub struct RequestHeader { + pub version: u32, + pub format: DocumentFormat, + pub input_bytes: usize, + pub input_sha256: String, +} + +impl RequestHeader { + pub fn from_bytes(format: DocumentFormat, bytes: &[u8]) -> Result { + ensure!( + bytes.len() <= MAX_INPUT_BYTES, + "Document input exceeds byte budget" + ); + Ok(Self { + version: VERSION, + format, + input_bytes: bytes.len(), + input_sha256: digest(bytes), + }) + } + + /// Validate before admitting/allocating the declared raw payload. + pub fn validate(&self) -> Result<()> { + ensure!( + self.version == VERSION, + "Unsupported extraction protocol version" + ); + ensure!( + self.input_bytes <= MAX_INPUT_BYTES, + "Document input exceeds byte budget" + ); + ensure!( + self.input_sha256.len() == 64 + && self + .input_sha256 + .bytes() + .all(|b| b.is_ascii_digit() || (b'a'..=b'f').contains(&b)), + "Invalid input SHA-256" + ); + Ok(()) + } + + pub fn decode(bytes: &[u8]) -> Result { + ensure!( + bytes.len() <= MAX_HEADER_BYTES, + "Extraction header exceeds byte budget" + ); + let value: Self = serde_json::from_slice(bytes)?; + value.validate()?; + Ok(value) + } + + pub fn validate_payload(&self, bytes: &[u8]) -> Result<()> { + self.validate()?; + ensure!( + bytes.len() == self.input_bytes && digest(bytes) == self.input_sha256, + "Extraction payload mismatch" + ); + Ok(()) + } +} + +#[derive(Debug, Serialize, Deserialize)] +#[serde(deny_unknown_fields)] +pub struct DocumentPassage { + pub text: String, + pub source: SourceLocation, +} + +#[derive(Debug, Serialize, Deserialize)] +#[serde(tag = "status", rename_all = "snake_case", deny_unknown_fields)] +pub enum ExtractionOutcome { + Extracted { passages: Vec }, + Rejected { reason: String }, +} + +#[derive(Debug, Serialize, Deserialize)] +#[serde(deny_unknown_fields)] +pub struct ExtractionResponse { + pub version: u32, + pub input_sha256: String, + pub outcome: ExtractionOutcome, +} + +impl ExtractionResponse { + pub fn validate(&self, request: &RequestHeader) -> Result<()> { + request.validate()?; + ensure!( + self.version == VERSION, + "Unsupported extraction response version" + ); + ensure!( + self.input_sha256 == request.input_sha256, + "Extraction input digest mismatch" + ); + match &self.outcome { + ExtractionOutcome::Rejected { reason } => { + ensure!( + !reason.trim().is_empty() && reason.len() <= MAX_REJECTION_BYTES, + "Invalid extraction rejection" + ); + } + ExtractionOutcome::Extracted { passages } => { + ensure!( + passages.len() <= MAX_PASSAGES, + "Extraction passage count exceeds budget" + ); + let mut total = 0usize; + for passage in passages { + total = total + .checked_add(passage.text.len()) + .ok_or_else(|| anyhow::anyhow!("Extraction size overflow"))?; + ensure!( + total <= MAX_TEXT_BYTES, + "Extraction text exceeds byte budget" + ); + let valid = match request.format { + DocumentFormat::Pdf => { + passage.source.kind == "page" + && (1..=1000).contains(&passage.source.number) + } + DocumentFormat::Docx => { + passage.source.kind == "paragraph" + && passage.source.number > 0 + && passage.source.number <= MAX_PASSAGES as u32 + } + }; + ensure!(valid, "Invalid document source location"); + } + } + } + Ok(()) + } + + pub fn decode(bytes: &[u8], request: &RequestHeader) -> Result { + ensure!( + bytes.len() <= MAX_RESPONSE_BYTES, + "Extraction response exceeds byte budget" + ); + let response: Self = serde_json::from_slice(bytes)?; + response.validate(request)?; + Ok(response) + } +} + +/// Execute against precisely the admitted bytes. Protocol violations are errors; +/// parser failures are explicit rejections. No filesystem access is performed. +pub fn extract_document(request: &RequestHeader, bytes: &[u8]) -> Result { + request.validate_payload(bytes)?; + let outcome = match crate::extraction::document_bytes(request.format, bytes) { + Ok(passages) => ExtractionOutcome::Extracted { + passages: passages + .into_iter() + .map(|p| DocumentPassage { + text: p.text, + source: p.source, + }) + .collect(), + }, + Err(error) => { + let message = format!("{error:#}"); + let mut end = message.len().min(MAX_REJECTION_BYTES); + while !message.is_char_boundary(end) { + end -= 1; + } + ExtractionOutcome::Rejected { + reason: message[..end].to_owned(), + } + } + }; + let response = ExtractionResponse { + version: VERSION, + input_sha256: request.input_sha256.clone(), + outcome, + }; + response.validate(request)?; + Ok(response) +} + +fn digest(bytes: &[u8]) -> String { + format!("{:x}", Sha256::digest(bytes)) +} + +#[cfg(test)] +mod tests { + use super::*; + use std::io::{Cursor, Write}; + + #[test] + fn extracts_real_docx_buffer_with_paragraph_and_exact_digest() -> Result<()> { + let mut archive = zip::ZipWriter::new(Cursor::new(Vec::new())); + archive.start_file( + "word/document.xml", + zip::write::SimpleFileOptions::default(), + )?; + archive.write_all(br#"supplier cancellation"#)?; + let bytes = archive.finish()?.into_inner(); + let request = RequestHeader::from_bytes(DocumentFormat::Docx, &bytes)?; + let response = extract_document(&request, &bytes)?; + let ExtractionOutcome::Extracted { passages } = &response.outcome else { + anyhow::bail!("Valid DOCX rejected"); + }; + assert_eq!(passages.len(), 1); + assert_eq!(passages[0].text, "supplier cancellation"); + assert_eq!(passages[0].source.kind, "paragraph"); + assert_eq!(passages[0].source.number, 1); + ExtractionResponse::decode(&serde_json::to_vec(&response)?, &request)?; + Ok(()) + } + + #[test] + fn rejects_version_length_hash_and_oversized_header_before_parsing() -> Result<()> { + let mut request = RequestHeader::from_bytes(DocumentFormat::Pdf, b"bad pdf")?; + assert!(extract_document(&request, b"new pdf").is_err()); + request.input_bytes = MAX_INPUT_BYTES + 1; + assert!(request.validate().is_err()); + request.input_bytes = 7; + request.version += 1; + assert!(request.validate().is_err()); + assert!(RequestHeader::decode(&[b' '; MAX_HEADER_BYTES + 1]).is_err()); + Ok(()) + } + + #[test] + fn parser_rejection_retains_exact_digest_and_round_trips() -> Result<()> { + for format in [DocumentFormat::Pdf, DocumentFormat::Docx] { + let request = RequestHeader::from_bytes(format, b"invalid document")?; + let response = extract_document(&request, b"invalid document")?; + assert!(matches!( + response.outcome, + ExtractionOutcome::Rejected { .. } + )); + ExtractionResponse::decode(&serde_json::to_vec(&response)?, &request)?; + } + Ok(()) + } + + #[test] + fn validates_response_identity_locations_and_aggregate_budget() -> Result<()> { + let request = RequestHeader::from_bytes(DocumentFormat::Pdf, b"fixture")?; + let mut response = ExtractionResponse { + version: VERSION, + input_sha256: request.input_sha256.clone(), + outcome: ExtractionOutcome::Extracted { + passages: vec![DocumentPassage { + text: "passage".into(), + source: SourceLocation { + kind: "page".into(), + number: 0, + }, + }], + }, + }; + assert!(response.validate(&request).is_err()); + if let ExtractionOutcome::Extracted { passages } = &mut response.outcome { + passages[0].source.number = 1; + } + response.validate(&request)?; + response.version += 1; + assert!(response.validate(&request).is_err()); + response.version = VERSION; + assert!(ExtractionResponse::decode(&vec![b' '; MAX_RESPONSE_BYTES + 1], &request).is_err()); + let mut value = serde_json::to_value(&response)?; + value["outcome"]["passages"][0]["source"]["unexpected"] = serde_json::json!(true); + assert!(ExtractionResponse::decode(&serde_json::to_vec(&value)?, &request).is_err()); + response.input_sha256 = "0".repeat(64); + assert!(response.validate(&request).is_err()); + response.input_sha256 = request.input_sha256.clone(); + if let ExtractionOutcome::Extracted { passages } = &mut response.outcome { + passages[0].text = "x".repeat(MAX_TEXT_BYTES + 1); + } + assert!(response.validate(&request).is_err()); + Ok(()) + } +} diff --git a/crates/searchi-core/src/extraction_transport.rs b/crates/searchi-core/src/extraction_transport.rs new file mode 100644 index 0000000..7dd4ff2 --- /dev/null +++ b/crates/searchi-core/src/extraction_transport.rs @@ -0,0 +1,135 @@ +//! Bounded single-request framing. Process deadlines, EOF/status validation and +//! cancellation belong to the caller; these blocking I/O helpers do not isolate +//! parsing or make an unresponsive pipe safe on their own. +use crate::extraction_protocol::{ + ExtractionResponse, RequestHeader, MAX_HEADER_BYTES, MAX_RESPONSE_BYTES, +}; +use anyhow::{ensure, Context, Result}; +use std::io::{Read, Write}; + +fn read_frame(reader: &mut impl Read, limit: usize) -> Result> { + let mut prefix = [0; 4]; + reader + .read_exact(&mut prefix) + .context("Missing extraction frame length")?; + let length = u32::from_le_bytes(prefix) as usize; + ensure!( + length > 0 && length <= limit, + "Extraction frame exceeds budget" + ); + let mut bytes = Vec::new(); + bytes + .try_reserve_exact(length) + .context("Cannot allocate bounded extraction frame")?; + bytes.resize(length, 0); + reader + .read_exact(&mut bytes) + .context("Truncated extraction frame")?; + Ok(bytes) +} + +fn write_frame(writer: &mut impl Write, bytes: &[u8], limit: usize) -> Result<()> { + ensure!( + !bytes.is_empty() && bytes.len() <= limit, + "Extraction frame exceeds budget" + ); + writer.write_all(&u32::try_from(bytes.len())?.to_le_bytes())?; + writer.write_all(bytes)?; + Ok(()) +} + +/// Header length + JSON header + exact raw payload. No filename is transmitted. +pub fn write_request(writer: &mut impl Write, header: &RequestHeader, input: &[u8]) -> Result<()> { + header.validate_payload(input)?; + write_frame(writer, &serde_json::to_vec(header)?, MAX_HEADER_BYTES)?; + writer.write_all(input)?; + Ok(()) +} + +pub fn read_request(reader: &mut impl Read) -> Result<(RequestHeader, Vec)> { + let header = RequestHeader::decode(&read_frame(reader, MAX_HEADER_BYTES)?)?; + // Validate the declaration before allocating or reading any raw payload. + let mut input = Vec::new(); + input + .try_reserve_exact(header.input_bytes) + .context("Cannot allocate bounded document input")?; + input.resize(header.input_bytes, 0); + reader + .read_exact(&mut input) + .context("Truncated document input")?; + header.validate_payload(&input)?; + Ok((header, input)) +} + +pub fn write_response( + writer: &mut impl Write, + response: &ExtractionResponse, + request: &RequestHeader, +) -> Result<()> { + response.validate(request)?; + write_frame(writer, &serde_json::to_vec(response)?, MAX_RESPONSE_BYTES) +} + +pub fn read_response( + reader: &mut impl Read, + request: &RequestHeader, +) -> Result { + ExtractionResponse::decode(&read_frame(reader, MAX_RESPONSE_BYTES)?, request) +} + +#[cfg(test)] +mod tests { + use super::*; + use crate::extraction_protocol::{DocumentFormat, ExtractionOutcome, VERSION}; + use std::io::Cursor; + + #[test] + fn binary_input_preserves_every_byte_without_base64_or_paths() -> Result<()> { + let input = b"\0\xff\r\nPDF binary input"; + let header = RequestHeader::from_bytes(DocumentFormat::Pdf, input)?; + let mut wire = Vec::new(); + write_request(&mut wire, &header, input)?; + assert!(wire.ends_with(input)); + let (decoded, bytes) = read_request(&mut Cursor::new(wire))?; + assert_eq!(bytes, input); + assert_eq!(decoded.input_sha256, header.input_sha256); + Ok(()) + } + + #[test] + fn oversized_and_truncated_frames_fail_without_payload_allocation() { + let oversized = u32::MAX.to_le_bytes(); + assert!(read_request(&mut Cursor::new(oversized)).is_err()); + assert!(read_request(&mut Cursor::new([1, 0])).is_err()); + let header = RequestHeader::from_bytes(DocumentFormat::Pdf, b"input").unwrap(); + let mut wire = Vec::new(); + write_request(&mut wire, &header, b"input").unwrap(); + wire.pop(); + assert!(read_request(&mut Cursor::new(wire)).is_err()); + } + + #[test] + fn wrong_input_digest_is_rejected_and_writer_emits_nothing() { + let header = RequestHeader::from_bytes(DocumentFormat::Pdf, b"first").unwrap(); + let mut wire = Vec::new(); + assert!(write_request(&mut wire, &header, b"other").is_err()); + assert!(wire.is_empty()); + } + + #[test] + fn response_identity_and_size_are_checked() -> Result<()> { + let header = RequestHeader::from_bytes(DocumentFormat::Pdf, b"input")?; + let response = ExtractionResponse { + version: VERSION, + input_sha256: header.input_sha256.clone(), + outcome: ExtractionOutcome::Rejected { + reason: "Invalid PDF".into(), + }, + }; + let mut wire = Vec::new(); + write_response(&mut wire, &response, &header)?; + read_response(&mut Cursor::new(wire), &header)?; + assert!(read_response(&mut Cursor::new(u32::MAX.to_le_bytes()), &header).is_err()); + Ok(()) + } +} diff --git a/crates/searchi-core/src/lib.rs b/crates/searchi-core/src/lib.rs index 1a0e5f5..9b37da2 100644 --- a/crates/searchi-core/src/lib.rs +++ b/crates/searchi-core/src/lib.rs @@ -5,6 +5,8 @@ //! Worker isolation remains a production milestone. mod code; mod extraction; +pub mod extraction_protocol; +pub mod extraction_transport; pub use code::{code_file_predicate, is_code_path, CODE_EXTENSIONS}; pub mod metrics; use anyhow::{bail, Context, Result}; @@ -81,7 +83,8 @@ pub struct SearchResult { /// Verified source location. Document numbers are one based; timestamp_ms /// numbers are milliseconds from media start, and image numbers are zero. -#[derive(Debug, Clone, Serialize)] +#[derive(Debug, Clone, Serialize, serde::Deserialize)] +#[serde(deny_unknown_fields)] pub struct SourceLocation { pub kind: String, pub number: u32, diff --git a/docs/EXTRACTION-ISOLATION.md b/docs/EXTRACTION-ISOLATION.md index dc64ba8..664a76c 100644 --- a/docs/EXTRACTION-ISOLATION.md +++ b/docs/EXTRACTION-ISOLATION.md @@ -42,7 +42,7 @@ result; retain the original parser-input digest, not a substituted later hash. Source files and SQLite are not one atomic transaction. Preserve semantic and agent excerpt provenance checks and clear old passages on parser rejection. -- [ ] Buffer API and typed protocol with independent limits and genuine fixtures. +- [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. - [ ] Shared-engine document adapter, explicit failures and post-parser guards. @@ -56,3 +56,24 @@ invalid locations, blocked stdin, worker crash/timeout, missing helper, cancelled reconciliation, corrupt/encrypted documents and a source changing during parsing. Shipping only protocol types or unit fixtures does not close the isolated-extraction release gate. + +## Implemented first slice + +`searchi-core::extraction_protocol` exposes path-free PDF/DOCX requests and +validated responses. Version one admits at most 32 MiB input, a 1-KiB JSON +header, 13 MiB serialized response and 2 MiB aggregate extracted text. Exact +payload length/SHA, response version/digest and document locations are checked; +parser rejection is explicit. Unknown nested source fields are rejected. + +`extraction_transport` frames bounded JSON lengths as four-byte little-endian +integers followed by raw input bytes, with no base64 or filesystem paths. +It rejects oversized lengths before allocation and mismatched input before +writing. It is blocking I/O: the parent still must supply deadlines, cancellation, +EOF/process-status validation and owned-child cleanup. + +All 50 core tests passed on Windows (18 unit and 32 integration), including +actual DOCX buffer extraction, corrupt PDF/DOCX rejection, raw binary byte +preservation, version/hash/location/size failures and truncated frames. Strict +all-target Clippy, Rust formatting and diff checks passed. Product extraction +defaults and code/text parsing are unchanged. No separate worker was launched; +process isolation and resource ceilings remain open.