//! 6-layer malware scanning pipeline for file uploads. //! //! Layers 1-4 always run (in-process, deterministic). Layers 5-6 are optional //! (external services). //! //! **Error policy is per-layer**, declared at each layer's source file as //! `pub const ERROR_POLICY`. The aggregator `final_status` consults each //! layer's policy via `error_policy_for`. In-process layers are `FailClosed` //! (an error is a structural defect); external layers are `FailOpen` (an //! error is an outage that must not block the platform). See //! `docs/scan-pipeline-audit.md` for the rationale. //! //! See also: `/docs/tech/content-protection` pub mod archive; pub mod clamav; pub mod content_type; pub mod hash_lookup; pub mod metadefender; pub mod signing_linux; pub mod signing_macos; pub mod signing_windows; pub mod spool; pub mod structural; pub mod urlhaus; pub mod worker; pub mod yara; use serde::Serialize; use sha2::{Digest, Sha256}; use crate::config::ScanConfig; use crate::db::FileScanStatus; use crate::storage::FileType; /// Per-layer scan verdict #[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize)] #[serde(rename_all = "lowercase")] pub enum LayerVerdict { Pass, Fail, Skip, Error, } /// Policy for how a layer's `Error` verdict feeds into the pipeline's final status. /// /// - `FailClosed`, an `Error` from this layer holds the upload for admin review. /// Appropriate for deterministic in-process layers where an `Error` indicates a /// real bug or a structurally suspicious file. /// - `FailOpen`, an `Error` from this layer is treated as `Skip` for aggregation. /// Appropriate for external services (network, daemons) where an outage on a /// third party must not take down the platform's upload pipeline. /// /// Each layer declares its own `ERROR_POLICY` const; the aggregator in /// `ScanPipeline::final_status` consults the declaration via `error_policy_for`. #[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize)] #[serde(rename_all = "snake_case")] pub enum ErrorPolicy { FailClosed, FailOpen, } /// Result from a single scanning layer #[derive(Debug, Clone, Serialize)] pub struct LayerResult { pub layer: &'static str, pub verdict: LayerVerdict, pub detail: Option, } /// Look up a layer's declared error policy by name. Defaults to `FailClosed` /// for unknown layers, a defensive choice that surfaces uninstrumented /// additions during testing rather than silently fail-opening them. fn error_policy_for(layer: &str) -> ErrorPolicy { match layer { "content_type" => content_type::ERROR_POLICY, "structural" => structural::ERROR_POLICY, "archive" => archive::ERROR_POLICY, // Recursive interior scan of an archive's entries. A decompressed entry // we could not fully inspect (un-openable, over budget, or nested deeper // than the scan depth) errors here and must fail closed, a payload // hidden in a zip-in-a-zip is held for review, never passed Clean. This // is the structural close of the nested-archive recursion chronic. "archive_nested" => ErrorPolicy::FailClosed, "yara" => yara::ERROR_POLICY, "clamav" => clamav::ERROR_POLICY, // A *reachable* clamd that couldn't fully scan (size/scan limit, // unparseable reply) is a coverage gap, not an outage, fail closed, // distinct from the FailOpen `clamav` layer used for an unreachable // daemon. Splitting the policy by layer identity is what keeps CHRONIC // S1 (size-limit → FailOpen → Clean) from recurring. "clamav_incomplete" => ErrorPolicy::FailClosed, "malwarebazaar" => hash_lookup::ERROR_POLICY, "urlhaus" => urlhaus::ERROR_POLICY, "signing_macos" => signing_macos::ERROR_POLICY, "signing_windows" => signing_windows::ERROR_POLICY, "signing_linux" => signing_linux::ERROR_POLICY, "metadefender" => metadefender::ERROR_POLICY, other => { tracing::error!( layer = other, "unknown scan layer; defaulting to FailClosed" ); ErrorPolicy::FailClosed } } } /// Decide whether the second-opinion (MetaDefender) layer should run, based /// on the verdicts of layers that have already completed. Any `Fail`, or any /// `Error` from a fail-closed in-process layer, counts as "suspicious enough /// to escalate". Pure `Error`s from fail-open external layers do not, those /// are operational noise, not malware signals. fn suspicion_present(layers: &[LayerResult]) -> bool { layers.iter().any(|l| match l.verdict { LayerVerdict::Fail => true, LayerVerdict::Error => error_policy_for(l.layer) == ErrorPolicy::FailClosed, _ => false, }) } /// Run an external hash-lookup layer under the aggregate external-lookup deadline, /// degrading to a Skip (FailOpen) on timeout. Shared by the buffered `scan()` and /// the streaming `scan_stream()` so BOTH paths get the same defense-in-depth /// deadline (Perf-S1), previously only `scan_stream` wrapped these calls, so a /// small upload had only the per-request client timeout with no aggregate ceiling. async fn lookup_with_timeout( layer: &'static str, fut: impl std::future::Future, ) -> LayerResult { let lookup_timeout = std::time::Duration::from_secs(crate::constants::SCAN_EXTERNAL_LOOKUP_TIMEOUT_SECS); match tokio::time::timeout(lookup_timeout, fut).await { Ok(layer_result) => layer_result, Err(_) => { tracing::warn!("{layer} lookup timed out; treating as Skip (FailOpen)"); LayerResult { layer, verdict: LayerVerdict::Skip, detail: Some(format!("{layer} lookup timed out")), } } } } /// Aggregate per-layer results into a final scan status. /// /// - Any layer `Fail` → `Quarantined` (terminal). /// - Any layer `Error` whose policy is `FailClosed` → `HeldForReview`. /// - `FailOpen` errors are treated as Skip-equivalent for aggregation; they're /// still surfaced in the per-layer detail so admins and PoM can see degraded /// layers in the dashboard. /// - Otherwise → `Clean`. fn final_status(layers: &[LayerResult]) -> FileScanStatus { if layers.iter().any(|l| l.verdict == LayerVerdict::Fail) { return FileScanStatus::Quarantined; } let has_fail_closed_error = layers.iter().any(|l| { l.verdict == LayerVerdict::Error && error_policy_for(l.layer) == ErrorPolicy::FailClosed }); if has_fail_closed_error { FileScanStatus::HeldForReview } else { FileScanStatus::Clean } } /// Whether a file's tail would go unscanned, true when it exceeds the YARA /// prefix cap ([`crate::constants::SCAN_YARA_MAX_BYTES`]) and ClamAV does not /// provide a *full-file* backstop for it. Such a file must be held for review /// rather than certified Clean on a prefix-only scan. /// /// ClamAV is a full-file backstop only up to the operator-declared /// [`ScanConfig::clamav_max_scan_bytes`]: clamd silently scans only the first /// `MaxScanSize`/`MaxFileSize` bytes and returns `OK` with no error, so trusting /// "socket configured" alone let a large file with a payload past the YARA /// prefix certify Clean (ultra-fuzz Run #24 Security MODERATE). `None` (coverage /// undeclared) is fail-closed: no backstop, hold for review. fn yara_tail_unscanned(data_len: usize, clamav_backstop_bytes: Option) -> bool { if data_len <= crate::constants::SCAN_YARA_MAX_BYTES { return false; } // Backstop only covers the file if the operator declared coverage that // reaches its size. Undeclared coverage does NOT count. let covered = clamav_backstop_bytes.is_some_and(|max| data_len as u64 <= max); !covered } /// Fail-closed result for when a CPU scan layer panics on crafted input. /// /// The in-process parsers (goblin, yara-x, zip, content-type) run over fully /// attacker-controlled bytes; a malformed file can panic them. Such a file is /// held for admin review, never passed as `Clean`. Critically, building a /// result here lets the caller `return` normally instead of `.expect()`-ing the /// join handle and unwinding the scan-worker task, the worker pool is spawned /// once and never respawns, so a propagated panic would permanently shrink it /// (two crafted uploads could disable scanning entirely). `scan_panic` is not a /// registered layer, so `error_policy_for` defaults it to `FailClosed`. fn panicked_sync_result(file_size: u64) -> ScanResult { let layer = LayerResult { layer: "scan_panic", verdict: LayerVerdict::Error, detail: Some("a CPU scan layer panicked on this input".to_string()), }; ScanResult { status: final_status(std::slice::from_ref(&layer)), layers: vec![layer], sha256: String::new(), file_size, } } /// Fail-closed result for a scan whose CPU layers exceeded the wall-clock /// deadline (`SCAN_CPU_LAYERS_TIMEOUT_SECS`), e.g. a pathological slow-codec /// archive decompress. Held for review rather than passed, and the worker is /// freed so a crafted upload can't monopolize the two-worker scan pool /// (fuzz 2026-07-06 F1). fn timed_out_sync_result(file_size: u64) -> ScanResult { let layer = LayerResult { layer: "scan_timeout", verdict: LayerVerdict::Error, detail: Some("CPU scan layers exceeded the wall-clock deadline".to_string()), }; ScanResult { status: final_status(std::slice::from_ref(&layer)), layers: vec![layer], sha256: String::new(), file_size, } } /// Fail-closed result for a file too large to spool for scanning. /// /// Files above [`crate::constants::SCAN_SPOOL_MAX_BYTES`] cannot be spooled to /// disk for the CPU layers (the platform accepts videos up to /// `MAX_VIDEO_SIZE`, which exceeds the spool ceiling). Rather than fail the scan /// job, which left the upload stuck `Pending` and retried forever by the /// reaper, hold it for admin review under an explicit, documented policy. If /// auto-scanning large videos is wanted instead, raise `SCAN_SPOOL_MAX_BYTES` /// to cover the max accepted upload (at a proportional scratch-disk cost). fn too_large_to_scan(file_size: u64) -> ScanResult { let layer = LayerResult { layer: "scan_size_limit", verdict: LayerVerdict::Error, detail: Some(format!( "file size {file_size} exceeds the scan spool ceiling ({} bytes); held for review", crate::constants::SCAN_SPOOL_MAX_BYTES )), }; ScanResult { status: final_status(std::slice::from_ref(&layer)), layers: vec![layer], sha256: String::new(), file_size, } } /// Scan-then-promote closing move (C1): copy a Clean object from its unserved /// `staging/{uuid}` key to the immutable, content-addressed /// `{owner}/c/{sha256}.{ext}` key, repoint the entity (and, for CDN images, its /// materialized public URL) at the content key, mark it Clean, and enqueue the /// staging object for durable deletion. The bytes a buyer is served are then /// provably the bytes that were scanned: the content key is named by the scanned /// hash, the owner holds no presign to it, and the (owner-less, random) staging /// key they *can* re-PUT to is unserved and deleted. /// /// Shared by the scan worker's Clean path and the admin approve-held path so the /// two promote sites can't diverge. Ordering is fail-safe: the S3 copy runs /// first, then a single transaction repoints the row and enqueues the staging /// delete. If the copy or the DB write fails, the row keeps pointing at the /// still-present (unserved) staging key and the caller leaves the work un-done, /// so a retry re-promotes, the copy is idempotent (hash-named destination), and /// an already-promoted `{owner}/c/...` key short-circuits. #[allow(clippy::too_many_arguments)] pub async fn promote_staging_to_content( db: &sqlx::PgPool, backend: &dyn crate::storage::StorageBackend, public_backend: Option<&dyn crate::storage::StorageBackend>, cdn_base_url: &str, kind: crate::db::scan_jobs::ScanTargetKind, file_type: FileType, target_id: uuid::Uuid, owner: crate::db::UserId, staging_key: &str, sha256: &str, bucket: crate::storage::S3Bucket, ) -> crate::error::Result<()> { use crate::db::scan_jobs::ScanTargetKind; use crate::storage::{S3Client, S3Key}; if sha256.is_empty() { return Err(crate::error::AppError::Storage(format!( "cannot promote {staging_key}: scan recorded no content hash" ))); } // Already promoted (worker/admin retry, admin bulk over a mixed set): a // content key is `{owner}/c/...`, never `staging/...`. Nothing to copy or repoint. if !staging_key.starts_with("staging/") { return Ok(()); } let ext = crate::storage::key_extension(staging_key); let content = S3Client::content_key(owner, sha256, ext); let content_str = content.as_str().to_string(); // The three CDN-unsigned image kinds have their content object served from // the PUBLIC bucket; everything else (gated media, presigned insertions, // OTA) keeps its content object in the same private bucket the staging // object lives in. Staging is always private, so a public-bucket promote is // a CROSS-bucket copy (private staging -> public content) issued on the // public backend with the private bucket as the copy source. let to_public = kind.content_served_from_public_bucket(); let staging = S3Key::from_stored(staging_key); // The source size picks the copy strategy: S3 rejects a single-part // `CopyObject` above 5 GiB, so a large source has to go through ranged // multipart `UploadPartCopy` or the promote fails *after* a successful // upload and scan, the worst position to fail in. The S3 object is the // authoritative size (same posture as blob confirm), so read it here rather // than threading a recorded size through all three call sites; one // HeadObject is negligible next to the object copy that follows. The source // always lives in `backend`'s (private, staging) bucket, including on the // cross-bucket public promote. let src_size = backend .object_size(staging_key) .await? .ok_or_else(|| { crate::error::AppError::Storage(format!( "cannot promote {staging_key}: staging object not found in storage" )) })? .max(0) as u64; let needs_multipart = src_size > crate::constants::S3_SINGLE_COPY_MAX_BYTES; // Only consulted on the multipart branch; a single CopyObject carries the // source's content type across on its own. let content_type = crate::storage::content_type_for(file_type, ext); // 1. Promote the object in storage (copy staging -> content). Idempotent. if to_public { let public = public_backend.ok_or_else(|| { crate::error::AppError::Storage(format!( "cannot promote CDN image {staging_key}: public bucket not configured" )) })?; if needs_multipart { public .copy_object_multipart( backend.bucket(), &staging, &content, content_type, src_size, None, ) .await?; } else { public .copy_object_from(backend.bucket(), &staging, &content) .await?; } } else if needs_multipart { // Same-bucket multipart promote: the source bucket is this backend's own. backend .copy_object_multipart( backend.bucket(), &staging, &content, content_type, src_size, None, ) .await?; } else { backend.copy_object(&staging, &content).await?; } // 2. Repoint the row + enqueue the staging delete atomically. let mut tx = db.begin().await?; if kind.is_cdn_served_without_gate() { let content_url = if matches!(kind, ScanTargetKind::ContentInsertion) { // No materialized URL column (insertions are served presigned). String::new() } else { // Content object is in the public bucket, which is exactly what the // CDN base fronts, so the URL is `{cdn}/{content_key}`. crate::storage::build_project_image_url(cdn_base_url, &content_str) }; crate::db::scanning::promote_cdn_image_by_key( &mut tx, staging_key, &content_str, &content_url, ) .await?; } else { crate::db::scanning::promote_gated(&mut *tx, kind, file_type, target_id, &content_str) .await?; } crate::db::pending_s3_deletions::enqueue_deletions( &mut *tx, &[(staging_key.to_string(), bucket.as_str().to_string())], "scan_promote_staging", ) .await?; tx.commit().await?; Ok(()) } /// Aggregate scan result across all layers #[derive(Debug, Clone)] pub struct ScanResult { pub status: FileScanStatus, pub layers: Vec, pub sha256: String, pub file_size: u64, } /// Pre-compiled scanning pipeline. Initialized once at startup and shared via Arc. pub struct ScanPipeline { yara_rules: Option, /// Number of YARA rule files that compiled, and the configured health floor. yara_rule_count: usize, yara_min_rule_files: usize, clamav_socket: Option, /// Operator-declared ClamAV per-object scan coverage; see /// [`ScanConfig::clamav_max_scan_bytes`]. `None` = coverage unknown = /// ClamAV is not treated as a full-file backstop (fail-closed). clamav_max_scan_bytes: Option, malwarebazaar_enabled: bool, urlhaus_enabled: bool, abuse_ch_auth_key: Option, metadefender_api_key: Option, } /// How the bytes to scan are sourced. This is the ONLY thing that differs between /// the buffered and streaming scan paths, all scan POLICY (CPU layers off the /// runtime, ClamAV, URLhaus host extraction, external lookups, byte caps) runs in /// the single private [`ScanPipeline::run_scan`], so a policy cannot drift between /// the two entry points (ultra-fuzz CHRONIC: scan()/scan_stream() twin divergence, /// closed Run 9). enum ScanInput { /// Whole object held in memory, small uploads. Buffered(bytes::Bytes), /// Object spooled to a tempfile and scanned via a memory map. The retained /// `SpoolHandle` keeps the file on disk (and unlinks it on drop) so ClamAV can /// stream it by path while the CPU layers read the map. Spooled { map: std::sync::Arc, spool: spool::SpoolHandle, }, } /// A cheaply-cloneable, `Send + 'static` view of the scan bytes for the /// `spawn_blocking` CPU work (a `Bytes` refcount bump or an `Arc` clone). #[derive(Clone)] enum ScanBytes { Buffered(bytes::Bytes), Mapped(std::sync::Arc), } impl ScanBytes { fn as_slice(&self) -> &[u8] { match self { ScanBytes::Buffered(b) => b, ScanBytes::Mapped(m) => &m[..], } } } /// Where ClamAV reads its bytes: the in-memory buffer, or the spool path it /// streams via INSTREAM frames. enum ClamavSource { Buffered(bytes::Bytes), Path(std::path::PathBuf), } impl ScanInput { /// A `Send + 'static` byte handle for `spawn_blocking` CPU work. fn byte_handle(&self) -> ScanBytes { match self { ScanInput::Buffered(b) => ScanBytes::Buffered(b.clone()), ScanInput::Spooled { map, .. } => ScanBytes::Mapped(std::sync::Arc::clone(map)), } } fn len(&self) -> usize { match self { ScanInput::Buffered(b) => b.len(), ScanInput::Spooled { map, .. } => map.len(), } } } impl ScanPipeline { /// Create a new pipeline, compiling YARA rules from the configured directory. pub fn new(config: &ScanConfig) -> Result { let (yara_rules, yara_rule_count) = yara::compile_rules_from_dir(&config.yara_rules_dir)?; Ok(ScanPipeline { yara_rules, yara_rule_count, yara_min_rule_files: config.yara_min_rule_files, clamav_socket: config.clamav_socket.clone(), clamav_max_scan_bytes: config.clamav_max_scan_bytes, malwarebazaar_enabled: config.malwarebazaar_enabled, urlhaus_enabled: config.urlhaus_enabled, abuse_ch_auth_key: config.abuse_ch_auth_key.clone(), metadefender_api_key: config.metadefender_api_key.clone(), }) } /// The configured ClamAV socket path, if any. Used to spawn the runtime /// liveness probe (`worker::spawn_clamav_health_probe`). pub fn clamav_socket(&self) -> Option<&str> { self.clamav_socket.as_deref() } /// Assert at startup that at least one real AV layer is live. Refuse to /// boot otherwise, ClamAV's FailOpen policy means a dead clamd /// silently passes every upload as Clean, and a YARA-rules-empty deploy /// gives the same false sense of coverage. If the operator configured /// scanning, a misconfiguration must be loud at boot, not silent at runtime. pub async fn assert_live(&self) -> Result<(), String> { let mut live_layers: Vec<&str> = Vec::new(); if let Some(ref socket) = self.clamav_socket { match clamav::ping(socket).await { Ok(()) => live_layers.push("clamav"), Err(e) => { return Err(format!("ClamAV socket {socket} unreachable: {e}")); } } // ClamAV being reachable does not mean it scans whole objects: clamd // silently truncates at MaxScanSize/MaxFileSize and reports OK, and // those limits aren't queryable over the socket. Surface loudly at // boot when the operator hasn't declared coverage reaching the spool // ceiling, above the declared coverage, large files are held for // review rather than certified Clean (ultra-fuzz Run #24 Security). match self.clamav_max_scan_bytes { None => tracing::warn!( yara_prefix = crate::constants::SCAN_YARA_MAX_BYTES, "CLAMAV_MAX_SCAN_BYTES is not set, ClamAV is NOT treated as a full-file backstop; \ files larger than the YARA prefix will be held for admin review. Set it to your \ clamd min(MaxScanSize, MaxFileSize, StreamMaxLength) to auto-clear large uploads." ), Some(max) if max < crate::constants::SCAN_SPOOL_MAX_BYTES => tracing::warn!( declared_coverage = max, spool_ceiling = crate::constants::SCAN_SPOOL_MAX_BYTES, "CLAMAV_MAX_SCAN_BYTES is below the scan spool ceiling, uploads between the declared \ ClamAV coverage and the spool ceiling will be held for admin review, not auto-cleared." ), Some(_) => {} } } if self.yara_rules.is_some() { // Expected-rule-count floor: a corpus that quietly dropped below the // operator-declared size (e.g. a yara-x upgrade made N rules // uncompilable) is degraded coverage masquerading as a live layer. // Fail boot loudly when a floor is set and we're under it. if self.yara_min_rule_files > 0 && self.yara_rule_count < self.yara_min_rule_files { return Err(format!( "YARA corpus degraded: {} rule files compiled, below the configured \ floor of {} (YARA_MIN_RULE_FILES). Refusing to boot, a silently \ shrunken rule set is false coverage.", self.yara_rule_count, self.yara_min_rule_files, )); } live_layers.push("yara"); } if self.malwarebazaar_enabled { live_layers.push("malwarebazaar"); } if self.urlhaus_enabled { live_layers.push("urlhaus"); } if self.metadefender_api_key.is_some() { live_layers.push("metadefender"); } if live_layers.is_empty() { return Err( "Scanning configured but no AV layer is live (no ClamAV socket, \ no YARA rules, no remote API keys). Refusing to boot, the \ FailOpen policy would pass every upload as Clean." .to_string(), ); } tracing::info!(layers = ?live_layers, "scan pipeline live layers asserted"); Ok(()) } /// Buffered scan entry point, small uploads held in memory. A thin adapter /// over [`run_scan`](Self::run_scan); all scan policy lives there. pub(crate) async fn scan( self: std::sync::Arc, data: impl Into, file_type: FileType, ) -> ScanResult { // `bytes::Bytes` is already a cheaply-cloneable refcounted buffer, so the // download hands its single aggregated allocation straight here with no // extra copy (the buffered path previously aggregated then `to_vec`'d the // body, transiently doubling to ~200 MB for a 100 MB file, Run #2). self.run_scan(ScanInput::Buffered(data.into()), file_type) .await } /// Streaming scan entry point, large uploads spooled to a tempfile and /// scanned via a memory map, so the >100 MB case doesn't hold the whole object /// in RAM. A thin adapter over [`run_scan`](Self::run_scan). pub(crate) async fn scan_stream( self: std::sync::Arc, spool: spool::SpoolHandle, file_type: FileType, ) -> ScanResult { let map = match spool::mmap_read(spool.path()) { Ok(m) => std::sync::Arc::new(m), Err(e) => { let file_size = std::fs::metadata(spool.path()).map_or(0, |m| m.len()); let layer = LayerResult { layer: "spool", verdict: LayerVerdict::Error, detail: Some(e), }; return ScanResult { status: final_status(std::slice::from_ref(&layer)), layers: vec![layer], sha256: String::new(), file_size, }; } }; self.run_scan(ScanInput::Spooled { map, spool }, file_type) .await } /// The single scan body shared by [`scan`](Self::scan) and /// [`scan_stream`](Self::scan_stream). Every scan policy lives here exactly /// once; the entry points differ only in how `input` sources its bytes, so a /// policy (CPU work off the runtime, ClamAV, URLhaus host extraction, external /// lookups, byte caps) can no longer drift between buffered and streaming /// (ultra-fuzz CHRONIC, closed Run 9). /// /// CPU-bound layers (sha256, content-type, structural, archive, yara) run on a /// blocking-pool thread via `spawn_blocking`; ClamAV and URLhaus run /// concurrently with them via `tokio::join!`. async fn run_scan( self: std::sync::Arc, input: ScanInput, file_type: FileType, ) -> ScanResult { let file_size = input.len() as u64; // CPU layers + hash, off the runtime, under a wall-clock deadline. The // per-layer timeouts (yara 30s) did not cover the archive decompress // walk, so a slow-codec archive could pin a scan worker for its full // decompress time (fuzz 2026-07-06 F1). The timeout frees the worker on // elapse; the blocking thread runs to completion on the large blocking // pool (spawn_blocking is not cancellable), so the memory-budget ceiling // still holds while the two-worker pool is no longer monopolized. let sync_bytes = input.byte_handle(); let sync_self = std::sync::Arc::clone(&self); let sync_fut = tokio::time::timeout( std::time::Duration::from_secs(crate::constants::SCAN_CPU_LAYERS_TIMEOUT_SECS), tokio::task::spawn_blocking(move || { sync_self.run_sync_layers(sync_bytes.as_slice(), file_type) }), ); // ClamAV: buffered scans the in-memory bytes; spooled streams the file by // path (INSTREAM frames) so a >100 MB object isn't re-buffered. The // retained SpoolHandle in `input` keeps the file alive across this join. let clamav_socket = self.clamav_socket.clone(); let clamav_source = match &input { ScanInput::Buffered(b) => ClamavSource::Buffered(b.clone()), ScanInput::Spooled { spool, .. } => ClamavSource::Path(spool.path().to_path_buf()), }; let clamav_fut = async move { let Some(socket) = clamav_socket else { return LayerResult { layer: "clamav", verdict: LayerVerdict::Skip, detail: Some("ClamAV not configured".to_string()), }; }; match clamav_source { ClamavSource::Buffered(data) => clamav::scan_with_clamav(&socket, &data).await, ClamavSource::Path(path) => match tokio::fs::File::open(&path).await { Ok(file) => clamav::scan_with_clamav_stream(&socket, file).await, Err(e) => LayerResult { layer: "clamav", verdict: LayerVerdict::Error, detail: Some(format!("open spool for clamav: {e}")), }, }, } }; // URLhaus: extract candidate hosts on the blocking pool (the byte walk can // page-fault on the mmap), then do only the network lookups async. This is // ONE policy for both entry points, the exact step that used to drift // between scan() and scan_stream() (ultra-fuzz CHRONIC). let urlhaus_bytes = input.byte_handle(); let urlhaus_enabled = self.urlhaus_enabled; let urlhaus_key = self.abuse_ch_auth_key.clone(); let urlhaus_fut = async move { if urlhaus_enabled { let hosts = tokio::task::spawn_blocking(move || { urlhaus::extract_unique_hosts( urlhaus_bytes.as_slice(), urlhaus::MAX_HOSTS_PER_FILE, ) }) .await .unwrap_or_default(); urlhaus::check_urlhaus_hosts(hosts, urlhaus_key.as_deref()).await } else { LayerResult { layer: "urlhaus", verdict: LayerVerdict::Skip, detail: Some("URLhaus lookups disabled".to_string()), } } }; let (sync_result, clamav_result, urlhaus_result) = tokio::join!(sync_fut, clamav_fut, urlhaus_fut); let (mut layers, sha256) = match sync_result { Ok(Ok(v)) => v, Ok(Err(join_err)) => { tracing::error!( error = %join_err, is_panic = join_err.is_panic(), "scan sync layers panicked; holding file for review (worker survives)" ); return panicked_sync_result(file_size); } Err(_elapsed) => { tracing::error!( timeout_secs = crate::constants::SCAN_CPU_LAYERS_TIMEOUT_SECS, "scan CPU layers exceeded the wall-clock deadline; holding file for \ review (worker freed; blocking thread runs to completion)" ); return timed_out_sync_result(file_size); } }; layers.push(clamav_result); layers.push(urlhaus_result); self.push_external_lookups(&mut layers, &sha256).await; let status = final_status(&layers); // Dropping `input` here releases the mmap and unlinks the spool tempfile // (if any), after ClamAV has finished streaming it. drop(input); ScanResult { status, layers, sha256, file_size, } } /// Append the post-sync, hash-keyed external-lookup layers to `layers`: /// Layer 6 MalwareBazaar (by hash) then Layer 9 MetaDefender (a /// suspicion-gated second opinion that only fires when a prior layer flagged /// the file, to stay within the free-tier quota). Each runs under the shared /// aggregate external-lookup deadline (`lookup_with_timeout`, Perf-S1). Shared /// by `scan` and `scan_stream` so the buffered and streaming paths can't drift. async fn push_external_lookups(&self, layers: &mut Vec, sha256: &str) { layers.push(if self.malwarebazaar_enabled { lookup_with_timeout( "malwarebazaar", hash_lookup::check_malwarebazaar(sha256, self.abuse_ch_auth_key.as_deref()), ) .await } else { LayerResult { layer: "malwarebazaar", verdict: LayerVerdict::Skip, detail: Some("MalwareBazaar lookups disabled".to_string()), } }); layers.push(if suspicion_present(layers) { lookup_with_timeout( "metadefender", metadefender::check_metadefender(sha256, self.metadefender_api_key.as_deref()), ) .await } else { LayerResult { layer: "metadefender", verdict: LayerVerdict::Skip, detail: Some("No prior suspicion; second-opinion not invoked".to_string()), } }); } /// CPU-bound layers + SHA-256. Pure sync; safe to call from `spawn_blocking`. fn run_sync_layers(&self, data: &[u8], file_type: FileType) -> (Vec, String) { let mut layers = Vec::with_capacity(5); // SHA-256 hash for audit + MalwareBazaar lookup let sha256 = { let mut hasher = Sha256::new(); hasher.update(data); hex::encode(hasher.finalize()) }; layers.push(content_type::verify_content_type(data, file_type)); layers.push(structural::analyze_binary(data, file_type)); // Single-pass archive inspection: the bomb-defense walk ("archive") and // the recursive interior content scan ("archive_nested") are derived from // ONE decompression of each entry. Previously each entry was decompressed // twice, once to count for bomb defense, once to buffer+scan its content // (Run #2 Performance SERIOUS). No-op (Skip/Skip) for non-archives. let (archive_layer, archive_nested_layer) = archive::inspect_archive(data, file_type, self.yara_rules.as_ref()); layers.push(archive_layer); layers.push(archive_nested_layer); layers.push(match self.yara_rules { Some(ref rules) => { // Cap YARA's input: it walks the whole slice, faulting the entire // mmap resident. ClamAV (streamed, uncapped) is the full-file // backstop, so scanning a generous prefix bounds peak RAM without // surrendering the floor. if data.len() > crate::constants::SCAN_YARA_MAX_BYTES { if yara_tail_unscanned(data.len(), self.clamav_max_scan_bytes) { // No full-file backstop reaches this size, so the bytes // past the YARA prefix would go entirely unscanned, a // tail-of-file evasion (append the payload past the cap // and it passes Clean). This is true both when ClamAV is // absent and when it is present but its declared // MaxScanSize doesn't cover the file (Run #24 Security). // Hold the file rather than certify an unscanned tail // (ultra-fuzz Run 13 Storage). `yara` is FailClosed, so // this Error → HeldForReview. tracing::warn!( full_bytes = data.len(), capped_bytes = crate::constants::SCAN_YARA_MAX_BYTES, "YARA input would be capped but no ClamAV full-file backstop is configured; holding file (tail unscanned)" ); LayerResult { layer: "yara", verdict: LayerVerdict::Error, detail: Some(format!( "tail unscanned: {} B exceeds the {} B YARA prefix and no ClamAV full-file scan is configured", data.len(), crate::constants::SCAN_YARA_MAX_BYTES )), } } else { tracing::warn!( full_bytes = data.len(), capped_bytes = crate::constants::SCAN_YARA_MAX_BYTES, "YARA input capped; scanning prefix only (ClamAV scans the full file)" ); yara::scan_with_yara(rules, &data[..crate::constants::SCAN_YARA_MAX_BYTES]) } } else { yara::scan_with_yara(rules, data) } } None => LayerResult { layer: "yara", verdict: LayerVerdict::Skip, detail: Some("No YARA rules loaded".to_string()), }, }); layers.push(signing_macos::verify_apple_signature(data, file_type)); layers.push(signing_windows::verify_authenticode(data, file_type)); layers.push(signing_linux::verify_appimage_signature(data, file_type)); (layers, sha256) } } #[cfg(test)] mod tests { use super::*; #[test] fn layer_verdict_serializes_lowercase() { assert_eq!( serde_json::to_string(&LayerVerdict::Pass).unwrap(), "\"pass\"" ); assert_eq!( serde_json::to_string(&LayerVerdict::Fail).unwrap(), "\"fail\"" ); assert_eq!( serde_json::to_string(&LayerVerdict::Skip).unwrap(), "\"skip\"" ); assert_eq!( serde_json::to_string(&LayerVerdict::Error).unwrap(), "\"error\"" ); } #[test] fn scan_result_quarantined_on_any_fail() { let layers = [ LayerResult { layer: "test1", verdict: LayerVerdict::Pass, detail: None, }, LayerResult { layer: "test2", verdict: LayerVerdict::Fail, detail: Some("bad".to_string()), }, ]; let has_fail = layers.iter().any(|l| l.verdict == LayerVerdict::Fail); assert!(has_fail); } #[test] fn panicked_sync_layer_is_held_for_review_not_clean() { // A file that panics a CPU parser must never come back Clean, it is // held for admin review, and the panic is contained in a returnable // result (not an `.expect` that would unwind the scan worker). let r = panicked_sync_result(4096); assert_eq!(r.status, FileScanStatus::HeldForReview); assert_eq!(r.file_size, 4096); assert!(r.sha256.is_empty()); assert_eq!(r.layers.len(), 1); assert_eq!(r.layers[0].verdict, LayerVerdict::Error); // The synthetic layer must resolve to a FailClosed policy (the default // for unregistered names), which is what makes final_status hold it. assert_eq!(error_policy_for(r.layers[0].layer), ErrorPolicy::FailClosed); } #[test] fn oversize_file_is_held_for_review_not_failed() { // A file above the spool ceiling is held for admin review (explicit, // documented policy) rather than failing the job and stranding the // upload in Pending. Verdict resolves to FailClosed -> HeldForReview. let huge = crate::constants::SCAN_SPOOL_MAX_BYTES + 1; let r = too_large_to_scan(huge); assert_eq!(r.status, FileScanStatus::HeldForReview); assert_eq!(r.file_size, huge); assert!(r.sha256.is_empty()); assert_eq!(r.layers[0].layer, "scan_size_limit"); assert_eq!(error_policy_for(r.layers[0].layer), ErrorPolicy::FailClosed); } #[test] fn sha256_computation() { let mut hasher = Sha256::new(); hasher.update(b"hello"); let hash = hex::encode(hasher.finalize()); assert_eq!( hash, "2cf24dba5fb0a30e26e83b2ac5b9e29e1b161e5c1fa7425e73043362938b9824" ); } // Pipeline integration tests /// Create a minimal ScanPipeline with no external deps (no YARA, no ClamAV, no MalwareBazaar). /// Wrapped in `Arc` because `scan` consumes `Arc` (see `pub async fn scan`). fn make_pipeline() -> std::sync::Arc { std::sync::Arc::new(ScanPipeline { yara_rules: None, yara_rule_count: 0, yara_min_rule_files: 0, clamav_socket: None, clamav_max_scan_bytes: None, malwarebazaar_enabled: false, urlhaus_enabled: false, abuse_ch_auth_key: None, metadefender_api_key: None, }) } /// SEC-S2: a corpus that compiled fewer rule files than the configured floor /// is degraded coverage masquerading as a live layer, boot must fail closed. #[tokio::test] async fn assert_live_refuses_degraded_yara_corpus() { let mut compiler = yara_x::Compiler::new(); compiler .add_source(r#"rule r { strings: $a = "x" condition: $a }"#) .unwrap(); let pipeline = std::sync::Arc::new(ScanPipeline { yara_rules: Some(compiler.build()), yara_rule_count: 1, yara_min_rule_files: 2, clamav_socket: None, clamav_max_scan_bytes: None, malwarebazaar_enabled: false, urlhaus_enabled: false, abuse_ch_auth_key: None, metadefender_api_key: None, }); let err = pipeline .assert_live() .await .expect_err("degraded corpus must refuse boot"); assert!( err.contains("YARA corpus degraded"), "unexpected error: {err}" ); } /// The complement: a corpus meeting the floor boots, with YARA counted live. #[tokio::test] async fn assert_live_accepts_corpus_meeting_floor() { let mut compiler = yara_x::Compiler::new(); compiler .add_source(r#"rule r { strings: $a = "x" condition: $a }"#) .unwrap(); let pipeline = std::sync::Arc::new(ScanPipeline { yara_rules: Some(compiler.build()), yara_rule_count: 3, yara_min_rule_files: 3, clamav_socket: None, clamav_max_scan_bytes: None, malwarebazaar_enabled: false, urlhaus_enabled: false, abuse_ch_auth_key: None, metadefender_api_key: None, }); pipeline .assert_live() .await .expect("a corpus meeting the floor must boot"); } #[tokio::test] async fn pipeline_clean_download_passes() { let pipeline = make_pipeline(); let result = pipeline .clone() .scan(b"just some file content".to_vec(), FileType::Download) .await; assert_eq!(result.status, FileScanStatus::Clean); assert_eq!(result.file_size, 22); assert!(!result.sha256.is_empty()); assert_eq!(result.layers.len(), 12); } #[tokio::test] async fn pipeline_unrecognized_audio_quarantined() { let pipeline = make_pipeline(); // Unrecognized data claimed as audio should be rejected by content_type layer let result = pipeline .clone() .scan(b"audio data here".to_vec(), FileType::Audio) .await; assert_eq!(result.status, FileScanStatus::Quarantined); } #[tokio::test] async fn pipeline_clean_cover_passes() { let pipeline = make_pipeline(); // PNG magic bytes let png = [0x89, 0x50, 0x4E, 0x47, 0x0D, 0x0A, 0x1A, 0x0A]; let result = pipeline.clone().scan(png.to_vec(), FileType::Cover).await; assert_eq!(result.status, FileScanStatus::Clean); } #[tokio::test] async fn pipeline_pe_as_audio_quarantined() { let pipeline = make_pipeline(); // PE magic bytes, content-type layer should detect application/* and fail let pe_header = b"MZ\x90\x00\x03\x00\x00\x00"; let result = pipeline .clone() .scan(pe_header.to_vec(), FileType::Audio) .await; assert_eq!(result.status, FileScanStatus::Quarantined); // Verify content_type layer produced the fail let content_type_layer = result .layers .iter() .find(|l| l.layer == "content_type") .unwrap(); assert_eq!(content_type_layer.verdict, LayerVerdict::Fail); } #[tokio::test] async fn pipeline_pe_as_cover_quarantined() { let pipeline = make_pipeline(); let pe_header = b"MZ\x90\x00\x03\x00\x00\x00"; let result = pipeline .clone() .scan(pe_header.to_vec(), FileType::Cover) .await; assert_eq!(result.status, FileScanStatus::Quarantined); } #[tokio::test] async fn pipeline_sha256_is_deterministic() { let pipeline = make_pipeline(); let data = b"deterministic hash test"; let r1 = pipeline .clone() .scan(data.to_vec(), FileType::Download) .await; let r2 = pipeline .clone() .scan(data.to_vec(), FileType::Download) .await; assert_eq!(r1.sha256, r2.sha256); } #[tokio::test] async fn pipeline_skips_optional_layers_when_unconfigured() { let pipeline = make_pipeline(); let result = pipeline .clone() .scan(b"test".to_vec(), FileType::Download) .await; let yara = result.layers.iter().find(|l| l.layer == "yara").unwrap(); assert_eq!(yara.verdict, LayerVerdict::Skip); let clamav = result.layers.iter().find(|l| l.layer == "clamav").unwrap(); assert_eq!(clamav.verdict, LayerVerdict::Skip); let mb = result .layers .iter() .find(|l| l.layer == "malwarebazaar") .unwrap(); assert_eq!(mb.verdict, LayerVerdict::Skip); let uh = result.layers.iter().find(|l| l.layer == "urlhaus").unwrap(); assert_eq!(uh.verdict, LayerVerdict::Skip); } #[tokio::test] async fn pipeline_always_produces_12_layers() { let pipeline = make_pipeline(); for file_type in [FileType::Audio, FileType::Cover, FileType::Download] { let result = pipeline.clone().scan(b"data".to_vec(), file_type).await; // 11 base layers + the recursive archive-interior layer (archive_nested). assert_eq!( result.layers.len(), 12, "Expected 12 layers for {file_type:?}" ); } } #[test] fn suspicion_present_on_fail() { let layers = vec![pass("content_type"), fail("yara")]; assert!(suspicion_present(&layers)); } #[test] fn suspicion_present_on_fail_closed_error() { let layers = vec![pass("content_type"), err("archive")]; assert!(suspicion_present(&layers)); } #[test] fn no_suspicion_when_fail_open_error_only() { // External-layer errors are operational noise, not malware signals; // they must not invoke MetaDefender. let layers = vec![pass("content_type"), err("malwarebazaar"), err("urlhaus")]; assert!(!suspicion_present(&layers)); } #[test] fn no_suspicion_when_all_clean() { let layers = vec![pass("content_type"), skip("yara"), pass("structural")]; assert!(!suspicion_present(&layers)); } #[tokio::test] async fn pipeline_errors_held_for_review() { // Errors from fail-closed layers (archive is in-process deterministic) // should hold the file for admin review. let pipeline = make_pipeline(); // Corrupted ZIP magic bytes, archive layer returns Error let mut data = vec![0x50, 0x4B, 0x03, 0x04]; data.extend_from_slice(&[0xFF; 100]); let result = pipeline.clone().scan(data, FileType::Download).await; let archive = result.layers.iter().find(|l| l.layer == "archive").unwrap(); assert_eq!(archive.verdict, LayerVerdict::Error); assert_eq!(result.status, FileScanStatus::HeldForReview); } // Per-layer fail policy tests fn err(layer: &'static str) -> LayerResult { LayerResult { layer, verdict: LayerVerdict::Error, detail: None, } } fn pass(layer: &'static str) -> LayerResult { LayerResult { layer, verdict: LayerVerdict::Pass, detail: None, } } fn skip(layer: &'static str) -> LayerResult { LayerResult { layer, verdict: LayerVerdict::Skip, detail: None, } } fn fail(layer: &'static str) -> LayerResult { LayerResult { layer, verdict: LayerVerdict::Fail, detail: None, } } #[test] fn yara_tail_unscanned_only_without_backstop() { use crate::constants::SCAN_YARA_MAX_BYTES; let over = SCAN_YARA_MAX_BYTES + 1; // Over the cap, no declared ClamAV coverage → tail unscanned, must hold. assert!(yara_tail_unscanned(over, None)); // Over the cap, ClamAV coverage present but SHORT of the file (Run #24 // MODERATE: clamd reachable but MaxScanSize doesn't reach) → still hold. assert!(yara_tail_unscanned(over, Some(over as u64 - 1))); // Over the cap, declared coverage reaches the file → ClamAV is a real // full-file backstop, don't hold. assert!(!yara_tail_unscanned(over, Some(over as u64))); assert!(!yara_tail_unscanned(over, Some(u64::MAX))); // Within the YARA cap → whole file scanned by YARA regardless of ClamAV. assert!(!yara_tail_unscanned(SCAN_YARA_MAX_BYTES, None)); assert!(!yara_tail_unscanned(1024, None)); } #[test] fn final_status_clean_when_all_pass() { let layers = vec![ pass("content_type"), pass("structural"), pass("archive"), skip("yara"), skip("clamav"), skip("malwarebazaar"), ]; assert_eq!(final_status(&layers), FileScanStatus::Clean); } #[test] fn final_status_quarantined_on_any_fail() { let layers = vec![pass("content_type"), fail("yara"), skip("clamav")]; assert_eq!(final_status(&layers), FileScanStatus::Quarantined); } #[test] fn final_status_fail_beats_error() { // A Fail anywhere supersedes any Error, regardless of policy. let layers = vec![err("malwarebazaar"), fail("yara")]; assert_eq!(final_status(&layers), FileScanStatus::Quarantined); } #[test] fn final_status_held_on_fail_closed_error() { // archive is FailClosed, its Error must hold the file. let layers = vec![pass("content_type"), err("archive"), skip("clamav")]; assert_eq!(final_status(&layers), FileScanStatus::HeldForReview); } #[test] fn final_status_clean_on_fail_open_error_only() { // malwarebazaar is FailOpen, its Error must NOT hold the file. // This is the regression of 2026-05-10 that motivated the audit. let layers = vec![ pass("content_type"), pass("structural"), pass("archive"), skip("yara"), skip("clamav"), err("malwarebazaar"), ]; assert_eq!(final_status(&layers), FileScanStatus::Clean); } #[test] fn final_status_clean_when_all_external_layers_error() { // Worst-case external-services outage: every network/daemon layer // erroring at once. As long as the in-process layers pass, the file // is Clean. Health is surfaced separately via per-layer monitoring. let layers = vec![ pass("content_type"), pass("structural"), pass("archive"), skip("yara"), err("clamav"), err("malwarebazaar"), ]; assert_eq!(final_status(&layers), FileScanStatus::Clean); } #[test] fn clamav_incomplete_is_fail_closed_and_holds() { // CHRONIC S1: a reachable-but-incomplete clamav scan is emitted under the // `clamav_incomplete` layer, which must be FailClosed → HeldForReview, // distinct from the FailOpen `clamav` layer used for an unreachable daemon. assert_eq!( error_policy_for("clamav_incomplete"), ErrorPolicy::FailClosed ); assert_eq!(error_policy_for("clamav"), ErrorPolicy::FailOpen); let layers = vec![ pass("content_type"), pass("structural"), pass("archive"), skip("yara"), err("clamav_incomplete"), ]; assert_eq!(final_status(&layers), FileScanStatus::HeldForReview); } #[test] fn final_status_held_on_unknown_layer_error() { // Defensive default: an unknown layer name that errors falls through // to FailClosed. This is what catches a new layer added without // wiring its policy into `error_policy_for`. let layers = vec![ pass("content_type"), err("brand_new_layer_someone_forgot_to_register"), ]; assert_eq!(final_status(&layers), FileScanStatus::HeldForReview); } #[test] fn error_policy_for_all_known_layers() { // Every layer name produced by the pipeline must have an explicit // declaration in `error_policy_for`. The default branch is reserved // for genuine programmer error (new layer, forgot to register). for name in [ "content_type", "structural", "archive", "yara", "clamav", "malwarebazaar", ] { let policy = error_policy_for(name); // Both values are valid; we just want this to not hit the default. // If a layer is renamed without updating `error_policy_for`, this // test still passes (the rename produces a new unknown name) //, but the per-layer name tests below catch that. let _ = policy; } } #[test] fn content_type_is_fail_closed() { assert_eq!(error_policy_for("content_type"), ErrorPolicy::FailClosed); } #[test] fn structural_is_fail_closed() { assert_eq!(error_policy_for("structural"), ErrorPolicy::FailClosed); } #[test] fn archive_is_fail_closed() { assert_eq!(error_policy_for("archive"), ErrorPolicy::FailClosed); } #[test] fn yara_is_fail_closed() { assert_eq!(error_policy_for("yara"), ErrorPolicy::FailClosed); } #[test] fn clamav_is_fail_open() { assert_eq!(error_policy_for("clamav"), ErrorPolicy::FailOpen); } #[test] fn malwarebazaar_is_fail_open() { assert_eq!(error_policy_for("malwarebazaar"), ErrorPolicy::FailOpen); } #[test] fn urlhaus_is_fail_open() { assert_eq!(error_policy_for("urlhaus"), ErrorPolicy::FailOpen); } #[test] fn signing_macos_is_fail_open() { assert_eq!(error_policy_for("signing_macos"), ErrorPolicy::FailOpen); } #[test] fn metadefender_is_fail_open() { assert_eq!(error_policy_for("metadefender"), ErrorPolicy::FailOpen); } #[test] fn signing_windows_is_fail_open() { assert_eq!(error_policy_for("signing_windows"), ErrorPolicy::FailOpen); } #[test] fn signing_linux_is_fail_open() { assert_eq!(error_policy_for("signing_linux"), ErrorPolicy::FailOpen); } }