diff --git a/CHANGELOG.md b/CHANGELOG.md index 163ce3f..4b7cf56 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -17,12 +17,20 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 - Preserve richer HTML semantics in Markdown, including collision-safe code fences, footnotes, math, callouts, highlights, strikethrough, image sources, figure captions, task checkboxes, and valid tables with formatted cells. -- Add a pluggable post-download `ContentProcessor` registry and built-in - `PdfProcessor` that extracts Markdown from text-based PDFs with local - `pdf-inspector` processing and explicit OCR guidance for unsupported pages. -- Add a built-in `OfficeProcessor` that converts Word (DOCX), PowerPoint (PPTX) - and spreadsheet (XLSX, XLSM, XLS, ODS) documents to Markdown locally, with - zip-bomb limits checked before any part is inflated. +- Add a pluggable post-download `ContentProcessor` registry. +- Add pluggable document conversion: `DocumentProcessor` runs an ordered, + per-format chain of `DocumentConverter` backends configured with + `DocumentConversion` (formats, chain order, 25 MB document cap, remote opt-in) + through `ToolBuilder`, the CLI/MCP flags `--convert`, `--converter`, + `--allow-remote-conversion`, `--max-document-bytes`, and Python keyword arguments. +- Built-in converters behind Cargo features: `pdf-inspector` for text-based PDFs + (`docs-pdf`, default) and `office` for Word (DOCX), PowerPoint (PPTX) and + spreadsheets (XLSX, XLSM, XLSB, XLS, ODS) with zip-bomb limits (`docs-office`, + default), plus opt-in remote `CloudflareConverter` (Workers AI `toMarkdown`, + `docs-cloudflare`) and generic `HttpConverter` (`docs-http`). Remote converters + run only when explicitly allowed and never under `ToolBuilder::hardened()`. +- `ToolBuilder::content_processor` adds custom processors ahead of the built-ins, + and `FetchOptions` carries `document_conversion` and `content_processors`. - Discover agent-facing resources from HTTP and HTML declarations plus bounded conventional probes, including `llms.txt`, `auth.md`, OAuth metadata, MCP server cards, A2A agent cards, and Agent Skills indexes. Markdown responses diff --git a/README.md b/README.md index 60cc502..afa018a 100644 --- a/README.md +++ b/README.md @@ -46,16 +46,59 @@ requests. - `HtmlProcessor` - metadata and focused-content extraction followed by native Markdown or text conversion; accepts a custom `HtmlToMarkdownConverter` -- `PdfProcessor` - text-based PDF classification and Markdown extraction via - [`pdf-inspector`](https://github.com/firecrawl/pdf-inspector); scanned or - image-only pages are reported as requiring OCR -- `OfficeProcessor` - Word (DOCX), PowerPoint (PPTX) and spreadsheet (XLSX, XLS, - ODS) documents to Markdown, parsed locally with zip-bomb limits +- `DocumentProcessor` - PDF and Office documents to Markdown through a + configurable chain of document converters (see [Document conversion](#document-conversion)) -Custom processors implement `ContentProcessor` and can be registered in a -`ContentProcessorRegistry`. Pass that registry to -`FetcherRegistry::with_content_processors` to retain the built-in fetchers while -customizing post-download processing. +Custom processors implement `ContentProcessor` and are added with +`ToolBuilder::content_processor` (checked before the built-ins) or registered in a +`ContentProcessorRegistry` passed to `FetcherRegistry::with_content_processors`. + +## Document Conversion + +Documents are always downloaded by fetchkit itself, under the same URL policy, +DNS pinning, transport and timeouts as any other fetch. `DocumentProcessor` then +detects the format (Content-Type, then file extension for generic binary types, +confirmed by magic bytes) and runs an ordered chain of `DocumentConverter` +backends. The next converter runs when one fails, does not support the input, or +returns nothing usable (for example a scanned PDF that needs OCR). + +| Converter | Cargo feature | Default | Where it runs | Formats | +| --- | --- | --- | --- | --- | +| `PdfInspectorConverter` (`pdf-inspector`) | `docs-pdf` | on | local | PDF (text layer; reports OCR-only pages) | +| `OfficeConverter` (`office`) | `docs-office` | on | local | DOCX, PPTX, XLSX, XLSM, XLSB, XLS, ODS | +| `CloudflareConverter` (`cloudflare`) | `docs-cloudflare` | off | remote | PDF, DOCX, XLSX, XLSB, XLS, ODT, ODS | +| `HttpConverter` (any name) | `docs-http` | off | remote (configurable) | any format your service accepts | + +`docs-all` enables every backend. Remote converters upload document bytes to a +third party, so they only run when remote conversion is explicitly allowed, and +never under `ToolBuilder::hardened()`. + +```rust +use fetchkit::documents::CloudflareConverter; +use fetchkit::{DocumentConversion, DocumentFormat, ToolBuilder}; + +let tool = ToolBuilder::new() + .document_conversion( + DocumentConversion::default() // pdf-inspector + office + .converter(CloudflareConverter::new(account_id, api_token)) + .allow_remote(true) // opt in to uploads + .chain(DocumentFormat::Pdf, ["pdf-inspector", "cloudflare"]) + .formats([DocumentFormat::Pdf, DocumentFormat::Docx, DocumentFormat::Xlsx]) + .max_document_bytes(25 * 1024 * 1024), // separate from max_body_size + ) + .build(); +``` + +Extension points: implement `DocumentConverter` for a new backend, +`FormatDetector` to change format detection, `ConverterClient` to route remote +uploads through your own egress, or `ContentProcessor` to bypass the chain. + +CLI and MCP server flags: `--convert pdf,docx`, `--converter office,cloudflare`, +`--allow-remote-conversion`, `--max-document-bytes`. The Cloudflare converter +reads `CLOUDFLARE_ACCOUNT_ID` and `CLOUDFLARE_API_TOKEN` (build with +`--features docs-cloudflare`). Python: `FetchkitTool(document_formats=..., +converters=..., allow_remote_conversion=..., max_document_bytes=..., +cloudflare_account_id=..., cloudflare_api_token=...)`. ## Installation @@ -336,14 +379,16 @@ Partial content is returned on body timeout or body-size limit with `truncated: ### PDF And Binary Content -When Markdown is requested, text-based PDFs are downloaded within the configured -body limit and converted to Markdown. PDF parsing runs locally; no OCR service or +When Markdown is requested, text-based PDFs are downloaded within the document +size limit (25 MB default) and converted to Markdown. PDF parsing runs locally; no OCR service or additional network request is used. Scanned/image-only PDFs report `use_ocr` as the suggested next action. Office documents are converted the same way, locally: Word headings, lists and tables, one section per PowerPoint slide, and one Markdown table per spreadsheet -sheet (DOCX, PPTX, XLSX, XLS, ODS). Legacy DOC and PPT files are not supported. +sheet (DOCX, PPTX, XLSX, XLSM, XLSB, XLS, ODS). Legacy DOC and PPT files are not +converted by the built-in local converters; route them to a remote or custom +converter if needed. Other binary content returns metadata only: - Images, audio, video, fonts diff --git a/crates/fetchkit-cli/Cargo.toml b/crates/fetchkit-cli/Cargo.toml index 07e434a..b48078d 100644 --- a/crates/fetchkit-cli/Cargo.toml +++ b/crates/fetchkit-cli/Cargo.toml @@ -20,6 +20,9 @@ doc = false default = [] bot-auth = ["fetchkit/bot-auth"] render-rakers = ["fetchkit/render-rakers"] +docs-cloudflare = ["fetchkit/docs-cloudflare"] +docs-http = ["fetchkit/docs-http"] +docs-all = ["fetchkit/docs-all"] [dependencies] fetchkit = { path = "../fetchkit", version = "0.5.0" } diff --git a/crates/fetchkit-cli/src/main.rs b/crates/fetchkit-cli/src/main.rs index 84d081c..ec7c1eb 100644 --- a/crates/fetchkit-cli/src/main.rs +++ b/crates/fetchkit-cli/src/main.rs @@ -13,8 +13,8 @@ mod mcp; -use clap::{Parser, Subcommand, ValueEnum}; -use fetchkit::{CrawlPage, FetchRequest, Tool, TOOL_LLMTXT}; +use clap::{Args, Parser, Subcommand, ValueEnum}; +use fetchkit::{CrawlPage, DocumentConversionSettings, FetchRequest, Tool, TOOL_LLMTXT}; use std::io::{self, Write}; /// Output format for fetch subcommand @@ -27,6 +27,41 @@ enum OutputFormat { Json, } +/// Document (PDF, Office) conversion options shared by `fetch` and `mcp`. +#[derive(Args, Debug, Default, Clone)] +struct DocumentArgs { + /// Document formats to convert, comma-separated (e.g. pdf,docx,xlsx) + #[arg(long = "convert", value_delimiter = ',')] + formats: Option>, + + /// Converter order, comma-separated (pdf-inspector, office, cloudflare). Cloudflare + /// reads CLOUDFLARE_ACCOUNT_ID and CLOUDFLARE_API_TOKEN. + #[arg(long = "converter", value_delimiter = ',')] + converters: Option>, + + /// Allow converters that upload documents to a third-party service + #[arg(long)] + allow_remote_conversion: bool, + + /// Maximum document download size in bytes (default 25 MB) + #[arg(long)] + max_document_bytes: Option, +} + +impl DocumentArgs { + fn settings(&self, respect_proxy_env: bool) -> DocumentConversionSettings { + DocumentConversionSettings { + formats: self.formats.clone(), + converters: self.converters.clone(), + allow_remote: self.allow_remote_conversion, + max_document_bytes: self.max_document_bytes, + cloudflare_account_id: std::env::var("CLOUDFLARE_ACCOUNT_ID").ok(), + cloudflare_api_token: std::env::var("CLOUDFLARE_API_TOKEN").ok(), + respect_proxy_env, + } + } +} + /// Fetchkit - AI-friendly web content fetching tool #[derive(Parser, Debug)] #[command(name = "fetchkit")] @@ -59,6 +94,9 @@ enum Commands { /// Agent FQDN for Signature-Agent header (requires --bot-auth-key) #[arg(long)] bot_auth_agent: Option, + + #[command(flatten)] + documents: DocumentArgs, }, /// Fetch URL and output as markdown with metadata frontmatter Fetch { @@ -104,6 +142,9 @@ enum Commands { /// Render HTML with the rakers backend before conversion #[arg(long)] render_rakers: bool, + + #[command(flatten)] + documents: DocumentArgs, }, } @@ -123,6 +164,7 @@ async fn main() { allow_env_proxy, bot_auth_key, bot_auth_agent, + documents, }) => { mcp::run_server(build_tool( None, @@ -131,6 +173,7 @@ async fn main() { bot_auth_key, bot_auth_agent, false, + &documents, )) .await; } @@ -146,8 +189,10 @@ async fn main() { crawl, max_pages, render_rakers, + documents, }) => { let options = FetchCommandOptions { + documents, output, user_agent, hardened, @@ -177,9 +222,19 @@ fn build_tool( bot_auth_key: Option, bot_auth_agent: Option, render_rakers: bool, + documents: &DocumentArgs, ) -> Tool { let mut builder = Tool::builder().enable_markdown(true); + let settings = documents.settings(allow_env_proxy); + if !settings.is_default() { + let conversion = settings.build().unwrap_or_else(|e| { + eprintln!("Error: {e}"); + std::process::exit(1); + }); + builder = builder.document_conversion(conversion); + } + if hardened { builder = builder.hardened(); } @@ -229,6 +284,7 @@ fn build_tool( } struct FetchCommandOptions { + documents: DocumentArgs, output: OutputFormat, user_agent: Option, hardened: bool, @@ -260,6 +316,7 @@ async fn run_fetch(url: &str, options: FetchCommandOptions) { options.bot_auth_key, options.bot_auth_agent, options.render_rakers, + &options.documents, ); // Execute request diff --git a/crates/fetchkit-python/Cargo.toml b/crates/fetchkit-python/Cargo.toml index b98b22a..c9f9ab0 100644 --- a/crates/fetchkit-python/Cargo.toml +++ b/crates/fetchkit-python/Cargo.toml @@ -12,6 +12,11 @@ publish = false name = "fetchkit_py" crate-type = ["cdylib"] +[features] +default = [] +docs-cloudflare = ["fetchkit/docs-cloudflare"] +docs-all = ["fetchkit/docs-all"] + [dependencies] fetchkit = { path = "../fetchkit" } pyo3 = { version = "0.29", features = ["extension-module"] } diff --git a/crates/fetchkit-python/src/lib.rs b/crates/fetchkit-python/src/lib.rs index 52f9bcd..5dc7a45 100644 --- a/crates/fetchkit-python/src/lib.rs +++ b/crates/fetchkit-python/src/lib.rs @@ -12,7 +12,10 @@ //! print(response.content) //! ``` -use fetchkit::{FetchError, FetchRequest, FetchResponse, HttpMethod, Tool, ToolBuilder}; +use fetchkit::{ + DocumentConversionSettings, FetchError, FetchRequest, FetchResponse, HttpMethod, Tool, + ToolBuilder, +}; use pyo3::exceptions::{PyDeprecationWarning, PyValueError}; use pyo3::prelude::*; @@ -214,7 +217,13 @@ impl PyFetchkitTool { allowed_ports=None, blocked_hosts=None, same_host_redirects_only=None, - hardened=false + hardened=false, + document_formats=None, + converters=None, + allow_remote_conversion=false, + max_document_bytes=None, + cloudflare_account_id=None, + cloudflare_api_token=None ))] fn new( enable_markdown: bool, @@ -229,6 +238,12 @@ impl PyFetchkitTool { blocked_hosts: Option>, same_host_redirects_only: Option, hardened: bool, + document_formats: Option>, + converters: Option>, + allow_remote_conversion: bool, + max_document_bytes: Option, + cloudflare_account_id: Option, + cloudflare_api_token: Option, ) -> PyResult { let mut builder = ToolBuilder::new() .enable_markdown(enable_markdown) @@ -279,6 +294,20 @@ impl PyFetchkitTool { } } + let documents = DocumentConversionSettings { + formats: document_formats, + converters, + allow_remote: allow_remote_conversion, + max_document_bytes, + cloudflare_account_id, + cloudflare_api_token, + respect_proxy_env, + }; + if !documents.is_default() { + builder = + builder.document_conversion(documents.build().map_err(PyValueError::new_err)?); + } + let runtime = tokio::runtime::Runtime::new() .map_err(|e| PyValueError::new_err(format!("Failed to create runtime: {}", e)))?; @@ -406,6 +435,12 @@ fn deprecated_fetch_kit_tool( blocked_hosts, same_host_redirects_only, hardened, + None, + None, + false, + None, + None, + None, ) } @@ -423,7 +458,8 @@ fn fetch( max_pages: Option, ) -> PyResult { let tool = PyFetchkitTool::new( - true, true, None, None, None, None, true, false, None, None, None, false, + true, true, None, None, None, None, true, false, None, None, None, false, None, None, + false, None, None, None, )?; tool.fetch( url, diff --git a/crates/fetchkit/Cargo.toml b/crates/fetchkit/Cargo.toml index 5489dc9..3c9808b 100644 --- a/crates/fetchkit/Cargo.toml +++ b/crates/fetchkit/Cargo.toml @@ -11,7 +11,13 @@ categories.workspace = true readme = "../../README.md" [features] -default = [] +default = ["docs-pdf", "docs-office"] +# Document conversion backends (see knowledge/foundations/fetchers.md) +docs-pdf = ["dep:pdf-inspector"] +docs-office = ["dep:calamine", "dep:zip", "dep:quick-xml"] +docs-cloudflare = [] +docs-http = [] +docs-all = ["docs-pdf", "docs-office", "docs-cloudflare", "docs-http"] bot-auth = ["dep:ed25519-dalek", "dep:base64", "dep:sha2", "dep:rand"] live-tests = [] render-rakers = ["dep:rakers"] @@ -30,10 +36,10 @@ futures = { workspace = true } bytes = { workspace = true } async-trait = { workspace = true } tower = { workspace = true } -pdf-inspector = { workspace = true } -calamine = { workspace = true } -zip = { workspace = true } -quick-xml = { workspace = true } +pdf-inspector = { workspace = true, optional = true } +calamine = { workspace = true, optional = true } +zip = { workspace = true, optional = true } +quick-xml = { workspace = true, optional = true } libc = { workspace = true } rakers = { workspace = true, optional = true } diff --git a/crates/fetchkit/src/client.rs b/crates/fetchkit/src/client.rs index 09ee2f3..8cc1927 100644 --- a/crates/fetchkit/src/client.rs +++ b/crates/fetchkit/src/client.rs @@ -5,7 +5,9 @@ #[cfg(feature = "bot-auth")] use crate::bot_auth::BotAuthConfig; +use crate::content::ContentProcessor; use crate::dns::DnsPolicy; +use crate::documents::DocumentConversion; use crate::error::FetchError; use crate::fetchers::FetcherRegistry; use crate::transport::{HttpTransport, ReqwestTransport}; @@ -57,6 +59,11 @@ pub struct FetchOptions { /// validation, DNS policy resolution, redirect following, and content handling; /// only the socket-level send is delegated. pub transport: Option>, + /// Document (PDF, Office) conversion config. `None` uses + /// [`DocumentConversion::default`]: built-in local converters only. + pub document_conversion: Option, + /// Custom content processors, consulted before the built-in ones. + pub content_processors: Vec>, } // Manual Debug: `transport` is a trait object that does not implement Debug. @@ -86,6 +93,15 @@ impl std::fmt::Debug for FetchOptions { .as_ref() .map(|_| "") .unwrap_or("None"), + ) + .field("document_conversion", &self.document_conversion) + .field( + "content_processors", + &self + .content_processors + .iter() + .map(|processor| processor.name()) + .collect::>(), ); d.finish() } @@ -205,8 +221,8 @@ pub async fn fetch(req: FetchRequest) -> Result { /// Fetch a URL with custom options /// -/// Uses the default fetcher registry with all built-in fetchers. -/// For custom fetcher configuration, use [`FetcherRegistry`] directly. +/// Uses all built-in fetchers, with content processors and document conversion +/// taken from `options`. For custom fetchers, use [`FetcherRegistry`] directly. pub async fn fetch_with_options( mut req: FetchRequest, options: FetchOptions, @@ -221,8 +237,7 @@ pub async fn fetch_with_options( return crate::crawl::crawl_fetch_with_options(req, options).await; } - // Use registry with default fetchers - let registry = FetcherRegistry::with_defaults(); + let registry = FetcherRegistry::for_options(&options); registry.fetch(req, options).await } @@ -292,7 +307,7 @@ pub async fn batch_fetch_with_options( .map(|(idx, req)| { let options = Arc::clone(&options); async move { - let registry = FetcherRegistry::with_defaults(); + let registry = FetcherRegistry::for_options(&options); let result = registry.fetch(req, (*options).clone()).await; (idx, result) } diff --git a/crates/fetchkit/src/content.rs b/crates/fetchkit/src/content.rs index 6d693ed..fb7e073 100644 --- a/crates/fetchkit/src/content.rs +++ b/crates/fetchkit/src/content.rs @@ -3,28 +3,20 @@ //! Fetchers own retrieval and network policy. Content processors operate only on //! bounded response bytes, so format-specific extraction cannot bypass egress controls. -use std::sync::{Arc, LazyLock}; +use std::sync::Arc; use async_trait::async_trait; use bytes::Bytes; -use pdf_inspector::process_pdf_mem; use thiserror::Error; -use tokio::sync::Semaphore; use url::Url; use crate::convert::{ extract_headings, extract_metadata, extract_readable_content, filter_excessive_newlines, html_to_text, strip_boilerplate, BuiltinHtmlToMarkdownConverter, HtmlToMarkdownConverter, }; -use crate::office::OfficeProcessor; +use crate::documents::{DocumentConversion, DocumentProcessor}; use crate::{PageMetadata, PageQuality}; -// pdf-inspector uses Rayon internally, and every document parser is CPU-bound; bound -// concurrent documents (PDF and Office alike) so batch fetches cannot multiply parser -// work without limit. -pub(crate) static DOCUMENT_PROCESSING_LIMIT: LazyLock> = - LazyLock::new(|| Arc::new(Semaphore::new(2))); - /// Bounded response bytes and metadata passed to a [`ContentProcessor`]. pub struct ContentProcessorInput { /// Final URL after redirects. @@ -119,6 +111,12 @@ pub trait ContentProcessor: Send + Sync { false } + /// Download cap for responses this processor handles, overriding + /// `max_body_size`. Output is still capped by `max_body_size`. + fn max_input_bytes(&self) -> Option { + None + } + /// Process an already-downloaded, bounded response body. async fn process( &self, @@ -126,6 +124,36 @@ pub trait ContentProcessor: Send + Sync { ) -> Result; } +#[async_trait] +impl ContentProcessor for Arc { + fn name(&self) -> &'static str { + (**self).name() + } + + fn matches(&self, url: &Url, content_type: Option<&str>) -> bool { + (**self).matches(url, content_type) + } + + fn supports_output(&self, output_format: ContentOutputFormat) -> bool { + (**self).supports_output(output_format) + } + + fn accepts_truncated_input(&self) -> bool { + (**self).accepts_truncated_input() + } + + fn max_input_bytes(&self) -> Option { + (**self).max_input_bytes() + } + + async fn process( + &self, + input: ContentProcessorInput, + ) -> Result { + (**self).process(input).await + } +} + /// Ordered registry of response content processors. pub struct ContentProcessorRegistry { processors: Vec>, @@ -145,15 +173,36 @@ impl ContentProcessorRegistry { } } - /// Create a registry containing fetchkit's built-in processors. + /// Create a registry containing fetchkit's built-in processors: document + /// conversion with [`DocumentConversion::default`], then HTML. pub fn with_defaults() -> Self { + Self::with_document_conversion(DocumentConversion::default()) + } + + /// Create the built-in registry with a custom document conversion config. + pub fn with_document_conversion(conversion: DocumentConversion) -> Self { let mut registry = Self::new(); - registry.register(Box::new(PdfProcessor)); - registry.register(Box::new(OfficeProcessor)); + registry.register(Box::new(DocumentProcessor::new(conversion))); registry.register(Box::new(HtmlProcessor::default())); registry } + /// Build the registry described by fetch options: custom processors first + /// (first match wins), then document conversion, then HTML. + pub fn for_options(options: &crate::FetchOptions) -> Self { + let mut registry = Self::new(); + for processor in &options.content_processors { + registry.register(Box::new(Arc::clone(processor))); + } + registry.register(Box::new(DocumentProcessor::new( + options.document_conversion.clone().unwrap_or_default(), + ))); + if !registry.contains_named("html") { + registry.register(Box::new(HtmlProcessor::default())); + } + registry + } + /// Append a processor. The first matching processor wins. pub fn register(&mut self, processor: Box) { self.processors.push(processor); @@ -271,122 +320,8 @@ impl ContentProcessor for HtmlProcessor { } } -/// Extracts native text PDFs as Markdown using `pdf-inspector`. -pub struct PdfProcessor; - -pub(crate) async fn run_document_task( - limit: Arc, - task: F, -) -> Result -where - T: Send + 'static, - F: FnOnce() -> T + Send + 'static, -{ - let permit = limit.acquire_owned().await.map_err(|error| { - ContentProcessorError(format!("document processing unavailable: {error}")) - })?; - tokio::task::spawn_blocking(move || { - // Keep the permit in the blocking task: cancelling its async caller must not - // admit another parse while this non-cancellable work is still running. - let _permit = permit; - task() - }) - .await - .map_err(|error| ContentProcessorError(format!("document processor task failed: {error}"))) -} - -#[async_trait] -impl ContentProcessor for PdfProcessor { - fn name(&self) -> &'static str { - "pdf" - } - - fn matches(&self, url: &Url, content_type: Option<&str>) -> bool { - let media_type = content_type - .and_then(|value| value.split(';').next()) - .map(str::trim) - .unwrap_or_default(); - if media_type.eq_ignore_ascii_case("application/pdf") - || media_type.eq_ignore_ascii_case("application/x-pdf") - { - return true; - } - - (media_type.is_empty() || media_type.eq_ignore_ascii_case("application/octet-stream")) - && url.path().to_ascii_lowercase().ends_with(".pdf") - } - - async fn process( - &self, - input: ContentProcessorInput, - ) -> Result { - let result = run_document_task(Arc::clone(&DOCUMENT_PROCESSING_LIMIT), move || { - process_pdf_mem(&input.body) - }) - .await? - .map_err(|error| ContentProcessorError(error.to_string()))?; - - let needs_ocr = !result.pages_needing_ocr.is_empty(); - let mut warnings = Vec::new(); - if needs_ocr { - warnings.push("pdf_requires_ocr".to_string()); - } - if result.has_encoding_issues { - warnings.push("pdf_encoding_issues".to_string()); - } - - let content = result.markdown.filter(|value| !value.trim().is_empty()); - let error = content.is_none().then(|| { - if needs_ocr { - format!( - "PDF requires OCR for pages: {}", - comma_separated_pages(&result.pages_needing_ocr) - ) - } else { - "PDF contains no extractable text".to_string() - } - }); - let metadata = PageMetadata { - title: result.title, - extraction_method: Some("pdf_inspector".to_string()), - ..Default::default() - }; - let quality = PageQuality { - score: if content.is_none() { - 0.0 - } else if warnings.is_empty() { - 1.0 - } else { - 0.6 - }, - warnings, - extraction_method: Some("pdf_inspector".to_string()), - suggested_next_action: needs_ocr.then(|| "use_ocr".to_string()), - ..Default::default() - }; - - Ok(ProcessedContent { - format: content.as_ref().map(|_| "markdown".to_string()), - content, - metadata: Some(metadata), - quality: Some(quality), - error, - }) - } -} - -fn comma_separated_pages(pages: &[u32]) -> String { - pages - .iter() - .map(u32::to_string) - .collect::>() - .join(", ") -} - #[cfg(test)] mod tests { - use std::sync::mpsc; - use super::*; struct StubMarkdownConverter; @@ -397,6 +332,7 @@ mod tests { } } + #[cfg(feature = "docs-pdf")] fn text_pdf(text: &str) -> Bytes { let escaped = text .replace('\\', "\\\\") @@ -433,53 +369,20 @@ mod tests { Bytes::from(pdf) } - #[test] - fn pdf_processor_matches_media_type_or_binary_pdf_path() { - let processor = PdfProcessor; - let extensionless = Url::parse("https://example.com/download?id=1").unwrap(); - let pdf_path = Url::parse("https://example.com/report.PDF").unwrap(); - - assert!(processor.matches(&extensionless, Some("application/pdf; charset=binary"))); - assert!(processor.matches(&pdf_path, Some("application/octet-stream"))); - assert!(!processor.matches(&pdf_path, Some("text/html"))); - assert!(!processor.matches(&extensionless, Some("application/octet-stream"))); - } - - #[tokio::test(flavor = "multi_thread", worker_threads = 2)] - async fn cancelled_pdf_caller_does_not_release_running_task_permit() { - let limit = Arc::new(Semaphore::new(1)); - let (started_tx, started_rx) = mpsc::channel(); - let (release_tx, release_rx) = mpsc::channel(); - let handle = tokio::spawn(run_document_task(Arc::clone(&limit), move || { - started_tx.send(()).unwrap(); - release_rx.recv().unwrap(); - })); - - started_rx.recv().unwrap(); - handle.abort(); - assert_eq!(limit.available_permits(), 0); - - release_tx.send(()).unwrap(); - tokio::time::timeout(std::time::Duration::from_secs(1), async { - while limit.available_permits() == 0 { - tokio::task::yield_now().await; - } - }) - .await - .unwrap(); - } - #[test] fn registry_uses_first_matching_processor() { let registry = ContentProcessorRegistry::with_defaults(); let url = Url::parse("https://example.com/document").unwrap(); + #[cfg(feature = "docs-pdf")] assert_eq!( registry .find(&url, Some("application/pdf")) .map(ContentProcessor::name), - Some("pdf") + Some("document") ); + #[cfg(not(feature = "docs-pdf"))] + assert!(registry.find(&url, Some("application/pdf")).is_none()); assert_eq!( registry .find_for_output( @@ -562,8 +465,9 @@ mod tests { } #[tokio::test] - async fn pdf_processor_extracts_markdown_from_memory() { - let processor = PdfProcessor; + #[cfg(feature = "docs-pdf")] + async fn default_document_processor_extracts_pdf_markdown() { + let processor = DocumentProcessor::default(); let result = processor .process(ContentProcessorInput { url: Url::parse("https://example.com/document").unwrap(), diff --git a/crates/fetchkit/src/crawl.rs b/crates/fetchkit/src/crawl.rs index 18add32..08f93d6 100644 --- a/crates/fetchkit/src/crawl.rs +++ b/crates/fetchkit/src/crawl.rs @@ -23,7 +23,7 @@ pub(crate) async fn crawl_fetch_with_options( .unwrap_or(DEFAULT_CRAWL_MAX_PAGES) .clamp(1, MAX_CRAWL_MAX_PAGES); - let registry = FetcherRegistry::with_defaults(); + let registry = FetcherRegistry::for_options(&options); let mut seed_request = request.clone(); seed_request.crawl = None; seed_request.max_pages = None; diff --git a/crates/fetchkit/src/documents/cloudflare.rs b/crates/fetchkit/src/documents/cloudflare.rs new file mode 100644 index 0000000..7e3d61b --- /dev/null +++ b/crates/fetchkit/src/documents/cloudflare.rs @@ -0,0 +1,323 @@ +// Decisions: +// - Uses the Workers AI `toMarkdown` REST endpoint with one file per request, so a +// failure maps cleanly to one document. +// - Images are not in the default format set: Cloudflare bills image conversion, +// and fetchkit's document chain only routes non-image formats anyway. +// - Formats follow Cloudflare's published list (no PPTX or legacy .doc). Override +// with `CloudflareConverter::formats` when Cloudflare adds more. +//! Remote backend: Cloudflare Workers AI `toMarkdown`. + +use std::sync::Arc; +use std::time::Duration; + +use async_trait::async_trait; +use serde::Deserialize; +use url::Url; + +use super::remote::{ + multipart_file, status_error, ConverterClient, ConverterRequest, ReqwestConverterClient, + SecretString, DEFAULT_REMOTE_TIMEOUT, MAX_REMOTE_RESPONSE_BYTES, +}; +use super::{ + ConvertError, ConvertedDocument, DocumentConverter, DocumentFormat, DocumentInput, Locality, +}; + +const DEFAULT_API_BASE: &str = "https://api.cloudflare.com/client/v4"; + +const DEFAULT_FORMATS: &[DocumentFormat] = &[ + DocumentFormat::Pdf, + DocumentFormat::Docx, + DocumentFormat::Xlsx, + DocumentFormat::Xlsb, + DocumentFormat::Xls, + DocumentFormat::Odt, + DocumentFormat::Ods, +]; + +/// Converts documents with Cloudflare Workers AI `toMarkdown`. +/// +/// Requires an account id and an API token with Workers AI permission. This is a +/// [`Locality::Remote`] converter: it runs only when remote conversion is allowed. +#[derive(Clone)] +pub struct CloudflareConverter { + account_id: String, + api_token: SecretString, + api_base: Url, + formats: Vec, + timeout: Duration, + client: Arc, +} + +impl std::fmt::Debug for CloudflareConverter { + fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { + f.debug_struct("CloudflareConverter") + .field("account_id", &self.account_id) + .field("api_token", &self.api_token) + .field("api_base", &self.api_base.as_str()) + .field("formats", &self.formats) + .field("timeout", &self.timeout) + .finish() + } +} + +impl CloudflareConverter { + /// Create a converter for a Cloudflare account. + pub fn new(account_id: impl Into, api_token: impl Into) -> Self { + Self { + account_id: account_id.into(), + api_token: api_token.into(), + api_base: Url::parse(DEFAULT_API_BASE).expect("valid default Cloudflare API URL"), + formats: DEFAULT_FORMATS.to_vec(), + timeout: DEFAULT_REMOTE_TIMEOUT, + client: Arc::new(ReqwestConverterClient::new()), + } + } + + /// Read `CLOUDFLARE_ACCOUNT_ID` and `CLOUDFLARE_API_TOKEN`. Intended for CLIs; + /// libraries should pass credentials explicitly. + pub fn from_env() -> Option { + let account_id = std::env::var("CLOUDFLARE_ACCOUNT_ID").ok()?; + let api_token = std::env::var("CLOUDFLARE_API_TOKEN").ok()?; + (!account_id.trim().is_empty() && !api_token.trim().is_empty()) + .then(|| Self::new(account_id.trim(), api_token.trim())) + } + + /// Override the API base URL (for tests or a gateway in front of Cloudflare). + pub fn api_base(mut self, api_base: Url) -> Self { + self.api_base = api_base; + self + } + + /// Override the formats sent to Cloudflare. + pub fn formats(mut self, formats: impl IntoIterator) -> Self { + self.formats = formats.into_iter().collect(); + self + } + + /// Per-document request timeout. + pub fn timeout(mut self, timeout: Duration) -> Self { + self.timeout = timeout; + self + } + + /// Route uploads through a custom client (e.g. a host egress proxy). + pub fn client(mut self, client: Arc) -> Self { + self.client = client; + self + } + + fn endpoint(&self) -> Result { + let base = self.api_base.as_str().trim_end_matches('/'); + let account: String = + url::form_urlencoded::byte_serialize(self.account_id.as_bytes()).collect(); + Url::parse(&format!("{base}/accounts/{account}/ai/tomarkdown")) + .map_err(|error| ConvertError::Failed(format!("cloudflare endpoint: {error}"))) + } +} + +#[derive(Deserialize)] +struct Envelope { + #[serde(default)] + success: bool, + #[serde(default)] + result: Vec, + #[serde(default)] + errors: Vec, +} + +#[derive(Deserialize)] +struct FileResult { + #[serde(default)] + format: String, + #[serde(default)] + data: Option, + #[serde(default)] + error: Option, +} + +#[derive(Deserialize)] +struct ApiMessage { + #[serde(default)] + message: String, +} + +#[async_trait] +impl DocumentConverter for CloudflareConverter { + fn name(&self) -> &str { + "cloudflare" + } + + fn locality(&self) -> Locality { + Locality::Remote + } + + fn supports(&self, format: DocumentFormat) -> bool { + self.formats.contains(&format) + } + + async fn convert(&self, input: DocumentInput) -> Result { + let (content_type, body) = multipart_file( + "files", + &input.file_name, + input.format.media_type(), + &input.body, + ); + let response = self + .client + .post(ConverterRequest { + url: self.endpoint()?, + headers: vec![ + ("Content-Type".to_string(), content_type), + ( + "Authorization".to_string(), + format!("Bearer {}", self.api_token.expose()), + ), + ("Accept".to_string(), "application/json".to_string()), + ], + body, + timeout: self.timeout, + max_response_bytes: MAX_REMOTE_RESPONSE_BYTES, + }) + .await?; + + let envelope: Option = serde_json::from_slice(&response.body).ok(); + if let Some(error) = status_error("cloudflare", response.status, &response.body) { + let api_message = envelope + .as_ref() + .and_then(|envelope| envelope.errors.first()) + .map(|message| message.message.clone()) + .filter(|message| !message.is_empty()); + let auth_failure = matches!(response.status, 401 | 403); + return Err(match (error, api_message) { + (ConvertError::Failed(_), Some(message)) if !auth_failure => { + ConvertError::Failed(format!("cloudflare: HTTP {}: {message}", response.status)) + } + (error, _) => error, + }); + } + let envelope = envelope.ok_or_else(|| { + ConvertError::Failed("cloudflare: response is not valid JSON".to_string()) + })?; + if !envelope.success { + let message = envelope + .errors + .iter() + .map(|message| message.message.as_str()) + .collect::>() + .join("; "); + return Err(ConvertError::Failed(format!("cloudflare: {message}"))); + } + let file = envelope + .result + .into_iter() + .next() + .ok_or_else(|| ConvertError::Failed("cloudflare: empty result".to_string()))?; + if file.format != "markdown" { + return Err(ConvertError::Failed(format!( + "cloudflare: {}", + file.error + .unwrap_or_else(|| "conversion failed".to_string()) + ))); + } + Ok(ConvertedDocument { + markdown: file.data.unwrap_or_default(), + extraction_method: Some("cloudflare_tomarkdown".to_string()), + ..Default::default() + }) + } +} + +#[cfg(test)] +mod tests { + use bytes::Bytes; + use wiremock::matchers::{header, header_regex, method, path}; + use wiremock::{Mock, MockServer, ResponseTemplate}; + + use super::*; + + fn input() -> DocumentInput { + DocumentInput { + url: Url::parse("https://example.com/report.docx").unwrap(), + format: DocumentFormat::Docx, + content_type: None, + file_name: "report.docx".to_string(), + body: Bytes::from_static(b"PK\x03\x04docx"), + } + } + + async fn converter(server: &MockServer) -> CloudflareConverter { + CloudflareConverter::new("acct", "secret-token") + .api_base(Url::parse(&server.uri()).unwrap()) + } + + #[tokio::test] + async fn uploads_file_and_returns_markdown() { + let server = MockServer::start().await; + Mock::given(method("POST")) + .and(path("/accounts/acct/ai/tomarkdown")) + .and(header("authorization", "Bearer secret-token")) + .and(header_regex("content-type", "^multipart/form-data; boundary=")) + .respond_with(ResponseTemplate::new(200).set_body_json(serde_json::json!({ + "success": true, + "result": [{"name": "report.docx", "mimeType": "x", "format": "markdown", "tokens": 3, "data": "# Report"}] + }))) + .expect(1) + .mount(&server) + .await; + + let result = converter(&server).await.convert(input()).await.unwrap(); + assert_eq!(result.markdown, "# Report"); + assert_eq!( + result.extraction_method.as_deref(), + Some("cloudflare_tomarkdown") + ); + let requests = server.received_requests().await.unwrap(); + let body = String::from_utf8_lossy(&requests[0].body); + assert!(body.contains("filename=\"report.docx\"")); + assert!(body.contains("wordprocessingml")); + } + + #[tokio::test] + async fn per_file_error_is_reported() { + let server = MockServer::start().await; + Mock::given(method("POST")) + .respond_with(ResponseTemplate::new(200).set_body_json(serde_json::json!({ + "success": true, + "result": [{"name": "report.docx", "format": "error", "error": "corrupt file"}] + }))) + .mount(&server) + .await; + + let error = converter(&server).await.convert(input()).await.unwrap_err(); + assert_eq!( + error, + ConvertError::Failed("cloudflare: corrupt file".into()) + ); + } + + #[tokio::test] + async fn auth_failure_does_not_leak_token() { + let server = MockServer::start().await; + Mock::given(method("POST")) + .respond_with(ResponseTemplate::new(403).set_body_json(serde_json::json!({ + "success": false, "errors": [{"code": 10000, "message": "Authentication error"}] + }))) + .mount(&server) + .await; + + let converter = converter(&server).await; + let error = converter.convert(input()).await.unwrap_err().to_string(); + assert!(error.contains("authentication failed")); + assert!(!error.contains("secret-token")); + assert!(!format!("{converter:?}").contains("secret-token")); + } + + #[test] + fn default_formats_exclude_unsupported_office_types() { + let converter = CloudflareConverter::new("a", "b"); + assert!(converter.supports(DocumentFormat::Xlsx)); + assert!(!converter.supports(DocumentFormat::Pptx)); + assert!(!converter.supports(DocumentFormat::Doc)); + assert_eq!(converter.locality(), Locality::Remote); + } +} diff --git a/crates/fetchkit/src/documents/http.rs b/crates/fetchkit/src/documents/http.rs new file mode 100644 index 0000000..9ffd1a3 --- /dev/null +++ b/crates/fetchkit/src/documents/http.rs @@ -0,0 +1,258 @@ +//! Remote backend: any HTTP service that accepts a multipart upload and returns +//! Markdown, such as a self-hosted Docling or MarkItDown wrapper. + +use std::sync::Arc; +use std::time::Duration; + +use async_trait::async_trait; +use url::Url; + +use super::remote::{ + multipart_file, status_error, ConverterClient, ConverterRequest, ReqwestConverterClient, + SecretString, DEFAULT_REMOTE_TIMEOUT, MAX_REMOTE_RESPONSE_BYTES, +}; +use super::{ + ConvertError, ConvertedDocument, DocumentConverter, DocumentFormat, DocumentInput, Locality, +}; + +/// Generic multipart-upload converter. +/// +/// Sends the document as one multipart file field and reads Markdown either from +/// the whole response body or, with [`json_pointer`](Self::json_pointer), from a +/// string inside a JSON response. +#[derive(Clone)] +pub struct HttpConverter { + name: String, + endpoint: Url, + field: String, + headers: Vec<(String, SecretString)>, + formats: Vec, + json_pointer: Option, + locality: Locality, + timeout: Duration, + max_input_bytes: Option, + client: Arc, +} + +impl std::fmt::Debug for HttpConverter { + fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { + f.debug_struct("HttpConverter") + .field("name", &self.name) + .field("endpoint", &self.endpoint.as_str()) + .field("field", &self.field) + .field( + "headers", + &self + .headers + .iter() + .map(|(name, _)| name.as_str()) + .collect::>(), + ) + .field("formats", &self.formats) + .field("json_pointer", &self.json_pointer) + .field("locality", &self.locality) + .finish() + } +} + +impl HttpConverter { + /// Create a converter named `name` that POSTs to `endpoint`. + /// + /// Defaults: field `file`, all formats, raw-body Markdown response, remote. + pub fn new(name: impl Into, endpoint: Url) -> Self { + Self { + name: name.into(), + endpoint, + field: "file".to_string(), + headers: Vec::new(), + formats: DocumentFormat::ALL.to_vec(), + json_pointer: None, + locality: Locality::Remote, + timeout: DEFAULT_REMOTE_TIMEOUT, + max_input_bytes: None, + client: Arc::new(ReqwestConverterClient::new()), + } + } + + /// Multipart field name for the file. + pub fn field(mut self, field: impl Into) -> Self { + self.field = field.into(); + self + } + + /// Add a request header. Values are redacted from `Debug` output. + pub fn header(mut self, name: impl Into, value: impl Into) -> Self { + self.headers.push((name.into(), value.into())); + self + } + + /// Formats this service accepts. + pub fn formats(mut self, formats: impl IntoIterator) -> Self { + self.formats = formats.into_iter().collect(); + self + } + + /// Read Markdown from this JSON pointer (e.g. `/markdown`) instead of the raw body. + pub fn json_pointer(mut self, pointer: impl Into) -> Self { + self.json_pointer = Some(pointer.into()); + self + } + + /// Mark the service as local. Only for services you operate on the same trust + /// boundary: local converters run even when remote conversion is not allowed. + pub fn locality(mut self, locality: Locality) -> Self { + self.locality = locality; + self + } + + /// Per-document request timeout. + pub fn timeout(mut self, timeout: Duration) -> Self { + self.timeout = timeout; + self + } + + /// Largest document this service accepts. + pub fn max_input_bytes(mut self, bytes: usize) -> Self { + self.max_input_bytes = Some(bytes); + self + } + + /// Route uploads through a custom client. + pub fn client(mut self, client: Arc) -> Self { + self.client = client; + self + } +} + +#[async_trait] +impl DocumentConverter for HttpConverter { + fn name(&self) -> &str { + &self.name + } + + fn locality(&self) -> Locality { + self.locality + } + + fn supports(&self, format: DocumentFormat) -> bool { + self.formats.contains(&format) + } + + fn max_input_bytes(&self) -> Option { + self.max_input_bytes + } + + async fn convert(&self, input: DocumentInput) -> Result { + let (content_type, body) = multipart_file( + &self.field, + &input.file_name, + input.format.media_type(), + &input.body, + ); + let mut headers = vec![("Content-Type".to_string(), content_type)]; + headers.extend( + self.headers + .iter() + .map(|(name, value)| (name.clone(), value.expose().to_string())), + ); + let response = self + .client + .post(ConverterRequest { + url: self.endpoint.clone(), + headers, + body, + timeout: self.timeout, + max_response_bytes: MAX_REMOTE_RESPONSE_BYTES, + }) + .await?; + if let Some(error) = status_error(&self.name, response.status, &response.body) { + return Err(error); + } + + let markdown = match &self.json_pointer { + None => String::from_utf8_lossy(&response.body).into_owned(), + Some(pointer) => { + let json: serde_json::Value = + serde_json::from_slice(&response.body).map_err(|error| { + ConvertError::Failed(format!("{}: invalid JSON: {error}", self.name)) + })?; + json.pointer(pointer) + .and_then(serde_json::Value::as_str) + .ok_or_else(|| { + ConvertError::Failed(format!("{}: no string at {pointer}", self.name)) + })? + .to_string() + } + }; + Ok(ConvertedDocument::markdown(markdown)) + } +} + +#[cfg(test)] +mod tests { + use bytes::Bytes; + use wiremock::matchers::{header, method, path}; + use wiremock::{Mock, MockServer, ResponseTemplate}; + + use super::*; + + fn input() -> DocumentInput { + DocumentInput { + url: Url::parse("https://example.com/deck.pptx").unwrap(), + format: DocumentFormat::Pptx, + content_type: None, + file_name: "deck.pptx".to_string(), + body: Bytes::from_static(b"PK\x03\x04pptx"), + } + } + + #[tokio::test] + async fn reads_markdown_from_json_pointer() { + let server = MockServer::start().await; + Mock::given(method("POST")) + .and(path("/convert")) + .and(header("x-api-key", "k")) + .respond_with( + ResponseTemplate::new(200) + .set_body_json(serde_json::json!({"document": {"md": "# Slides"}})), + ) + .mount(&server) + .await; + + let converter = HttpConverter::new( + "docling", + Url::parse(&format!("{}/convert", server.uri())).unwrap(), + ) + .header("x-api-key", "k") + .json_pointer("/document/md"); + let result = converter.convert(input()).await.unwrap(); + assert_eq!(result.markdown, "# Slides"); + assert!(!format!("{converter:?}").contains("\"k\"")); + let requests = server.received_requests().await.unwrap(); + assert!(String::from_utf8_lossy(&requests[0].body).contains("name=\"file\"")); + } + + #[tokio::test] + async fn raw_body_and_http_errors() { + let server = MockServer::start().await; + Mock::given(method("POST")) + .and(path("/ok")) + .respond_with(ResponseTemplate::new(200).set_body_string("plain md")) + .mount(&server) + .await; + Mock::given(method("POST")) + .and(path("/fail")) + .respond_with(ResponseTemplate::new(500).set_body_string("kaput")) + .mount(&server) + .await; + + let ok = HttpConverter::new("svc", Url::parse(&format!("{}/ok", server.uri())).unwrap()); + assert_eq!(ok.convert(input()).await.unwrap().markdown, "plain md"); + let fail = HttpConverter::new( + "svc", + Url::parse(&format!("{}/fail", server.uri())).unwrap(), + ); + let error = fail.convert(input()).await.unwrap_err().to_string(); + assert!(error.contains("HTTP 500: kaput")); + } +} diff --git a/crates/fetchkit/src/documents/mod.rs b/crates/fetchkit/src/documents/mod.rs new file mode 100644 index 0000000..0f8b891 --- /dev/null +++ b/crates/fetchkit/src/documents/mod.rs @@ -0,0 +1,1064 @@ +// Decisions: +// - Document conversion is a backend chain behind ONE ContentProcessor +// (`DocumentProcessor`). Fetchers keep owning retrieval and egress policy; a +// converter only ever receives bounded bytes fetchkit already downloaded, never a +// URL to fetch itself. +// - Remote converters (Locality::Remote) send those bytes to a third party, so they +// are skipped unless `DocumentConversion::allow_remote(true)` is set explicitly. +// `ToolBuilder::hardened()` forces remote conversion off (TM-LEAK-006). +// - Per-format chain order is by converter name. A converter falls through to the +// next one on unsupported input, an error, or an empty / OCR-required result, so +// "local first, remote for scanned PDFs" needs no special case. +// - Format detection runs on response metadata before download (Content-Type, then +// URL extension for missing/generic types) and is confirmed against magic bytes +// after download, so a mislabelled body never reaches a parser for another format. +//! Pluggable document (PDF, Office, OpenDocument) to Markdown conversion. +//! +//! [`DocumentProcessor`] is a [`ContentProcessor`] that detects the document format +//! and runs an ordered chain of [`DocumentConverter`] backends configured by +//! [`DocumentConversion`]. Implement [`DocumentConverter`] to add a backend, or +//! [`FormatDetector`] to change how formats are recognised. + +#[cfg(feature = "docs-cloudflare")] +mod cloudflare; +#[cfg(feature = "docs-http")] +mod http; +#[cfg(feature = "docs-office")] +mod office; +#[cfg(feature = "docs-pdf")] +mod pdf_inspector; +#[cfg(any(feature = "docs-cloudflare", feature = "docs-http"))] +mod remote; +mod settings; + +use std::collections::BTreeMap; +use std::fmt; +use std::sync::Arc; + +use async_trait::async_trait; +use bytes::Bytes; +use tracing::{debug, warn}; +use url::Url; + +use crate::content::{ + ContentOutputFormat, ContentProcessor, ContentProcessorError, ContentProcessorInput, + ProcessedContent, +}; +use crate::{PageMetadata, PageQuality}; + +#[cfg(feature = "docs-cloudflare")] +pub use cloudflare::CloudflareConverter; +#[cfg(feature = "docs-http")] +pub use http::HttpConverter; +#[cfg(feature = "docs-office")] +pub use office::OfficeConverter; +#[cfg(feature = "docs-pdf")] +pub use pdf_inspector::PdfInspectorConverter; +#[cfg(any(feature = "docs-cloudflare", feature = "docs-http"))] +pub use remote::{ + ConverterClient, ConverterRequest, ConverterResponse, ReqwestConverterClient, SecretString, +}; +pub use settings::{DocumentConversionSettings, BUILTIN_CONVERTERS}; + +/// Default maximum size of a downloaded document (25 MB). +/// +/// Documents are commonly larger than HTML pages, so this is separate from +/// `max_body_size`. Converted output is still capped by `max_body_size`. +pub const DEFAULT_MAX_DOCUMENT_BYTES: usize = 25 * 1024 * 1024; + +/// Binary document formats fetchkit can route to a converter. +#[derive(Debug, Clone, Copy, PartialEq, Eq, PartialOrd, Ord, Hash)] +#[non_exhaustive] +pub enum DocumentFormat { + /// Portable Document Format. + Pdf, + /// Word (Office Open XML). + Docx, + /// Word 97-2003. + Doc, + /// Excel (Office Open XML), including macro-enabled `.xlsm`. + Xlsx, + /// Excel binary workbook. + Xlsb, + /// Excel 97-2003. + Xls, + /// PowerPoint (Office Open XML). + Pptx, + /// PowerPoint 97-2003. + Ppt, + /// OpenDocument text. + Odt, + /// OpenDocument spreadsheet. + Ods, + /// OpenDocument presentation. + Odp, + /// Rich Text Format. + Rtf, + /// EPUB e-book. + Epub, +} + +/// How a container format is recognised from its first bytes. +#[derive(Clone, Copy)] +enum Magic { + Pdf, + Zip, + Ole2, + Rtf, +} + +impl DocumentFormat { + /// Every known format. + pub const ALL: &'static [DocumentFormat] = &[ + Self::Pdf, + Self::Docx, + Self::Doc, + Self::Xlsx, + Self::Xlsb, + Self::Xls, + Self::Pptx, + Self::Ppt, + Self::Odt, + Self::Ods, + Self::Odp, + Self::Rtf, + Self::Epub, + ]; + + /// Stable lowercase identifier, used in config and CLI flags. + pub fn as_str(self) -> &'static str { + match self { + Self::Pdf => "pdf", + Self::Docx => "docx", + Self::Doc => "doc", + Self::Xlsx => "xlsx", + Self::Xlsb => "xlsb", + Self::Xls => "xls", + Self::Pptx => "pptx", + Self::Ppt => "ppt", + Self::Odt => "odt", + Self::Ods => "ods", + Self::Odp => "odp", + Self::Rtf => "rtf", + Self::Epub => "epub", + } + } + + /// Canonical media type. + pub fn media_type(self) -> &'static str { + self.media_types()[0] + } + + /// Media types that identify this format. + pub fn media_types(self) -> &'static [&'static str] { + match self { + Self::Pdf => &["application/pdf", "application/x-pdf"], + Self::Docx => { + &["application/vnd.openxmlformats-officedocument.wordprocessingml.document"] + } + Self::Doc => &["application/msword"], + Self::Xlsx => &[ + "application/vnd.openxmlformats-officedocument.spreadsheetml.sheet", + "application/vnd.ms-excel.sheet.macroenabled.12", + ], + Self::Xlsb => &["application/vnd.ms-excel.sheet.binary.macroenabled.12"], + Self::Xls => &["application/vnd.ms-excel"], + Self::Pptx => { + &["application/vnd.openxmlformats-officedocument.presentationml.presentation"] + } + Self::Ppt => &["application/vnd.ms-powerpoint"], + Self::Odt => &["application/vnd.oasis.opendocument.text"], + Self::Ods => &["application/vnd.oasis.opendocument.spreadsheet"], + Self::Odp => &["application/vnd.oasis.opendocument.presentation"], + Self::Rtf => &["application/rtf", "text/rtf"], + Self::Epub => &["application/epub+zip"], + } + } + + /// File extensions (lowercase, without dot) that identify this format. + pub fn extensions(self) -> &'static [&'static str] { + match self { + Self::Pdf => &["pdf"], + Self::Docx => &["docx"], + Self::Doc => &["doc"], + Self::Xlsx => &["xlsx", "xlsm"], + Self::Xlsb => &["xlsb"], + Self::Xls => &["xls"], + Self::Pptx => &["pptx"], + Self::Ppt => &["ppt"], + Self::Odt => &["odt"], + Self::Ods => &["ods"], + Self::Odp => &["odp"], + Self::Rtf => &["rtf"], + Self::Epub => &["epub"], + } + } + + /// Format for a media type (parameters such as `charset` are ignored). + pub fn from_media_type(content_type: &str) -> Option { + let media_type = media_type(content_type); + Self::ALL.iter().copied().find(|format| { + format + .media_types() + .iter() + .any(|known| known.eq_ignore_ascii_case(media_type)) + }) + } + + /// Format for a file extension (case-insensitive, without dot). + pub fn from_extension(extension: &str) -> Option { + Self::ALL.iter().copied().find(|format| { + format + .extensions() + .iter() + .any(|known| known.eq_ignore_ascii_case(extension)) + }) + } + + /// Whether `body` starts like this format's container. + pub fn matches_magic(self, body: &[u8]) -> bool { + match self.magic() { + // PDF readers accept a header within the first KiB. + Magic::Pdf => body + .get(..body.len().min(1024)) + .is_some_and(|head| head.windows(5).any(|window| window == b"%PDF-")), + Magic::Zip => body.starts_with(b"PK\x03\x04"), + Magic::Ole2 => body.starts_with(&[0xD0, 0xCF, 0x11, 0xE0, 0xA1, 0xB1, 0x1A, 0xE1]), + Magic::Rtf => body.starts_with(b"{\\rtf"), + } + } + + fn magic(self) -> Magic { + match self { + Self::Pdf => Magic::Pdf, + Self::Doc | Self::Xls | Self::Ppt => Magic::Ole2, + Self::Rtf => Magic::Rtf, + Self::Docx + | Self::Xlsx + | Self::Xlsb + | Self::Pptx + | Self::Odt + | Self::Ods + | Self::Odp + | Self::Epub => Magic::Zip, + } + } +} + +impl fmt::Display for DocumentFormat { + fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { + f.write_str(self.as_str()) + } +} + +impl std::str::FromStr for DocumentFormat { + type Err = String; + + fn from_str(value: &str) -> Result { + Self::from_extension(value.trim().trim_start_matches('.')) + .ok_or_else(|| format!("unknown document format: {value}")) + } +} + +fn media_type(content_type: &str) -> &str { + content_type.split(';').next().unwrap_or_default().trim() +} + +// Document parsers are CPU-bound (pdf-inspector also uses Rayon internally); bound +// concurrent local conversions so batch fetches cannot multiply parser work. +#[cfg(any(feature = "docs-pdf", feature = "docs-office"))] +static DOCUMENT_PROCESSING_LIMIT: std::sync::LazyLock> = + std::sync::LazyLock::new(|| Arc::new(tokio::sync::Semaphore::new(2))); + +/// Run CPU-bound conversion on the blocking pool under the shared document limit. +#[cfg(any(feature = "docs-pdf", feature = "docs-office"))] +pub(crate) async fn run_document_task(task: F) -> Result +where + T: Send + 'static, + F: FnOnce() -> T + Send + 'static, +{ + run_limited(Arc::clone(&DOCUMENT_PROCESSING_LIMIT), task).await +} + +#[cfg(any(feature = "docs-pdf", feature = "docs-office"))] +async fn run_limited(limit: Arc, task: F) -> Result +where + T: Send + 'static, + F: FnOnce() -> T + Send + 'static, +{ + let permit = limit.acquire_owned().await.map_err(|error| { + ConvertError::Failed(format!("document processing unavailable: {error}")) + })?; + tokio::task::spawn_blocking(move || { + // Keep the permit in the blocking task: cancelling its async caller must not + // admit another parse while this non-cancellable work is still running. + let _permit = permit; + task() + }) + .await + .map_err(|error| ConvertError::Failed(format!("document processor task failed: {error}"))) +} + +/// Recognises document formats from response metadata. +pub trait FormatDetector: Send + Sync { + /// Detect the format from the final URL and Content-Type, before download. + fn detect(&self, url: &Url, content_type: Option<&str>) -> Option; + + /// Confirm the detected format against the downloaded bytes. + fn confirm(&self, format: DocumentFormat, body: &[u8]) -> bool { + format.matches_magic(body) + } +} + +/// Content-Type first; URL extension only for missing or generic binary types. +#[derive(Debug, Default, Clone, Copy)] +pub struct DefaultFormatDetector; + +const GENERIC_BINARY_TYPES: &[&str] = &[ + "", + "application/octet-stream", + "binary/octet-stream", + "application/binary", + "application/download", + "application/force-download", + "application/x-download", + "application/zip", +]; + +impl FormatDetector for DefaultFormatDetector { + fn detect(&self, url: &Url, content_type: Option<&str>) -> Option { + let media_type = media_type(content_type.unwrap_or_default()); + if let Some(format) = DocumentFormat::from_media_type(media_type) { + return Some(format); + } + if !GENERIC_BINARY_TYPES + .iter() + .any(|generic| generic.eq_ignore_ascii_case(media_type)) + { + return None; + } + let file_name = url.path_segments()?.next_back()?; + let (_, extension) = file_name.rsplit_once('.')?; + DocumentFormat::from_extension(extension) + } +} + +/// Where a converter runs. Remote converters receive document bytes. +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub enum Locality { + /// Converts in-process or on the same host. + Local, + /// Sends document bytes to another service. + Remote, +} + +/// Bounded document bytes handed to a [`DocumentConverter`]. +#[derive(Debug, Clone)] +pub struct DocumentInput { + /// Final URL after redirects (informational; converters must not fetch it). + pub url: Url, + /// Detected and magic-byte-confirmed format. + pub format: DocumentFormat, + /// Response Content-Type, when present. + pub content_type: Option, + /// File name derived from the URL, with an extension matching `format`. + pub file_name: String, + /// Complete document bytes. Truncated downloads are never converted. + pub body: Bytes, +} + +/// Markdown produced by a [`DocumentConverter`]. +#[derive(Debug, Clone, Default)] +pub struct ConvertedDocument { + /// Converted Markdown. Empty means nothing extractable was found. + pub markdown: String, + /// Document title, when known. + pub title: Option, + /// Machine-readable warnings, such as `pdf_encoding_issues`. + pub warnings: Vec, + /// 1-based pages that need OCR to extract text. + pub ocr_pages: Vec, + /// Response `extraction_method`. Defaults to the converter name with `_` for `-`. + pub extraction_method: Option, +} + +impl ConvertedDocument { + /// Converted Markdown with no title or warnings. + pub fn markdown(markdown: impl Into) -> Self { + Self { + markdown: markdown.into(), + ..Default::default() + } + } + + fn is_usable(&self) -> bool { + !self.markdown.trim().is_empty() && self.ocr_pages.is_empty() + } +} + +/// Why a converter could not produce a document. +#[derive(Debug, Clone, PartialEq, Eq)] +pub enum ConvertError { + /// The converter does not handle this input; try the next one. + Unsupported(String), + /// Input exceeds this converter's size limit. + TooLarge { + /// Input size in bytes. + size: usize, + /// Converter limit in bytes. + limit: usize, + }, + /// The conversion failed. + Failed(String), +} + +impl fmt::Display for ConvertError { + fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { + match self { + Self::Unsupported(message) => write!(f, "unsupported: {message}"), + Self::TooLarge { size, limit } => { + write!(f, "document is {size} bytes, converter limit is {limit}") + } + Self::Failed(message) => f.write_str(message), + } + } +} + +impl std::error::Error for ConvertError {} + +/// One way to turn document bytes into Markdown. +#[async_trait] +pub trait DocumentConverter: Send + Sync { + /// Unique name used in chain config, logs and `extraction_method`. + fn name(&self) -> &str; + + /// Where the conversion runs. + fn locality(&self) -> Locality; + + /// Whether this converter handles `format`. + fn supports(&self, format: DocumentFormat) -> bool; + + /// Largest input this converter accepts, if it has its own limit. + fn max_input_bytes(&self) -> Option { + None + } + + /// Convert a complete, bounded document. + async fn convert(&self, input: DocumentInput) -> Result; +} + +/// Which formats are converted, by which backends, in what order. +/// +/// The default contains the built-in local converters enabled by Cargo features +/// (`docs-pdf` adds `PdfInspectorConverter`, `docs-office` adds `OfficeConverter`) +/// and accepts every format they support. +#[derive(Clone)] +pub struct DocumentConversion { + converters: Vec>, + formats: Option>, + chains: BTreeMap>, + max_document_bytes: usize, + allow_remote: bool, + detector: Arc, +} + +impl Default for DocumentConversion { + fn default() -> Self { + let conversion = Self::empty(); + #[cfg(feature = "docs-pdf")] + let conversion = conversion.converter(PdfInspectorConverter::new()); + #[cfg(feature = "docs-office")] + let conversion = conversion.converter(OfficeConverter::new()); + conversion + } +} + +impl fmt::Debug for DocumentConversion { + fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { + f.debug_struct("DocumentConversion") + .field( + "converters", + &self + .converters + .iter() + .map(|converter| converter.name().to_string()) + .collect::>(), + ) + .field("formats", &self.formats) + .field("chains", &self.chains) + .field("max_document_bytes", &self.max_document_bytes) + .field("allow_remote", &self.allow_remote) + .finish() + } +} + +impl DocumentConversion { + /// Configuration with no converters: every document is rejected as binary. + pub fn empty() -> Self { + Self { + converters: Vec::new(), + formats: None, + chains: BTreeMap::new(), + max_document_bytes: DEFAULT_MAX_DOCUMENT_BYTES, + allow_remote: false, + detector: Arc::new(DefaultFormatDetector), + } + } + + /// Register a converter. Without an explicit chain, converters run in + /// registration order. A converter with an existing name replaces it in place. + pub fn converter(self, converter: impl DocumentConverter + 'static) -> Self { + self.converter_arc(Arc::new(converter)) + } + + /// Register a shared converter. See [`converter`](Self::converter). + pub fn converter_arc(mut self, converter: Arc) -> Self { + match self + .converters + .iter_mut() + .find(|existing| existing.name() == converter.name()) + { + Some(existing) => *existing = converter, + None => self.converters.push(converter), + } + self + } + + /// Accept only these formats. Others are rejected as binary. + pub fn formats(mut self, formats: impl IntoIterator) -> Self { + self.formats = Some(formats.into_iter().collect()); + self + } + + /// Set the converter order for one format, by converter name. + pub fn chain(mut self, format: DocumentFormat, converters: I) -> Self + where + I: IntoIterator, + S: Into, + { + self.chains + .insert(format, converters.into_iter().map(Into::into).collect()); + self + } + + /// Maximum downloaded document size in bytes. + pub fn max_document_bytes(mut self, bytes: usize) -> Self { + self.max_document_bytes = bytes; + self + } + + /// Allow converters that send document bytes to another service. + pub fn allow_remote(mut self, allow: bool) -> Self { + self.allow_remote = allow; + self + } + + /// Replace format detection. + pub fn detector(mut self, detector: impl FormatDetector + 'static) -> Self { + self.detector = Arc::new(detector); + self + } + + /// Whether remote converters may run. + pub fn remote_allowed(&self) -> bool { + self.allow_remote + } + + /// Registered converter names in registration order. + pub fn converter_names(&self) -> Vec<&str> { + self.converters + .iter() + .map(|converter| converter.name()) + .collect() + } + + /// Converters that will run for `format`, in order, after remote filtering. + pub fn chain_for(&self, format: DocumentFormat) -> Vec> { + if self + .formats + .as_ref() + .is_some_and(|formats| !formats.contains(&format)) + { + return Vec::new(); + } + let candidates: Vec> = match self.chains.get(&format) { + Some(names) => names + .iter() + .filter_map(|name| { + let found = self + .converters + .iter() + .find(|converter| converter.name() == name); + if found.is_none() { + warn!(converter = %name, %format, "Unknown converter in document chain"); + } + found.cloned() + }) + .collect(), + None => self.converters.clone(), + }; + candidates + .into_iter() + .filter(|converter| converter.supports(format)) + .filter(|converter| self.allow_remote || converter.locality() == Locality::Local) + .collect() + } + + /// Detect a format this configuration will convert. + pub fn detect(&self, url: &Url, content_type: Option<&str>) -> Option { + self.detector + .detect(url, content_type) + .filter(|format| !self.chain_for(*format).is_empty()) + } +} + +/// [`ContentProcessor`] that converts documents through a [`DocumentConversion`] chain. +#[derive(Debug, Clone, Default)] +pub struct DocumentProcessor { + conversion: DocumentConversion, +} + +impl DocumentProcessor { + /// Create a processor for a conversion configuration. + pub fn new(conversion: DocumentConversion) -> Self { + Self { conversion } + } +} + +#[async_trait] +impl ContentProcessor for DocumentProcessor { + fn name(&self) -> &'static str { + "document" + } + + fn matches(&self, url: &Url, content_type: Option<&str>) -> bool { + self.conversion.detect(url, content_type).is_some() + } + + fn max_input_bytes(&self) -> Option { + Some(self.conversion.max_document_bytes) + } + + async fn process( + &self, + input: ContentProcessorInput, + ) -> Result { + debug_assert_eq!(input.output_format, ContentOutputFormat::Markdown); + let format = self + .conversion + .detect(&input.url, input.content_type.as_deref()) + .ok_or_else(|| ContentProcessorError("document format not enabled".to_string()))?; + if !self.conversion.detector.confirm(format, &input.body) { + return Err(ContentProcessorError(format!( + "response body is not a valid {} document", + format.as_str().to_ascii_uppercase() + ))); + } + + let document = DocumentInput { + file_name: file_name(&input.url, format), + url: input.url, + format, + content_type: input.content_type, + body: input.body, + }; + let size = document.body.len(); + let mut failures = Vec::new(); + let mut fallback: Option<(String, ConvertedDocument)> = None; + + for (index, converter) in self.conversion.chain_for(format).iter().enumerate() { + let name = converter.name().to_string(); + if let Some(limit) = converter.max_input_bytes().filter(|limit| size > *limit) { + failures.push(format!( + "{name}: {}", + ConvertError::TooLarge { size, limit } + )); + continue; + } + debug!(converter = %name, %format, size, "Converting document"); + match converter.convert(document.clone()).await { + Ok(mut converted) => { + if index > 0 { + converted.warnings.push("fallback_converter".to_string()); + } + if converted.is_usable() { + return Ok(processed(&name, converted)); + } + // Keep the first partial result (e.g. OCR-required) in case no + // later converter does better. + fallback.get_or_insert((name, converted)); + } + Err(error) => failures.push(format!("{name}: {error}")), + } + } + + match fallback { + Some((name, converted)) => Ok(processed(&name, converted)), + None if failures.is_empty() => Err(ContentProcessorError(format!( + "no converter enabled for {format}" + ))), + None => Err(ContentProcessorError(failures.join("; "))), + } + } +} + +fn file_name(url: &Url, format: DocumentFormat) -> String { + let last = url + .path_segments() + .and_then(|mut segments| segments.next_back()) + .map(|segment| { + percent_encoding::percent_decode_str(segment) + .decode_utf8_lossy() + .into_owned() + }) + .unwrap_or_default(); + let safe: String = last + .chars() + .filter(|ch| ch.is_ascii_alphanumeric() || matches!(ch, '.' | '-' | '_')) + .take(100) + .collect(); + let has_extension = safe.rsplit_once('.').is_some_and(|(stem, extension)| { + !stem.is_empty() && DocumentFormat::from_extension(extension) == Some(format) + }); + if has_extension { + safe + } else { + format!("document.{}", format.extensions()[0]) + } +} + +fn processed(converter: &str, converted: ConvertedDocument) -> ProcessedContent { + let extraction_method = converted + .extraction_method + .unwrap_or_else(|| converter.replace('-', "_")); + let needs_ocr = !converted.ocr_pages.is_empty(); + let mut warnings = converted.warnings; + if needs_ocr { + warnings.insert(0, "pdf_requires_ocr".to_string()); + } + let content = Some(converted.markdown).filter(|value| !value.trim().is_empty()); + let error = content.is_none().then(|| { + if needs_ocr { + format!( + "Document requires OCR for pages: {}", + converted + .ocr_pages + .iter() + .map(u32::to_string) + .collect::>() + .join(", ") + ) + } else { + "Document contains no extractable text".to_string() + } + }); + let degraded = needs_ocr || warnings.iter().any(|warning| warning.ends_with("_issues")); + let score = if content.is_none() { + 0.0 + } else if degraded { + 0.6 + } else if warnings + .iter() + .any(|warning| warning != "fallback_converter") + { + 0.8 + } else { + 1.0 + }; + + ProcessedContent { + format: content.as_ref().map(|_| "markdown".to_string()), + content, + metadata: Some(PageMetadata { + title: converted.title, + extraction_method: Some(extraction_method.clone()), + ..Default::default() + }), + quality: Some(PageQuality { + score, + warnings, + extraction_method: Some(extraction_method), + suggested_next_action: needs_ocr.then(|| "use_ocr".to_string()), + ..Default::default() + }), + error, + } +} + +#[cfg(test)] +mod tests { + use std::sync::atomic::{AtomicUsize, Ordering}; + + use super::*; + use crate::content::ContentFocus; + + struct Stub { + name: &'static str, + locality: Locality, + result: Result, + calls: AtomicUsize, + } + + impl Stub { + fn new(name: &'static str, result: Result) -> Self { + Self { + name, + locality: Locality::Local, + result, + calls: AtomicUsize::new(0), + } + } + + fn remote(mut self) -> Self { + self.locality = Locality::Remote; + self + } + } + + #[async_trait] + impl DocumentConverter for Stub { + fn name(&self) -> &str { + self.name + } + fn locality(&self) -> Locality { + self.locality + } + fn supports(&self, format: DocumentFormat) -> bool { + matches!(format, DocumentFormat::Pdf | DocumentFormat::Docx) + } + async fn convert(&self, input: DocumentInput) -> Result { + self.calls.fetch_add(1, Ordering::SeqCst); + assert!(input.file_name.ends_with(input.format.extensions()[0])); + self.result.clone() + } + } + + fn input(url: &str, content_type: &str, body: &'static [u8]) -> ContentProcessorInput { + ContentProcessorInput { + url: Url::parse(url).unwrap(), + content_type: Some(content_type.to_string()), + body: Bytes::from_static(body), + output_format: ContentOutputFormat::Markdown, + content_focus: ContentFocus::Full, + } + } + + const DOCX: &str = "application/vnd.openxmlformats-officedocument.wordprocessingml.document"; + + #[test] + fn detects_by_media_type_then_extension_for_generic_types() { + let detector = DefaultFormatDetector; + let url = Url::parse("https://example.com/files/Report.DOCX?x=1").unwrap(); + assert_eq!( + detector.detect(&url, Some("application/pdf; charset=binary")), + Some(DocumentFormat::Pdf) + ); + assert_eq!( + detector.detect(&url, Some("application/octet-stream")), + Some(DocumentFormat::Docx) + ); + assert_eq!(detector.detect(&url, None), Some(DocumentFormat::Docx)); + assert_eq!(detector.detect(&url, Some("text/html")), None); + assert_eq!( + detector.detect( + &Url::parse("https://example.com/a.xlsm").unwrap(), + Some("application/vnd.ms-excel.sheet.macroEnabled.12") + ), + Some(DocumentFormat::Xlsx) + ); + } + + #[test] + fn magic_bytes_confirm_container_type() { + assert!(DocumentFormat::Pdf.matches_magic(b"%PDF-1.7")); + assert!(DocumentFormat::Docx.matches_magic(b"PK\x03\x04rest")); + assert!(!DocumentFormat::Docx.matches_magic(b"%PDF-1.7")); + assert!( + DocumentFormat::Xls.matches_magic(&[0xD0, 0xCF, 0x11, 0xE0, 0xA1, 0xB1, 0x1A, 0xE1]) + ); + assert!(!DocumentFormat::Pdf.matches_magic(b"")); + } + + #[test] + fn format_parses_from_config_strings() { + assert_eq!("DOCX".parse::(), Ok(DocumentFormat::Docx)); + assert_eq!(".xlsm".parse::(), Ok(DocumentFormat::Xlsx)); + assert!("exe".parse::().is_err()); + } + + #[test] + fn remote_converters_are_skipped_unless_allowed() { + let conversion = DocumentConversion::empty() + .converter(Stub::new("remote", Ok(ConvertedDocument::markdown("x"))).remote()); + let url = Url::parse("https://example.com/a").unwrap(); + assert!(conversion.detect(&url, Some(DOCX)).is_none()); + let conversion = conversion.allow_remote(true); + assert_eq!( + conversion.detect(&url, Some(DOCX)), + Some(DocumentFormat::Docx) + ); + } + + #[test] + fn formats_filter_and_explicit_chain_order_apply() { + let conversion = DocumentConversion::empty() + .converter(Stub::new("a", Ok(ConvertedDocument::markdown("a")))) + .converter(Stub::new("b", Ok(ConvertedDocument::markdown("b")))) + .chain(DocumentFormat::Pdf, ["b", "missing", "a"]) + .formats([DocumentFormat::Pdf]); + let names: Vec = conversion + .chain_for(DocumentFormat::Pdf) + .iter() + .map(|converter| converter.name().to_string()) + .collect(); + assert_eq!(names, ["b", "a"]); + assert!(conversion.chain_for(DocumentFormat::Docx).is_empty()); + } + + #[tokio::test] + async fn falls_through_on_error_and_ocr_results() { + let ocr = ConvertedDocument { + ocr_pages: vec![1], + ..Default::default() + }; + let processor = DocumentProcessor::new( + DocumentConversion::empty() + .converter(Stub::new( + "broken", + Err(ConvertError::Failed("boom".into())), + )) + .converter(Stub::new("local-ocr", Ok(ocr))) + .converter(Stub::new("good", Ok(ConvertedDocument::markdown("# Done")))), + ); + let result = processor + .process(input( + "https://example.com/a.pdf", + "application/pdf", + b"%PDF-1.4", + )) + .await + .unwrap(); + assert_eq!(result.content.as_deref(), Some("# Done")); + let quality = result.quality.unwrap(); + assert_eq!(quality.extraction_method.as_deref(), Some("good")); + assert_eq!(quality.warnings, ["fallback_converter"]); + } + + #[tokio::test] + async fn returns_ocr_hint_when_no_converter_does_better() { + let ocr = ConvertedDocument { + ocr_pages: vec![2, 3], + ..Default::default() + }; + let processor = DocumentProcessor::new( + DocumentConversion::empty().converter(Stub::new("only", Ok(ocr))), + ); + let result = processor + .process(input( + "https://example.com/a", + "application/pdf", + b"%PDF-1.4", + )) + .await + .unwrap(); + assert!(result.content.is_none()); + assert_eq!( + result.error.as_deref(), + Some("Document requires OCR for pages: 2, 3") + ); + assert_eq!( + result.quality.unwrap().suggested_next_action.as_deref(), + Some("use_ocr") + ); + } + + #[tokio::test] + async fn rejects_body_that_does_not_match_detected_format() { + let stub = Arc::new(Stub::new("a", Ok(ConvertedDocument::markdown("x")))); + let processor = + DocumentProcessor::new(DocumentConversion::empty().converter_arc(stub.clone())); + let error = processor + .process(input( + "https://example.com/a", + DOCX, + b"not a docx", + )) + .await + .unwrap_err(); + assert!(error.to_string().contains("not a valid DOCX")); + assert_eq!(stub.calls.load(Ordering::SeqCst), 0); + } + + #[tokio::test] + async fn skips_converter_whose_limit_is_exceeded() { + struct Small; + #[async_trait] + impl DocumentConverter for Small { + fn name(&self) -> &str { + "small" + } + fn locality(&self) -> Locality { + Locality::Local + } + fn supports(&self, _format: DocumentFormat) -> bool { + true + } + fn max_input_bytes(&self) -> Option { + Some(2) + } + async fn convert( + &self, + _input: DocumentInput, + ) -> Result { + unreachable!("limit must be enforced before convert") + } + } + let processor = DocumentProcessor::new(DocumentConversion::empty().converter(Small)); + let error = processor + .process(input( + "https://example.com/a.pdf", + "application/pdf", + b"%PDF-1.4", + )) + .await + .unwrap_err(); + assert!(error.to_string().contains("converter limit is 2")); + } + + #[cfg(any(feature = "docs-pdf", feature = "docs-office"))] + #[tokio::test(flavor = "multi_thread", worker_threads = 2)] + async fn cancelled_caller_does_not_release_running_task_permit() { + use std::sync::mpsc; + + let limit = Arc::new(tokio::sync::Semaphore::new(1)); + let (started_tx, started_rx) = mpsc::channel(); + let (release_tx, release_rx) = mpsc::channel(); + let handle = tokio::spawn(run_limited(Arc::clone(&limit), move || { + started_tx.send(()).unwrap(); + release_rx.recv().unwrap(); + })); + + started_rx.recv().unwrap(); + handle.abort(); + assert_eq!(limit.available_permits(), 0); + + release_tx.send(()).unwrap(); + tokio::time::timeout(std::time::Duration::from_secs(1), async { + while limit.available_permits() == 0 { + tokio::task::yield_now().await; + } + }) + .await + .unwrap(); + } + + #[test] + fn file_name_is_sanitised_and_keeps_matching_extension() { + let url = Url::parse("https://example.com/a/Q3%20report%3B.docx").unwrap(); + assert_eq!(file_name(&url, DocumentFormat::Docx), "Q3report.docx"); + let url = Url::parse("https://example.com/download?id=7").unwrap(); + assert_eq!(file_name(&url, DocumentFormat::Xlsx), "document.xlsx"); + let url = Url::parse("https://example.com/.pdf").unwrap(); + assert_eq!(file_name(&url, DocumentFormat::Pdf), "document.pdf"); + } +} diff --git a/crates/fetchkit/src/office.rs b/crates/fetchkit/src/documents/office.rs similarity index 90% rename from crates/fetchkit/src/office.rs rename to crates/fetchkit/src/documents/office.rs index aa22f69..9cc4b8d 100644 --- a/crates/fetchkit/src/office.rs +++ b/crates/fetchkit/src/documents/office.rs @@ -24,13 +24,12 @@ use async_trait::async_trait; use calamine::{open_workbook_auto_from_rs, Data, Reader}; use quick_xml::events::{BytesStart, Event}; use quick_xml::Reader as XmlReader; -use url::Url; -use crate::content::{ - run_document_task, ContentProcessor, ContentProcessorError, ContentProcessorInput, - ProcessedContent, DOCUMENT_PROCESSING_LIMIT, +use super::{ + run_document_task, ConvertError, ConvertedDocument, DocumentConverter, DocumentFormat, + DocumentInput, Locality, }; -use crate::{PageMetadata, PageQuality}; +use crate::content::ContentProcessorError; /// Maximum number of entries in an Office zip archive. const MAX_ARCHIVE_ENTRIES: usize = 10_000; @@ -45,7 +44,7 @@ const MAX_SHEET_ROWS: usize = 1_000; /// Maximum columns rendered per spreadsheet sheet. const MAX_SHEET_COLUMNS: usize = 50; -/// Supported Office document kinds. +/// Office document kinds with distinct extraction paths. #[derive(Debug, Clone, Copy, PartialEq, Eq)] enum OfficeKind { Docx, @@ -54,54 +53,18 @@ enum OfficeKind { } impl OfficeKind { - fn from_media_type(media_type: &str) -> Option { - let media_type = media_type.to_ascii_lowercase(); - match media_type.as_str() { - "application/vnd.openxmlformats-officedocument.wordprocessingml.document" => { - Some(Self::Docx) - } - "application/vnd.openxmlformats-officedocument.presentationml.presentation" => { - Some(Self::Pptx) - } - "application/vnd.openxmlformats-officedocument.spreadsheetml.sheet" - | "application/vnd.ms-excel" - | "application/vnd.ms-excel.sheet.macroenabled.12" - | "application/vnd.oasis.opendocument.spreadsheet" => Some(Self::Spreadsheet), + fn from_format(format: DocumentFormat) -> Option { + match format { + DocumentFormat::Docx => Some(Self::Docx), + DocumentFormat::Pptx => Some(Self::Pptx), + DocumentFormat::Xlsx + | DocumentFormat::Xlsb + | DocumentFormat::Xls + | DocumentFormat::Ods => Some(Self::Spreadsheet), _ => None, } } - fn from_path(path: &str) -> Option { - let path = path.to_ascii_lowercase(); - let extension = path.rsplit_once('.')?.1; - match extension { - "docx" => Some(Self::Docx), - "pptx" => Some(Self::Pptx), - "xlsx" | "xlsm" | "xls" | "ods" => Some(Self::Spreadsheet), - _ => None, - } - } - - fn detect(url: &Url, content_type: Option<&str>) -> Option { - let media_type = content_type - .and_then(|value| value.split(';').next()) - .map(str::trim) - .unwrap_or_default(); - if let Some(kind) = Self::from_media_type(media_type) { - return Some(kind); - } - // Download endpoints often serve documents as generic binary; trust the - // extension only then, never over a specific media type. - let generic = media_type.is_empty() - || media_type.eq_ignore_ascii_case("application/octet-stream") - || media_type.eq_ignore_ascii_case("application/zip"); - if generic { - Self::from_path(url.path()) - } else { - None - } - } - fn extraction_method(self) -> &'static str { match self { Self::Docx => "office_docx", @@ -111,59 +74,45 @@ impl OfficeKind { } } -/// Extracts Word, PowerPoint and spreadsheet documents as Markdown. -pub struct OfficeProcessor; +/// Converts Word (DOCX), PowerPoint (PPTX) and spreadsheets (XLSX, XLSM, XLSB, +/// XLS, ODS) to Markdown in-process. +#[derive(Debug, Default, Clone, Copy)] +pub struct OfficeConverter; + +impl OfficeConverter { + /// Create the converter. + pub fn new() -> Self { + Self + } +} #[async_trait] -impl ContentProcessor for OfficeProcessor { - fn name(&self) -> &'static str { +impl DocumentConverter for OfficeConverter { + fn name(&self) -> &str { "office" } - fn matches(&self, url: &Url, content_type: Option<&str>) -> bool { - OfficeKind::detect(url, content_type).is_some() + fn locality(&self) -> Locality { + Locality::Local } - async fn process( - &self, - input: ContentProcessorInput, - ) -> Result { - let kind = OfficeKind::detect(&input.url, input.content_type.as_deref()) - .ok_or_else(|| ContentProcessorError("unsupported Office document".to_string()))?; - let body = input.body; - let extracted = run_document_task(DOCUMENT_PROCESSING_LIMIT.clone(), move || { - extract(kind, &body) - }) - .await??; + fn supports(&self, format: DocumentFormat) -> bool { + OfficeKind::from_format(format).is_some() + } - let content = Some(extracted.markdown).filter(|value| !value.trim().is_empty()); - let error = content - .is_none() - .then(|| "Document contains no extractable text".to_string()); - let metadata = PageMetadata { + async fn convert(&self, input: DocumentInput) -> Result { + let kind = OfficeKind::from_format(input.format) + .ok_or_else(|| ConvertError::Unsupported(input.format.to_string()))?; + let body = input.body; + let extracted = run_document_task(move || extract(kind, &body)) + .await? + .map_err(|error| ConvertError::Failed(error.0))?; + Ok(ConvertedDocument { + markdown: extracted.markdown, title: extracted.title, - extraction_method: Some(kind.extraction_method().to_string()), - ..Default::default() - }; - let quality = PageQuality { - score: if content.is_none() { - 0.0 - } else if extracted.warnings.is_empty() { - 1.0 - } else { - 0.8 - }, warnings: extracted.warnings, + ocr_pages: Vec::new(), extraction_method: Some(kind.extraction_method().to_string()), - ..Default::default() - }; - - Ok(ProcessedContent { - format: content.as_ref().map(|_| "markdown".to_string()), - content, - metadata: Some(metadata), - quality: Some(quality), - error, }) } } @@ -730,7 +679,16 @@ mod tests { use bytes::Bytes; use super::*; - use crate::content::{ContentFocus, ContentOutputFormat}; + use crate::content::{ + ContentFocus, ContentOutputFormat, ContentProcessor, ContentProcessorInput, + ProcessedContent, + }; + use crate::documents::{DocumentConversion, DocumentProcessor}; + use url::Url; + + fn processor() -> DocumentProcessor { + DocumentProcessor::new(DocumentConversion::empty().converter(OfficeConverter)) + } const DOCX: &str = "application/vnd.openxmlformats-officedocument.wordprocessingml.document"; const PPTX: &str = "application/vnd.openxmlformats-officedocument.presentationml.presentation"; @@ -751,7 +709,7 @@ mod tests { } async fn process(url: &str, content_type: &str, body: Bytes) -> ProcessedContent { - OfficeProcessor + processor() .process(ContentProcessorInput { url: Url::parse(url).unwrap(), content_type: Some(content_type.to_string()), @@ -836,7 +794,7 @@ mod tests { #[test] fn matches_media_types_and_generic_downloads_by_extension() { - let processor = OfficeProcessor; + let processor = processor(); let page = Url::parse("https://example.com/download").unwrap(); let xlsx = Url::parse("https://example.com/data.XLSX").unwrap(); assert!(processor.matches(&page, Some(DOCX))); diff --git a/crates/fetchkit/src/documents/pdf_inspector.rs b/crates/fetchkit/src/documents/pdf_inspector.rs new file mode 100644 index 0000000..2cacc54 --- /dev/null +++ b/crates/fetchkit/src/documents/pdf_inspector.rs @@ -0,0 +1,54 @@ +//! Local PDF backend using `pdf-inspector`. + +use async_trait::async_trait; +use pdf_inspector::process_pdf_mem; + +use super::{ + run_document_task, ConvertError, ConvertedDocument, DocumentConverter, DocumentFormat, + DocumentInput, Locality, +}; + +/// Extracts Markdown from text-based PDFs in-process. Does not perform OCR: +/// image-only pages are reported through [`ConvertedDocument::ocr_pages`]. +#[derive(Debug, Default, Clone, Copy)] +pub struct PdfInspectorConverter; + +impl PdfInspectorConverter { + /// Create the converter. + pub fn new() -> Self { + Self + } +} + +#[async_trait] +impl DocumentConverter for PdfInspectorConverter { + fn name(&self) -> &str { + "pdf-inspector" + } + + fn locality(&self) -> Locality { + Locality::Local + } + + fn supports(&self, format: DocumentFormat) -> bool { + format == DocumentFormat::Pdf + } + + async fn convert(&self, input: DocumentInput) -> Result { + let result = run_document_task(move || process_pdf_mem(&input.body)) + .await? + .map_err(|error| ConvertError::Failed(error.to_string()))?; + + let mut warnings = Vec::new(); + if result.has_encoding_issues { + warnings.push("pdf_encoding_issues".to_string()); + } + Ok(ConvertedDocument { + markdown: result.markdown.unwrap_or_default(), + title: result.title, + warnings, + ocr_pages: result.pages_needing_ocr, + extraction_method: None, + }) + } +} diff --git a/crates/fetchkit/src/documents/remote.rs b/crates/fetchkit/src/documents/remote.rs new file mode 100644 index 0000000..2ab00ef --- /dev/null +++ b/crates/fetchkit/src/documents/remote.rs @@ -0,0 +1,268 @@ +// Decisions: +// - Remote converters do not use `HttpTransport`: that trait is deliberately a +// single-hop GET/HEAD adapter for user URLs. Uploads go through `ConverterClient`, +// so a host can route them through its own egress boundary. +// - The converter endpoint is operator configuration, not user input, so it is not +// subject to request allow/block lists. It is never derived from the fetched URL. +// - Multipart bodies are encoded here so any `ConverterClient` only has to send bytes. +//! Shared plumbing for remote document converters. + +use std::fmt; +use std::sync::atomic::{AtomicU64, Ordering}; +use std::time::{Duration, SystemTime, UNIX_EPOCH}; + +use async_trait::async_trait; +use bytes::{Bytes, BytesMut}; +use url::Url; + +use super::ConvertError; + +/// Default timeout for one remote conversion. +pub(crate) const DEFAULT_REMOTE_TIMEOUT: Duration = Duration::from_secs(60); + +/// Default cap on a remote converter's response body (32 MB). +pub(crate) const MAX_REMOTE_RESPONSE_BYTES: usize = 32 * 1024 * 1024; + +/// A credential that never appears in `Debug` or `Display` output. +#[derive(Clone)] +pub struct SecretString(String); + +impl SecretString { + /// Wrap a secret value. + pub fn new(value: impl Into) -> Self { + Self(value.into()) + } + + /// Reveal the secret for use in a request header. + pub fn expose(&self) -> &str { + &self.0 + } +} + +impl fmt::Debug for SecretString { + fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { + f.write_str("SecretString()") + } +} + +impl From for SecretString { + fn from(value: String) -> Self { + Self(value) + } +} + +impl From<&str> for SecretString { + fn from(value: &str) -> Self { + Self(value.to_string()) + } +} + +/// One POST to a remote converter. +pub struct ConverterRequest { + /// Converter endpoint. + pub url: Url, + /// Request headers, including `Content-Type` and any credentials. + pub headers: Vec<(String, String)>, + /// Encoded request body. + pub body: Bytes, + /// Overall request timeout. + pub timeout: Duration, + /// Maximum response body size the client should read. + pub max_response_bytes: usize, +} + +impl fmt::Debug for ConverterRequest { + fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { + f.debug_struct("ConverterRequest") + .field("url", &self.url.as_str()) + .field( + "headers", + &self + .headers + .iter() + .map(|(name, _)| name.as_str()) + .collect::>(), + ) + .field("body_len", &self.body.len()) + .field("timeout", &self.timeout) + .finish() + } +} + +/// Response from a remote converter. +#[derive(Debug, Clone)] +pub struct ConverterResponse { + /// HTTP status code. + pub status: u16, + /// Response Content-Type, when present. + pub content_type: Option, + /// Response body, at most `max_response_bytes`. + pub body: Bytes, +} + +/// Sends converter uploads. Implement to route them through a host egress boundary. +#[async_trait] +pub trait ConverterClient: Send + Sync { + /// Send one POST request and return the bounded response. + async fn post(&self, request: ConverterRequest) -> Result; +} + +/// Default [`ConverterClient`] using reqwest. Redirects are not followed and +/// ambient proxy variables are ignored unless enabled. +#[derive(Debug, Clone, Default)] +pub struct ReqwestConverterClient { + respect_proxy_env: bool, +} + +impl ReqwestConverterClient { + /// Create a client that ignores `HTTP(S)_PROXY`. + pub fn new() -> Self { + Self::default() + } + + /// Honour `HTTP_PROXY`/`HTTPS_PROXY`/`NO_PROXY` for converter uploads. + pub fn respect_proxy_env(mut self, respect: bool) -> Self { + self.respect_proxy_env = respect; + self + } +} + +#[async_trait] +impl ConverterClient for ReqwestConverterClient { + async fn post(&self, request: ConverterRequest) -> Result { + let mut builder = reqwest::Client::builder() + .redirect(reqwest::redirect::Policy::none()) + .timeout(request.timeout); + if !self.respect_proxy_env { + builder = builder.no_proxy(); + } + let client = builder + .build() + .map_err(|error| ConvertError::Failed(format!("converter client: {error}")))?; + let mut post = client.post(request.url).body(request.body); + for (name, value) in &request.headers { + post = post.header(name, value); + } + let mut response = post + .send() + .await + .map_err(|error| ConvertError::Failed(format!("converter request failed: {error}")))?; + let status = response.status().as_u16(); + let content_type = response + .headers() + .get(reqwest::header::CONTENT_TYPE) + .and_then(|value| value.to_str().ok()) + .map(str::to_string); + let mut body = BytesMut::new(); + while let Some(chunk) = response + .chunk() + .await + .map_err(|error| ConvertError::Failed(format!("converter response: {error}")))? + { + if body.len() + chunk.len() > request.max_response_bytes { + return Err(ConvertError::Failed(format!( + "converter response exceeds {} bytes", + request.max_response_bytes + ))); + } + body.extend_from_slice(&chunk); + } + Ok(ConverterResponse { + status, + content_type, + body: body.freeze(), + }) + } +} + +/// Encode a single-file `multipart/form-data` body. Returns `(content_type, body)`. +pub(crate) fn multipart_file( + field: &str, + file_name: &str, + media_type: &str, + data: &[u8], +) -> (String, Bytes) { + static COUNTER: AtomicU64 = AtomicU64::new(0); + let nanos = SystemTime::now() + .duration_since(UNIX_EPOCH) + .map(|elapsed| elapsed.as_nanos()) + .unwrap_or_default(); + let mut boundary = format!( + "fetchkit-{nanos:x}-{:x}", + COUNTER.fetch_add(1, Ordering::Relaxed) + ); + // A boundary must not occur in the payload. + while data + .windows(boundary.len()) + .any(|window| window == boundary.as_bytes()) + { + boundary.push('x'); + } + + let quote = |value: &str| value.replace(['"', '\r', '\n'], "_"); + let mut body = BytesMut::with_capacity(data.len() + 256); + body.extend_from_slice( + format!( + "--{boundary}\r\nContent-Disposition: form-data; name=\"{}\"; filename=\"{}\"\r\nContent-Type: {media_type}\r\n\r\n", + quote(field), + quote(file_name) + ) + .as_bytes(), + ); + body.extend_from_slice(data); + body.extend_from_slice(format!("\r\n--{boundary}--\r\n").as_bytes()); + ( + format!("multipart/form-data; boundary={boundary}"), + body.freeze(), + ) +} + +/// Map an HTTP status to a converter error, or `None` on success. +pub(crate) fn status_error(name: &str, status: u16, body: &[u8]) -> Option { + if (200..300).contains(&status) { + return None; + } + let detail: String = String::from_utf8_lossy(body).chars().take(300).collect(); + Some(match status { + 401 | 403 => ConvertError::Failed(format!("{name}: authentication failed ({status})")), + 413 => ConvertError::Unsupported(format!("{name}: document too large ({status})")), + 415 => ConvertError::Unsupported(format!("{name}: format not accepted ({status})")), + _ => ConvertError::Failed(format!("{name}: HTTP {status}: {}", detail.trim())), + }) +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn secret_is_redacted() { + let secret = SecretString::new("token-123"); + assert_eq!(format!("{secret:?}"), "SecretString()"); + assert_eq!(secret.expose(), "token-123"); + } + + #[test] + fn multipart_wraps_file_with_unique_boundary() { + let (content_type, body) = multipart_file("files", "a\"b.pdf", "application/pdf", b"%PDF"); + let boundary = content_type.split("boundary=").nth(1).unwrap(); + let text = String::from_utf8(body.to_vec()).unwrap(); + assert!(text.starts_with(&format!("--{boundary}\r\n"))); + assert!(text.contains("name=\"files\"; filename=\"a_b.pdf\"")); + assert!(text.contains("Content-Type: application/pdf\r\n\r\n%PDF\r\n")); + assert!(text.ends_with(&format!("--{boundary}--\r\n"))); + } + + #[test] + fn status_errors_are_classified() { + assert!(status_error("x", 200, b"").is_none()); + assert!(matches!( + status_error("x", 401, b""), + Some(ConvertError::Failed(message)) if message.contains("authentication") + )); + assert!(matches!( + status_error("x", 413, b""), + Some(ConvertError::Unsupported(_)) + )); + } +} diff --git a/crates/fetchkit/src/documents/settings.rs b/crates/fetchkit/src/documents/settings.rs new file mode 100644 index 0000000..d2a5e76 --- /dev/null +++ b/crates/fetchkit/src/documents/settings.rs @@ -0,0 +1,203 @@ +//! String-based document conversion settings shared by the CLI, MCP server and +//! Python bindings. Libraries can use [`DocumentConversion`] directly instead. + +use super::{DocumentConversion, DocumentFormat}; + +/// Built-in converter names accepted by [`DocumentConversionSettings`]. +pub const BUILTIN_CONVERTERS: &[&str] = &["pdf-inspector", "office", "cloudflare"]; + +/// Plain settings resolved into a [`DocumentConversion`]. +/// +/// Credentials are passed explicitly; nothing is read from the environment. +#[derive(Debug, Default, Clone)] +pub struct DocumentConversionSettings { + /// Accepted formats (e.g. `["pdf", "docx"]`). `None` keeps every supported format. + pub formats: Option>, + /// Converter order applied to every format (e.g. `["office", "cloudflare"]`). + /// `None` keeps the built-in local converters in their default order. + pub converters: Option>, + /// Allow converters that send documents to another service. + pub allow_remote: bool, + /// Maximum downloaded document size in bytes. + pub max_document_bytes: Option, + /// Cloudflare account id, required when `converters` names `cloudflare`. + pub cloudflare_account_id: Option, + /// Cloudflare API token with Workers AI permission. + pub cloudflare_api_token: Option, + /// Honour proxy environment variables for remote converter uploads. + pub respect_proxy_env: bool, +} + +impl DocumentConversionSettings { + /// Whether every field is at its default, i.e. nothing to configure. + pub fn is_default(&self) -> bool { + self.formats.is_none() + && self.converters.is_none() + && !self.allow_remote + && self.max_document_bytes.is_none() + } + + /// Resolve the settings. Errors name the offending value. + pub fn build(&self) -> Result { + let mut conversion = DocumentConversion::default().allow_remote(self.allow_remote); + + if let Some(formats) = &self.formats { + let formats = formats + .iter() + .map(|value| value.parse::()) + .collect::, _>>()?; + conversion = conversion.formats(formats); + } + if let Some(bytes) = self.max_document_bytes { + conversion = conversion.max_document_bytes(bytes); + } + + if let Some(names) = &self.converters { + let names: Vec = names + .iter() + .map(|name| name.trim().to_ascii_lowercase()) + .filter(|name| !name.is_empty()) + .collect(); + if let Some(unknown) = names + .iter() + .find(|name| !BUILTIN_CONVERTERS.contains(&name.as_str())) + { + return Err(format!( + "unknown converter `{unknown}` (known: {})", + BUILTIN_CONVERTERS.join(", ") + )); + } + for name in &names { + conversion = self.add_builtin(conversion, name)?; + } + for format in DocumentFormat::ALL { + conversion = conversion.chain(*format, names.iter().cloned()); + } + } + Ok(conversion) + } + + fn add_builtin( + &self, + conversion: DocumentConversion, + name: &str, + ) -> Result { + match name { + "pdf-inspector" => { + if cfg!(feature = "docs-pdf") { + Ok(conversion) + } else { + Err("converter `pdf-inspector` requires the docs-pdf feature".to_string()) + } + } + "office" => { + if cfg!(feature = "docs-office") { + Ok(conversion) + } else { + Err("converter `office` requires the docs-office feature".to_string()) + } + } + "cloudflare" => self.cloudflare(conversion), + _ => unreachable!("validated against BUILTIN_CONVERTERS"), + } + } + + #[cfg(feature = "docs-cloudflare")] + fn cloudflare(&self, conversion: DocumentConversion) -> Result { + use std::sync::Arc; + + if !self.allow_remote { + return Err( + "converter `cloudflare` uploads documents to Cloudflare; allow remote conversion to use it" + .to_string(), + ); + } + let (Some(account_id), Some(api_token)) = ( + self.cloudflare_account_id + .as_deref() + .filter(|value| !value.trim().is_empty()), + self.cloudflare_api_token + .as_deref() + .filter(|value| !value.trim().is_empty()), + ) else { + return Err( + "converter `cloudflare` needs a Cloudflare account id and API token".to_string(), + ); + }; + let client = super::ReqwestConverterClient::new().respect_proxy_env(self.respect_proxy_env); + Ok(conversion.converter( + super::CloudflareConverter::new(account_id.trim(), api_token.trim()) + .client(Arc::new(client)), + )) + } + + #[cfg(not(feature = "docs-cloudflare"))] + fn cloudflare(&self, _conversion: DocumentConversion) -> Result { + Err("converter `cloudflare` requires the docs-cloudflare feature".to_string()) + } +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn rejects_unknown_names_and_formats() { + let settings = DocumentConversionSettings { + converters: Some(vec!["magic".into()]), + ..Default::default() + }; + assert!(settings + .build() + .unwrap_err() + .contains("unknown converter `magic`")); + let settings = DocumentConversionSettings { + formats: Some(vec!["exe".into()]), + ..Default::default() + }; + assert!(settings.build().unwrap_err().contains("exe")); + } + + #[test] + fn formats_and_limits_apply() { + let settings = DocumentConversionSettings { + formats: Some(vec!["docx".into()]), + max_document_bytes: Some(10), + ..Default::default() + }; + assert!(!settings.is_default()); + let conversion = settings.build().unwrap(); + assert!(conversion.chain_for(DocumentFormat::Pdf).is_empty()); + } + + #[cfg(feature = "docs-cloudflare")] + #[test] + fn cloudflare_requires_remote_opt_in_and_credentials() { + let mut settings = DocumentConversionSettings { + converters: Some(vec!["cloudflare".into()]), + ..Default::default() + }; + assert!(settings.build().unwrap_err().contains("allow remote")); + settings.allow_remote = true; + assert!(settings.build().unwrap_err().contains("account id")); + settings.cloudflare_account_id = Some("acct".into()); + settings.cloudflare_api_token = Some("tok".into()); + let conversion = settings.build().unwrap(); + assert_eq!(conversion.chain_for(DocumentFormat::Docx).len(), 1); + assert!(conversion.chain_for(DocumentFormat::Pptx).is_empty()); + } + + #[cfg(not(feature = "docs-cloudflare"))] + #[test] + fn cloudflare_without_feature_is_explained() { + let settings = DocumentConversionSettings { + converters: Some(vec!["cloudflare".into()]), + allow_remote: true, + ..Default::default() + }; + assert!(settings + .build() + .unwrap_err() + .contains("docs-cloudflare feature")); + } +} diff --git a/crates/fetchkit/src/fetchers/default.rs b/crates/fetchkit/src/fetchers/default.rs index 35db5b4..c4aa466 100644 --- a/crates/fetchkit/src/fetchers/default.rs +++ b/crates/fetchkit/src/fetchers/default.rs @@ -16,6 +16,7 @@ use crate::convert::{ classify_agent_resource, filter_excessive_newlines, is_html, is_markdown_content_type, is_plain_text_content_type, }; +use crate::documents::{DefaultFormatDetector, FormatDetector}; use crate::error::FetchError; use crate::fetchers::Fetcher; use crate::file_saver::FileSaver; @@ -86,6 +87,8 @@ const BINARY_PREFIXES: &[&str] = &[ "application/vnd.ms-", "application/vnd.openxmlformats", "application/vnd.oasis.opendocument", + "application/msword", + "application/epub+zip", "font/", ]; @@ -544,6 +547,7 @@ impl Fetcher for DefaultFetcher { // Check for unsupported binary content if let Some(ref ct) = meta.content_type { if metadata_processor.is_none() && is_binary_content_type(ct) { + let error = unsupported_binary_message(&parsed_final_url, ct); return Ok(FetchResponse { url: final_url, status_code, @@ -553,10 +557,7 @@ impl Fetcher for DefaultFetcher { etag: meta.etag, filename: meta.filename, redirect_chain, - error: Some( - "Binary content is not supported. Only textual content (HTML, text, JSON, etc.) can be fetched." - .to_string(), - ), + error: Some(error), quality: Some(binary_quality_signal()), ..Default::default() }); @@ -565,8 +566,13 @@ impl Fetcher for DefaultFetcher { // THREAT[TM-DOS-001]: Read body with timeout and size limit // THREAT[TM-DOS-003]: Size limit also protects against compressed content bombs + // Document processors may raise the download cap (e.g. 25 MB documents); + // conversion output is still capped by max_body_size below. + let input_limit = metadata_processor + .and_then(|processor| processor.max_input_bytes()) + .unwrap_or(max_body_size); let (body, input_truncated) = - read_body_with_timeout(response, BODY_TIMEOUT, max_body_size).await?; + read_body_with_timeout(response, BODY_TIMEOUT, input_limit).await?; let size = body.len() as u64; // Convert to string @@ -1198,6 +1204,21 @@ impl Drop for DenyProxy { } } +const UNSUPPORTED_BINARY_MESSAGE: &str = + "Binary content is not supported. Only textual content (HTML, text, JSON, etc.) can be fetched."; + +/// Rejection message, naming the document format when one was recognised so +/// operators know conversion can be enabled. +fn unsupported_binary_message(url: &Url, content_type: &str) -> String { + match DefaultFormatDetector.detect(url, Some(content_type)) { + Some(format) => format!( + "{UNSUPPORTED_BINARY_MESSAGE} {} conversion is not enabled in this fetchkit configuration; enable a document converter to convert it.", + format.as_str().to_ascii_uppercase() + ), + None => UNSUPPORTED_BINARY_MESSAGE.to_string(), + } +} + /// Check if content type indicates binary content fn is_binary_content_type(content_type: &str) -> bool { let ct_lower = content_type.to_lowercase(); diff --git a/crates/fetchkit/src/fetchers/mod.rs b/crates/fetchkit/src/fetchers/mod.rs index 76cabf0..347437d 100644 --- a/crates/fetchkit/src/fetchers/mod.rs +++ b/crates/fetchkit/src/fetchers/mod.rs @@ -163,6 +163,13 @@ impl FetcherRegistry { Self::with_content_processors(ContentProcessorRegistry::with_defaults()) } + /// Create the default fetchers with content processors built from `options` + /// (custom processors, then [`DocumentConversion`](crate::DocumentConversion), + /// then HTML). + pub fn for_options(options: &FetchOptions) -> Self { + Self::with_content_processors(ContentProcessorRegistry::for_options(options)) + } + /// Create the default fetcher registry with a custom content processor registry. pub fn with_content_processors(content_processors: ContentProcessorRegistry) -> Self { let mut registry = Self::new(); diff --git a/crates/fetchkit/src/lib.rs b/crates/fetchkit/src/lib.rs index e33288a..d6389a1 100644 --- a/crates/fetchkit/src/lib.rs +++ b/crates/fetchkit/src/lib.rs @@ -82,10 +82,10 @@ pub mod content; mod convert; mod crawl; mod dns; +pub mod documents; mod error; pub mod fetchers; pub mod file_saver; -mod office; mod tool; pub mod transport; mod types; @@ -93,7 +93,7 @@ mod types; pub use client::{batch_fetch, batch_fetch_with_options, fetch, fetch_with_options, FetchOptions}; pub use content::{ ContentFocus, ContentOutputFormat, ContentProcessor, ContentProcessorError, - ContentProcessorInput, ContentProcessorRegistry, HtmlProcessor, PdfProcessor, ProcessedContent, + ContentProcessorInput, ContentProcessorRegistry, HtmlProcessor, ProcessedContent, }; pub use convert::{ extract_headings, extract_metadata, extract_readable_content, html_to_markdown, @@ -101,6 +101,11 @@ pub use convert::{ BuiltinHtmlToMarkdownConverter, HtmlToMarkdownConverter, }; pub use dns::DnsPolicy; +pub use documents::{ + ConvertError, ConvertedDocument, DefaultFormatDetector, DocumentConversion, + DocumentConversionSettings, DocumentConverter, DocumentFormat, DocumentInput, + DocumentProcessor, FormatDetector, Locality, +}; pub use error::{FetchError, ToolError}; pub use fetchers::{ ArXivFetcher, DefaultFetcher, DocsSiteFetcher, Fetcher, FetcherRegistry, GitHubCodeFetcher, @@ -108,7 +113,6 @@ pub use fetchers::{ RSSFeedFetcher, StackOverflowFetcher, TwitterFetcher, WikipediaFetcher, YouTubeFetcher, }; pub use file_saver::{FileSaveError, FileSaver, LocalFileSaver, SaveResult}; -pub use office::OfficeProcessor; pub use tool::{ Tool, ToolBuilder, ToolExecution, ToolImage, ToolOutput, ToolOutputMetadata, ToolService, ToolStatus, diff --git a/crates/fetchkit/src/tool.rs b/crates/fetchkit/src/tool.rs index c073a0e..2f02123 100644 --- a/crates/fetchkit/src/tool.rs +++ b/crates/fetchkit/src/tool.rs @@ -6,7 +6,9 @@ #[cfg(feature = "bot-auth")] use crate::bot_auth::BotAuthConfig; use crate::client::{fetch_with_options, FetchOptions}; +use crate::content::ContentProcessor; use crate::dns::DnsPolicy; +use crate::documents::DocumentConversion; use crate::error::{FetchError, ToolError}; use crate::fetchers::FetcherRegistry; use crate::file_saver::FileSaver; @@ -151,6 +153,12 @@ pub struct ToolBuilder { bot_auth: Option, /// Pluggable HTTP transport. None => default ReqwestTransport. transport: Option>, + /// Document conversion config. None => DocumentConversion::default(). + document_conversion: Option, + /// Custom content processors, consulted before built-in ones. + content_processors: Vec>, + /// Set by `hardened()`: remote document converters never run. + deny_remote_conversion: bool, } // Manual Debug: `transport` is a trait object that does not implement Debug @@ -181,6 +189,15 @@ impl std::fmt::Debug for ToolBuilder { .as_ref() .map(|_| "") .unwrap_or("None"), + ) + .field("document_conversion", &self.document_conversion) + .field( + "content_processors", + &self + .content_processors + .iter() + .map(|processor| processor.name()) + .collect::>(), ); d.finish() } @@ -357,12 +374,41 @@ impl ToolBuilder { self } + /// Configure document (PDF, Office) conversion: formats, converter backends, + /// per-format chain order, size cap, and whether remote converters may run. + /// + /// ``` + /// use fetchkit::{DocumentConversion, DocumentFormat, ToolBuilder}; + /// + /// let tool = ToolBuilder::new() + /// .document_conversion( + /// DocumentConversion::default() + /// .formats([DocumentFormat::Pdf]) + /// .max_document_bytes(5 * 1024 * 1024), + /// ) + /// .build(); + /// # let _ = tool; + /// ``` + pub fn document_conversion(mut self, conversion: DocumentConversion) -> Self { + self.document_conversion = Some(conversion); + self + } + + /// Add a custom content processor. Custom processors are consulted before the + /// built-in document and HTML processors, in registration order. + pub fn content_processor(mut self, processor: Arc) -> Self { + self.content_processors.push(processor); + self + } + /// Apply a production-oriented hardening profile. /// /// This preset keeps private IP blocking enabled, ignores ambient proxy /// environment variables, restricts outbound traffic to ports 80 and 443, - /// blocks common internal DNS suffixes, and only follows same-host redirects. + /// blocks common internal DNS suffixes, only follows same-host redirects, and + /// never runs remote document converters. pub fn hardened(mut self) -> Self { + self.deny_remote_conversion = true; self = self .block_private_ips(true) .use_env_proxy(false) @@ -400,6 +446,17 @@ impl ToolBuilder { #[cfg(feature = "bot-auth")] bot_auth: self.bot_auth.clone(), transport: self.transport.clone(), + document_conversion: self.resolved_document_conversion(), + content_processors: self.content_processors.clone(), + } + } + + fn resolved_document_conversion(&self) -> Option { + let conversion = self.document_conversion.clone(); + if self.deny_remote_conversion { + Some(conversion.unwrap_or_default().allow_remote(false)) + } else { + conversion } } @@ -468,6 +525,8 @@ pub struct Tool { #[cfg(feature = "bot-auth")] bot_auth: Option, transport: Option>, + document_conversion: Option, + content_processors: Vec>, } // Manual Debug: `transport` is a trait object that does not implement Debug @@ -501,6 +560,15 @@ impl std::fmt::Debug for Tool { .as_ref() .map(|_| "") .unwrap_or("None"), + ) + .field("document_conversion", &self.document_conversion) + .field( + "content_processors", + &self + .content_processors + .iter() + .map(|processor| processor.name()) + .collect::>(), ); d.finish() } @@ -649,7 +717,7 @@ impl Tool { let saver = saver.ok_or(FetchError::SaverNotAvailable)?; let options = self.build_options(); - let registry = FetcherRegistry::with_defaults(); + let registry = FetcherRegistry::for_options(&options); registry.fetch_to_file(req, options, saver).await } else { self.execute(req).await @@ -677,6 +745,8 @@ impl Tool { // None => default ReqwestTransport; hosts inject a custom transport via // ToolBuilder::transport and every Tool execution path honors it. transport: self.transport.clone(), + document_conversion: self.document_conversion.clone(), + content_processors: self.content_processors.clone(), } } } diff --git a/crates/fetchkit/tests/documents.rs b/crates/fetchkit/tests/documents.rs new file mode 100644 index 0000000..536808c --- /dev/null +++ b/crates/fetchkit/tests/documents.rs @@ -0,0 +1,277 @@ +//! End-to-end document conversion through the Tool surface (wiremock, no network). + +use std::sync::Arc; + +use async_trait::async_trait; +use fetchkit::{ + ContentProcessor, ContentProcessorError, ContentProcessorInput, ConvertError, + ConvertedDocument, DocumentConversion, DocumentConverter, DocumentFormat, DocumentInput, + FetchRequest, Locality, ProcessedContent, Tool, ToolBuilder, +}; +use url::Url; +use wiremock::matchers::{method, path}; +use wiremock::{Mock, MockServer, ResponseTemplate}; + +const DOCX: &str = "application/vnd.openxmlformats-officedocument.wordprocessingml.document"; + +struct EchoDocx { + name: &'static str, + locality: Locality, +} + +#[async_trait] +impl DocumentConverter for EchoDocx { + fn name(&self) -> &str { + self.name + } + fn locality(&self) -> Locality { + self.locality + } + fn supports(&self, format: DocumentFormat) -> bool { + format == DocumentFormat::Docx + } + async fn convert(&self, input: DocumentInput) -> Result { + Ok(ConvertedDocument { + markdown: format!("# {} ({} bytes)", input.file_name, input.body.len()), + title: Some("Quarterly".to_string()), + ..Default::default() + }) + } +} + +fn local() -> EchoDocx { + EchoDocx { + name: "echo", + locality: Locality::Local, + } +} + +fn docx_body(len: usize) -> Vec { + let mut body = b"PK\x03\x04".to_vec(); + body.resize(len, b'x'); + body +} + +async fn serve_docx(len: usize) -> MockServer { + let server = MockServer::start().await; + Mock::given(method("GET")) + .and(path("/files/q3.docx")) + .respond_with( + ResponseTemplate::new(200) + .insert_header("content-type", DOCX) + .set_body_bytes(docx_body(len)), + ) + .mount(&server) + .await; + server +} + +fn builder() -> ToolBuilder { + Tool::builder().block_private_ips(false) +} + +fn request(server: &MockServer) -> FetchRequest { + FetchRequest::new(format!("{}/files/q3.docx", server.uri())).as_markdown() +} + +#[tokio::test] +async fn unconverted_format_is_rejected_with_actionable_message() { + let server = MockServer::start().await; + Mock::given(method("GET")) + .and(path("/files/old.doc")) + .respond_with( + ResponseTemplate::new(200) + .insert_header("content-type", "application/msword") + .set_body_bytes(vec![0xD0, 0xCF, 0x11, 0xE0, 0xA1, 0xB1, 0x1A, 0xE1]), + ) + .mount(&server) + .await; + let response = builder() + .build() + .execute(FetchRequest::new(format!("{}/files/old.doc", server.uri())).as_markdown()) + .await + .unwrap(); + + assert!(response.content.is_none()); + let error = response.error.unwrap(); + assert!(error.contains("DOC conversion is not enabled"), "{error}"); +} + +#[tokio::test] +async fn empty_conversion_disables_built_in_converters() { + let server = serve_docx(64).await; + let response = builder() + .document_conversion(DocumentConversion::empty()) + .build() + .execute(request(&server)) + .await + .unwrap(); + assert!(response + .error + .unwrap() + .contains("DOCX conversion is not enabled")); +} + +#[tokio::test] +async fn custom_converter_converts_docx_through_tool() { + let server = serve_docx(64).await; + let tool = builder() + .document_conversion(DocumentConversion::empty().converter(local())) + .build(); + let response = tool.execute(request(&server)).await.unwrap(); + + assert_eq!(response.format.as_deref(), Some("markdown")); + assert_eq!(response.content.as_deref(), Some("# q3.docx (64 bytes)")); + let metadata = response.metadata.unwrap(); + assert_eq!(metadata.title.as_deref(), Some("Quarterly")); + assert_eq!(metadata.extraction_method.as_deref(), Some("echo")); +} + +#[tokio::test] +async fn remote_converter_needs_explicit_opt_in_and_hardened_denies_it() { + let server = serve_docx(64).await; + let remote = || { + DocumentConversion::empty().converter(EchoDocx { + name: "remote-echo", + locality: Locality::Remote, + }) + }; + + let not_allowed = builder().document_conversion(remote()).build(); + let response = not_allowed.execute(request(&server)).await.unwrap(); + assert!(response.content.is_none()); + + let allowed = builder() + .document_conversion(remote().allow_remote(true)) + .build(); + let response = allowed.execute(request(&server)).await.unwrap(); + assert!(response.content.unwrap().starts_with("# q3.docx")); + + // hardened() wins even when remote conversion was allowed; loopback is + // re-enabled so only the conversion policy is under test. + let hardened = Tool::builder() + .document_conversion(remote().allow_remote(true)) + .hardened() + .block_private_ips(false) + .build(); + assert!(format!("{hardened:?}").contains("allow_remote: false")); +} + +#[tokio::test] +async fn document_cap_is_separate_from_body_cap() { + let server = serve_docx(4096).await; + let small_body_cap = || builder().max_body_size(1024); + + let converted = small_body_cap() + .document_conversion(DocumentConversion::empty().converter(local())) + .build() + .execute(request(&server)) + .await + .unwrap(); + assert_eq!(converted.content.as_deref(), Some("# q3.docx (4096 bytes)")); + + let too_large = small_body_cap() + .document_conversion( + DocumentConversion::empty() + .converter(local()) + .max_document_bytes(2048), + ) + .build() + .execute(request(&server)) + .await + .unwrap(); + assert!(too_large.content.is_none()); + assert!(too_large + .error + .unwrap() + .contains("cannot process a partial response body")); +} + +#[tokio::test] +async fn format_allowlist_limits_conversion() { + let server = serve_docx(64).await; + let tool = builder() + .document_conversion( + DocumentConversion::empty() + .converter(local()) + .formats([DocumentFormat::Pdf]), + ) + .build(); + let response = tool.execute(request(&server)).await.unwrap(); + assert!(response.content.is_none()); +} + +struct Shout; + +#[async_trait] +impl ContentProcessor for Shout { + fn name(&self) -> &'static str { + "shout" + } + fn matches(&self, _url: &Url, content_type: Option<&str>) -> bool { + content_type.is_some_and(|value| value.starts_with(DOCX)) + } + async fn process( + &self, + input: ContentProcessorInput, + ) -> Result { + Ok(ProcessedContent { + format: Some("markdown".to_string()), + content: Some(format!("CUSTOM {}", input.body.len())), + ..Default::default() + }) + } +} + +#[tokio::test] +async fn custom_content_processor_runs_before_document_chain() { + let server = serve_docx(10).await; + let tool = builder() + .document_conversion(DocumentConversion::empty().converter(local())) + .content_processor(Arc::new(Shout)) + .build(); + let response = tool.execute(request(&server)).await.unwrap(); + assert_eq!(response.content.as_deref(), Some("CUSTOM 10")); +} + +#[cfg(feature = "docs-cloudflare")] +#[tokio::test] +async fn cloudflare_backend_converts_downloaded_docx() { + use fetchkit::documents::CloudflareConverter; + use wiremock::matchers::header; + + let origin = serve_docx(32).await; + let cloudflare = MockServer::start().await; + Mock::given(method("POST")) + .and(path("/accounts/acct/ai/tomarkdown")) + .and(header("authorization", "Bearer tok")) + .respond_with(ResponseTemplate::new(200).set_body_json(serde_json::json!({ + "success": true, + "result": [{"name": "q3.docx", "format": "markdown", "data": "# From Cloudflare"}] + }))) + .expect(1) + .mount(&cloudflare) + .await; + + let converter = + CloudflareConverter::new("acct", "tok").api_base(Url::parse(&cloudflare.uri()).unwrap()); + let tool = builder() + .document_conversion( + DocumentConversion::empty() + .converter(converter) + .allow_remote(true), + ) + .build(); + let response = tool.execute(request(&origin)).await.unwrap(); + + assert_eq!(response.content.as_deref(), Some("# From Cloudflare")); + assert_eq!( + response.metadata.unwrap().extraction_method.as_deref(), + Some("cloudflare_tomarkdown") + ); + let uploads = cloudflare.received_requests().await.unwrap(); + assert!(uploads[0] + .body + .windows(4) + .any(|window| window == b"PK\x03\x04")); +} diff --git a/crates/fetchkit/tests/integration.rs b/crates/fetchkit/tests/integration.rs index 31ab0a8..80d9edb 100644 --- a/crates/fetchkit/tests/integration.rs +++ b/crates/fetchkit/tests/integration.rs @@ -465,6 +465,7 @@ async fn test_pdf_content_processor_returns_markdown() { assert!(resp.error.is_none()); } +#[cfg(feature = "docs-office")] fn docx_with_paragraph(text: &str) -> Vec { use std::io::Write; @@ -484,6 +485,7 @@ fn docx_with_paragraph(text: &str) -> Vec { buffer.into_inner() } +#[cfg(feature = "docs-office")] #[tokio::test] async fn test_docx_content_processor_returns_markdown() { let mock_server = MockServer::start().await; diff --git a/docs/hardening.md b/docs/hardening.md index 7c88c89..350602a 100644 --- a/docs/hardening.md +++ b/docs/hardening.md @@ -95,3 +95,21 @@ For Internet-facing fetching from untrusted input: - only opt in to `respect_proxy_env(true)` when the proxy is deliberate and hardened If you must fetch nonstandard ports or internal-looking public domains, start from `.hardened()` and then add the narrowest exception you need. + +## Document conversion + +PDF and Office documents are converted by a chain of document converters. The +default converters (`pdf-inspector`, `office`) run in-process and make no network +requests. Remote converters (`CloudflareConverter`, `HttpConverter`) upload the +downloaded document to another service: + +- They run only when `DocumentConversion::allow_remote(true)` (CLI: + `--allow-remote-conversion`) is set, and never under `ToolBuilder::hardened()`. +- Their endpoint is operator configuration; fetched content can never choose it. +- Route uploads through your egress boundary by passing your own + `ConverterClient`, or enable proxy environment variables for them with + `ReqwestConverterClient::respect_proxy_env(true)`. +- Documents are downloaded up to `max_document_bytes` (25 MB default), separate + from `max_body_size`. Lower it, or restrict `formats`, for tighter budgets. +- Local parsing of hostile documents is CPU-bound; isolate the process and apply + outer CPU and memory limits for high-volume untrusted input (TM-DOS-007/008). diff --git a/knowledge/foundations/fetchers.md b/knowledge/foundations/fetchers.md index f794cf0..596cc05 100644 --- a/knowledge/foundations/fetchers.md +++ b/knowledge/foundations/fetchers.md @@ -56,9 +56,14 @@ Each content processor must implement: requested output, and content focus. Input bytes have already passed fetchkit's timeout and decompressed-size limits. +6. **`max_input_bytes()`** - Optional download cap overriding `max_body_size` for + matched responses. `None` is the default. + `ContentProcessorRegistry` is ordered and uses the first matching processor. It provides -`new()`, `with_defaults()`, `register()`, `find()`, and `find_for_output()`. Processors do not perform network -requests. `FetcherRegistry::with_content_processors()` retains the built-in fetchers while +`new()`, `with_defaults()`, `with_document_conversion()`, `for_options()`, `register()`, +`find()`, and `find_for_output()`. Custom processors from `FetchOptions::content_processors` +(`ToolBuilder::content_processor`) are consulted before the built-ins. Processors other +than remote document converters do not perform network requests. `FetcherRegistry::with_content_processors()` retains the built-in fetchers while allowing callers to replace the default content processor registry. #### HtmlProcessor @@ -76,31 +81,63 @@ allowing callers to replace the default content processor registry. - Accepts truncated input and returns useful partial content with truncation signals. - Runs after optional rendering; it performs no rendering or network requests. -#### PdfProcessor - -- Matches PDF response media types, including extensionless download URLs. A `.pdf` - suffix is a fallback only for missing Content-Type or `application/octet-stream`. -- Uses `pdf-inspector` to classify and extract Markdown from bounded in-memory bytes. -- Runs CPU work through `spawn_blocking`, with at most two PDF documents processed - concurrently per process. The blocking task owns its concurrency permit, so caller +#### DocumentProcessor and document converters + +`DocumentProcessor` (name `document`) is the single content processor for binary +documents. It delegates to an ordered chain of `DocumentConverter` backends +configured by `DocumentConversion`. + +- **Detection**: `FormatDetector` maps Content-Type to a `DocumentFormat`; the URL + extension is used only for missing or generic binary types (`application/octet-stream`, + `application/zip`, download types). After download the format is confirmed by magic + bytes (`%PDF-`, ZIP, OLE2, RTF); a mismatch is an error and no converter runs. +- **Chain**: per-format order by converter name (`chain(format, names)`), otherwise + registration order. Converters that do not support the format are skipped. The next + converter runs on `Unsupported`, `TooLarge`, `Failed`, or a result with no text or + OCR-required pages. The first partial result is returned when nothing does better. + Later converters add a `fallback_converter` quality warning. +- **Remote gating**: converters report `Locality`. `Remote` converters run only with + `allow_remote(true)`; `ToolBuilder::hardened()` forces it off at build time. +- **Limits**: `max_document_bytes` (default 25 MB) replaces `max_body_size` as the + download cap for responses `DocumentProcessor` matches, via + `ContentProcessor::max_input_bytes`. Output is still capped by `max_body_size`; + truncated downloads are never converted. A converter may declare its own + `max_input_bytes`. +- **Concurrency**: local converters share one blocking-task path with at most two + documents parsed per process; the blocking task owns its permit so caller cancellation cannot admit replacement work while parsing continues. -- Does not perform OCR. Responses identify OCR-required or encoding-problem pages - through quality warnings and `suggested_next_action: "use_ocr"`. -- Extracted output is capped by the same configured `max_body_size`; partial PDF input - is not passed to the parser. - -#### OfficeProcessor - -- Matches Word (DOCX), PowerPoint (PPTX) and spreadsheet (XLSX, XLSM, XLS, ODS) media - types. The file extension is a fallback only for missing Content-Type, - `application/octet-stream` or `application/zip`. -- Parses locally: DOCX and PPTX XML parts with `quick-xml`, spreadsheets with - `calamine`. Documents never leave the process; there is no hosted converter. -- Zip containers are checked against an entry-count and total uncompressed-size - budget from the central directory before any part is inflated, and each part it - reads itself is read through a size cap (TM-DOS-008). -- Shares the PDF processor's blocking-task path and two-document concurrency cap. -- Output caps: 1,000 slides, 1,000 rows and 50 columns per sheet, then `max_body_size`. +- **Configuration**: `ToolBuilder::document_conversion`, `FetchOptions::document_conversion`, + `DocumentConversionSettings` (string form used by CLI, MCP and Python), and + `ContentProcessorRegistry::with_document_conversion` / `for_options`. + `DocumentConversion::empty()` disables all built-in converters. + +Built-in converters (Cargo features): + +| Converter | Feature | Default | Locality | Formats | +| --- | --- | --- | --- | --- | +| `pdf-inspector` (`PdfInspectorConverter`) | `docs-pdf` | on | local | PDF | +| `office` (`OfficeConverter`) | `docs-office` | on | local | DOCX, PPTX, XLSX/XLSM, XLSB, XLS, ODS | +| `cloudflare` (`CloudflareConverter`) | `docs-cloudflare` | off | remote | PDF, DOCX, XLSX, XLSB, XLS, ODT, ODS | +| any name (`HttpConverter`) | `docs-http` | off | remote unless set local | configured | + +- `pdf-inspector`: classifies and extracts Markdown from bounded in-memory bytes; no + OCR. OCR-required pages produce `pdf_requires_ocr` and `suggested_next_action: "use_ocr"` + unless a later converter returns usable text. +- `office`: DOCX and PPTX XML parts via `quick-xml`, spreadsheets via `calamine`. Zip + containers are checked against an entry-count and total uncompressed-size budget + before any part is inflated, and each part is read through a size cap (TM-DOS-008). + Output caps: 1,000 slides, 1,000 rows and 50 columns per sheet. +- `cloudflare`: Workers AI `toMarkdown` REST endpoint, one multipart file per request, + bearer token held in `SecretString`. Images are not routed (billed by Cloudflare). +- `HttpConverter`: multipart upload to an operator-configured URL; Markdown from the raw + body or a JSON pointer. Covers self-hosted services such as Docling or MarkItDown. +- Remote uploads go through `ConverterClient` (default `ReqwestConverterClient`: + redirects off, proxy env ignored unless enabled, response capped at 32 MB), not + `HttpTransport`, which stays GET/HEAD for user URLs. Converter endpoints are operator + configuration and are never derived from fetched content (TM-LEAK-006). + +When no converter is enabled for a recognised document format, the binary rejection +error names the format and says conversion is not enabled. ### Built-in Fetchers @@ -404,8 +441,15 @@ Both built-in fetchers integrate resolve-then-check DNS validation: ``` crates/fetchkit/src/ -├── content.rs # ContentProcessor trait, registry, PDF processor -├── office.rs # Office (DOCX, PPTX, XLSX, XLS, ODS) processor +├── content.rs # ContentProcessor trait, registry, HTML processor +├── documents/ # DocumentProcessor, DocumentConverter chain and config +│ ├── mod.rs # formats, detection, chain, DocumentConversion +│ ├── settings.rs # string settings for CLI, MCP, Python +│ ├── pdf_inspector.rs # local PDF converter (docs-pdf) +│ ├── office.rs # local DOCX/PPTX/spreadsheet converter (docs-office) +│ ├── remote.rs # ConverterClient, multipart, SecretString +│ ├── cloudflare.rs # Cloudflare toMarkdown converter (docs-cloudflare) +│ └── http.rs # generic HTTP converter (docs-http) ├── dns.rs # DnsPolicy - SSRF prevention via resolve-then-check ├── file_saver.rs # FileSaver trait, LocalFileSaver, SaveResult, FileSaveError ├── fetchers/ diff --git a/knowledge/foundations/tool-contract.md b/knowledge/foundations/tool-contract.md index eb6f26e..23e2331 100644 --- a/knowledge/foundations/tool-contract.md +++ b/knowledge/foundations/tool-contract.md @@ -322,35 +322,36 @@ By default, Fetchkit blocks connections to private/reserved IP ranges: - `image/`, `audio/`, `video/`, `application/octet-stream`, `application/pdf`, `application/x-pdf`, `application/zip`, `application/gzip`, `application/x-tar`, `application/x-rar`, `application/x-7z`, `application/vnd.ms-`, `application/vnd.openxmlformats`, - `application/vnd.oasis.opendocument`, `font/`. + `application/vnd.oasis.opendocument`, `application/msword`, `application/epub+zip`, `font/`. - When `as_markdown` is requested, a matching registered content processor runs on the bounded response body before the generic binary response is returned. -- The built-in PDF processor: - - Matches `application/pdf`, `application/x-pdf`, or a `.pdf` URL served as - `application/octet-stream`/without Content-Type. - - Uses `pdf-inspector` on in-memory bytes and returns `format: "markdown"` for - PDFs with an extractable text layer. - - Performs no network requests and no OCR. Scanned/image-only pages produce - `pdf_requires_ocr` quality warnings and suggest `use_ocr`. - - Runs on a bounded blocking pool path, and both PDF input and extracted output - honor `max_body_size`. Partial PDF inputs are not parsed. -- The built-in Office processor: - - Matches DOCX, PPTX, XLSX/XLSM, XLS and ODS media types, or those file - extensions when served as `application/octet-stream`, `application/zip` or - without Content-Type. - - Returns `format: "markdown"`: Word headings, lists and tables; one - `## Slide N: title` section per slide in presentation order; one `## sheet` - Markdown table per spreadsheet sheet (first row as header, at most 1,000 rows - and 50 columns, with a `sheet_truncated` warning beyond that). - - Reports `office_docx`, `office_pptx` or `office_spreadsheet` as the extraction - method and `dc:title` from the document properties as the title. - - Performs no network requests. Zip containers are checked against an entry and - uncompressed-size budget before inflating (TM-DOS-008); shares the PDF - processor's blocking pool and two-document concurrency cap. Partial inputs are - not parsed. + Custom processors (`ToolBuilder::content_processor`) run before the built-ins. +- The built-in `DocumentProcessor` converts documents through the configured + converter chain (see [Fetcher System](fetchers.md)): + - Detects the format from Content-Type, or from the URL extension for missing or + generic binary types, and confirms it by magic bytes after download. + - Downloads up to `max_document_bytes` (25 MB default); output honors + `max_body_size`. Partial inputs are never converted. + - Default converters (`docs-pdf`, `docs-office`): `pdf-inspector` for text PDFs + (`pdf_inspector`; OCR-only pages produce `pdf_requires_ocr` and `use_ocr`) and + `office` for DOCX/PPTX/XLSX/XLSM/XLSB/XLS/ODS (`office_docx`, `office_pptx`, + `office_spreadsheet`; Word headings, lists and tables; one `## Slide N: title` + per slide; one `## sheet` table per sheet capped at 1,000 rows and 50 columns + with `sheet_truncated`). Titles come from document properties. + - Remote converters (`cloudflare` -> `cloudflare_tomarkdown`, `HttpConverter`) run + only when remote conversion is allowed and never under `hardened()`. + - Converter fallthrough adds `fallback_converter`; quality score is 1.0 clean, + 0.8 with warnings, 0.6 with OCR or encoding issues, 0.0 without text. +- Configuration surfaces: `ToolBuilder::document_conversion`, CLI/MCP + `--convert`, `--converter`, `--allow-remote-conversion`, `--max-document-bytes` + (Cloudflare credentials from `CLOUDFLARE_ACCOUNT_ID` / `CLOUDFLARE_API_TOKEN`), + Python `document_formats`, `converters`, `allow_remote_conversion`, + `max_document_bytes`, `cloudflare_account_id`, `cloudflare_api_token`. - For binary without a matching processor: - Return metadata (`content_type`, `size`, `filename`, `last_modified`) - - Include `error: "Binary content is not supported. Only textual content (HTML, text, JSON, etc.) can be fetched."` + - Include `error: "Binary content is not supported. Only textual content (HTML, text, JSON, etc.) can be fetched."`, + followed by `" conversion is not enabled in this fetchkit configuration; enable a document converter to convert it."` + when a document format was recognised - Omit `content`, `format`, `truncated` #### HEAD diff --git a/knowledge/log.md b/knowledge/log.md index 8c2d65e..cc23772 100644 --- a/knowledge/log.md +++ b/knowledge/log.md @@ -2,6 +2,7 @@ ## 2026-10-12 +* **Document conversion**: Replaced the PDF and Office processors with one `DocumentProcessor` running a configurable, per-format `DocumentConverter` chain; added feature-gated converters (`docs-pdf`, `docs-office`, `docs-cloudflare`, `docs-http`), a separate 25 MB document cap, remote-converter opt-in, and threats TM-LEAK-006/007. * **Office documents**: Added the local Office processor (DOCX, PPTX, XLSX, XLS, ODS) to the fetchers and tool contract, its zip-bomb threat TM-DOS-008, and the Markdown-first `Accept` header. ## 2026-08-08 diff --git a/knowledge/security/threat-model.md b/knowledge/security/threat-model.md index 6428670..ca47704 100644 --- a/knowledge/security/threat-model.md +++ b/knowledge/security/threat-model.md @@ -320,7 +320,7 @@ decompresses to a large size is caught by the same size limit that protects against unbounded responses (TM-DOS-001). **TM-DOS-007 — Adversarial PDF parsing (ACCEPTED):** -PDF content processors receive only bodies bounded by `max_body_size`; partial bodies +Document converters receive only bodies bounded by `max_document_bytes` (25 MB default); partial bodies are rejected before parsing, extracted output is capped again, and synchronous parsing runs off the async runtime with at most two documents active per process. The blocking task owns its concurrency permit, preventing cancelled callers from starting replacement @@ -331,12 +331,12 @@ request CPU/memory limits. Accepted because the in-process Rust parser has no re cancellation boundary once a blocking parse begins. **TM-DOS-008 — Zip bomb or adversarial Office document (MITIGATED):** -Office documents are zip archives (except legacy XLS), so a body within `max_body_size` +Office documents are zip archives (except legacy XLS), so a body within `max_document_bytes` can declare gigabytes of content. Before any part is inflated, the Office processor -rejects archives with more than 10,000 entries or more than 200 MB of declared +(`office` converter) rejects archives with more than 10,000 entries or more than 200 MB of declared uncompressed content, and every XML part it reads itself is read through a 64 MB cap, since declared sizes can lie. Spreadsheet output is capped at 1,000 rows and 50 columns -per sheet before the `max_body_size` output cap. Parsing shares the PDF processor's +per sheet before the `max_body_size` output cap. Parsing shares the local converters' blocking path and two-document concurrency cap. `calamine` inflates sheet parts itself; the declared-size budget bounds those as well. @@ -349,9 +349,25 @@ the declared-size budget bounds those as well. | TM-LEAK-003 | Response content leaks internal data | Low | Fetchkit returns content as-is; caller must filter sensitive data | **CALLER RISK** | | TM-LEAK-004 | User-Agent reveals software version | Info | Default UA `Everruns Fetchkit/1.0` reveals stack; configurable | **BY DESIGN** | | TM-LEAK-005 | Timing side-channels (connect time reveals network proximity) | Low | 1-second timeout masks some timing; not fully mitigated | **ACCEPTED** | +| TM-LEAK-006 | Fetched documents disclosed to a remote converter | Medium | Remote converters off by default, explicit `allow_remote(true)`, forced off by `hardened()`, backend named in `extraction_method` | MITIGATED | +| TM-LEAK-007 | Remote converter credentials exposed | Medium | Tokens held in `SecretString`, redacted from `Debug`, never echoed in errors or responses | MITIGATED | ### Mitigation Details +**TM-LEAK-006 — Document disclosure to remote converters (MITIGATED):** +Remote `DocumentConverter`s (Cloudflare `toMarkdown`, `HttpConverter`) upload the +downloaded document to another service. They never run unless the operator sets +`DocumentConversion::allow_remote(true)`; `ToolBuilder::hardened()` forces remote +conversion off at build time regardless of order. Converter endpoints come only from +operator configuration, never from fetched content, and the response's +`extraction_method` names the backend that produced it. Uploads go through +`ConverterClient`, so a host can route them through its own egress boundary. + +**TM-LEAK-007 — Converter credential exposure (MITIGATED):** +API tokens and configured header values are stored as `SecretString`, which redacts +itself from `Debug`; authentication failures are reported by status, without the +token or request headers. + **TM-LEAK-001 — Error message detail (MITIGATED):** Fetchkit's error types (`FetchError`) use generic messages that don't include resolved IP addresses or internal hostnames. Connect errors say "Failed to connect