Skip to main content

max / makenotwork

server: supervise scheduler, cap internal broadcast, unblock git2 walk, single-pass archive scan (ultra-fuzz Run 2 Perf CRITICAL + SERIOUS) CRITICAL: the scheduler loop ran every periodic job as a bare sequential .await inside one tokio::spawn; a panic in any job unwound the loop and permanently halted ALL background maintenance (cleanup, S3-orphan delete, webhook retry, build dispatch) with no alert (the WAM call lived inside the dead loop). The job sequence now runs in a supervised child task: a panic surfaces as a JoinError, is logged and (rate-limited) opens a WAM ticket from the surviving loop, the tick aborts, and the next tick re-runs everything. SERIOUS (broadcast): the internal CLI follower-broadcast had no recipient cap while the public twin enforces BROADCAST_MAX_RECIPIENTS; apply the same guard and roll back the 24h slot. SERIOUS (git2): the push-refs handler walked up to 50 commits with synchronous git2 disk I/O directly on a tokio worker thread; move the walk into spawn_blocking like every other git path. SERIOUS (scan): archive uploads were decompressed twice — once to count for bomb defense, once to buffer+scan interior content. inspect_archive now derives both the "archive" and "archive_nested" verdicts from a single tee_decompress per entry (full count for bomb, ≤100MB prefix for content); check_archive_safety, check_archive_safety_path, and scan_nested_contents delegate to it so the two layers can't drift. The buffered scan path also takes the S3 body as bytes::Bytes (download_object_buf) instead of an aggregate-then-to_vec double copy. Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
Author: Max Johnson <me@maxj.phd> · 2026-06-23 15:56 UTC
Commit: d49b61cd67a777828fa0eb1ba7311f4f02d0bc4f
Parent: a582754
12 files changed, +656 insertions, -66 deletions
M server/Cargo.lock +2
@@ -4312,6 +4312,7 @@ dependencies = [
4312 4312 "axum",
4313 4313 "axum-extra",
4314 4314 "base64 0.22.1",
4315 + "bytes",
4315 4316 "bzip2 0.4.4",
4316 4317 "chacha20poly1305",
4317 4318 "chrono",
@@ -6218,6 +6219,7 @@ version = "0.1.1"
6218 6219 dependencies = [
6219 6220 "aws-config",
6220 6221 "aws-sdk-s3",
6222 + "bytes",
6221 6223 "tokio",
6222 6224 "tracing",
6223 6225 ]
M server/Cargo.toml +1
@@ -103,6 +103,7 @@ log = "0.4"
103 103 # Error handling
104 104 thiserror = "2.0.18"
105 105 anyhow = "1.0.102"
106 + bytes = "1"
106 107
107 108 # Email validation (used at notify-me signup and guest-checkout entry points)
108 109 email_address = "0.2"
M server/src/routes/api/internal/cli_features.rs +12
@@ -182,6 +182,18 @@ pub(super) async fn send_broadcast(
182 182 let followers = db::follows::get_follower_emails(&state.db, req.user_id).await?;
183 183 let count = followers.len();
184 184
185 + if count > constants::BROADCAST_MAX_RECIPIENTS {
186 + // Mirror the public twin (routes/api/users/broadcast.rs): refuse to
187 + // materialize an unbounded recipient set in memory, and roll back the
188 + // 24h rate-limit slot so the creator can retry once the cap is lifted.
189 + let _ = db::users::clear_broadcast_at(&state.db, req.user_id).await;
190 + return Err(AppError::validation(format!(
191 + "Broadcast would reach {count} followers, above the per-send limit of {}. \
192 + Email info@makenot.work to lift the cap for your account.",
193 + constants::BROADCAST_MAX_RECIPIENTS
194 + )));
195 + }
196 +
185 197 if count > 0 {
186 198 let creator_name = db_user.display_name.as_deref()
187 199 .unwrap_or(&db_user.username)
M server/src/routes/git_issues/push_refs.rs +52 -41
@@ -132,49 +132,60 @@ pub(super) async fn process_push(
132 132 .await?
133 133 .ok_or(AppError::NotFound)?;
134 134
135 - // Open the bare repo and collect commit refs synchronously (git2 types are !Send)
136 - let commit_refs = {
137 - let root = repos_root(&state)?;
138 - let git_repo = crate::git::open_repo(&root, &req.repo_owner, &req.repo_name)
139 - .context("open git repo")?;
140 -
141 - let after_oid = git2::Oid::from_str(&req.after)
142 - .map_err(|e| AppError::BadRequest(format!("invalid 'after' oid: {}", e)))?;
143 -
144 - let mut revwalk = git_repo.revwalk()
145 - .context("revwalk init")?;
146 - revwalk.push(after_oid)
147 - .context("revwalk push")?;
148 -
149 - let is_new_branch = req.before.chars().all(|c| c == '0');
150 - if !is_new_branch
151 - && let Ok(before_oid) = git2::Oid::from_str(&req.before)
152 - {
153 - let _ = revwalk.hide(before_oid);
154 - }
155 -
156 - let max_commits = if is_new_branch { 1 } else { 50 };
157 - let mut collected: Vec<(String, String, Vec<IssueRef>)> = Vec::new();
135 + // Open the bare repo and collect commit refs on a blocking thread. git2 is
136 + // synchronous disk I/O and its types are !Send, so the whole walk runs in
137 + // spawn_blocking (every other git path does the same via crate::git) — doing
138 + // it inline parks a tokio worker thread for the duration (Run #2 Perf SERIOUS).
139 + // The walk creates and drops all git2 types internally and returns only the
140 + // owned, Send `Vec` of collected refs.
141 + let root = repos_root(&state)?;
142 + let repo_owner = req.repo_owner.clone();
143 + let repo_name = req.repo_name.clone();
144 + let after = req.after.clone();
145 + let before = req.before.clone();
146 + let commit_refs = tokio::task::spawn_blocking(
147 + move || -> Result<Vec<(String, String, Vec<IssueRef>)>> {
148 + let git_repo = crate::git::open_repo(&root, &repo_owner, &repo_name)
149 + .context("open git repo")?;
150 +
151 + let after_oid = git2::Oid::from_str(&after)
152 + .map_err(|e| AppError::BadRequest(format!("invalid 'after' oid: {}", e)))?;
153 +
154 + let mut revwalk = git_repo.revwalk().context("revwalk init")?;
155 + revwalk.push(after_oid).context("revwalk push")?;
156 +
157 + let is_new_branch = before.chars().all(|c| c == '0');
158 + if !is_new_branch
159 + && let Ok(before_oid) = git2::Oid::from_str(&before)
160 + {
161 + let _ = revwalk.hide(before_oid);
162 + }
158 163
159 - for oid_result in revwalk.take(max_commits) {
160 - let oid = match oid_result {
161 - Ok(o) => o,
162 - Err(_) => continue,
163 - };
164 - let commit = match git_repo.find_commit(oid) {
165 - Ok(c) => c,
166 - Err(_) => continue,
167 - };
168 - let message = commit.message().unwrap_or("");
169 - let refs = parse_issue_refs(message);
170 - if !refs.is_empty() {
171 - let oid_str = oid.to_string();
172 - let short_oid = oid_str[..7.min(oid_str.len())].to_string();
173 - collected.push((oid_str, short_oid, refs));
164 + let max_commits = if is_new_branch { 1 } else { 50 };
165 + let mut collected: Vec<(String, String, Vec<IssueRef>)> = Vec::new();
166 +
167 + for oid_result in revwalk.take(max_commits) {
168 + let oid = match oid_result {
169 + Ok(o) => o,
170 + Err(_) => continue,
171 + };
172 + let commit = match git_repo.find_commit(oid) {
173 + Ok(c) => c,
174 + Err(_) => continue,
175 + };
176 + let message = commit.message().unwrap_or("");
177 + let refs = parse_issue_refs(message);
178 + if !refs.is_empty() {
179 + let oid_str = oid.to_string();
180 + let short_oid = oid_str[..7.min(oid_str.len())].to_string();
181 + collected.push((oid_str, short_oid, refs));
182 + }
174 183 }
175 - }
176 - collected
177 - };
184 + Ok(collected)
185 + },
186 + )
187 + .await
188 + .context("git push-refs walk task")??;
178 189
179 190 // Now process DB operations (async-safe, git2 types are dropped)
180 191 let mut processed = 0u32;
M server/src/scanning/archive.rs +481 -8
@@ -83,22 +83,497 @@ fn has_zip_eocd(data: &[u8]) -> bool {
83 83 /// Check a file for archive / decompression-bomb safety issues.
84 84 /// Runs regardless of claimed type so a disguised archive is still inspected.
85 85 pub fn check_archive_safety(data: &[u8], file_type: FileType) -> LayerResult {
86 - // If an archive is disguised as a cover image, layer 1 (content_type)
87 - // handles the type mismatch; skip the archive walk for covers.
88 - if file_type == FileType::Cover {
89 - return skip("Archive check skipped for cover images");
86 + inspect_archive(data, file_type, None).0
87 + }
88 +
89 + /// Outcome of buffering a decompressed entry's prefix for the interior scan.
90 + enum ContentBuf {
91 + /// Fully buffered within the per-entry ceiling and shared budget.
92 + Buffered(Vec<u8>),
93 + /// Decompressed size exceeded `INTERIOR_ENTRY_MAX` — cannot content-scan.
94 + Overflow,
95 + /// The shared interior budget was exhausted before this entry finished.
96 + BudgetExceeded,
97 + }
98 +
99 + /// Decompress `reader` to completion (or until well past the bomb cap),
100 + /// returning the FULL decompressed byte count (for bomb accounting) and — when
101 + /// `want_content` is set — the buffered prefix (≤ `INTERIOR_ENTRY_MAX`, charged
102 + /// against `budget`) for the interior content scan. A single decompression now
103 + /// serves both the bomb-defense walk and the interior scan; they previously
104 + /// decompressed every entry independently (Run #2 Performance SERIOUS).
105 + ///
106 + /// `Err(detail)` is a mid-stream decode failure: the caller decides what that
107 + /// means for each dimension (ZIP bomb accounting uses a conservative estimate
108 + /// and continues; a single-stream compressor fails closed).
109 + fn tee_decompress(
110 + reader: &mut dyn Read,
111 + bomb_abs_limit: u64,
112 + want_content: bool,
113 + budget: &mut u64,
114 + ) -> Result<(u64, Option<ContentBuf>), (u64, String)> {
115 + let mut counted: u64 = 0;
116 + let mut buf = [0u8; 8192];
117 + let mut content = if want_content {
118 + Some(ContentBuf::Buffered(Vec::new()))
119 + } else {
120 + None
121 + };
122 + loop {
123 + match reader.read(&mut buf) {
124 + Ok(0) => break,
125 + Ok(n) => {
126 + counted += n as u64;
127 + if let Some(ContentBuf::Buffered(ref mut v)) = content {
128 + if v.len() + n > INTERIOR_ENTRY_MAX {
129 + content = Some(ContentBuf::Overflow);
130 + } else if (n as u64) > *budget {
131 + content = Some(ContentBuf::BudgetExceeded);
132 + } else {
133 + *budget -= n as u64;
134 + v.extend_from_slice(&buf[..n]);
135 + }
136 + }
137 + // Stop once the bomb cap is blown AND there's no more content to
138 + // buffer — reading further serves neither dimension.
139 + if counted > bomb_abs_limit
140 + && !matches!(content, Some(ContentBuf::Buffered(_)))
141 + {
142 + break;
143 + }
144 + }
145 + Err(e) => return Err((counted, format!("{e}"))),
146 + }
90 147 }
148 + Ok((counted, content))
149 + }
150 +
151 + /// Single-pass archive inspection. Decompresses each top-level entry ONCE and
152 + /// derives BOTH the bomb-defense verdict (`"archive"`) and the interior content
153 + /// verdict (`"archive_nested"`) from the same decompression. `check_archive_safety`,
154 + /// `check_archive_safety_path`, and `scan_nested_contents` all delegate here so
155 + /// the two layers can never drift apart.
156 + ///
157 + /// `yara_rules == None` still runs the interior content scan (content-type +
158 + /// structural layers); it only skips the YARA sub-check. Callers that want bomb
159 + /// defense only (`check_archive_safety`) take the first element and discard the
160 + /// second.
161 + pub fn inspect_archive(
162 + data: &[u8],
163 + file_type: FileType,
164 + yara_rules: Option<&yara_x::Rules>,
165 + ) -> (LayerResult, LayerResult) {
166 + // A cover-disguised archive: layer 1 (content_type) handles the type
167 + // mismatch, so the bomb walk is skipped — but the interior is still scanned
168 + // (the nested layer never gated on file type). `bomb` off, `content` on.
169 + let bomb = file_type != FileType::Cover;
91 170
92 171 match detect_kind(data) {
93 - Some(ArchiveKind::Zip) => inspect_zip(Cursor::new(data)),
94 - Some(stream) => inspect_compressed_stream(stream, data.len() as u64, Cursor::new(data)),
172 + Some(ArchiveKind::Zip) => walk_zip(Cursor::new(data), bomb, true, yara_rules),
173 + Some(stream) => walk_compressed(stream, data.len() as u64, Cursor::new(data), bomb, true, yara_rules),
95 174 None if has_zip_eocd(data) => {
96 175 // Prefixed / self-extracting ZIP: no offset-0 magic, but a real
97 176 // central directory at the tail. ZipArchive locates it from the end.
98 - inspect_zip(Cursor::new(data))
177 + walk_zip(Cursor::new(data), bomb, true, yara_rules)
178 + }
179 + None => (
180 + if bomb { skip("Not a recognized archive") } else { skip("Archive check skipped for cover images") },
181 + nested(LayerVerdict::Skip, "Not an archive; no interior to scan".to_string()),
182 + ),
183 + }
184 + }
185 +
186 + /// Bomb-defense result to surface when the bomb dimension is disabled (covers).
187 + fn bomb_disabled_archive() -> LayerResult {
188 + skip("Archive check skipped for cover images")
189 + }
190 +
191 + /// Walk a ZIP once, computing the bomb-defense verdict (when `bomb`) and the
192 + /// interior content verdict (when `content`). Returns `(archive, archive_nested)`.
193 + fn walk_zip<R: std::io::Read + std::io::Seek>(
194 + reader: R,
195 + bomb: bool,
196 + content: bool,
197 + yara_rules: Option<&yara_x::Rules>,
198 + ) -> (LayerResult, LayerResult) {
199 + let mut archive = match zip::ZipArchive::new(reader) {
200 + Ok(a) => a,
201 + Err(e) => {
202 + return (
203 + if bomb {
204 + LayerResult {
205 + layer: "archive",
206 + verdict: LayerVerdict::Error,
207 + detail: Some(format!("Failed to parse ZIP: {e}")),
208 + }
209 + } else {
210 + bomb_disabled_archive()
211 + },
212 + if content {
213 + nested(LayerVerdict::Error, format!("cannot open nested ZIP: {e}"))
214 + } else {
215 + nested(LayerVerdict::Skip, String::new())
216 + },
217 + );
218 + }
219 + };
220 +
221 + let count = archive.len();
222 +
223 + // `archive_res`/`nested_res` freeze the first non-clean verdict for each
224 + // dimension; the loop keeps running for the OTHER dimension so a bomb Fail
225 + // and a content Fail are both surfaced (and a later bomb can still upgrade
226 + // a merely-held file to a quarantine).
227 + let mut archive_res: Option<LayerResult> = None;
228 + let mut nested_res: Option<LayerResult> = None;
229 +
230 + if count > constants::SCAN_ZIP_MAX_ENTRIES {
231 + if bomb {
232 + archive_res = Some(LayerResult {
233 + layer: "archive",
234 + verdict: LayerVerdict::Fail,
235 + detail: Some(format!(
236 + "ZIP entry count {count} exceeds limit {}",
237 + constants::SCAN_ZIP_MAX_ENTRIES
238 + )),
239 + });
240 + }
241 + if content {
242 + nested_res = Some(nested(
243 + LayerVerdict::Error,
244 + format!(
245 + "nested ZIP entry count {count} exceeds limit {}",
246 + constants::SCAN_ZIP_MAX_ENTRIES
247 + ),
248 + ));
249 + }
250 + return (
251 + archive_res.unwrap_or_else(bomb_disabled_archive),
252 + nested_res.unwrap_or_else(|| nested(LayerVerdict::Skip, String::new())),
253 + );
254 + }
255 +
256 + let mut total_compressed: u64 = 0;
257 + let mut total_uncompressed: u64 = 0;
258 + let mut interior_budget: u64 = constants::SCAN_ZIP_MAX_UNCOMPRESSED;
259 +
260 + for i in 0..count {
261 + // Once both dimensions are settled, nothing more to learn.
262 + let bomb_active = bomb && archive_res.is_none();
263 + let content_active = content && nested_res.is_none();
264 + if !bomb_active && !content_active {
265 + break;
266 + }
267 +
268 + let (name, entry_compressed, claimed_size) = match archive.by_index_raw(i) {
269 + Ok(e) => (e.name().to_string(), e.compressed_size(), e.size()),
270 + Err(e) => {
271 + if bomb_active {
272 + archive_res = Some(LayerResult {
273 + layer: "archive",
274 + verdict: LayerVerdict::Error,
275 + detail: Some(format!("Failed to read ZIP entry {i}: {e}")),
276 + });
277 + }
278 + if content_active {
279 + nested_res = Some(nested(
280 + LayerVerdict::Error,
281 + format!("read nested ZIP entry {i}: {e}"),
282 + ));
283 + }
284 + continue;
285 + }
286 + };
287 +
288 + // Path traversal is a bomb-dimension failure (the nested walk never
289 + // checked entry names).
290 + if bomb_active {
291 + let name_lower = name.to_ascii_lowercase();
292 + if name.contains("../")
293 + || name.contains("..\\")
294 + || name_lower.contains("%2e%2e")
295 + || name.starts_with('/')
296 + || name.contains('\0')
297 + {
298 + archive_res = Some(LayerResult {
299 + layer: "archive",
300 + verdict: LayerVerdict::Fail,
301 + detail: Some(format!("Path traversal in entry: {name}")),
302 + });
303 + // Bomb verdict frozen; keep going only if content still needs us.
304 + if !content_active {
305 + break;
306 + }
307 + }
308 + }
309 +
310 + // Decompress the entry exactly once, teeing the full size (bomb) and the
311 + // buffered prefix (content).
312 + let want_content = content && nested_res.is_none();
313 + let (counted, content_buf, decode_err) = match archive.by_index(i) {
314 + Ok(mut entry) => match tee_decompress(
315 + &mut entry,
316 + constants::SCAN_ZIP_MAX_UNCOMPRESSED,
317 + want_content,
318 + &mut interior_budget,
319 + ) {
320 + Ok((c, cb)) => (c, cb, None),
321 + Err((c, why)) => (c, None, Some(why)),
322 + },
323 + // Could not open the entry to decompress: bomb uses a conservative
324 + // estimate; content is held for review.
325 + Err(e) => (
326 + claimed_size.saturating_mul(10).max(1024 * 1024),
327 + None,
328 + Some(format!("{e}")),
329 + ),
330 + };
331 +
332 + // Bomb accounting (skipped once the bomb verdict is frozen).
333 + if bomb && archive_res.is_none() {
334 + // A decode error mid-stream gets the conservative estimate rather
335 + // than trusting the attacker-controlled claimed size.
336 + let actual_size = if decode_err.is_some() {
337 + claimed_size.saturating_mul(10).max(1024 * 1024)
338 + } else {
339 + counted
340 + };
341 + if actual_size > constants::SCAN_ZIP_MAX_UNCOMPRESSED {
342 + archive_res = Some(LayerResult {
343 + layer: "archive",
344 + verdict: LayerVerdict::Fail,
345 + detail: Some(format!(
346 + "Actual decompressed size exceeds {} bytes (possible ZIP bomb)",
347 + constants::SCAN_ZIP_MAX_UNCOMPRESSED
348 + )),
349 + });
350 + } else {
351 + total_compressed += entry_compressed;
352 + total_uncompressed += actual_size;
353 + if entry_compressed > 0 && actual_size >= 1024 * 1024 {
354 + let entry_ratio = actual_size as f64 / entry_compressed as f64;
355 + if entry_ratio > constants::SCAN_ZIP_MAX_RATIO {
356 + archive_res = Some(LayerResult {
357 + layer: "archive",
358 + verdict: LayerVerdict::Fail,
359 + detail: Some(format!(
360 + "Entry {name} compression ratio {entry_ratio:.1}x exceeds limit of {:.0}x (possible ZIP bomb)",
361 + constants::SCAN_ZIP_MAX_RATIO
362 + )),
363 + });
364 + }
365 + }
366 + }
367 + }
368 +
369 + // Interior content scan (skipped once the nested verdict is frozen).
370 + if content && nested_res.is_none() {
371 + if let Some(why) = decode_err {
372 + nested_res = Some(nested(
373 + LayerVerdict::Error,
374 + format!("decode nested entry: {why}"),
375 + ));
376 + } else {
377 + match content_buf {
378 + Some(ContentBuf::Buffered(bytes)) => {
379 + if let Some(v) = scan_entry(
380 + &bytes,
381 + yara_rules,
382 + constants::SCAN_ZIP_MAX_DEPTH,
383 + &mut interior_budget,
384 + ) {
385 + nested_res = Some(v);
386 + }
387 + }
388 + Some(ContentBuf::Overflow) => {
389 + nested_res = Some(nested(
390 + LayerVerdict::Error,
391 + format!(
392 + "nested entry exceeds {INTERIOR_ENTRY_MAX}-byte interior scan ceiling"
393 + ),
394 + ));
395 + }
396 + Some(ContentBuf::BudgetExceeded) => {
397 + nested_res = Some(nested(
398 + LayerVerdict::Error,
399 + "nested archive content exceeds total interior scan budget".to_string(),
400 + ));
401 + }
402 + None => {}
403 + }
404 + }
99 405 }
100 - None => skip("Not a recognized archive"),
101 406 }
407 +
408 + // Finalize bomb verdict from the totals if nothing tripped mid-walk.
409 + let archive_result = match archive_res {
410 + Some(r) => r,
411 + None if !bomb => bomb_disabled_archive(),
412 + None => {
413 + if total_uncompressed > constants::SCAN_ZIP_MAX_UNCOMPRESSED {
414 + LayerResult {
415 + layer: "archive",
416 + verdict: LayerVerdict::Fail,
417 + detail: Some(format!(
418 + "Total uncompressed size {total_uncompressed} bytes exceeds limit of {} bytes",
419 + constants::SCAN_ZIP_MAX_UNCOMPRESSED
420 + )),
421 + }
422 + } else if total_compressed > 0
423 + && (total_uncompressed as f64 / total_compressed as f64) > constants::SCAN_ZIP_MAX_RATIO
424 + {
425 + LayerResult {
426 + layer: "archive",
427 + verdict: LayerVerdict::Fail,
428 + detail: Some(format!(
429 + "Compression ratio {:.1}x exceeds limit of {:.0}x (possible ZIP bomb)",
430 + total_uncompressed as f64 / total_compressed as f64,
431 + constants::SCAN_ZIP_MAX_RATIO
432 + )),
433 + }
434 + } else {
435 + LayerResult {
436 + layer: "archive",
437 + verdict: LayerVerdict::Pass,
438 + detail: Some(format!(
439 + "{count} entries, {:.1}x ratio",
440 + if total_compressed > 0 {
441 + total_uncompressed as f64 / total_compressed as f64
442 + } else {
443 + 0.0
444 + }
445 + )),
446 + }
447 + }
448 + }
449 + };
450 +
451 + let nested_result = match nested_res {
452 + Some(r) => r,
453 + None if !content => nested(LayerVerdict::Skip, String::new()),
454 + None => nested(
455 + LayerVerdict::Pass,
456 + "Archive interior fully scanned; no threats found".to_string(),
457 + ),
458 + };
459 +
460 + (archive_result, nested_result)
461 + }
462 +
463 + /// Walk a single-stream compressor (gzip/bzip2/xz/zstd) once, computing the
464 + /// bomb-defense verdict (when `bomb`) and the interior content verdict (when
465 + /// `content`). Returns `(archive, archive_nested)`.
466 + fn walk_compressed<R: Read>(
467 + kind: ArchiveKind,
468 + compressed_size: u64,
469 + reader: R,
470 + bomb: bool,
471 + content: bool,
472 + yara_rules: Option<&yara_x::Rules>,
473 + ) -> (LayerResult, LayerResult) {
474 + let mut decoder: Box<dyn Read> = match kind {
475 + ArchiveKind::Gzip => Box::new(flate2::read::MultiGzDecoder::new(reader)),
476 + ArchiveKind::Bzip2 => Box::new(bzip2::read::BzDecoder::new(reader)),
477 + ArchiveKind::Xz => Box::new(xz2::read::XzDecoder::new(reader)),
478 + ArchiveKind::Zstd => match zstd::stream::read::Decoder::new(reader) {
479 + Ok(d) => Box::new(d),
480 + Err(e) => {
481 + let why = format!("zstd init failed: {e}");
482 + return (
483 + if bomb { error(why.clone()) } else { bomb_disabled_archive() },
484 + if content {
485 + nested(LayerVerdict::Error, why)
486 + } else {
487 + nested(LayerVerdict::Skip, String::new())
488 + },
489 + );
490 + }
491 + },
492 + ArchiveKind::Zip => unreachable!("zip is handled by walk_zip"),
493 + };
494 +
495 + let abs_limit = constants::SCAN_ZIP_MAX_UNCOMPRESSED;
496 + let ratio_limit = compressed_size.saturating_mul(constants::SCAN_ZIP_MAX_RATIO as u64);
497 + let mut interior_budget: u64 = constants::SCAN_ZIP_MAX_UNCOMPRESSED;
498 +
499 + let (counted, content_buf, decode_err) =
500 + match tee_decompress(decoder.as_mut(), abs_limit, content, &mut interior_budget) {
501 + Ok((c, cb)) => (c, cb, None),
502 + Err((c, why)) => (c, None, Some(why)),
503 + };
504 +
505 + // Bomb verdict.
506 + let archive_result = if !bomb {
507 + bomb_disabled_archive()
508 + } else if let Some(ref why) = decode_err {
509 + // Mid-stream decode error is suspicious; fail closed for review.
510 + error(format!("{} decode error: {why}", kind.label()))
511 + } else if counted > abs_limit {
512 + LayerResult {
513 + layer: "archive",
514 + verdict: LayerVerdict::Fail,
515 + detail: Some(format!(
516 + "{} stream decompresses past {abs_limit} bytes (possible decompression bomb)",
517 + kind.label()
518 + )),
519 + }
520 + } else if compressed_size > 0 && counted > ratio_limit {
521 + LayerResult {
522 + layer: "archive",
523 + verdict: LayerVerdict::Fail,
524 + detail: Some(format!(
525 + "{} compression ratio exceeds {:.0}x (possible decompression bomb)",
526 + kind.label(),
527 + constants::SCAN_ZIP_MAX_RATIO
528 + )),
529 + }
530 + } else {
531 + LayerResult {
532 + layer: "archive",
533 + verdict: LayerVerdict::Pass,
534 + detail: Some(format!(
535 + "{} stream, {counted} bytes uncompressed ({:.1}x)",
536 + kind.label(),
537 + if compressed_size > 0 { counted as f64 / compressed_size as f64 } else { 0.0 }
538 + )),
539 + }
540 + };
541 +
542 + // Interior content verdict.
543 + let nested_result = if !content {
544 + nested(LayerVerdict::Skip, String::new())
545 + } else if let Some(why) = decode_err {
546 + nested(LayerVerdict::Error, format!("decode nested entry: {why}"))
547 + } else {
548 + match content_buf {
549 + Some(ContentBuf::Buffered(bytes)) => scan_entry(
550 + &bytes,
551 + yara_rules,
552 + constants::SCAN_ZIP_MAX_DEPTH,
553 + &mut interior_budget,
554 + )
555 + .unwrap_or_else(|| {
556 + nested(
557 + LayerVerdict::Pass,
558 + "Archive interior fully scanned; no threats found".to_string(),
559 + )
560 + }),
561 + Some(ContentBuf::Overflow) => nested(
562 + LayerVerdict::Error,
563 + format!("nested entry exceeds {INTERIOR_ENTRY_MAX}-byte interior scan ceiling"),
564 + ),
565 + Some(ContentBuf::BudgetExceeded) => nested(
566 + LayerVerdict::Error,
567 + "nested archive content exceeds total interior scan budget".to_string(),
568 + ),
569 + None => nested(
570 + LayerVerdict::Pass,
571 + "Archive interior fully scanned; no threats found".to_string(),
572 + ),
573 + }
574 + };
Lines truncated
M server/src/scanning/mod.rs +24 -15
@@ -294,16 +294,22 @@ impl ScanPipeline {
294 294 /// tokio runtime — the 4 in-process layers can each take seconds on a
295 295 /// 100 MB file. Network-bound layers (ClamAV, MalwareBazaar) run
296 296 /// concurrently with the sync block via `tokio::join!`.
297 - pub async fn scan(self: std::sync::Arc<Self>, data: Vec<u8>, file_type: FileType) -> ScanResult {
297 + pub async fn scan(
298 + self: std::sync::Arc<Self>,
299 + data: impl Into<bytes::Bytes>,
300 + file_type: FileType,
301 + ) -> ScanResult {
302 + // `bytes::Bytes` is already a cheaply-cloneable refcounted buffer, so the
303 + // download hands its single aggregated allocation straight here with no
304 + // extra copy — the buffered scan path previously aggregated then `to_vec`'d
305 + // the body, transiently doubling to ~200 MB for a 100 MB file (Run #2
306 + // Performance SERIOUS). Each consumer below clones the `Bytes` handle (a
307 + // refcount bump) and reads it as `&[u8]` by deref coercion.
308 + let data: bytes::Bytes = data.into();
298 309 let file_size = data.len() as u64;
299 - // Wrap the buffer in `Arc<Vec<u8>>` (moves the allocation in) rather than
300 - // `Arc::<[u8]>::from(Vec)`, which memcpy's the whole buffer into a fresh
301 - // refcounted allocation — a transient doubling to ~200 MB for a 100 MB
302 - // file. Every consumer below takes `&[u8]`, reached by deref coercion.
303 - let data = std::sync::Arc::new(data);
304 310
305 311 // Sync layers + hash, off the runtime
306 - let sync_data = std::sync::Arc::clone(&data);
312 + let sync_data = data.clone();
307 313 let sync_self = std::sync::Arc::clone(&self);
308 314 let sync_fut = tokio::task::spawn_blocking(move || sync_self.run_sync_layers(&sync_data, file_type));
309 315
@@ -311,7 +317,7 @@ impl ScanPipeline {
311 317 // but the hash is computed in the sync block, so we run MalwareBazaar
312 318 // after the sync block returns (its endpoint is fast). ClamAV can run
313 319 // concurrently with the sync block.
314 - let clamav_data = std::sync::Arc::clone(&data);
320 + let clamav_data = data.clone();
315 321 let clamav_socket = self.clamav_socket.clone();
316 322 let clamav_fut = async move {
317 323 match clamav_socket {
@@ -326,7 +332,7 @@ impl ScanPipeline {
326 332
327 333 // Layer 7: URLhaus — extract URLs from the bytes, query hosts.
328 334 // Runs concurrently with the sync block (independent of the hash).
329 - let urlhaus_data = std::sync::Arc::clone(&data);
335 + let urlhaus_data = data.clone();
330 336 let urlhaus_enabled = self.urlhaus_enabled;
331 337 let urlhaus_key = self.abuse_ch_auth_key.clone();
332 338 let urlhaus_fut = async move {
@@ -522,12 +528,15 @@ impl ScanPipeline {
522 528
523 529 layers.push(content_type::verify_content_type(data, file_type));
524 530 layers.push(structural::analyze_binary(data, file_type));
525 - layers.push(archive::check_archive_safety(data, file_type));
526 - // Recursively scan archive interiors: re-feed each entry's decompressed
527 - // bytes back through content-type/structural/YARA, descending into
528 - // nested archives. Covers the zip-in-a-zip seam the bomb-defense walk
529 - // above only counted. No-op (Skip) for non-archives.
530 - layers.push(archive::scan_nested_contents(data, self.yara_rules.as_ref()));
531 + // Single-pass archive inspection: the bomb-defense walk ("archive") and
532 + // the recursive interior content scan ("archive_nested") are derived from
533 + // ONE decompression of each entry. Previously each entry was decompressed
534 + // twice — once to count for bomb defense, once to buffer+scan its content
535 + // (Run #2 Performance SERIOUS). No-op (Skip/Skip) for non-archives.
536 + let (archive_layer, archive_nested_layer) =
537 + archive::inspect_archive(data, file_type, self.yara_rules.as_ref());
538 + layers.push(archive_layer);
539 + layers.push(archive_nested_layer);
531 540 layers.push(match self.yara_rules {
532 541 Some(ref rules) => yara::scan_with_yara(rules, data),
533 542 None => LayerResult {
M server/src/scanning/worker.rs +3 -1
@@ -277,7 +277,9 @@ async fn run_pipeline_and_decide(
277 277 );
278 278 super::too_large_to_scan(job.file_size_bytes as u64)
279 279 } else if (job.file_size_bytes as usize) < constants::SCAN_MAX_MEMORY_BYTES {
280 - let data = ctx.s3.download_object(&job.s3_key).await?;
280 + // `download_object_buf` returns the aggregated body as `Bytes` (no
281 + // `to_vec` copy); `scan` takes it directly (Run #2 Performance SERIOUS).
282 + let data = ctx.s3.download_object_buf(&job.s3_key).await?;
281 283 let _permit = ctx.scan_semaphore.acquire().await?;
282 284 Arc::clone(&ctx.pipeline).scan(data, file_type).await
283 285 } else {
M server/src/scheduler/mod.rs +46
@@ -79,6 +79,7 @@ pub fn spawn_scheduler(
79 79
80 80 let mut tick_count: u64 = 0;
81 81 let mut last_overrun_alert: Option<std::time::Instant> = None;
82 + let mut last_panic_alert: Option<std::time::Instant> = None;
82 83
83 84 loop {
84 85 tokio::select! {
@@ -118,6 +119,18 @@ pub fn spawn_scheduler(
118 119 continue;
119 120 }
120 121
122 + // Run every periodic job in a supervised child task. A panic in any
123 + // single job then aborts only this tick (the next tick re-runs the
124 + // full set); without supervision the panic unwinds the scheduler
125 + // loop and silently halts ALL background maintenance — including the
126 + // overrun/panic alerting below (Run #2 Performance CRITICAL). The
127 + // advisory lock stays held on `lock_conn` in this loop task for the
128 + // whole tick, so no other instance runs concurrently while the
129 + // child task works.
130 + let tick_jobs = {
131 + let state = state.clone();
132 + tokio::spawn(async move {
133 +
121 134 // Publish scheduled items. The announcement fan-out (4 DB
122 135 // roundtrips per item) gets spawned off the lock-held tick so a
123 136 // burst of releases doesn't extend the advisory-lock hold time.
@@ -303,6 +316,39 @@ pub fn spawn_scheduler(
303 316 }
304 317 }
305 318
319 + })
320 + };
321 +
322 + // Supervise the job task. A panic surfaces here as a JoinError
323 + // instead of unwinding the loop; log it and (rate-limited) open a
324 + // WAM ticket, then carry on so the next tick re-runs everything.
325 + if let Err(join_err) = tick_jobs.await {
326 + tracing::error!(
327 + tick = tick_count, error = ?join_err,
328 + "scheduler job task panicked; tick aborted, loop continues"
329 + );
330 + if let Some(ref wam) = state.wam {
331 + let cooldown_ok = last_panic_alert
332 + .is_none_or(|t| t.elapsed().as_secs() >= constants::ALERT_COOLDOWN_SECS);
333 + if cooldown_ok {
334 + let body = format!(
335 + "tick #{tick_count}: a scheduled job panicked ({join_err}). \
336 + The tick was aborted; the scheduler loop survived and will \
337 + re-run all jobs on the next tick."
338 + );
339 + wam.create_ticket(
340 + "Scheduler job panicked",
341 + Some(&body),
342 + "high",
343 + "scheduler-job-panic",
344 + None,
345 + )
346 + .await;
347 + last_panic_alert = Some(std::time::Instant::now());
348 + }
349 + }
350 + }
351 +
306 352 // Explicitly release the advisory lock (defense-in-depth: also released
307 353 // when lock_conn is dropped, but explicit unlock survives refactors that
308 354 // might move lock_conn into a shorter-lived scope).
M server/src/storage.rs +17
@@ -313,6 +313,10 @@ pub trait StorageBackend: Send + Sync {
313 313 async fn object_exists(&self, s3_key: &str) -> Result<bool>;
314 314 async fn object_size(&self, s3_key: &str) -> Result<Option<i64>>;
315 315 async fn download_object(&self, s3_key: &str) -> Result<Vec<u8>>;
316 + /// Download as `bytes::Bytes` — no `to_vec` copy of the aggregated body.
317 + /// Memory-sensitive callers (the scanner's buffered branch) use this so the
318 + /// payload isn't transiently doubled.
319 + async fn download_object_buf(&self, s3_key: &str) -> Result<bytes::Bytes>;
316 320 /// Stream the object body without buffering the whole payload. Callers
317 321 /// drive the stream to disk (scanner spool) or to a layer that consumes
318 322 /// chunks directly.
@@ -613,6 +617,15 @@ impl S3Client {
613 617 .map_err(AppError::Storage)
614 618 }
615 619
620 + /// Download an object as `bytes::Bytes` without the `to_vec` copy. See trait docs.
621 + pub async fn download_object_buf(&self, s3_key: &str) -> Result<bytes::Bytes> {
622 + self.inner
623 + .download_buf(s3_key)
624 + .await
625 + .map(|(bytes, _content_type)| bytes)
626 + .map_err(AppError::Storage)
627 + }
628 +
616 629 /// Stream an object's body from S3 without buffering. See trait docs.
617 630 pub async fn download_stream(&self, s3_key: &str) -> Result<s3_storage::ByteStream> {
618 631 self.inner
@@ -819,6 +832,10 @@ impl StorageBackend for S3Client {
819 832 self.download_object(s3_key).await
820 833 }
821 834
835 + async fn download_object_buf(&self, s3_key: &str) -> Result<bytes::Bytes> {
836 + self.download_object_buf(s3_key).await
837 + }
838 +
822 839 async fn download_stream(&self, s3_key: &str) -> Result<s3_storage::ByteStream> {
823 840 self.download_stream(s3_key).await
824 841 }
M shared/s3-storage/Cargo.lock +1
@@ -1655,6 +1655,7 @@ version = "0.1.1"
1655 1655 dependencies = [
1656 1656 "aws-config",
1657 1657 "aws-sdk-s3",
1658 + "bytes",
1658 1659 "tokio",
1659 1660 "tracing",
1660 1661 ]
M shared/s3-storage/Cargo.toml +1
@@ -6,5 +6,6 @@ edition = "2024"
6 6 [dependencies]
7 7 aws-sdk-s3 = "1.131"
8 8 aws-config = { version = "1.8", features = ["behavior-version-latest"] }
9 + bytes = "1"
9 10 tokio = { version = "1", features = ["fs", "io-util"] }
10 11 tracing = "0.1"
M shared/s3-storage/src/lib.rs +16 -1
@@ -103,7 +103,22 @@ impl S3Client {
103 103 }
104 104
105 105 /// Download bytes from S3. Returns `(data, content_type)`.
106 + ///
107 + /// Convenience `Vec<u8>` form; for the zero-extra-copy path use
108 + /// [`download_buf`](Self::download_buf), which returns the aggregated
109 + /// `Bytes` directly.
106 110 pub async fn download(&self, key: &str) -> Result<(Vec<u8>, String), String> {
111 + let (bytes, content_type) = self.download_buf(key).await?;
112 + Ok((bytes.to_vec(), content_type))
113 + }
114 +
115 + /// Download an object as `bytes::Bytes`, returning `(data, content_type)`.
116 + ///
117 + /// Unlike [`download`](Self::download) this does not copy the aggregated
118 + /// body into a fresh `Vec` — the caller gets the SDK's buffer directly. Use
119 + /// it on memory-sensitive paths (e.g. the scanner's buffered branch) where
120 + /// the extra `to_vec` would transiently double the footprint.
121 + pub async fn download_buf(&self, key: &str) -> Result<(bytes::Bytes, String), String> {
107 122 let resp = self
108 123 .client
109 124 .get_object()
@@ -124,7 +139,7 @@ impl S3Client {
124 139 .await
125 140 .map_err(|e| format!("S3 read body failed: {e}"))?;
126 141
127 - Ok((bytes.into_bytes().to_vec(), content_type))
142 + Ok((bytes.into_bytes(), content_type))
128 143 }
129 144
130 145 /// Stream an object's body from S3 without buffering. Caller drives the