Skip to main content

max / makenotwork

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