Skip to main content

max / makenotwork

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