Add bounded exact-byte document extraction protocol and transport

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"

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<Vec<Passage>> {
+    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<Option<SourceInput>> {
     let extension = path

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<Self> {
+        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<Self> {
+        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<DocumentPassage> },
+    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<Self> {
+        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<ExtractionResponse> {
+    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#"<w:document xmlns:w="http://schemas.openxmlformats.org/wordprocessingml/2006/main"><w:body><w:p><w:r><w:t>supplier cancellation</w:t></w:r></w:p></w:body></w:document>"#)?;
+        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(())
+    }
+}

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<Vec<u8>> {
+    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<u8>)> {
+    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> {
+    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(())
+    }
+}

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,