Skip to main content

max / makenotwork

23.2 KB · 611 lines History Blame Raw
1 //! Fetch the prod backup that `migration_dry_run` runs against.
2 //!
3 //! Sources supported:
4 //! - `file:///abs/path/to/dump.sql.gz` — local copy (dev).
5 //! - `rsync://host/module/path` — rsync daemon protocol.
6 //! - `ssh://user@host[:port]/path/file.sql.gz` — rsync-over-ssh. Used to pull
7 //! prod backups from `backup-puller@alpha-west-1`.
8 //!
9 //! The fetch is command-driven: the operator triggers it via /backup/fetch, it
10 //! is not implicit in promote. That keeps the slowest, most failure-prone step
11 //! visible in the TUI rather than buried inside a deploy.
12
13 use crate::config::Config;
14 use crate::topology::Topology;
15 use anyhow::{Context, Result, bail};
16 use chrono::Utc;
17 use ops_exec::{CapabilitySet, Executor, SshExec, SyncOpts};
18 use sqlx::SqlitePool;
19 use std::path::Path;
20 use std::sync::Arc;
21 use tokio::process::Command;
22
23 #[derive(Debug, Clone)]
24 pub struct FetchedBackup {
25 pub source: String,
26 pub local_path: String,
27 pub byte_size: Option<i64>,
28 }
29
30 /// Parsed `backup.source` URL. Owned strings so the parsed form outlives the
31 /// (possibly transient) URL we read from config.
32 #[derive(Debug, Clone, PartialEq, Eq)]
33 pub(crate) enum BackupSource {
34 /// Local file copy. Path follows the `file://` prefix.
35 File { path: String },
36 /// rsync daemon protocol. Full URL stays intact (rsync handles it).
37 RsyncDaemon { url: String },
38 /// rsync-over-ssh. Port is optional.
39 Ssh {
40 user_host: String,
41 port: Option<u16>,
42 path: String,
43 },
44 }
45
46 /// Parse a `backup.source` URL into a `BackupSource`. Rejects unsupported
47 /// schemes and malformed `ssh://` URLs (no path part).
48 pub(crate) fn parse_source(s: &str) -> Result<BackupSource> {
49 if let Some(rest) = s.strip_prefix("file://") {
50 if rest.is_empty() {
51 bail!("file:// URL is missing a path: {s}");
52 }
53 return Ok(BackupSource::File { path: rest.into() });
54 }
55 if s.starts_with("rsync://") {
56 return Ok(BackupSource::RsyncDaemon { url: s.into() });
57 }
58 if let Some(rest) = s.strip_prefix("ssh://") {
59 let (user_host_port, path_rest) = rest
60 .split_once('/')
61 .with_context(|| format!("ssh:// URL missing path: {s}"))?;
62 if user_host_port.is_empty() {
63 bail!("ssh:// URL missing user@host: {s}");
64 }
65 let path = format!("/{path_rest}");
66 let (user_host, port) = match user_host_port.rsplit_once(':') {
67 Some((uh, p)) => {
68 // Heuristic: trailing `:digits` after the final `:` is the port.
69 // Anything else (IPv6 literal, etc.) gets left alone.
70 match p.parse::<u16>() {
71 Ok(n) => (uh.to_string(), Some(n)),
72 Err(_) => (user_host_port.to_string(), None),
73 }
74 }
75 None => (user_host_port.to_string(), None),
76 };
77 if user_host.is_empty() {
78 bail!("ssh:// URL has empty host (port {port:?})");
79 }
80 return Ok(BackupSource::Ssh {
81 user_host,
82 port,
83 path,
84 });
85 }
86 bail!("unsupported backup source scheme: {s}");
87 }
88
89 /// Absolute floor: backups smaller than this are an empty or header-only file
90 /// (an rsync that wrote zero bytes, an empty source). This is the fallback when
91 /// there is no prior backup to compare against (the first-ever fetch).
92 const MIN_BACKUP_BYTES: u64 = 64;
93
94 /// Plausibility floor as a fraction of the last verified backup's size. A real
95 /// dump never abruptly halves; a source-side truncation that still closed a
96 /// valid gzip would pass `gzip -t` and the absolute floor, but not this. Every
97 /// row in `backups` is a previously-verified dump, so the last one is a sound
98 /// reference. Integer-halved at the call site.
99 ///
100 /// A dump *can* legitimately halve, though — a retention prune landing, a bloated
101 /// table finally being swept. When that happens the floor is self-sealing: no new
102 /// row is written unless a fetch clears it, so the reference can never advance and
103 /// every later fetch fails on the same stale number. That is not hypothetical;
104 /// Sando sat wedged from 2026-06-12 to 2026-07-27 after MNW started deleting
105 /// expired `tower_sessions` rows and the dump dropped 43 MB -> 6 MB overnight.
106 /// `force` (operator-supplied, via `POST /backup/fetch {"force":true}`) is the way
107 /// out: it drops to the absolute floor for one fetch, so the accepted dump becomes
108 /// the new reference. It never skips `gzip -t` — a truncated file is still refused.
109 const MIN_BACKUP_FRACTION_DENOM: i64 = 2;
110
111 /// Verify a freshly-downloaded backup before it is allowed to become the live
112 /// dump: reject anything below `min_bytes` (the plausibility floor derived from
113 /// history), and for a gzip require a complete, valid stream (`gzip -t` fails on
114 /// truncation/corruption). `is_gz` is taken from the *destination* name, not the
115 /// temp path (which carries a `.partial` suffix).
116 async fn verify_backup(tmp_path: &str, is_gz: bool, min_bytes: u64) -> Result<()> {
117 let meta = tokio::fs::metadata(tmp_path)
118 .await
119 .with_context(|| format!("stat fetched backup {tmp_path}"))?;
120 anyhow::ensure!(
121 meta.len() >= min_bytes,
122 "fetched backup {tmp_path} is implausibly small ({} bytes, floor {min_bytes}); \
123 treating as a failed/truncated transfer",
124 meta.len(),
125 );
126 if is_gz {
127 let out = Command::new("gzip")
128 .arg("-t")
129 .arg(tmp_path)
130 .output()
131 .await
132 .with_context(|| format!("spawning gzip -t {tmp_path}"))?;
133 anyhow::ensure!(
134 out.status.success(),
135 "fetched backup {tmp_path} failed gzip integrity check (truncated/corrupt): {}",
136 String::from_utf8_lossy(&out.stderr),
137 );
138 }
139 Ok(())
140 }
141
142 /// Pull the configured prod dump into `topo.backup.local_path`.
143 ///
144 /// `force` re-baselines the plausibility floor: see `MIN_BACKUP_FRACTION_DENOM`.
145 /// Pass `false` for anything automated — it is an operator escape hatch, not a
146 /// retry strategy.
147 pub async fn fetch(
148 pool: &SqlitePool,
149 _cfg: &Arc<Config>,
150 topo: &Arc<Topology>,
151 force: bool,
152 ) -> Result<FetchedBackup> {
153 let source = topo.backup.source.clone();
154 let local_path = topo.backup.local_path.clone();
155
156 if let Some(parent) = Path::new(&local_path).parent() {
157 tokio::fs::create_dir_all(parent).await?;
158 }
159
160 // Download to a sibling temp path, verify integrity, then atomically rename
161 // into place. The live `local_path` is never the write target, so a partial
162 // or corrupt transfer can never become the backup `migration_dry_run`
163 // restores (CF4). `--inplace`/`--partial` are deliberately NOT used — those
164 // keep a truncated file on failure, the opposite of what we want here.
165 let tmp_path = format!("{local_path}.partial");
166 let is_gz = std::path::Path::new(&local_path)
167 .extension()
168 .is_some_and(|ext| ext.eq_ignore_ascii_case("gz"));
169
170 // Plausibility floor: half the last verified backup's size, never below the
171 // absolute floor. The first-ever fetch (no prior row) falls back to the
172 // absolute floor.
173 let last_size: Option<i64> =
174 sqlx::query_scalar("SELECT byte_size FROM backups ORDER BY fetched_at DESC LIMIT 1")
175 .fetch_optional(pool)
176 .await?;
177 let min_bytes = if force {
178 tracing::warn!(
179 last_verified_bytes = last_size,
180 "force: re-baselining the backup plausibility floor to the absolute minimum; \
181 this fetch's size becomes the new reference"
182 );
183 MIN_BACKUP_BYTES
184 } else {
185 last_size.map_or(MIN_BACKUP_BYTES, |s| {
186 ((s / MIN_BACKUP_FRACTION_DENOM) as u64).max(MIN_BACKUP_BYTES)
187 })
188 };
189
190 let parsed = parse_source(&source)?;
191 let downloaded: Result<()> = async {
192 match parsed {
193 BackupSource::File { path } => {
194 tokio::fs::copy(&path, &tmp_path)
195 .await
196 .with_context(|| format!("copy {path} -> {tmp_path}"))?;
197 }
198 BackupSource::RsyncDaemon { url } => {
199 let out = Command::new("rsync")
200 .args(["-az", &url, &tmp_path])
201 .output()
202 .await
203 .context("spawning rsync")?;
204 anyhow::ensure!(
205 out.status.success(),
206 "rsync (daemon) failed: {}",
207 String::from_utf8_lossy(&out.stderr),
208 );
209 }
210 BackupSource::Ssh {
211 user_host,
212 path,
213 port,
214 } => {
215 // Through the executor rather than a hand-rolled `rsync -e ssh`:
216 // one transport, one set of SSH flags. The sync plane is now
217 // gated (ops_exec::gate_pull): grant `observe:artifact` and
218 // confine to the dump's own directory. `path` is operator config
219 // (the BackupSource in sando.toml), not attacker input, so the
220 // parent-dir root is a formality here — but it keeps this pull
221 // fail-closed like every other, rather than an open read of the
222 // remote host.
223 let caps = CapabilitySet::from_tokens(Vec::<&str>::new(), ["artifact"]);
224 let exec = SshExec::new(user_host, caps).with_port(port);
225 let exec = match Path::new(&path).parent() {
226 Some(dir) if !dir.as_os_str().is_empty() => exec.with_pull_root(dir),
227 // A bare filename with no directory: confine to the path
228 // itself (starts_with is reflexive), still fail-closed.
229 _ => exec.with_pull_root(&path),
230 };
231 // NOT `--partial`: a truncated leftover here is dangerous, not
232 // useful — a resumed fetch could splice two different dumps into
233 // one plausible-looking file (CF4). NOT `-z` either: the dump is
234 // already compressed.
235 let opts = SyncOpts {
236 compress: false,
237 partial: false,
238 ..SyncOpts::default()
239 };
240 exec.pull_file(Path::new(&path), Path::new(&tmp_path), &opts)
241 .await
242 .context("rsync (ssh) failed")?;
243 }
244 }
245 verify_backup(&tmp_path, is_gz, min_bytes).await
246 }
247 .await;
248
249 // On any download/verify failure, remove the temp file so a corrupt
250 // `.partial` never lingers, and leave the existing live backup untouched.
251 if let Err(e) = downloaded {
252 let _ = tokio::fs::remove_file(&tmp_path).await;
253 return Err(e);
254 }
255
256 tokio::fs::rename(&tmp_path, &local_path)
257 .await
258 .with_context(|| format!("atomic rename {tmp_path} -> {local_path}"))?;
259
260 let meta = tokio::fs::metadata(&local_path).await?;
261 let size = meta.len() as i64;
262
263 // Record the new backup and prune stale rows in one transaction, so a crash
264 // between the two can't leave the insert without the prune (or, worse, lose
265 // the insert while keeping a half-applied delete). The on-disk file is
266 // overwritten each fetch (single `local_path`), so rows older than 30 days
267 // reference a path that no longer exists — keep the table from growing.
268 let mut tx = pool.begin().await?;
269 sqlx::query(
270 "INSERT INTO backups (fetched_at, source, local_path, byte_size) VALUES (?, ?, ?, ?)",
271 )
272 .bind(Utc::now().to_rfc3339())
273 .bind(&source)
274 .bind(&local_path)
275 .bind(size)
276 .execute(&mut *tx)
277 .await?;
278 sqlx::query("DELETE FROM backups WHERE fetched_at < datetime('now', '-30 days')")
279 .execute(&mut *tx)
280 .await?;
281 tx.commit().await?;
282
283 Ok(FetchedBackup {
284 source,
285 local_path,
286 byte_size: Some(size),
287 })
288 }
289
290 #[cfg(test)]
291 mod tests {
292 use super::*;
293 use crate::topology::{BackupConfig, RepoConfig};
294
295 // ---- CF4: atomic write + integrity ----
296
297 async fn mem_pool() -> SqlitePool {
298 let pool = sqlx::sqlite::SqlitePoolOptions::new()
299 .max_connections(1)
300 .connect("sqlite::memory:")
301 .await
302 .unwrap();
303 crate::db::migrate(&pool).await.unwrap();
304 pool
305 }
306
307 fn topo_with_backup(source: String, local_path: String) -> Topology {
308 Topology {
309 repo: RepoConfig {
310 bare_path: "/tmp/x.git".into(),
311 branch: "main".into(),
312 upstream: None,
313 },
314 backup: BackupConfig { source, local_path },
315 tiers: vec![],
316 aux_repos: Vec::new(),
317 }
318 }
319
320 /// ~4 KB of poorly-compressible bytes so a gzip of it stays well above the
321 /// size floor and a half-truncation lands mid-stream (failing `gzip -t`).
322 fn incompressible(n: usize) -> Vec<u8> {
323 (0..n)
324 .map(|i| (i.wrapping_mul(2_654_435_761) >> 13) as u8)
325 .collect()
326 }
327
328 async fn write_valid_gz(path: &Path) {
329 let plain = path.with_extension("plain");
330 tokio::fs::write(&plain, incompressible(4096))
331 .await
332 .unwrap();
333 let out = Command::new("sh")
334 .arg("-c")
335 .arg(format!("gzip -c {} > {}", plain.display(), path.display()))
336 .output()
337 .await
338 .unwrap();
339 assert!(out.status.success(), "gzip shim failed");
340 }
341
342 #[tokio::test]
343 async fn fetch_file_source_writes_atomically_and_records_row() {
344 let tmp = tempfile::tempdir().unwrap();
345 let src = tmp.path().join("src.sql.gz");
346 write_valid_gz(&src).await;
347 let dest = tmp.path().join("backups/latest.sql.gz");
348 let topo = Arc::new(topo_with_backup(
349 format!("file://{}", src.display()),
350 dest.to_string_lossy().into_owned(),
351 ));
352 let pool = mem_pool().await;
353 let cfg = Arc::new(Config::for_tests());
354
355 let fb = fetch(&pool, &cfg, &topo, false).await.unwrap();
356 assert!(dest.exists(), "live backup written");
357 assert!(
358 !dest.with_file_name("latest.sql.gz.partial").exists(),
359 "temp file consumed by the atomic rename",
360 );
361 assert!(fb.byte_size.unwrap() > 0);
362 let count: (i64,) = sqlx::query_as("SELECT COUNT(*) FROM backups")
363 .fetch_one(&pool)
364 .await
365 .unwrap();
366 assert_eq!(count.0, 1, "a row is recorded for a successful fetch");
367 }
368
369 #[tokio::test]
370 async fn fetch_rejects_a_dump_far_below_the_last_backup_size() {
371 // Plausibility floor: a complete, valid gzip that is far smaller than the
372 // last verified backup is a likely source-side truncation and is rejected
373 // even though `gzip -t` passes — the gap the 64-byte absolute floor missed.
374 let tmp = tempfile::tempdir().unwrap();
375 let src = tmp.path().join("src.sql.gz");
376 write_valid_gz(&src).await; // ~4 KB valid gzip
377 let dest = tmp.path().join("backups/latest.sql.gz");
378 let topo = Arc::new(topo_with_backup(
379 format!("file://{}", src.display()),
380 dest.to_string_lossy().into_owned(),
381 ));
382 let pool = mem_pool().await;
383 let cfg = Arc::new(Config::for_tests());
384
385 // Seed a prior verified backup far larger than the incoming one; the floor
386 // becomes 500_000, well above the ~4 KB dump.
387 sqlx::query("INSERT INTO backups (fetched_at, source, local_path, byte_size) VALUES (?, 'x', ?, 1000000)")
388 .bind(Utc::now().to_rfc3339())
389 .bind(dest.to_string_lossy().into_owned())
390 .execute(&pool).await.unwrap();
391
392 let err = fetch(&pool, &cfg, &topo, false)
393 .await
394 .unwrap_err()
395 .to_string();
396 assert!(err.contains("implausibly small"), "{err}");
397 assert!(
398 !dest.exists(),
399 "a rejected dump never becomes the live backup"
400 );
401 let count: (i64,) = sqlx::query_as("SELECT COUNT(*) FROM backups")
402 .fetch_one(&pool)
403 .await
404 .unwrap();
405 assert_eq!(count.0, 1, "the rejected fetch records no new row");
406 }
407
408 #[tokio::test]
409 async fn force_rebaselines_the_floor_after_a_legitimate_shrink() {
410 // The wedge this exists for: the floor is derived from a row that only a
411 // passing fetch can replace, so a dump that legitimately halves locks the
412 // fetch out permanently. `force` accepts one undersized dump and makes it
413 // the new reference, unwedging the next ordinary fetch.
414 let tmp = tempfile::tempdir().unwrap();
415 let src = tmp.path().join("src.sql.gz");
416 write_valid_gz(&src).await;
417 let dest = tmp.path().join("backups/latest.sql.gz");
418 let topo = Arc::new(topo_with_backup(
419 format!("file://{}", src.display()),
420 dest.to_string_lossy().into_owned(),
421 ));
422 let pool = mem_pool().await;
423 let cfg = Arc::new(Config::for_tests());
424 sqlx::query("INSERT INTO backups (fetched_at, source, local_path, byte_size) VALUES (?, 'x', ?, 1000000)")
425 .bind(Utc::now().to_rfc3339())
426 .bind(dest.to_string_lossy().into_owned())
427 .execute(&pool).await.unwrap();
428
429 // Same dump the un-forced fetch rejects above.
430 let fb = fetch(&pool, &cfg, &topo, true).await.unwrap();
431 assert!(dest.exists(), "the forced dump becomes the live backup");
432 let recorded = fb.byte_size.unwrap();
433 assert!(
434 recorded < 1_000_000,
435 "the accepted dump really is the small one"
436 );
437
438 // The new row is now the reference, so the next fetch passes unforced.
439 fetch(&pool, &cfg, &topo, false)
440 .await
441 .expect("floor re-baselined to the forced fetch's size");
442 }
443
444 #[tokio::test]
445 async fn force_still_rejects_a_corrupt_gzip() {
446 // `force` relaxes the size floor only. A truncated dump is refused either
447 // way — otherwise the escape hatch would be a way to install a broken
448 // backup as the thing migration_dry_run restores.
449 let tmp = tempfile::tempdir().unwrap();
450 let src = tmp.path().join("src.sql.gz");
451 write_valid_gz(&src).await;
452 let whole = tokio::fs::read(&src).await.unwrap();
453 tokio::fs::write(&src, &whole[..whole.len() / 2])
454 .await
455 .unwrap();
456 let dest = tmp.path().join("backups/latest.sql.gz");
457 let topo = Arc::new(topo_with_backup(
458 format!("file://{}", src.display()),
459 dest.to_string_lossy().into_owned(),
460 ));
461 let pool = mem_pool().await;
462 let cfg = Arc::new(Config::for_tests());
463
464 let err = fetch(&pool, &cfg, &topo, true)
465 .await
466 .unwrap_err()
467 .to_string();
468 assert!(err.contains("gzip integrity check"), "{err}");
469 assert!(
470 !dest.exists(),
471 "a corrupt dump never becomes the live backup"
472 );
473 }
474
475 #[tokio::test]
476 async fn fetch_rejects_truncated_gz_and_leaves_no_live_file() {
477 let tmp = tempfile::tempdir().unwrap();
478 let src = tmp.path().join("src.sql.gz");
479 write_valid_gz(&src).await;
480 // Truncate to half: a valid gzip prefix that fails `gzip -t` mid-stream.
481 let full = tokio::fs::read(&src).await.unwrap();
482 assert!(
483 full.len() / 2 > MIN_BACKUP_BYTES as usize,
484 "half must clear the size floor to exercise gzip -t"
485 );
486 tokio::fs::write(&src, &full[..full.len() / 2])
487 .await
488 .unwrap();
489
490 let dest = tmp.path().join("backups/latest.sql.gz");
491 let topo = Arc::new(topo_with_backup(
492 format!("file://{}", src.display()),
493 dest.to_string_lossy().into_owned(),
494 ));
495 let pool = mem_pool().await;
496 let cfg = Arc::new(Config::for_tests());
497
498 let res = fetch(&pool, &cfg, &topo, false).await;
499 assert!(res.is_err(), "a truncated gzip must fail the fetch");
500 assert!(
501 !dest.exists(),
502 "no live backup file results from a failed fetch"
503 );
504 assert!(
505 !dest.with_file_name("latest.sql.gz.partial").exists(),
506 "the corrupt temp file is cleaned up",
507 );
508 let count: (i64,) = sqlx::query_as("SELECT COUNT(*) FROM backups")
509 .fetch_one(&pool)
510 .await
511 .unwrap();
512 assert_eq!(count.0, 0, "no row recorded for a failed fetch");
513 }
514
515 #[test]
516 fn parses_file_url() {
517 let s = parse_source("file:///opt/backups/latest.sql.gz").unwrap();
518 assert_eq!(
519 s,
520 BackupSource::File {
521 path: "/opt/backups/latest.sql.gz".into()
522 }
523 );
524 }
525
526 #[test]
527 fn file_url_without_path_errors() {
528 assert!(parse_source("file://").is_err());
529 }
530
531 #[test]
532 fn parses_rsync_daemon_url() {
533 let s = parse_source("rsync://astra/mnw/latest.sql.gz").unwrap();
534 assert_eq!(
535 s,
536 BackupSource::RsyncDaemon {
537 url: "rsync://astra/mnw/latest.sql.gz".into()
538 }
539 );
540 }
541
542 #[test]
543 fn parses_ssh_url_with_port() {
544 let s = parse_source("ssh://backup-puller@alpha-west-1:2200/latest.sql.gz").unwrap();
545 assert_eq!(
546 s,
547 BackupSource::Ssh {
548 user_host: "backup-puller@alpha-west-1".into(),
549 port: Some(2200),
550 path: "/latest.sql.gz".into(),
551 }
552 );
553 }
554
555 #[test]
556 fn parses_ssh_url_without_port() {
557 let s = parse_source("ssh://max@astra/opt/backups/mnw/latest.sql.gz").unwrap();
558 assert_eq!(
559 s,
560 BackupSource::Ssh {
561 user_host: "max@astra".into(),
562 port: None,
563 path: "/opt/backups/mnw/latest.sql.gz".into(),
564 }
565 );
566 }
567
568 #[test]
569 fn ssh_url_without_path_errors() {
570 // `split_once('/')` — `ssh://user@host` has no `/` after the scheme.
571 assert!(parse_source("ssh://backup-puller@alpha-west-1").is_err());
572 }
573
574 #[test]
575 fn ssh_url_without_user_host_errors() {
576 // Empty user@host: `ssh:///foo`. Caught by the empty-prefix check.
577 assert!(parse_source("ssh:///latest.sql.gz").is_err());
578 }
579
580 #[test]
581 fn ssh_url_with_non_numeric_after_colon_treats_as_part_of_host() {
582 // `host:notaport` should NOT parse `notaport` as a port. Leave the
583 // colon part of user_host; libssh/rsync will reject if truly wrong.
584 let s = parse_source("ssh://user@host:notaport/path").unwrap();
585 assert_eq!(
586 s,
587 BackupSource::Ssh {
588 user_host: "user@host:notaport".into(),
589 port: None,
590 path: "/path".into(),
591 }
592 );
593 }
594
595 #[test]
596 fn rejects_unknown_scheme() {
597 assert!(parse_source("ftp://example.com/file").is_err());
598 assert!(parse_source("just-a-path.sql.gz").is_err());
599 assert!(parse_source("").is_err());
600 }
601
602 #[test]
603 fn ssh_url_preserves_multi_segment_path() {
604 let s = parse_source("ssh://a@b:22/opt/foo/bar/baz.sql.gz").unwrap();
605 match s {
606 BackupSource::Ssh { path, .. } => assert_eq!(path, "/opt/foo/bar/baz.sql.gz"),
607 _ => panic!("wrong variant"),
608 }
609 }
610 }
611