Skip to main content

max / makenotwork

56.0 KB · 1402 lines History Blame Raw
1 //! 6-layer malware scanning pipeline for file uploads.
2 //!
3 //! Layers 1-4 always run (in-process, deterministic). Layers 5-6 are optional
4 //! (external services).
5 //!
6 //! **Error policy is per-layer**, declared at each layer's source file as
7 //! `pub const ERROR_POLICY`. The aggregator `final_status` consults each
8 //! layer's policy via `error_policy_for`. In-process layers are `FailClosed`
9 //! (an error is a structural defect); external layers are `FailOpen` (an
10 //! error is an outage that must not block the platform). See
11 //! `docs/scan-pipeline-audit.md` for the rationale.
12 //!
13 //! See also: `/docs/tech/content-protection`
14
15 pub mod archive;
16 pub mod clamav;
17 pub mod content_type;
18 pub mod hash_lookup;
19 pub mod metadefender;
20 pub mod signing_linux;
21 pub mod signing_macos;
22 pub mod signing_windows;
23 pub mod spool;
24 pub mod structural;
25 pub mod urlhaus;
26 pub mod worker;
27 pub mod yara;
28
29 use serde::Serialize;
30 use sha2::{Digest, Sha256};
31
32 use crate::config::ScanConfig;
33 use crate::db::FileScanStatus;
34 use crate::storage::FileType;
35
36 /// Per-layer scan verdict
37 #[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize)]
38 #[serde(rename_all = "lowercase")]
39 pub enum LayerVerdict {
40 Pass,
41 Fail,
42 Skip,
43 Error,
44 }
45
46 /// Policy for how a layer's `Error` verdict feeds into the pipeline's final status.
47 ///
48 /// - `FailClosed`, an `Error` from this layer holds the upload for admin review.
49 /// Appropriate for deterministic in-process layers where an `Error` indicates a
50 /// real bug or a structurally suspicious file.
51 /// - `FailOpen`, an `Error` from this layer is treated as `Skip` for aggregation.
52 /// Appropriate for external services (network, daemons) where an outage on a
53 /// third party must not take down the platform's upload pipeline.
54 ///
55 /// Each layer declares its own `ERROR_POLICY` const; the aggregator in
56 /// `ScanPipeline::final_status` consults the declaration via `error_policy_for`.
57 #[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize)]
58 #[serde(rename_all = "snake_case")]
59 pub enum ErrorPolicy {
60 FailClosed,
61 FailOpen,
62 }
63
64 /// Result from a single scanning layer
65 #[derive(Debug, Clone, Serialize)]
66 pub struct LayerResult {
67 pub layer: &'static str,
68 pub verdict: LayerVerdict,
69 pub detail: Option<String>,
70 }
71
72 /// Look up a layer's declared error policy by name. Defaults to `FailClosed`
73 /// for unknown layers, a defensive choice that surfaces uninstrumented
74 /// additions during testing rather than silently fail-opening them.
75 fn error_policy_for(layer: &str) -> ErrorPolicy {
76 match layer {
77 "content_type" => content_type::ERROR_POLICY,
78 "structural" => structural::ERROR_POLICY,
79 "archive" => archive::ERROR_POLICY,
80 // Recursive interior scan of an archive's entries. A decompressed entry
81 // we could not fully inspect (un-openable, over budget, or nested deeper
82 // than the scan depth) errors here and must fail closed, a payload
83 // hidden in a zip-in-a-zip is held for review, never passed Clean. This
84 // is the structural close of the nested-archive recursion chronic.
85 "archive_nested" => ErrorPolicy::FailClosed,
86 "yara" => yara::ERROR_POLICY,
87 "clamav" => clamav::ERROR_POLICY,
88 // A *reachable* clamd that couldn't fully scan (size/scan limit,
89 // unparseable reply) is a coverage gap, not an outage, fail closed,
90 // distinct from the FailOpen `clamav` layer used for an unreachable
91 // daemon. Splitting the policy by layer identity is what keeps CHRONIC
92 // S1 (size-limit → FailOpen → Clean) from recurring.
93 "clamav_incomplete" => ErrorPolicy::FailClosed,
94 "malwarebazaar" => hash_lookup::ERROR_POLICY,
95 "urlhaus" => urlhaus::ERROR_POLICY,
96 "signing_macos" => signing_macos::ERROR_POLICY,
97 "signing_windows" => signing_windows::ERROR_POLICY,
98 "signing_linux" => signing_linux::ERROR_POLICY,
99 "metadefender" => metadefender::ERROR_POLICY,
100 other => {
101 tracing::error!(
102 layer = other,
103 "unknown scan layer; defaulting to FailClosed"
104 );
105 ErrorPolicy::FailClosed
106 }
107 }
108 }
109
110 /// Decide whether the second-opinion (MetaDefender) layer should run, based
111 /// on the verdicts of layers that have already completed. Any `Fail`, or any
112 /// `Error` from a fail-closed in-process layer, counts as "suspicious enough
113 /// to escalate". Pure `Error`s from fail-open external layers do not, those
114 /// are operational noise, not malware signals.
115 fn suspicion_present(layers: &[LayerResult]) -> bool {
116 layers.iter().any(|l| match l.verdict {
117 LayerVerdict::Fail => true,
118 LayerVerdict::Error => error_policy_for(l.layer) == ErrorPolicy::FailClosed,
119 _ => false,
120 })
121 }
122
123 /// Run an external hash-lookup layer under the aggregate external-lookup deadline,
124 /// degrading to a Skip (FailOpen) on timeout. Shared by the buffered `scan()` and
125 /// the streaming `scan_stream()` so BOTH paths get the same defense-in-depth
126 /// deadline (Perf-S1), previously only `scan_stream` wrapped these calls, so a
127 /// small upload had only the per-request client timeout with no aggregate ceiling.
128 async fn lookup_with_timeout(
129 layer: &'static str,
130 fut: impl std::future::Future<Output = LayerResult>,
131 ) -> LayerResult {
132 let lookup_timeout =
133 std::time::Duration::from_secs(crate::constants::SCAN_EXTERNAL_LOOKUP_TIMEOUT_SECS);
134 match tokio::time::timeout(lookup_timeout, fut).await {
135 Ok(layer_result) => layer_result,
136 Err(_) => {
137 tracing::warn!("{layer} lookup timed out; treating as Skip (FailOpen)");
138 LayerResult {
139 layer,
140 verdict: LayerVerdict::Skip,
141 detail: Some(format!("{layer} lookup timed out")),
142 }
143 }
144 }
145 }
146
147 /// Aggregate per-layer results into a final scan status.
148 ///
149 /// - Any layer `Fail` → `Quarantined` (terminal).
150 /// - Any layer `Error` whose policy is `FailClosed` → `HeldForReview`.
151 /// - `FailOpen` errors are treated as Skip-equivalent for aggregation; they're
152 /// still surfaced in the per-layer detail so admins and PoM can see degraded
153 /// layers in the dashboard.
154 /// - Otherwise → `Clean`.
155 fn final_status(layers: &[LayerResult]) -> FileScanStatus {
156 if layers.iter().any(|l| l.verdict == LayerVerdict::Fail) {
157 return FileScanStatus::Quarantined;
158 }
159 let has_fail_closed_error = layers.iter().any(|l| {
160 l.verdict == LayerVerdict::Error && error_policy_for(l.layer) == ErrorPolicy::FailClosed
161 });
162 if has_fail_closed_error {
163 FileScanStatus::HeldForReview
164 } else {
165 FileScanStatus::Clean
166 }
167 }
168
169 /// Whether a file's tail would go unscanned, true when it exceeds the YARA
170 /// prefix cap ([`crate::constants::SCAN_YARA_MAX_BYTES`]) and ClamAV does not
171 /// provide a *full-file* backstop for it. Such a file must be held for review
172 /// rather than certified Clean on a prefix-only scan.
173 ///
174 /// ClamAV is a full-file backstop only up to the operator-declared
175 /// [`ScanConfig::clamav_max_scan_bytes`]: clamd silently scans only the first
176 /// `MaxScanSize`/`MaxFileSize` bytes and returns `OK` with no error, so trusting
177 /// "socket configured" alone let a large file with a payload past the YARA
178 /// prefix certify Clean (ultra-fuzz Run #24 Security MODERATE). `None` (coverage
179 /// undeclared) is fail-closed: no backstop, hold for review.
180 fn yara_tail_unscanned(data_len: usize, clamav_backstop_bytes: Option<u64>) -> bool {
181 if data_len <= crate::constants::SCAN_YARA_MAX_BYTES {
182 return false;
183 }
184 // Backstop only covers the file if the operator declared coverage that
185 // reaches its size. Undeclared coverage does NOT count.
186 let covered = clamav_backstop_bytes.is_some_and(|max| data_len as u64 <= max);
187 !covered
188 }
189
190 /// Fail-closed result for when a CPU scan layer panics on crafted input.
191 ///
192 /// The in-process parsers (goblin, yara-x, zip, content-type) run over fully
193 /// attacker-controlled bytes; a malformed file can panic them. Such a file is
194 /// held for admin review, never passed as `Clean`. Critically, building a
195 /// result here lets the caller `return` normally instead of `.expect()`-ing the
196 /// join handle and unwinding the scan-worker task, the worker pool is spawned
197 /// once and never respawns, so a propagated panic would permanently shrink it
198 /// (two crafted uploads could disable scanning entirely). `scan_panic` is not a
199 /// registered layer, so `error_policy_for` defaults it to `FailClosed`.
200 fn panicked_sync_result(file_size: u64) -> ScanResult {
201 let layer = LayerResult {
202 layer: "scan_panic",
203 verdict: LayerVerdict::Error,
204 detail: Some("a CPU scan layer panicked on this input".to_string()),
205 };
206 ScanResult {
207 status: final_status(std::slice::from_ref(&layer)),
208 layers: vec![layer],
209 sha256: String::new(),
210 file_size,
211 }
212 }
213
214 /// Fail-closed result for a scan whose CPU layers exceeded the wall-clock
215 /// deadline (`SCAN_CPU_LAYERS_TIMEOUT_SECS`), e.g. a pathological slow-codec
216 /// archive decompress. Held for review rather than passed, and the worker is
217 /// freed so a crafted upload can't monopolize the two-worker scan pool
218 /// (fuzz 2026-07-06 F1).
219 fn timed_out_sync_result(file_size: u64) -> ScanResult {
220 let layer = LayerResult {
221 layer: "scan_timeout",
222 verdict: LayerVerdict::Error,
223 detail: Some("CPU scan layers exceeded the wall-clock deadline".to_string()),
224 };
225 ScanResult {
226 status: final_status(std::slice::from_ref(&layer)),
227 layers: vec![layer],
228 sha256: String::new(),
229 file_size,
230 }
231 }
232
233 /// Fail-closed result for a file too large to spool for scanning.
234 ///
235 /// Files above [`crate::constants::SCAN_SPOOL_MAX_BYTES`] cannot be spooled to
236 /// disk for the CPU layers (the platform accepts videos up to
237 /// `MAX_VIDEO_SIZE`, which exceeds the spool ceiling). Rather than fail the scan
238 /// job, which left the upload stuck `Pending` and retried forever by the
239 /// reaper, hold it for admin review under an explicit, documented policy. If
240 /// auto-scanning large videos is wanted instead, raise `SCAN_SPOOL_MAX_BYTES`
241 /// to cover the max accepted upload (at a proportional scratch-disk cost).
242 fn too_large_to_scan(file_size: u64) -> ScanResult {
243 let layer = LayerResult {
244 layer: "scan_size_limit",
245 verdict: LayerVerdict::Error,
246 detail: Some(format!(
247 "file size {file_size} exceeds the scan spool ceiling ({} bytes); held for review",
248 crate::constants::SCAN_SPOOL_MAX_BYTES
249 )),
250 };
251 ScanResult {
252 status: final_status(std::slice::from_ref(&layer)),
253 layers: vec![layer],
254 sha256: String::new(),
255 file_size,
256 }
257 }
258
259 /// Scan-then-promote closing move (C1): copy a Clean object from its unserved
260 /// `staging/{uuid}` key to the immutable, content-addressed
261 /// `{owner}/c/{sha256}.{ext}` key, repoint the entity (and, for CDN images, its
262 /// materialized public URL) at the content key, mark it Clean, and enqueue the
263 /// staging object for durable deletion. The bytes a buyer is served are then
264 /// provably the bytes that were scanned: the content key is named by the scanned
265 /// hash, the owner holds no presign to it, and the (owner-less, random) staging
266 /// key they *can* re-PUT to is unserved and deleted.
267 ///
268 /// Shared by the scan worker's Clean path and the admin approve-held path so the
269 /// two promote sites can't diverge. Ordering is fail-safe: the S3 copy runs
270 /// first, then a single transaction repoints the row and enqueues the staging
271 /// delete. If the copy or the DB write fails, the row keeps pointing at the
272 /// still-present (unserved) staging key and the caller leaves the work un-done,
273 /// so a retry re-promotes, the copy is idempotent (hash-named destination), and
274 /// an already-promoted `{owner}/c/...` key short-circuits.
275 #[allow(clippy::too_many_arguments)]
276 pub async fn promote_staging_to_content(
277 db: &sqlx::PgPool,
278 backend: &dyn crate::storage::StorageBackend,
279 public_backend: Option<&dyn crate::storage::StorageBackend>,
280 cdn_base_url: &str,
281 kind: crate::db::scan_jobs::ScanTargetKind,
282 file_type: FileType,
283 target_id: uuid::Uuid,
284 owner: crate::db::UserId,
285 staging_key: &str,
286 sha256: &str,
287 bucket: crate::storage::S3Bucket,
288 ) -> crate::error::Result<()> {
289 use crate::db::scan_jobs::ScanTargetKind;
290 use crate::storage::{S3Client, S3Key};
291
292 if sha256.is_empty() {
293 return Err(crate::error::AppError::Storage(format!(
294 "cannot promote {staging_key}: scan recorded no content hash"
295 )));
296 }
297 // Already promoted (worker/admin retry, admin bulk over a mixed set): a
298 // content key is `{owner}/c/...`, never `staging/...`. Nothing to copy or repoint.
299 if !staging_key.starts_with("staging/") {
300 return Ok(());
301 }
302
303 let ext = crate::storage::key_extension(staging_key);
304 let content = S3Client::content_key(owner, sha256, ext);
305 let content_str = content.as_str().to_string();
306
307 // The three CDN-unsigned image kinds have their content object served from
308 // the PUBLIC bucket; everything else (gated media, presigned insertions,
309 // OTA) keeps its content object in the same private bucket the staging
310 // object lives in. Staging is always private, so a public-bucket promote is
311 // a CROSS-bucket copy (private staging -> public content) issued on the
312 // public backend with the private bucket as the copy source.
313 let to_public = kind.content_served_from_public_bucket();
314
315 let staging = S3Key::from_stored(staging_key);
316
317 // The source size picks the copy strategy: S3 rejects a single-part
318 // `CopyObject` above 5 GiB, so a large source has to go through ranged
319 // multipart `UploadPartCopy` or the promote fails *after* a successful
320 // upload and scan, the worst position to fail in. The S3 object is the
321 // authoritative size (same posture as blob confirm), so read it here rather
322 // than threading a recorded size through all three call sites; one
323 // HeadObject is negligible next to the object copy that follows. The source
324 // always lives in `backend`'s (private, staging) bucket, including on the
325 // cross-bucket public promote.
326 let src_size = backend
327 .object_size(staging_key)
328 .await?
329 .ok_or_else(|| {
330 crate::error::AppError::Storage(format!(
331 "cannot promote {staging_key}: staging object not found in storage"
332 ))
333 })?
334 .max(0) as u64;
335 let needs_multipart = src_size > crate::constants::S3_SINGLE_COPY_MAX_BYTES;
336 // Only consulted on the multipart branch; a single CopyObject carries the
337 // source's content type across on its own.
338 let content_type = crate::storage::content_type_for(file_type, ext);
339
340 // 1. Promote the object in storage (copy staging -> content). Idempotent.
341 if to_public {
342 let public = public_backend.ok_or_else(|| {
343 crate::error::AppError::Storage(format!(
344 "cannot promote CDN image {staging_key}: public bucket not configured"
345 ))
346 })?;
347 if needs_multipart {
348 public
349 .copy_object_multipart(
350 backend.bucket(),
351 &staging,
352 &content,
353 content_type,
354 src_size,
355 None,
356 )
357 .await?;
358 } else {
359 public
360 .copy_object_from(backend.bucket(), &staging, &content)
361 .await?;
362 }
363 } else if needs_multipart {
364 // Same-bucket multipart promote: the source bucket is this backend's own.
365 backend
366 .copy_object_multipart(
367 backend.bucket(),
368 &staging,
369 &content,
370 content_type,
371 src_size,
372 None,
373 )
374 .await?;
375 } else {
376 backend.copy_object(&staging, &content).await?;
377 }
378
379 // 2. Repoint the row + enqueue the staging delete atomically.
380 let mut tx = db.begin().await?;
381 if kind.is_cdn_served_without_gate() {
382 let content_url = if matches!(kind, ScanTargetKind::ContentInsertion) {
383 // No materialized URL column (insertions are served presigned).
384 String::new()
385 } else {
386 // Content object is in the public bucket, which is exactly what the
387 // CDN base fronts, so the URL is `{cdn}/{content_key}`.
388 crate::storage::build_project_image_url(cdn_base_url, &content_str)
389 };
390 crate::db::scanning::promote_cdn_image_by_key(
391 &mut tx,
392 staging_key,
393 &content_str,
394 &content_url,
395 )
396 .await?;
397 } else {
398 crate::db::scanning::promote_gated(&mut *tx, kind, file_type, target_id, &content_str)
399 .await?;
400 }
401 crate::db::pending_s3_deletions::enqueue_deletions(
402 &mut *tx,
403 &[(staging_key.to_string(), bucket.as_str().to_string())],
404 "scan_promote_staging",
405 )
406 .await?;
407 tx.commit().await?;
408 Ok(())
409 }
410
411 /// Aggregate scan result across all layers
412 #[derive(Debug, Clone)]
413 pub struct ScanResult {
414 pub status: FileScanStatus,
415 pub layers: Vec<LayerResult>,
416 pub sha256: String,
417 pub file_size: u64,
418 }
419
420 /// Pre-compiled scanning pipeline. Initialized once at startup and shared via Arc.
421 pub struct ScanPipeline {
422 yara_rules: Option<yara_x::Rules>,
423 /// Number of YARA rule files that compiled, and the configured health floor.
424 yara_rule_count: usize,
425 yara_min_rule_files: usize,
426 clamav_socket: Option<String>,
427 /// Operator-declared ClamAV per-object scan coverage; see
428 /// [`ScanConfig::clamav_max_scan_bytes`]. `None` = coverage unknown =
429 /// ClamAV is not treated as a full-file backstop (fail-closed).
430 clamav_max_scan_bytes: Option<u64>,
431 malwarebazaar_enabled: bool,
432 urlhaus_enabled: bool,
433 abuse_ch_auth_key: Option<String>,
434 metadefender_api_key: Option<String>,
435 }
436
437 /// How the bytes to scan are sourced. This is the ONLY thing that differs between
438 /// the buffered and streaming scan paths, all scan POLICY (CPU layers off the
439 /// runtime, ClamAV, URLhaus host extraction, external lookups, byte caps) runs in
440 /// the single private [`ScanPipeline::run_scan`], so a policy cannot drift between
441 /// the two entry points (ultra-fuzz CHRONIC: scan()/scan_stream() twin divergence,
442 /// closed Run 9).
443 enum ScanInput {
444 /// Whole object held in memory, small uploads.
445 Buffered(bytes::Bytes),
446 /// Object spooled to a tempfile and scanned via a memory map. The retained
447 /// `SpoolHandle` keeps the file on disk (and unlinks it on drop) so ClamAV can
448 /// stream it by path while the CPU layers read the map.
449 Spooled {
450 map: std::sync::Arc<memmap2::Mmap>,
451 spool: spool::SpoolHandle,
452 },
453 }
454
455 /// A cheaply-cloneable, `Send + 'static` view of the scan bytes for the
456 /// `spawn_blocking` CPU work (a `Bytes` refcount bump or an `Arc<Mmap>` clone).
457 #[derive(Clone)]
458 enum ScanBytes {
459 Buffered(bytes::Bytes),
460 Mapped(std::sync::Arc<memmap2::Mmap>),
461 }
462
463 impl ScanBytes {
464 fn as_slice(&self) -> &[u8] {
465 match self {
466 ScanBytes::Buffered(b) => b,
467 ScanBytes::Mapped(m) => &m[..],
468 }
469 }
470 }
471
472 /// Where ClamAV reads its bytes: the in-memory buffer, or the spool path it
473 /// streams via INSTREAM frames.
474 enum ClamavSource {
475 Buffered(bytes::Bytes),
476 Path(std::path::PathBuf),
477 }
478
479 impl ScanInput {
480 /// A `Send + 'static` byte handle for `spawn_blocking` CPU work.
481 fn byte_handle(&self) -> ScanBytes {
482 match self {
483 ScanInput::Buffered(b) => ScanBytes::Buffered(b.clone()),
484 ScanInput::Spooled { map, .. } => ScanBytes::Mapped(std::sync::Arc::clone(map)),
485 }
486 }
487
488 fn len(&self) -> usize {
489 match self {
490 ScanInput::Buffered(b) => b.len(),
491 ScanInput::Spooled { map, .. } => map.len(),
492 }
493 }
494 }
495
496 impl ScanPipeline {
497 /// Create a new pipeline, compiling YARA rules from the configured directory.
498 pub fn new(config: &ScanConfig) -> Result<Self, String> {
499 let (yara_rules, yara_rule_count) = yara::compile_rules_from_dir(&config.yara_rules_dir)?;
500
501 Ok(ScanPipeline {
502 yara_rules,
503 yara_rule_count,
504 yara_min_rule_files: config.yara_min_rule_files,
505 clamav_socket: config.clamav_socket.clone(),
506 clamav_max_scan_bytes: config.clamav_max_scan_bytes,
507 malwarebazaar_enabled: config.malwarebazaar_enabled,
508 urlhaus_enabled: config.urlhaus_enabled,
509 abuse_ch_auth_key: config.abuse_ch_auth_key.clone(),
510 metadefender_api_key: config.metadefender_api_key.clone(),
511 })
512 }
513
514 /// The configured ClamAV socket path, if any. Used to spawn the runtime
515 /// liveness probe (`worker::spawn_clamav_health_probe`).
516 pub fn clamav_socket(&self) -> Option<&str> {
517 self.clamav_socket.as_deref()
518 }
519
520 /// Assert at startup that at least one real AV layer is live. Refuse to
521 /// boot otherwise, ClamAV's FailOpen policy means a dead clamd
522 /// silently passes every upload as Clean, and a YARA-rules-empty deploy
523 /// gives the same false sense of coverage. If the operator configured
524 /// scanning, a misconfiguration must be loud at boot, not silent at runtime.
525 pub async fn assert_live(&self) -> Result<(), String> {
526 let mut live_layers: Vec<&str> = Vec::new();
527 if let Some(ref socket) = self.clamav_socket {
528 match clamav::ping(socket).await {
529 Ok(()) => live_layers.push("clamav"),
530 Err(e) => {
531 return Err(format!("ClamAV socket {socket} unreachable: {e}"));
532 }
533 }
534 // ClamAV being reachable does not mean it scans whole objects: clamd
535 // silently truncates at MaxScanSize/MaxFileSize and reports OK, and
536 // those limits aren't queryable over the socket. Surface loudly at
537 // boot when the operator hasn't declared coverage reaching the spool
538 // ceiling, above the declared coverage, large files are held for
539 // review rather than certified Clean (ultra-fuzz Run #24 Security).
540 match self.clamav_max_scan_bytes {
541 None => tracing::warn!(
542 yara_prefix = crate::constants::SCAN_YARA_MAX_BYTES,
543 "CLAMAV_MAX_SCAN_BYTES is not set, ClamAV is NOT treated as a full-file backstop; \
544 files larger than the YARA prefix will be held for admin review. Set it to your \
545 clamd min(MaxScanSize, MaxFileSize, StreamMaxLength) to auto-clear large uploads."
546 ),
547 Some(max) if max < crate::constants::SCAN_SPOOL_MAX_BYTES => tracing::warn!(
548 declared_coverage = max,
549 spool_ceiling = crate::constants::SCAN_SPOOL_MAX_BYTES,
550 "CLAMAV_MAX_SCAN_BYTES is below the scan spool ceiling, uploads between the declared \
551 ClamAV coverage and the spool ceiling will be held for admin review, not auto-cleared."
552 ),
553 Some(_) => {}
554 }
555 }
556 if self.yara_rules.is_some() {
557 // Expected-rule-count floor: a corpus that quietly dropped below the
558 // operator-declared size (e.g. a yara-x upgrade made N rules
559 // uncompilable) is degraded coverage masquerading as a live layer.
560 // Fail boot loudly when a floor is set and we're under it.
561 if self.yara_min_rule_files > 0 && self.yara_rule_count < self.yara_min_rule_files {
562 return Err(format!(
563 "YARA corpus degraded: {} rule files compiled, below the configured \
564 floor of {} (YARA_MIN_RULE_FILES). Refusing to boot, a silently \
565 shrunken rule set is false coverage.",
566 self.yara_rule_count, self.yara_min_rule_files,
567 ));
568 }
569 live_layers.push("yara");
570 }
571 if self.malwarebazaar_enabled {
572 live_layers.push("malwarebazaar");
573 }
574 if self.urlhaus_enabled {
575 live_layers.push("urlhaus");
576 }
577 if self.metadefender_api_key.is_some() {
578 live_layers.push("metadefender");
579 }
580 if live_layers.is_empty() {
581 return Err(
582 "Scanning configured but no AV layer is live (no ClamAV socket, \
583 no YARA rules, no remote API keys). Refusing to boot, the \
584 FailOpen policy would pass every upload as Clean."
585 .to_string(),
586 );
587 }
588 tracing::info!(layers = ?live_layers, "scan pipeline live layers asserted");
589 Ok(())
590 }
591
592 /// Buffered scan entry point, small uploads held in memory. A thin adapter
593 /// over [`run_scan`](Self::run_scan); all scan policy lives there.
594 pub(crate) async fn scan(
595 self: std::sync::Arc<Self>,
596 data: impl Into<bytes::Bytes>,
597 file_type: FileType,
598 ) -> ScanResult {
599 // `bytes::Bytes` is already a cheaply-cloneable refcounted buffer, so the
600 // download hands its single aggregated allocation straight here with no
601 // extra copy (the buffered path previously aggregated then `to_vec`'d the
602 // body, transiently doubling to ~200 MB for a 100 MB file, Run #2).
603 self.run_scan(ScanInput::Buffered(data.into()), file_type)
604 .await
605 }
606
607 /// Streaming scan entry point, large uploads spooled to a tempfile and
608 /// scanned via a memory map, so the >100 MB case doesn't hold the whole object
609 /// in RAM. A thin adapter over [`run_scan`](Self::run_scan).
610 pub(crate) async fn scan_stream(
611 self: std::sync::Arc<Self>,
612 spool: spool::SpoolHandle,
613 file_type: FileType,
614 ) -> ScanResult {
615 let map = match spool::mmap_read(spool.path()) {
616 Ok(m) => std::sync::Arc::new(m),
617 Err(e) => {
618 let file_size = std::fs::metadata(spool.path()).map_or(0, |m| m.len());
619 let layer = LayerResult {
620 layer: "spool",
621 verdict: LayerVerdict::Error,
622 detail: Some(e),
623 };
624 return ScanResult {
625 status: final_status(std::slice::from_ref(&layer)),
626 layers: vec![layer],
627 sha256: String::new(),
628 file_size,
629 };
630 }
631 };
632 self.run_scan(ScanInput::Spooled { map, spool }, file_type)
633 .await
634 }
635
636 /// The single scan body shared by [`scan`](Self::scan) and
637 /// [`scan_stream`](Self::scan_stream). Every scan policy lives here exactly
638 /// once; the entry points differ only in how `input` sources its bytes, so a
639 /// policy (CPU work off the runtime, ClamAV, URLhaus host extraction, external
640 /// lookups, byte caps) can no longer drift between buffered and streaming
641 /// (ultra-fuzz CHRONIC, closed Run 9).
642 ///
643 /// CPU-bound layers (sha256, content-type, structural, archive, yara) run on a
644 /// blocking-pool thread via `spawn_blocking`; ClamAV and URLhaus run
645 /// concurrently with them via `tokio::join!`.
646 async fn run_scan(
647 self: std::sync::Arc<Self>,
648 input: ScanInput,
649 file_type: FileType,
650 ) -> ScanResult {
651 let file_size = input.len() as u64;
652
653 // CPU layers + hash, off the runtime, under a wall-clock deadline. The
654 // per-layer timeouts (yara 30s) did not cover the archive decompress
655 // walk, so a slow-codec archive could pin a scan worker for its full
656 // decompress time (fuzz 2026-07-06 F1). The timeout frees the worker on
657 // elapse; the blocking thread runs to completion on the large blocking
658 // pool (spawn_blocking is not cancellable), so the memory-budget ceiling
659 // still holds while the two-worker pool is no longer monopolized.
660 let sync_bytes = input.byte_handle();
661 let sync_self = std::sync::Arc::clone(&self);
662 let sync_fut = tokio::time::timeout(
663 std::time::Duration::from_secs(crate::constants::SCAN_CPU_LAYERS_TIMEOUT_SECS),
664 tokio::task::spawn_blocking(move || {
665 sync_self.run_sync_layers(sync_bytes.as_slice(), file_type)
666 }),
667 );
668
669 // ClamAV: buffered scans the in-memory bytes; spooled streams the file by
670 // path (INSTREAM frames) so a >100 MB object isn't re-buffered. The
671 // retained SpoolHandle in `input` keeps the file alive across this join.
672 let clamav_socket = self.clamav_socket.clone();
673 let clamav_source = match &input {
674 ScanInput::Buffered(b) => ClamavSource::Buffered(b.clone()),
675 ScanInput::Spooled { spool, .. } => ClamavSource::Path(spool.path().to_path_buf()),
676 };
677 let clamav_fut = async move {
678 let Some(socket) = clamav_socket else {
679 return LayerResult {
680 layer: "clamav",
681 verdict: LayerVerdict::Skip,
682 detail: Some("ClamAV not configured".to_string()),
683 };
684 };
685 match clamav_source {
686 ClamavSource::Buffered(data) => clamav::scan_with_clamav(&socket, &data).await,
687 ClamavSource::Path(path) => match tokio::fs::File::open(&path).await {
688 Ok(file) => clamav::scan_with_clamav_stream(&socket, file).await,
689 Err(e) => LayerResult {
690 layer: "clamav",
691 verdict: LayerVerdict::Error,
692 detail: Some(format!("open spool for clamav: {e}")),
693 },
694 },
695 }
696 };
697
698 // URLhaus: extract candidate hosts on the blocking pool (the byte walk can
699 // page-fault on the mmap), then do only the network lookups async. This is
700 // ONE policy for both entry points, the exact step that used to drift
701 // between scan() and scan_stream() (ultra-fuzz CHRONIC).
702 let urlhaus_bytes = input.byte_handle();
703 let urlhaus_enabled = self.urlhaus_enabled;
704 let urlhaus_key = self.abuse_ch_auth_key.clone();
705 let urlhaus_fut = async move {
706 if urlhaus_enabled {
707 let hosts = tokio::task::spawn_blocking(move || {
708 urlhaus::extract_unique_hosts(
709 urlhaus_bytes.as_slice(),
710 urlhaus::MAX_HOSTS_PER_FILE,
711 )
712 })
713 .await
714 .unwrap_or_default();
715 urlhaus::check_urlhaus_hosts(hosts, urlhaus_key.as_deref()).await
716 } else {
717 LayerResult {
718 layer: "urlhaus",
719 verdict: LayerVerdict::Skip,
720 detail: Some("URLhaus lookups disabled".to_string()),
721 }
722 }
723 };
724
725 let (sync_result, clamav_result, urlhaus_result) =
726 tokio::join!(sync_fut, clamav_fut, urlhaus_fut);
727 let (mut layers, sha256) = match sync_result {
728 Ok(Ok(v)) => v,
729 Ok(Err(join_err)) => {
730 tracing::error!(
731 error = %join_err,
732 is_panic = join_err.is_panic(),
733 "scan sync layers panicked; holding file for review (worker survives)"
734 );
735 return panicked_sync_result(file_size);
736 }
737 Err(_elapsed) => {
738 tracing::error!(
739 timeout_secs = crate::constants::SCAN_CPU_LAYERS_TIMEOUT_SECS,
740 "scan CPU layers exceeded the wall-clock deadline; holding file for \
741 review (worker freed; blocking thread runs to completion)"
742 );
743 return timed_out_sync_result(file_size);
744 }
745 };
746 layers.push(clamav_result);
747 layers.push(urlhaus_result);
748 self.push_external_lookups(&mut layers, &sha256).await;
749
750 let status = final_status(&layers);
751
752 // Dropping `input` here releases the mmap and unlinks the spool tempfile
753 // (if any), after ClamAV has finished streaming it.
754 drop(input);
755
756 ScanResult {
757 status,
758 layers,
759 sha256,
760 file_size,
761 }
762 }
763
764 /// Append the post-sync, hash-keyed external-lookup layers to `layers`:
765 /// Layer 6 MalwareBazaar (by hash) then Layer 9 MetaDefender (a
766 /// suspicion-gated second opinion that only fires when a prior layer flagged
767 /// the file, to stay within the free-tier quota). Each runs under the shared
768 /// aggregate external-lookup deadline (`lookup_with_timeout`, Perf-S1). Shared
769 /// by `scan` and `scan_stream` so the buffered and streaming paths can't drift.
770 async fn push_external_lookups(&self, layers: &mut Vec<LayerResult>, sha256: &str) {
771 layers.push(if self.malwarebazaar_enabled {
772 lookup_with_timeout(
773 "malwarebazaar",
774 hash_lookup::check_malwarebazaar(sha256, self.abuse_ch_auth_key.as_deref()),
775 )
776 .await
777 } else {
778 LayerResult {
779 layer: "malwarebazaar",
780 verdict: LayerVerdict::Skip,
781 detail: Some("MalwareBazaar lookups disabled".to_string()),
782 }
783 });
784
785 layers.push(if suspicion_present(layers) {
786 lookup_with_timeout(
787 "metadefender",
788 metadefender::check_metadefender(sha256, self.metadefender_api_key.as_deref()),
789 )
790 .await
791 } else {
792 LayerResult {
793 layer: "metadefender",
794 verdict: LayerVerdict::Skip,
795 detail: Some("No prior suspicion; second-opinion not invoked".to_string()),
796 }
797 });
798 }
799
800 /// CPU-bound layers + SHA-256. Pure sync; safe to call from `spawn_blocking`.
801 fn run_sync_layers(&self, data: &[u8], file_type: FileType) -> (Vec<LayerResult>, String) {
802 let mut layers = Vec::with_capacity(5);
803
804 // SHA-256 hash for audit + MalwareBazaar lookup
805 let sha256 = {
806 let mut hasher = Sha256::new();
807 hasher.update(data);
808 hex::encode(hasher.finalize())
809 };
810
811 layers.push(content_type::verify_content_type(data, file_type));
812 layers.push(structural::analyze_binary(data, file_type));
813 // Single-pass archive inspection: the bomb-defense walk ("archive") and
814 // the recursive interior content scan ("archive_nested") are derived from
815 // ONE decompression of each entry. Previously each entry was decompressed
816 // twice, once to count for bomb defense, once to buffer+scan its content
817 // (Run #2 Performance SERIOUS). No-op (Skip/Skip) for non-archives.
818 let (archive_layer, archive_nested_layer) =
819 archive::inspect_archive(data, file_type, self.yara_rules.as_ref());
820 layers.push(archive_layer);
821 layers.push(archive_nested_layer);
822 layers.push(match self.yara_rules {
823 Some(ref rules) => {
824 // Cap YARA's input: it walks the whole slice, faulting the entire
825 // mmap resident. ClamAV (streamed, uncapped) is the full-file
826 // backstop, so scanning a generous prefix bounds peak RAM without
827 // surrendering the floor.
828 if data.len() > crate::constants::SCAN_YARA_MAX_BYTES {
829 if yara_tail_unscanned(data.len(), self.clamav_max_scan_bytes) {
830 // No full-file backstop reaches this size, so the bytes
831 // past the YARA prefix would go entirely unscanned, a
832 // tail-of-file evasion (append the payload past the cap
833 // and it passes Clean). This is true both when ClamAV is
834 // absent and when it is present but its declared
835 // MaxScanSize doesn't cover the file (Run #24 Security).
836 // Hold the file rather than certify an unscanned tail
837 // (ultra-fuzz Run 13 Storage). `yara` is FailClosed, so
838 // this Error → HeldForReview.
839 tracing::warn!(
840 full_bytes = data.len(),
841 capped_bytes = crate::constants::SCAN_YARA_MAX_BYTES,
842 "YARA input would be capped but no ClamAV full-file backstop is configured; holding file (tail unscanned)"
843 );
844 LayerResult {
845 layer: "yara",
846 verdict: LayerVerdict::Error,
847 detail: Some(format!(
848 "tail unscanned: {} B exceeds the {} B YARA prefix and no ClamAV full-file scan is configured",
849 data.len(),
850 crate::constants::SCAN_YARA_MAX_BYTES
851 )),
852 }
853 } else {
854 tracing::warn!(
855 full_bytes = data.len(),
856 capped_bytes = crate::constants::SCAN_YARA_MAX_BYTES,
857 "YARA input capped; scanning prefix only (ClamAV scans the full file)"
858 );
859 yara::scan_with_yara(rules, &data[..crate::constants::SCAN_YARA_MAX_BYTES])
860 }
861 } else {
862 yara::scan_with_yara(rules, data)
863 }
864 }
865 None => LayerResult {
866 layer: "yara",
867 verdict: LayerVerdict::Skip,
868 detail: Some("No YARA rules loaded".to_string()),
869 },
870 });
871 layers.push(signing_macos::verify_apple_signature(data, file_type));
872 layers.push(signing_windows::verify_authenticode(data, file_type));
873 layers.push(signing_linux::verify_appimage_signature(data, file_type));
874
875 (layers, sha256)
876 }
877 }
878
879 #[cfg(test)]
880 mod tests {
881 use super::*;
882
883 #[test]
884 fn layer_verdict_serializes_lowercase() {
885 assert_eq!(
886 serde_json::to_string(&LayerVerdict::Pass).unwrap(),
887 "\"pass\""
888 );
889 assert_eq!(
890 serde_json::to_string(&LayerVerdict::Fail).unwrap(),
891 "\"fail\""
892 );
893 assert_eq!(
894 serde_json::to_string(&LayerVerdict::Skip).unwrap(),
895 "\"skip\""
896 );
897 assert_eq!(
898 serde_json::to_string(&LayerVerdict::Error).unwrap(),
899 "\"error\""
900 );
901 }
902
903 #[test]
904 fn scan_result_quarantined_on_any_fail() {
905 let layers = [
906 LayerResult {
907 layer: "test1",
908 verdict: LayerVerdict::Pass,
909 detail: None,
910 },
911 LayerResult {
912 layer: "test2",
913 verdict: LayerVerdict::Fail,
914 detail: Some("bad".to_string()),
915 },
916 ];
917 let has_fail = layers.iter().any(|l| l.verdict == LayerVerdict::Fail);
918 assert!(has_fail);
919 }
920
921 #[test]
922 fn panicked_sync_layer_is_held_for_review_not_clean() {
923 // A file that panics a CPU parser must never come back Clean, it is
924 // held for admin review, and the panic is contained in a returnable
925 // result (not an `.expect` that would unwind the scan worker).
926 let r = panicked_sync_result(4096);
927 assert_eq!(r.status, FileScanStatus::HeldForReview);
928 assert_eq!(r.file_size, 4096);
929 assert!(r.sha256.is_empty());
930 assert_eq!(r.layers.len(), 1);
931 assert_eq!(r.layers[0].verdict, LayerVerdict::Error);
932 // The synthetic layer must resolve to a FailClosed policy (the default
933 // for unregistered names), which is what makes final_status hold it.
934 assert_eq!(error_policy_for(r.layers[0].layer), ErrorPolicy::FailClosed);
935 }
936
937 #[test]
938 fn oversize_file_is_held_for_review_not_failed() {
939 // A file above the spool ceiling is held for admin review (explicit,
940 // documented policy) rather than failing the job and stranding the
941 // upload in Pending. Verdict resolves to FailClosed -> HeldForReview.
942 let huge = crate::constants::SCAN_SPOOL_MAX_BYTES + 1;
943 let r = too_large_to_scan(huge);
944 assert_eq!(r.status, FileScanStatus::HeldForReview);
945 assert_eq!(r.file_size, huge);
946 assert!(r.sha256.is_empty());
947 assert_eq!(r.layers[0].layer, "scan_size_limit");
948 assert_eq!(error_policy_for(r.layers[0].layer), ErrorPolicy::FailClosed);
949 }
950
951 #[test]
952 fn sha256_computation() {
953 let mut hasher = Sha256::new();
954 hasher.update(b"hello");
955 let hash = hex::encode(hasher.finalize());
956 assert_eq!(
957 hash,
958 "2cf24dba5fb0a30e26e83b2ac5b9e29e1b161e5c1fa7425e73043362938b9824"
959 );
960 }
961
962 // Pipeline integration tests
963
964 /// Create a minimal ScanPipeline with no external deps (no YARA, no ClamAV, no MalwareBazaar).
965 /// Wrapped in `Arc` because `scan` consumes `Arc<Self>` (see `pub async fn scan`).
966 fn make_pipeline() -> std::sync::Arc<ScanPipeline> {
967 std::sync::Arc::new(ScanPipeline {
968 yara_rules: None,
969 yara_rule_count: 0,
970 yara_min_rule_files: 0,
971 clamav_socket: None,
972 clamav_max_scan_bytes: None,
973 malwarebazaar_enabled: false,
974 urlhaus_enabled: false,
975 abuse_ch_auth_key: None,
976 metadefender_api_key: None,
977 })
978 }
979
980 /// SEC-S2: a corpus that compiled fewer rule files than the configured floor
981 /// is degraded coverage masquerading as a live layer, boot must fail closed.
982 #[tokio::test]
983 async fn assert_live_refuses_degraded_yara_corpus() {
984 let mut compiler = yara_x::Compiler::new();
985 compiler
986 .add_source(r#"rule r { strings: $a = "x" condition: $a }"#)
987 .unwrap();
988 let pipeline = std::sync::Arc::new(ScanPipeline {
989 yara_rules: Some(compiler.build()),
990 yara_rule_count: 1,
991 yara_min_rule_files: 2,
992 clamav_socket: None,
993 clamav_max_scan_bytes: None,
994 malwarebazaar_enabled: false,
995 urlhaus_enabled: false,
996 abuse_ch_auth_key: None,
997 metadefender_api_key: None,
998 });
999 let err = pipeline
1000 .assert_live()
1001 .await
1002 .expect_err("degraded corpus must refuse boot");
1003 assert!(
1004 err.contains("YARA corpus degraded"),
1005 "unexpected error: {err}"
1006 );
1007 }
1008
1009 /// The complement: a corpus meeting the floor boots, with YARA counted live.
1010 #[tokio::test]
1011 async fn assert_live_accepts_corpus_meeting_floor() {
1012 let mut compiler = yara_x::Compiler::new();
1013 compiler
1014 .add_source(r#"rule r { strings: $a = "x" condition: $a }"#)
1015 .unwrap();
1016 let pipeline = std::sync::Arc::new(ScanPipeline {
1017 yara_rules: Some(compiler.build()),
1018 yara_rule_count: 3,
1019 yara_min_rule_files: 3,
1020 clamav_socket: None,
1021 clamav_max_scan_bytes: None,
1022 malwarebazaar_enabled: false,
1023 urlhaus_enabled: false,
1024 abuse_ch_auth_key: None,
1025 metadefender_api_key: None,
1026 });
1027 pipeline
1028 .assert_live()
1029 .await
1030 .expect("a corpus meeting the floor must boot");
1031 }
1032
1033 #[tokio::test]
1034 async fn pipeline_clean_download_passes() {
1035 let pipeline = make_pipeline();
1036 let result = pipeline
1037 .clone()
1038 .scan(b"just some file content".to_vec(), FileType::Download)
1039 .await;
1040 assert_eq!(result.status, FileScanStatus::Clean);
1041 assert_eq!(result.file_size, 22);
1042 assert!(!result.sha256.is_empty());
1043 assert_eq!(result.layers.len(), 12);
1044 }
1045
1046 #[tokio::test]
1047 async fn pipeline_unrecognized_audio_quarantined() {
1048 let pipeline = make_pipeline();
1049 // Unrecognized data claimed as audio should be rejected by content_type layer
1050 let result = pipeline
1051 .clone()
1052 .scan(b"audio data here".to_vec(), FileType::Audio)
1053 .await;
1054 assert_eq!(result.status, FileScanStatus::Quarantined);
1055 }
1056
1057 #[tokio::test]
1058 async fn pipeline_clean_cover_passes() {
1059 let pipeline = make_pipeline();
1060 // PNG magic bytes
1061 let png = [0x89, 0x50, 0x4E, 0x47, 0x0D, 0x0A, 0x1A, 0x0A];
1062 let result = pipeline.clone().scan(png.to_vec(), FileType::Cover).await;
1063 assert_eq!(result.status, FileScanStatus::Clean);
1064 }
1065
1066 #[tokio::test]
1067 async fn pipeline_pe_as_audio_quarantined() {
1068 let pipeline = make_pipeline();
1069 // PE magic bytes, content-type layer should detect application/* and fail
1070 let pe_header = b"MZ\x90\x00\x03\x00\x00\x00";
1071 let result = pipeline
1072 .clone()
1073 .scan(pe_header.to_vec(), FileType::Audio)
1074 .await;
1075 assert_eq!(result.status, FileScanStatus::Quarantined);
1076 // Verify content_type layer produced the fail
1077 let content_type_layer = result
1078 .layers
1079 .iter()
1080 .find(|l| l.layer == "content_type")
1081 .unwrap();
1082 assert_eq!(content_type_layer.verdict, LayerVerdict::Fail);
1083 }
1084
1085 #[tokio::test]
1086 async fn pipeline_pe_as_cover_quarantined() {
1087 let pipeline = make_pipeline();
1088 let pe_header = b"MZ\x90\x00\x03\x00\x00\x00";
1089 let result = pipeline
1090 .clone()
1091 .scan(pe_header.to_vec(), FileType::Cover)
1092 .await;
1093 assert_eq!(result.status, FileScanStatus::Quarantined);
1094 }
1095
1096 #[tokio::test]
1097 async fn pipeline_sha256_is_deterministic() {
1098 let pipeline = make_pipeline();
1099 let data = b"deterministic hash test";
1100 let r1 = pipeline
1101 .clone()
1102 .scan(data.to_vec(), FileType::Download)
1103 .await;
1104 let r2 = pipeline
1105 .clone()
1106 .scan(data.to_vec(), FileType::Download)
1107 .await;
1108 assert_eq!(r1.sha256, r2.sha256);
1109 }
1110
1111 #[tokio::test]
1112 async fn pipeline_skips_optional_layers_when_unconfigured() {
1113 let pipeline = make_pipeline();
1114 let result = pipeline
1115 .clone()
1116 .scan(b"test".to_vec(), FileType::Download)
1117 .await;
1118
1119 let yara = result.layers.iter().find(|l| l.layer == "yara").unwrap();
1120 assert_eq!(yara.verdict, LayerVerdict::Skip);
1121
1122 let clamav = result.layers.iter().find(|l| l.layer == "clamav").unwrap();
1123 assert_eq!(clamav.verdict, LayerVerdict::Skip);
1124
1125 let mb = result
1126 .layers
1127 .iter()
1128 .find(|l| l.layer == "malwarebazaar")
1129 .unwrap();
1130 assert_eq!(mb.verdict, LayerVerdict::Skip);
1131
1132 let uh = result.layers.iter().find(|l| l.layer == "urlhaus").unwrap();
1133 assert_eq!(uh.verdict, LayerVerdict::Skip);
1134 }
1135
1136 #[tokio::test]
1137 async fn pipeline_always_produces_12_layers() {
1138 let pipeline = make_pipeline();
1139 for file_type in [FileType::Audio, FileType::Cover, FileType::Download] {
1140 let result = pipeline.clone().scan(b"data".to_vec(), file_type).await;
1141 // 11 base layers + the recursive archive-interior layer (archive_nested).
1142 assert_eq!(
1143 result.layers.len(),
1144 12,
1145 "Expected 12 layers for {file_type:?}"
1146 );
1147 }
1148 }
1149
1150 #[test]
1151 fn suspicion_present_on_fail() {
1152 let layers = vec![pass("content_type"), fail("yara")];
1153 assert!(suspicion_present(&layers));
1154 }
1155
1156 #[test]
1157 fn suspicion_present_on_fail_closed_error() {
1158 let layers = vec![pass("content_type"), err("archive")];
1159 assert!(suspicion_present(&layers));
1160 }
1161
1162 #[test]
1163 fn no_suspicion_when_fail_open_error_only() {
1164 // External-layer errors are operational noise, not malware signals;
1165 // they must not invoke MetaDefender.
1166 let layers = vec![pass("content_type"), err("malwarebazaar"), err("urlhaus")];
1167 assert!(!suspicion_present(&layers));
1168 }
1169
1170 #[test]
1171 fn no_suspicion_when_all_clean() {
1172 let layers = vec![pass("content_type"), skip("yara"), pass("structural")];
1173 assert!(!suspicion_present(&layers));
1174 }
1175
1176 #[tokio::test]
1177 async fn pipeline_errors_held_for_review() {
1178 // Errors from fail-closed layers (archive is in-process deterministic)
1179 // should hold the file for admin review.
1180 let pipeline = make_pipeline();
1181 // Corrupted ZIP magic bytes, archive layer returns Error
1182 let mut data = vec![0x50, 0x4B, 0x03, 0x04];
1183 data.extend_from_slice(&[0xFF; 100]);
1184 let result = pipeline.clone().scan(data, FileType::Download).await;
1185 let archive = result.layers.iter().find(|l| l.layer == "archive").unwrap();
1186 assert_eq!(archive.verdict, LayerVerdict::Error);
1187 assert_eq!(result.status, FileScanStatus::HeldForReview);
1188 }
1189
1190 // Per-layer fail policy tests
1191
1192 fn err(layer: &'static str) -> LayerResult {
1193 LayerResult {
1194 layer,
1195 verdict: LayerVerdict::Error,
1196 detail: None,
1197 }
1198 }
1199 fn pass(layer: &'static str) -> LayerResult {
1200 LayerResult {
1201 layer,
1202 verdict: LayerVerdict::Pass,
1203 detail: None,
1204 }
1205 }
1206 fn skip(layer: &'static str) -> LayerResult {
1207 LayerResult {
1208 layer,
1209 verdict: LayerVerdict::Skip,
1210 detail: None,
1211 }
1212 }
1213 fn fail(layer: &'static str) -> LayerResult {
1214 LayerResult {
1215 layer,
1216 verdict: LayerVerdict::Fail,
1217 detail: None,
1218 }
1219 }
1220
1221 #[test]
1222 fn yara_tail_unscanned_only_without_backstop() {
1223 use crate::constants::SCAN_YARA_MAX_BYTES;
1224 let over = SCAN_YARA_MAX_BYTES + 1;
1225 // Over the cap, no declared ClamAV coverage → tail unscanned, must hold.
1226 assert!(yara_tail_unscanned(over, None));
1227 // Over the cap, ClamAV coverage present but SHORT of the file (Run #24
1228 // MODERATE: clamd reachable but MaxScanSize doesn't reach) → still hold.
1229 assert!(yara_tail_unscanned(over, Some(over as u64 - 1)));
1230 // Over the cap, declared coverage reaches the file → ClamAV is a real
1231 // full-file backstop, don't hold.
1232 assert!(!yara_tail_unscanned(over, Some(over as u64)));
1233 assert!(!yara_tail_unscanned(over, Some(u64::MAX)));
1234 // Within the YARA cap → whole file scanned by YARA regardless of ClamAV.
1235 assert!(!yara_tail_unscanned(SCAN_YARA_MAX_BYTES, None));
1236 assert!(!yara_tail_unscanned(1024, None));
1237 }
1238
1239 #[test]
1240 fn final_status_clean_when_all_pass() {
1241 let layers = vec![
1242 pass("content_type"),
1243 pass("structural"),
1244 pass("archive"),
1245 skip("yara"),
1246 skip("clamav"),
1247 skip("malwarebazaar"),
1248 ];
1249 assert_eq!(final_status(&layers), FileScanStatus::Clean);
1250 }
1251
1252 #[test]
1253 fn final_status_quarantined_on_any_fail() {
1254 let layers = vec![pass("content_type"), fail("yara"), skip("clamav")];
1255 assert_eq!(final_status(&layers), FileScanStatus::Quarantined);
1256 }
1257
1258 #[test]
1259 fn final_status_fail_beats_error() {
1260 // A Fail anywhere supersedes any Error, regardless of policy.
1261 let layers = vec![err("malwarebazaar"), fail("yara")];
1262 assert_eq!(final_status(&layers), FileScanStatus::Quarantined);
1263 }
1264
1265 #[test]
1266 fn final_status_held_on_fail_closed_error() {
1267 // archive is FailClosed, its Error must hold the file.
1268 let layers = vec![pass("content_type"), err("archive"), skip("clamav")];
1269 assert_eq!(final_status(&layers), FileScanStatus::HeldForReview);
1270 }
1271
1272 #[test]
1273 fn final_status_clean_on_fail_open_error_only() {
1274 // malwarebazaar is FailOpen, its Error must NOT hold the file.
1275 // This is the regression of 2026-05-10 that motivated the audit.
1276 let layers = vec![
1277 pass("content_type"),
1278 pass("structural"),
1279 pass("archive"),
1280 skip("yara"),
1281 skip("clamav"),
1282 err("malwarebazaar"),
1283 ];
1284 assert_eq!(final_status(&layers), FileScanStatus::Clean);
1285 }
1286
1287 #[test]
1288 fn final_status_clean_when_all_external_layers_error() {
1289 // Worst-case external-services outage: every network/daemon layer
1290 // erroring at once. As long as the in-process layers pass, the file
1291 // is Clean. Health is surfaced separately via per-layer monitoring.
1292 let layers = vec![
1293 pass("content_type"),
1294 pass("structural"),
1295 pass("archive"),
1296 skip("yara"),
1297 err("clamav"),
1298 err("malwarebazaar"),
1299 ];
1300 assert_eq!(final_status(&layers), FileScanStatus::Clean);
1301 }
1302
1303 #[test]
1304 fn clamav_incomplete_is_fail_closed_and_holds() {
1305 // CHRONIC S1: a reachable-but-incomplete clamav scan is emitted under the
1306 // `clamav_incomplete` layer, which must be FailClosed → HeldForReview,
1307 // distinct from the FailOpen `clamav` layer used for an unreachable daemon.
1308 assert_eq!(
1309 error_policy_for("clamav_incomplete"),
1310 ErrorPolicy::FailClosed
1311 );
1312 assert_eq!(error_policy_for("clamav"), ErrorPolicy::FailOpen);
1313 let layers = vec![
1314 pass("content_type"),
1315 pass("structural"),
1316 pass("archive"),
1317 skip("yara"),
1318 err("clamav_incomplete"),
1319 ];
1320 assert_eq!(final_status(&layers), FileScanStatus::HeldForReview);
1321 }
1322
1323 #[test]
1324 fn final_status_held_on_unknown_layer_error() {
1325 // Defensive default: an unknown layer name that errors falls through
1326 // to FailClosed. This is what catches a new layer added without
1327 // wiring its policy into `error_policy_for`.
1328 let layers = vec![
1329 pass("content_type"),
1330 err("brand_new_layer_someone_forgot_to_register"),
1331 ];
1332 assert_eq!(final_status(&layers), FileScanStatus::HeldForReview);
1333 }
1334
1335 #[test]
1336 fn error_policy_for_all_known_layers() {
1337 // Every layer name produced by the pipeline must have an explicit
1338 // declaration in `error_policy_for`. The default branch is reserved
1339 // for genuine programmer error (new layer, forgot to register).
1340 for name in [
1341 "content_type",
1342 "structural",
1343 "archive",
1344 "yara",
1345 "clamav",
1346 "malwarebazaar",
1347 ] {
1348 let policy = error_policy_for(name);
1349 // Both values are valid; we just want this to not hit the default.
1350 // If a layer is renamed without updating `error_policy_for`, this
1351 // test still passes (the rename produces a new unknown name)
1352 //, but the per-layer name tests below catch that.
1353 let _ = policy;
1354 }
1355 }
1356
1357 #[test]
1358 fn content_type_is_fail_closed() {
1359 assert_eq!(error_policy_for("content_type"), ErrorPolicy::FailClosed);
1360 }
1361 #[test]
1362 fn structural_is_fail_closed() {
1363 assert_eq!(error_policy_for("structural"), ErrorPolicy::FailClosed);
1364 }
1365 #[test]
1366 fn archive_is_fail_closed() {
1367 assert_eq!(error_policy_for("archive"), ErrorPolicy::FailClosed);
1368 }
1369 #[test]
1370 fn yara_is_fail_closed() {
1371 assert_eq!(error_policy_for("yara"), ErrorPolicy::FailClosed);
1372 }
1373 #[test]
1374 fn clamav_is_fail_open() {
1375 assert_eq!(error_policy_for("clamav"), ErrorPolicy::FailOpen);
1376 }
1377 #[test]
1378 fn malwarebazaar_is_fail_open() {
1379 assert_eq!(error_policy_for("malwarebazaar"), ErrorPolicy::FailOpen);
1380 }
1381 #[test]
1382 fn urlhaus_is_fail_open() {
1383 assert_eq!(error_policy_for("urlhaus"), ErrorPolicy::FailOpen);
1384 }
1385 #[test]
1386 fn signing_macos_is_fail_open() {
1387 assert_eq!(error_policy_for("signing_macos"), ErrorPolicy::FailOpen);
1388 }
1389 #[test]
1390 fn metadefender_is_fail_open() {
1391 assert_eq!(error_policy_for("metadefender"), ErrorPolicy::FailOpen);
1392 }
1393 #[test]
1394 fn signing_windows_is_fail_open() {
1395 assert_eq!(error_policy_for("signing_windows"), ErrorPolicy::FailOpen);
1396 }
1397 #[test]
1398 fn signing_linux_is_fail_open() {
1399 assert_eq!(error_policy_for("signing_linux"), ErrorPolicy::FailOpen);
1400 }
1401 }
1402