Skip to main content

max / makenotwork

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