Skip to main content

max / makenotwork

28.4 KB · 731 lines History Blame Raw
1 //! 6-layer malware scanning pipeline for file uploads.
2 //!
3 //! Layers 1-4 always run (in-process, deterministic). Layers 5-6 are optional
4 //! (external services).
5 //!
6 //! **Error policy is per-layer**, declared at each layer's source file as
7 //! `pub const ERROR_POLICY`. The aggregator `final_status` consults each
8 //! layer's policy via `error_policy_for`. In-process layers are `FailClosed`
9 //! (an error is a structural defect); external layers are `FailOpen` (an
10 //! error is an outage that must not block the platform). See
11 //! `docs/scan-pipeline-audit.md` for the rationale.
12 //!
13 //! See also: `/docs/tech/content-protection`
14
15 pub mod archive;
16 pub mod clamav;
17 pub mod content_type;
18 pub mod hash_lookup;
19 pub mod metadefender;
20 pub mod signing_linux;
21 pub mod signing_macos;
22 pub mod signing_windows;
23 pub mod spool;
24 pub mod structural;
25 pub mod urlhaus;
26 pub mod worker;
27 pub mod yara;
28
29 use serde::Serialize;
30 use sha2::{Digest, Sha256};
31
32 use crate::config::ScanConfig;
33 use crate::db::FileScanStatus;
34 use crate::storage::FileType;
35
36 /// Per-layer scan verdict
37 #[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize)]
38 #[serde(rename_all = "lowercase")]
39 pub enum LayerVerdict {
40 Pass,
41 Fail,
42 Skip,
43 Error,
44 }
45
46 /// Policy for how a layer's `Error` verdict feeds into the pipeline's final status.
47 ///
48 /// - `FailClosed` — an `Error` from this layer holds the upload for admin review.
49 /// Appropriate for deterministic in-process layers where an `Error` indicates a
50 /// real bug or a structurally suspicious file.
51 /// - `FailOpen` — an `Error` from this layer is treated as `Skip` for aggregation.
52 /// Appropriate for external services (network, daemons) where an outage on a
53 /// third party must not take down the platform's upload pipeline.
54 ///
55 /// Each layer declares its own `ERROR_POLICY` const; the aggregator in
56 /// `ScanPipeline::final_status` consults the declaration via `error_policy_for`.
57 #[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize)]
58 #[serde(rename_all = "snake_case")]
59 pub enum ErrorPolicy {
60 FailClosed,
61 FailOpen,
62 }
63
64 /// Result from a single scanning layer
65 #[derive(Debug, Clone, Serialize)]
66 pub struct LayerResult {
67 pub layer: &'static str,
68 pub verdict: LayerVerdict,
69 pub detail: Option<String>,
70 }
71
72 /// Look up a layer's declared error policy by name. Defaults to `FailClosed`
73 /// for unknown layers — a defensive choice that surfaces uninstrumented
74 /// additions during testing rather than silently fail-opening them.
75 fn error_policy_for(layer: &str) -> ErrorPolicy {
76 match layer {
77 "content_type" => content_type::ERROR_POLICY,
78 "structural" => structural::ERROR_POLICY,
79 "archive" => archive::ERROR_POLICY,
80 "yara" => yara::ERROR_POLICY,
81 "clamav" => clamav::ERROR_POLICY,
82 "malwarebazaar" => hash_lookup::ERROR_POLICY,
83 "urlhaus" => urlhaus::ERROR_POLICY,
84 "signing_macos" => signing_macos::ERROR_POLICY,
85 "signing_windows" => signing_windows::ERROR_POLICY,
86 "signing_linux" => signing_linux::ERROR_POLICY,
87 "metadefender" => metadefender::ERROR_POLICY,
88 other => {
89 tracing::error!(layer = other, "unknown scan layer; defaulting to FailClosed");
90 ErrorPolicy::FailClosed
91 }
92 }
93 }
94
95 /// Decide whether the second-opinion (MetaDefender) layer should run, based
96 /// on the verdicts of layers that have already completed. Any `Fail`, or any
97 /// `Error` from a fail-closed in-process layer, counts as "suspicious enough
98 /// to escalate". Pure `Error`s from fail-open external layers do not — those
99 /// are operational noise, not malware signals.
100 fn suspicion_present(layers: &[LayerResult]) -> bool {
101 layers.iter().any(|l| {
102 match l.verdict {
103 LayerVerdict::Fail => true,
104 LayerVerdict::Error => error_policy_for(l.layer) == ErrorPolicy::FailClosed,
105 _ => false,
106 }
107 })
108 }
109
110 /// Aggregate per-layer results into a final scan status.
111 ///
112 /// - Any layer `Fail` → `Quarantined` (terminal).
113 /// - Any layer `Error` whose policy is `FailClosed` → `HeldForReview`.
114 /// - `FailOpen` errors are treated as Skip-equivalent for aggregation; they're
115 /// still surfaced in the per-layer detail so admins and PoM can see degraded
116 /// layers in the dashboard.
117 /// - Otherwise → `Clean`.
118 fn final_status(layers: &[LayerResult]) -> FileScanStatus {
119 if layers.iter().any(|l| l.verdict == LayerVerdict::Fail) {
120 return FileScanStatus::Quarantined;
121 }
122 let has_fail_closed_error = layers.iter().any(|l| {
123 l.verdict == LayerVerdict::Error && error_policy_for(l.layer) == ErrorPolicy::FailClosed
124 });
125 if has_fail_closed_error {
126 FileScanStatus::HeldForReview
127 } else {
128 FileScanStatus::Clean
129 }
130 }
131
132 /// Aggregate scan result across all layers
133 #[derive(Debug, Clone)]
134 pub struct ScanResult {
135 pub status: FileScanStatus,
136 pub layers: Vec<LayerResult>,
137 pub sha256: String,
138 pub file_size: u64,
139 }
140
141 /// Pre-compiled scanning pipeline. Initialized once at startup and shared via Arc.
142 pub struct ScanPipeline {
143 yara_rules: Option<yara_x::Rules>,
144 clamav_socket: Option<String>,
145 malwarebazaar_enabled: bool,
146 urlhaus_enabled: bool,
147 abuse_ch_auth_key: Option<String>,
148 metadefender_api_key: Option<String>,
149 }
150
151 impl ScanPipeline {
152 /// Create a new pipeline, compiling YARA rules from the configured directory.
153 pub fn new(config: &ScanConfig) -> Result<Self, String> {
154 let yara_rules = yara::compile_rules_from_dir(&config.yara_rules_dir)?;
155
156 Ok(ScanPipeline {
157 yara_rules,
158 clamav_socket: config.clamav_socket.clone(),
159 malwarebazaar_enabled: config.malwarebazaar_enabled,
160 urlhaus_enabled: config.urlhaus_enabled,
161 abuse_ch_auth_key: config.abuse_ch_auth_key.clone(),
162 metadefender_api_key: config.metadefender_api_key.clone(),
163 })
164 }
165
166 /// Assert at startup that at least one real AV layer is live. Refuse to
167 /// boot otherwise — ClamAV's FailOpen policy means a dead clamd
168 /// silently passes every upload as Clean, and a YARA-rules-empty deploy
169 /// gives the same false sense of coverage. If the operator configured
170 /// scanning, a misconfiguration must be loud at boot, not silent at runtime.
171 pub async fn assert_live(&self) -> Result<(), String> {
172 let mut live_layers: Vec<&str> = Vec::new();
173 if let Some(ref socket) = self.clamav_socket {
174 match clamav::ping(socket).await {
175 Ok(()) => live_layers.push("clamav"),
176 Err(e) => {
177 return Err(format!("ClamAV socket {socket} unreachable: {e}"));
178 }
179 }
180 }
181 if self.yara_rules.is_some() {
182 live_layers.push("yara");
183 }
184 if self.malwarebazaar_enabled {
185 live_layers.push("malwarebazaar");
186 }
187 if self.urlhaus_enabled {
188 live_layers.push("urlhaus");
189 }
190 if self.metadefender_api_key.is_some() {
191 live_layers.push("metadefender");
192 }
193 if live_layers.is_empty() {
194 return Err(
195 "Scanning configured but no AV layer is live (no ClamAV socket, \
196 no YARA rules, no remote API keys). Refusing to boot — the \
197 FailOpen policy would pass every upload as Clean."
198 .to_string(),
199 );
200 }
201 tracing::info!(layers = ?live_layers, "scan pipeline live layers asserted");
202 Ok(())
203 }
204
205 /// Run all applicable scanning layers against file data.
206 ///
207 /// CPU-bound layers (sha256, content-type, structural, archive, yara) run
208 /// on a blocking-pool thread via `spawn_blocking` so they don't stall the
209 /// tokio runtime — the 4 in-process layers can each take seconds on a
210 /// 100 MB file. Network-bound layers (ClamAV, MalwareBazaar) run
211 /// concurrently with the sync block via `tokio::join!`.
212 pub async fn scan(self: std::sync::Arc<Self>, data: Vec<u8>, file_type: FileType) -> ScanResult {
213 let file_size = data.len() as u64;
214 let data = std::sync::Arc::<[u8]>::from(data);
215
216 // Sync layers + hash, off the runtime
217 let sync_data = std::sync::Arc::clone(&data);
218 let sync_self = std::sync::Arc::clone(&self);
219 let sync_fut = tokio::task::spawn_blocking(move || sync_self.run_sync_layers(&sync_data, file_type));
220
221 // Async layers — ClamAV needs the bytes, MalwareBazaar needs only the hash
222 // but the hash is computed in the sync block, so we run MalwareBazaar
223 // after the sync block returns (its endpoint is fast). ClamAV can run
224 // concurrently with the sync block.
225 let clamav_data = std::sync::Arc::clone(&data);
226 let clamav_socket = self.clamav_socket.clone();
227 let clamav_fut = async move {
228 match clamav_socket {
229 Some(socket) => clamav::scan_with_clamav(&socket, &clamav_data).await,
230 None => LayerResult {
231 layer: "clamav",
232 verdict: LayerVerdict::Skip,
233 detail: Some("ClamAV not configured".to_string()),
234 },
235 }
236 };
237
238 // Layer 7: URLhaus — extract URLs from the bytes, query hosts.
239 // Runs concurrently with the sync block (independent of the hash).
240 let urlhaus_data = std::sync::Arc::clone(&data);
241 let urlhaus_enabled = self.urlhaus_enabled;
242 let urlhaus_key = self.abuse_ch_auth_key.clone();
243 let urlhaus_fut = async move {
244 if urlhaus_enabled {
245 urlhaus::check_urlhaus(&urlhaus_data, urlhaus_key.as_deref()).await
246 } else {
247 LayerResult {
248 layer: "urlhaus",
249 verdict: LayerVerdict::Skip,
250 detail: Some("URLhaus lookups disabled".to_string()),
251 }
252 }
253 };
254
255 let (sync_result, clamav_result, urlhaus_result) =
256 tokio::join!(sync_fut, clamav_fut, urlhaus_fut);
257 let (mut layers, sha256) = sync_result
258 .expect("scan_sync spawn_blocking panicked");
259 layers.push(clamav_result);
260 layers.push(urlhaus_result);
261
262 // Layer 6: MalwareBazaar — needs the hash from the sync block
263 layers.push(if self.malwarebazaar_enabled {
264 hash_lookup::check_malwarebazaar(&sha256, self.abuse_ch_auth_key.as_deref()).await
265 } else {
266 LayerResult {
267 layer: "malwarebazaar",
268 verdict: LayerVerdict::Skip,
269 detail: Some("MalwareBazaar lookups disabled".to_string()),
270 }
271 });
272
273 // Layer 9: MetaDefender (second-opinion). Only invoked when a prior
274 // layer flagged the file as suspicious — keeps us within the free-tier
275 // quota and avoids spending budget on uncontroversial uploads.
276 layers.push(if suspicion_present(&layers) {
277 metadefender::check_metadefender(&sha256, self.metadefender_api_key.as_deref()).await
278 } else {
279 LayerResult {
280 layer: "metadefender",
281 verdict: LayerVerdict::Skip,
282 detail: Some("No prior suspicion; second-opinion not invoked".to_string()),
283 }
284 });
285
286 let status = final_status(&layers);
287
288 ScanResult {
289 status,
290 layers,
291 sha256,
292 file_size,
293 }
294 }
295
296 /// Streaming counterpart to `scan`. Runs against a spooled tempfile so
297 /// the >100 MB case doesn't hold the whole object in RAM. The CPU
298 /// layers operate on a memory mapping (`spool::mmap_read`) — pages are
299 /// demand-paged by the kernel as goblin / yara-x / archive walk them —
300 /// and ClamAV streams the file via INSTREAM frames.
301 pub async fn scan_stream(
302 self: std::sync::Arc<Self>,
303 spool: spool::SpoolHandle,
304 file_type: FileType,
305 ) -> ScanResult {
306 let file_size = std::fs::metadata(spool.path())
307 .map(|m| m.len())
308 .unwrap_or(0);
309
310 let map = match spool::mmap_read(spool.path()) {
311 Ok(m) => std::sync::Arc::new(m),
312 Err(e) => {
313 let layer = LayerResult {
314 layer: "spool",
315 verdict: LayerVerdict::Error,
316 detail: Some(e),
317 };
318 return ScanResult {
319 status: final_status(std::slice::from_ref(&layer)),
320 layers: vec![layer],
321 sha256: String::new(),
322 file_size,
323 };
324 }
325 };
326
327 let sync_map = std::sync::Arc::clone(&map);
328 let sync_self = std::sync::Arc::clone(&self);
329 let sync_fut = tokio::task::spawn_blocking(move || sync_self.run_sync_layers(&sync_map, file_type));
330
331 let clamav_socket = self.clamav_socket.clone();
332 let clamav_path = spool.path().to_path_buf();
333 let clamav_fut = async move {
334 match clamav_socket {
335 Some(socket) => match tokio::fs::File::open(&clamav_path).await {
336 Ok(file) => clamav::scan_with_clamav_stream(&socket, file).await,
337 Err(e) => LayerResult {
338 layer: "clamav",
339 verdict: LayerVerdict::Error,
340 detail: Some(format!("open spool for clamav: {e}")),
341 },
342 },
343 None => LayerResult {
344 layer: "clamav",
345 verdict: LayerVerdict::Skip,
346 detail: Some("ClamAV not configured".to_string()),
347 },
348 }
349 };
350
351 let urlhaus_map = std::sync::Arc::clone(&map);
352 let urlhaus_enabled = self.urlhaus_enabled;
353 let urlhaus_key = self.abuse_ch_auth_key.clone();
354 let urlhaus_fut = async move {
355 if urlhaus_enabled {
356 urlhaus::check_urlhaus(&urlhaus_map, urlhaus_key.as_deref()).await
357 } else {
358 LayerResult {
359 layer: "urlhaus",
360 verdict: LayerVerdict::Skip,
361 detail: Some("URLhaus lookups disabled".to_string()),
362 }
363 }
364 };
365
366 let (sync_result, clamav_result, urlhaus_result) =
367 tokio::join!(sync_fut, clamav_fut, urlhaus_fut);
368 let (mut layers, sha256) = sync_result
369 .expect("scan_stream sync spawn_blocking panicked");
370 layers.push(clamav_result);
371 layers.push(urlhaus_result);
372
373 layers.push(if self.malwarebazaar_enabled {
374 hash_lookup::check_malwarebazaar(&sha256, self.abuse_ch_auth_key.as_deref()).await
375 } else {
376 LayerResult {
377 layer: "malwarebazaar",
378 verdict: LayerVerdict::Skip,
379 detail: Some("MalwareBazaar lookups disabled".to_string()),
380 }
381 });
382
383 layers.push(if suspicion_present(&layers) {
384 metadefender::check_metadefender(&sha256, self.metadefender_api_key.as_deref()).await
385 } else {
386 LayerResult {
387 layer: "metadefender",
388 verdict: LayerVerdict::Skip,
389 detail: Some("No prior suspicion; second-opinion not invoked".to_string()),
390 }
391 });
392
393 let status = final_status(&layers);
394 drop(map);
395 drop(spool);
396
397 ScanResult {
398 status,
399 layers,
400 sha256,
401 file_size,
402 }
403 }
404
405 /// CPU-bound layers + SHA-256. Pure sync; safe to call from `spawn_blocking`.
406 fn run_sync_layers(&self, data: &[u8], file_type: FileType) -> (Vec<LayerResult>, String) {
407 let mut layers = Vec::with_capacity(5);
408
409 // SHA-256 hash for audit + MalwareBazaar lookup
410 let sha256 = {
411 let mut hasher = Sha256::new();
412 hasher.update(data);
413 hex::encode(hasher.finalize())
414 };
415
416 layers.push(content_type::verify_content_type(data, file_type));
417 layers.push(structural::analyze_binary(data, file_type));
418 layers.push(archive::check_archive_safety(data, file_type));
419 layers.push(match self.yara_rules {
420 Some(ref rules) => yara::scan_with_yara(rules, data),
421 None => LayerResult {
422 layer: "yara",
423 verdict: LayerVerdict::Skip,
424 detail: Some("No YARA rules loaded".to_string()),
425 },
426 });
427 layers.push(signing_macos::verify_apple_signature(data, file_type));
428 layers.push(signing_windows::verify_authenticode(data, file_type));
429 layers.push(signing_linux::verify_appimage_signature(data, file_type));
430
431 (layers, sha256)
432 }
433 }
434
435 #[cfg(test)]
436 mod tests {
437 use super::*;
438
439 #[test]
440 fn layer_verdict_serializes_lowercase() {
441 assert_eq!(
442 serde_json::to_string(&LayerVerdict::Pass).unwrap(),
443 "\"pass\""
444 );
445 assert_eq!(
446 serde_json::to_string(&LayerVerdict::Fail).unwrap(),
447 "\"fail\""
448 );
449 assert_eq!(
450 serde_json::to_string(&LayerVerdict::Skip).unwrap(),
451 "\"skip\""
452 );
453 assert_eq!(
454 serde_json::to_string(&LayerVerdict::Error).unwrap(),
455 "\"error\""
456 );
457 }
458
459 #[test]
460 fn scan_result_quarantined_on_any_fail() {
461 let layers = [
462 LayerResult {
463 layer: "test1",
464 verdict: LayerVerdict::Pass,
465 detail: None,
466 },
467 LayerResult {
468 layer: "test2",
469 verdict: LayerVerdict::Fail,
470 detail: Some("bad".to_string()),
471 },
472 ];
473 let has_fail = layers.iter().any(|l| l.verdict == LayerVerdict::Fail);
474 assert!(has_fail);
475 }
476
477 #[test]
478 fn sha256_computation() {
479 let mut hasher = Sha256::new();
480 hasher.update(b"hello");
481 let hash = hex::encode(hasher.finalize());
482 assert_eq!(
483 hash,
484 "2cf24dba5fb0a30e26e83b2ac5b9e29e1b161e5c1fa7425e73043362938b9824"
485 );
486 }
487
488 // -- Pipeline integration tests --
489
490 /// Create a minimal ScanPipeline with no external deps (no YARA, no ClamAV, no MalwareBazaar).
491 /// Wrapped in `Arc` because `scan` consumes `Arc<Self>` (see `pub async fn scan`).
492 fn make_pipeline() -> std::sync::Arc<ScanPipeline> {
493 std::sync::Arc::new(ScanPipeline {
494 yara_rules: None,
495 clamav_socket: None,
496 malwarebazaar_enabled: false,
497 urlhaus_enabled: false,
498 abuse_ch_auth_key: None,
499 metadefender_api_key: None,
500 })
501 }
502
503 #[tokio::test]
504 async fn pipeline_clean_download_passes() {
505 let pipeline = make_pipeline();
506 let result = pipeline.clone().scan(b"just some file content".to_vec(), FileType::Download).await;
507 assert_eq!(result.status, FileScanStatus::Clean);
508 assert_eq!(result.file_size, 22);
509 assert!(!result.sha256.is_empty());
510 assert_eq!(result.layers.len(), 11);
511 }
512
513 #[tokio::test]
514 async fn pipeline_unrecognized_audio_quarantined() {
515 let pipeline = make_pipeline();
516 // Unrecognized data claimed as audio should be rejected by content_type layer
517 let result = pipeline.clone().scan(b"audio data here".to_vec(), FileType::Audio).await;
518 assert_eq!(result.status, FileScanStatus::Quarantined);
519 }
520
521 #[tokio::test]
522 async fn pipeline_clean_cover_passes() {
523 let pipeline = make_pipeline();
524 // PNG magic bytes
525 let png = [0x89, 0x50, 0x4E, 0x47, 0x0D, 0x0A, 0x1A, 0x0A];
526 let result = pipeline.clone().scan(png.to_vec(), FileType::Cover).await;
527 assert_eq!(result.status, FileScanStatus::Clean);
528 }
529
530 #[tokio::test]
531 async fn pipeline_pe_as_audio_quarantined() {
532 let pipeline = make_pipeline();
533 // PE magic bytes — content-type layer should detect application/* and fail
534 let pe_header = b"MZ\x90\x00\x03\x00\x00\x00";
535 let result = pipeline.clone().scan(pe_header.to_vec(), FileType::Audio).await;
536 assert_eq!(result.status, FileScanStatus::Quarantined);
537 // Verify content_type layer produced the fail
538 let content_type_layer = result.layers.iter().find(|l| l.layer == "content_type").unwrap();
539 assert_eq!(content_type_layer.verdict, LayerVerdict::Fail);
540 }
541
542 #[tokio::test]
543 async fn pipeline_pe_as_cover_quarantined() {
544 let pipeline = make_pipeline();
545 let pe_header = b"MZ\x90\x00\x03\x00\x00\x00";
546 let result = pipeline.clone().scan(pe_header.to_vec(), FileType::Cover).await;
547 assert_eq!(result.status, FileScanStatus::Quarantined);
548 }
549
550 #[tokio::test]
551 async fn pipeline_sha256_is_deterministic() {
552 let pipeline = make_pipeline();
553 let data = b"deterministic hash test";
554 let r1 = pipeline.clone().scan(data.to_vec(), FileType::Download).await;
555 let r2 = pipeline.clone().scan(data.to_vec(), FileType::Download).await;
556 assert_eq!(r1.sha256, r2.sha256);
557 }
558
559 #[tokio::test]
560 async fn pipeline_skips_optional_layers_when_unconfigured() {
561 let pipeline = make_pipeline();
562 let result = pipeline.clone().scan(b"test".to_vec(), FileType::Download).await;
563
564 let yara = result.layers.iter().find(|l| l.layer == "yara").unwrap();
565 assert_eq!(yara.verdict, LayerVerdict::Skip);
566
567 let clamav = result.layers.iter().find(|l| l.layer == "clamav").unwrap();
568 assert_eq!(clamav.verdict, LayerVerdict::Skip);
569
570 let mb = result.layers.iter().find(|l| l.layer == "malwarebazaar").unwrap();
571 assert_eq!(mb.verdict, LayerVerdict::Skip);
572
573 let uh = result.layers.iter().find(|l| l.layer == "urlhaus").unwrap();
574 assert_eq!(uh.verdict, LayerVerdict::Skip);
575 }
576
577 #[tokio::test]
578 async fn pipeline_always_produces_11_layers() {
579 let pipeline = make_pipeline();
580 for file_type in [FileType::Audio, FileType::Cover, FileType::Download] {
581 let result = pipeline.clone().scan(b"data".to_vec(), file_type).await;
582 assert_eq!(result.layers.len(), 11, "Expected 11 layers for {:?}", file_type);
583 }
584 }
585
586 #[test]
587 fn suspicion_present_on_fail() {
588 let layers = vec![pass("content_type"), fail("yara")];
589 assert!(suspicion_present(&layers));
590 }
591
592 #[test]
593 fn suspicion_present_on_fail_closed_error() {
594 let layers = vec![pass("content_type"), err("archive")];
595 assert!(suspicion_present(&layers));
596 }
597
598 #[test]
599 fn no_suspicion_when_fail_open_error_only() {
600 // External-layer errors are operational noise, not malware signals;
601 // they must not invoke MetaDefender.
602 let layers = vec![pass("content_type"), err("malwarebazaar"), err("urlhaus")];
603 assert!(!suspicion_present(&layers));
604 }
605
606 #[test]
607 fn no_suspicion_when_all_clean() {
608 let layers = vec![pass("content_type"), skip("yara"), pass("structural")];
609 assert!(!suspicion_present(&layers));
610 }
611
612 #[tokio::test]
613 async fn pipeline_errors_held_for_review() {
614 // Errors from fail-closed layers (archive is in-process deterministic)
615 // should hold the file for admin review.
616 let pipeline = make_pipeline();
617 // Corrupted ZIP magic bytes — archive layer returns Error
618 let mut data = vec![0x50, 0x4B, 0x03, 0x04];
619 data.extend_from_slice(&[0xFF; 100]);
620 let result = pipeline.clone().scan(data, FileType::Download).await;
621 let archive = result.layers.iter().find(|l| l.layer == "archive").unwrap();
622 assert_eq!(archive.verdict, LayerVerdict::Error);
623 assert_eq!(result.status, FileScanStatus::HeldForReview);
624 }
625
626 // -- Per-layer fail policy tests --
627
628 fn err(layer: &'static str) -> LayerResult {
629 LayerResult { layer, verdict: LayerVerdict::Error, detail: None }
630 }
631 fn pass(layer: &'static str) -> LayerResult {
632 LayerResult { layer, verdict: LayerVerdict::Pass, detail: None }
633 }
634 fn skip(layer: &'static str) -> LayerResult {
635 LayerResult { layer, verdict: LayerVerdict::Skip, detail: None }
636 }
637 fn fail(layer: &'static str) -> LayerResult {
638 LayerResult { layer, verdict: LayerVerdict::Fail, detail: None }
639 }
640
641 #[test]
642 fn final_status_clean_when_all_pass() {
643 let layers = vec![pass("content_type"), pass("structural"), pass("archive"), skip("yara"), skip("clamav"), skip("malwarebazaar")];
644 assert_eq!(final_status(&layers), FileScanStatus::Clean);
645 }
646
647 #[test]
648 fn final_status_quarantined_on_any_fail() {
649 let layers = vec![pass("content_type"), fail("yara"), skip("clamav")];
650 assert_eq!(final_status(&layers), FileScanStatus::Quarantined);
651 }
652
653 #[test]
654 fn final_status_fail_beats_error() {
655 // A Fail anywhere supersedes any Error, regardless of policy.
656 let layers = vec![err("malwarebazaar"), fail("yara")];
657 assert_eq!(final_status(&layers), FileScanStatus::Quarantined);
658 }
659
660 #[test]
661 fn final_status_held_on_fail_closed_error() {
662 // archive is FailClosed — its Error must hold the file.
663 let layers = vec![pass("content_type"), err("archive"), skip("clamav")];
664 assert_eq!(final_status(&layers), FileScanStatus::HeldForReview);
665 }
666
667 #[test]
668 fn final_status_clean_on_fail_open_error_only() {
669 // malwarebazaar is FailOpen — its Error must NOT hold the file.
670 // This is the regression of 2026-05-10 that motivated the audit.
671 let layers = vec![pass("content_type"), pass("structural"), pass("archive"), skip("yara"), skip("clamav"), err("malwarebazaar")];
672 assert_eq!(final_status(&layers), FileScanStatus::Clean);
673 }
674
675 #[test]
676 fn final_status_clean_when_all_external_layers_error() {
677 // Worst-case external-services outage: every network/daemon layer
678 // erroring at once. As long as the in-process layers pass, the file
679 // is Clean. Health is surfaced separately via per-layer monitoring.
680 let layers = vec![pass("content_type"), pass("structural"), pass("archive"), skip("yara"), err("clamav"), err("malwarebazaar")];
681 assert_eq!(final_status(&layers), FileScanStatus::Clean);
682 }
683
684 #[test]
685 fn final_status_held_on_unknown_layer_error() {
686 // Defensive default: an unknown layer name that errors falls through
687 // to FailClosed. This is what catches a new layer added without
688 // wiring its policy into `error_policy_for`.
689 let layers = vec![pass("content_type"), err("brand_new_layer_someone_forgot_to_register")];
690 assert_eq!(final_status(&layers), FileScanStatus::HeldForReview);
691 }
692
693 #[test]
694 fn error_policy_for_all_known_layers() {
695 // Every layer name produced by the pipeline must have an explicit
696 // declaration in `error_policy_for`. The default branch is reserved
697 // for genuine programmer error (new layer, forgot to register).
698 for name in ["content_type", "structural", "archive", "yara", "clamav", "malwarebazaar"] {
699 let policy = error_policy_for(name);
700 // Both values are valid; we just want this to not hit the default.
701 // If a layer is renamed without updating `error_policy_for`, this
702 // test still passes (the rename produces a new unknown name)
703 // — but the per-layer name tests below catch that.
704 let _ = policy;
705 }
706 }
707
708 #[test]
709 fn content_type_is_fail_closed() { assert_eq!(error_policy_for("content_type"), ErrorPolicy::FailClosed); }
710 #[test]
711 fn structural_is_fail_closed() { assert_eq!(error_policy_for("structural"), ErrorPolicy::FailClosed); }
712 #[test]
713 fn archive_is_fail_closed() { assert_eq!(error_policy_for("archive"), ErrorPolicy::FailClosed); }
714 #[test]
715 fn yara_is_fail_closed() { assert_eq!(error_policy_for("yara"), ErrorPolicy::FailClosed); }
716 #[test]
717 fn clamav_is_fail_open() { assert_eq!(error_policy_for("clamav"), ErrorPolicy::FailOpen); }
718 #[test]
719 fn malwarebazaar_is_fail_open() { assert_eq!(error_policy_for("malwarebazaar"), ErrorPolicy::FailOpen); }
720 #[test]
721 fn urlhaus_is_fail_open() { assert_eq!(error_policy_for("urlhaus"), ErrorPolicy::FailOpen); }
722 #[test]
723 fn signing_macos_is_fail_open() { assert_eq!(error_policy_for("signing_macos"), ErrorPolicy::FailOpen); }
724 #[test]
725 fn metadefender_is_fail_open() { assert_eq!(error_policy_for("metadefender"), ErrorPolicy::FailOpen); }
726 #[test]
727 fn signing_windows_is_fail_open() { assert_eq!(error_policy_for("signing_windows"), ErrorPolicy::FailOpen); }
728 #[test]
729 fn signing_linux_is_fail_open() { assert_eq!(error_policy_for("signing_linux"), ErrorPolicy::FailOpen); }
730 }
731