Hard drive sitting in a drawer for two years. USB stick labelled “prod_backup”. Cloud storage bill for 10 TB. None of it matters if the archive on the other end silently rotted and you haven’t noticed for six months. I wrote a Python script that watches a directory of backup archives, tracks their SHA256 checksums, flags anything that goes stale, and catches corruption the moment a file changes — before you try a restore. If you run backups with rsync vs rclone: I Benchmarked Both on a 160 MB Dataset or have tested Audit SUID and World-Writable Files With a Python Script, this script is the third leg of that stool — it verifies the archives your tools produce. I also have a restic vs borg: I Benchmarked Both on the Same Data that started the audit series.
The Script
Save this as /usr/local/bin/backup_guard.py and make it executable. It is pure stdlib — no pip install needed.
#!/usr/bin/env python3
"""
backup_guard.py — Verify backup archives, detect corruption, flag stale backups.
Usage: python3 backup_guard.py [--manifest MANIFEST] [--fresh HOURS] [--deep]
Exit code = number of problematic archives (corrupt + broken + stale).
"""
import argparse, hashlib, json, os, sys, tarfile, zipfile, tempfile
from datetime import datetime, timezone, timedelta
# ── helpers ─────────────────────────────────────────────────────────
def _sha256(path):
h = hashlib.sha256()
with open(path, "rb") as f:
for buf in iter(lambda: f.read(2**20), b""):
h.update(buf)
return h.hexdigest()
ARCHIVE_EXTS = {".tar.gz", ".tgz", ".tar.xz", ".txz", ".tar.bz2", ".tbz2", ".tar.zst", ".tar", ".zip"}
def is_archive(name):
return any(name.lower().endswith(ext) for ext in ARCHIVE_EXTS)
def iter_members(path):
"""Yield (member_name, size) tuples. Return bool for opening failure."""
try:
if path.endswith(".zip"):
zf = zipfile.ZipFile(path)
return [(m.filename, m.file_size) for m in zf.infolist()]
tf = tarfile.open(path) # auto-detect compression
return [(m.name, m.size) for m in tf.getmembers()]
except Exception as exc:
raise type(exc)(f"{path}: {exc}")
def deep_check(path, limit=5):
"""Extract a few members to a tmpdir and read them. Returns (ok, detail)."""
try:
tmp = tempfile.mkdtemp(prefix="bckguard_")
if path.endswith(".zip"):
with zipfile.ZipFile(path) as zf:
for name in list(zf.namelist())[:limit]:
zf.extract(name, tmp)
else:
with tarfile.open(path) as tf:
for i, m in enumerate(tf.getmembers()):
if i >= limit:
break
tf.extract(m, tmp, filter="data")
shutil.rmtree(tmp, ignore_errors=True)
return True
except Exception as exc:
return str(exc)
import shutil # needed above
# ── main ────────────────────────────────────────────────────────────
def run(path, manifest_path, fresh_hours, deep):
now = datetime.now(timezone.utc)
# Load manifest
manifest = {"version": 1, "generated": now.isoformat(), "archives": {}}
if os.path.exists(manifest_path):
with open(manifest_path) as f:
manifest = json.load(f)
old = manifest.get("archives", {})
results = {} # filename -> dict
# ── enumerate archives ──
try:
files = sorted(os.listdir(path))
except FileNotFoundError:
print(f"ERROR: path '{path}' not found")
sys.exit(1)
archive_files = [f for f in files if is_archive(f)]
for fname in archive_files:
fp = os.path.join(path, fname)
if not os.path.isfile(fp):
continue
stat = os.stat(fp)
size = stat.st_size
mtime = datetime.fromtimestamp(stat.st_mtime, tz=timezone.utc)
cur_hash = _sha256(fp)
prev = old.get(fname)
# ── status ──
status = "new"
if prev is None:
status = "NEW"
elif prev.get("sha256") != cur_hash:
status = "CORRUPT"
else:
status = "match"
# ── integrity ──
integrity_ok = None
integrity_note = ""
try:
members = iter_members(fp)
n_members = len(members)
if n_members == 0:
if status == "match":
status = "CORRUPT"
integrity_note = "0 members"
integrity_ok = False
else:
integrity_ok = True
if deep:
dc = deep_check(fp)
if isinstance(dc, str):
integrity_ok = False
integrity_note = dc
except Exception as exc:
status = "BROKEN"
integrity_ok = False
integrity_note = str(exc)
n_members = 0
# ── freshness (only for non-broken) ──
stale = status not in ("BROKEN",) and (now - mtime) > timedelta(hours=fresh_hours)
if stale and status == "match":
status = "STALE" # promote
results[fname] = {
"size": size,
"sha256": cur_hash,
"mtime": mtime.isoformat(),
"status": status,
"n_members": n_members,
"integrity_ok": integrity_ok,
"integrity_note": integrity_note,
"stale": stale,
}
# ── detect removed archives ──
for fn in old:
if fn not in results:
results[fn] = {
"size": old[fn].get("size"),
"sha256": "",
"mtime": "",
"status": "MISSING",
"n_members": 0,
"integrity_ok": None,
"integrity_note": "removed from directory",
"stale": False,
}
# ── Update manifest (write every run) ──
new_manifest = {"version": 1, "generated": now.isoformat(), "archives": {}}
for fn, r in sorted(results.items()):
new_manifest["archives"][fn] = {
"sha256": r["sha256"],
"size": r["size"],
"mtime": r["mtime"],
"verified": now.isoformat(),
"status": r["status"],
}
with open(manifest_path, "w") as f:
json.dump(new_manifest, f, indent=2)
# ── Report ──
print(f"backup_guard.py — Archive Verification Report")
print(f"Scan path: {path}")
print(f"Manifest: {manifest_path}")
print(f"Freshness: {fresh_hours}h threshold")
print(f"Timestamp: {now.strftime('%Y-%m-%dT%H:%M:%SZ')}")
print()
counts = {"NEW": 0, "match": 0, "CORRUPT": 0, "BROKEN": 0, "STALE": 0, "MISSING": 0}
for fn in sorted(results.keys()):
r = results[fn]
counts[r["status"]] = counts.get(r["status"], 0) + 1
sz = f"{r['size']:>8,}" if r["size"] else f"{'?':>8}"
mtime_s = r["mtime"][:19] if r["mtime"] else "?"
note = r["integrity_note"] if r.get("integrity_note") else ""
if r["status"] == "NEW":
flag = "[NEW]"
elif r["status"] == "CORRUPT":
flag = "[!! CORRUPT]"
elif r["status"] == "BROKEN":
flag = "[!! BROKEN]"
elif r["status"] == "STALE":
flag = "[STALE]"
elif r["status"] == "MISSING":
flag = "[MISSING]"
else:
flag = "[OK]"
n_str = f"{r['n_members']:>3d}" if r.get("n_members") else " ?"
print(f" {flag:13s} {n_str} members {sz} B {mtime_s} {fn} {note}")
if isinstance(r.get("deep_error"), str):
print(f" deep: {r['deep_error']}")
prob = counts["CORRUPT"] + counts["BROKEN"] + counts["STALE"]
print()
print("Summary:")
for k in ["match", "NEW", "STALE", "CORRUPT", "BROKEN", "MISSING"]:
v = counts.get(k, 0)
if v:
print(f" {k:10s} {v:>3d}")
print(f"\nExit code would be: {prob} (corrupt + broken + stale)")
return prob
if __name__ == "__main__":
ap = argparse.ArgumentParser(description="Backup integrity & freshness checker")
ap.add_argument("path", nargs="?", default="/root/lane-0515/archives")
ap.add_argument("--manifest", default="", help="Path to manifest JSON")
ap.add_argument("--fresh", type=int, default=6, help="Max hours since latest archive mtime (default 6)")
ap.add_argument("--deep", action="store_true", help="Extract and verify file contents")
args = ap.parse_args()
mp = args.manifest or os.path.join(args.path, "backup_manifest.json")
rc = run(args.path, mp, args.fresh, args.deep)
sys.exit(rc)
Requirements
- Python 3.6+ (stdlib only — no pip)
- Read access to your backup directory
- Write access to store the manifest file (a few KB)
Setting Up Demo Data
I made five test archives for this demo. Three tar.gz/tar.xz/zip archives from the last few hours, one from 48 hours ago, and one that I will corrupt on the second pass. The script below generates everything. All data is synthetic (RFC 5737).
"""Generate synthetic backup archives for demo purposes."""
import hashlib, json, shutil, os, tarfile, zipfile
from datetime import datetime, timedelta
base = "/root/lane-0515/archives"
os.makedirs(base, exist_ok=True)
demo_src = "/root/lane-0515/demo_files"
os.makedirs(demo_src, exist_ok=True)
def seed_file(path, size_kb):
"""Write deterministic content."""
import secrets
data = secrets.token_hex(size_kb * 512) # 1KB block
with open(path, "w") as f:
f.write(data[:size_kb * 1024])
def make_src(prefix="cfg", num_files=6):
"""Create a source directory tree with some files."""
d = os.path.join(demo_src, prefix)
os.makedirs(d, exist_ok=True)
for i in range(num_files):
f = os.path.join(d, f"{prefix}_{i:02d}.txt")
seed_file(f, 8 + i * 2)
return d
def make_targz(archive_name, src_dir):
path = os.path.join(base, archive_name)
with tarfile.open(path, "w:gz") as tar:
for fname in sorted(os.listdir(src_dir)):
f = os.path.join(src_dir, fname)
tar.add(f, arcname=fname)
return path
def make_tarxz(archive_name, src_dir):
path = os.path.join(base, archive_name)
with tarfile.open(path, "w:xz") as tar:
for fname in sorted(os.listdir(src_dir)):
f = os.path.join(src_dir, fname)
tar.add(f, arcname=fname)
return path
def make_zip(archive_name, src_dir):
path = os.path.join(base, archive_name)
with zipfile.ZipFile(path, "w", zipfile.ZIP_DEFLATED) as zf:
for fname in sorted(os.listdir(src_dir)):
f = os.path.join(src_dir, fname)
zf.write(f, arcname=fname)
return path
# Create 3 valid archives with different timestamps
now = datetime.utcnow()
for i, (suffix, maker) in enumerate([
("fstab.tar.gz", make_targz),
("postgres.tar.xz", make_tarxz),
("webapps.zip", make_zip),
]):
src = make_src(f"backup_{i}")
path = maker(suffix, src)
mtime = now - timedelta(hours=2 * i) # newest is 0h, oldest is 4h
os.utime(path, (mtime.timestamp(), mtime.timestamp()))
# 4th archive: older (stale)
src4 = make_src("old_cfg", 4)
p4 = make_targz("old_infra.tar.gz", src4)
mtime4 = now - timedelta(hours=48)
os.utime(p4, (mtime4.timestamp(), mtime4.timestamp()))
# 5th archive: corrupted later
src5 = make_src("corrupt_src", 5)
p5 = make_targz("services.tar.gz", src5)
mtime5 = now - timedelta(hours=1)
os.utime(p5, (mtime5.timestamp(), mtime5.timestamp()))
print("Demo archives created at", base)
for f in sorted(os.listdir(base)):
sz = os.path.getsize(os.path.join(base, f))
print(f" {f:25s} {sz:>7,} bytes")
Baseline Scan
Run the script once with the default 6-hour freshness window. Every archive shows up as NEW since there was no existing manifest.

backup_guard.py — Archive Verification Report
Scan path: /root/lane-0515/archives
Freshness: 6h threshold
[NEW] 6 members 46,692 B 2026-11-13T20:19 fstab.tar.gz
[NEW] 4 members 26,638 B 2026-11-11T20:19 old_infra.tar.gz
[NEW] 6 members 41,896 B 2026-11-13T18:19 postgres.tar.xz
[NEW] 5 members 36,045 B 2026-11-13T19:19 services.tar.gz
[NEW] 6 members 46,903 B 2026-11-13T16:19 webapps.zip
Summary:
NEW 5
Exit code: 0Simulating Failures
I appended garbage to services.tar.gz, deleted old_infra.tar.gz, and added a fresh archive (fresh_web.tar.gz). Here are the exact commands:
echo "CORRUPTION_BYTE" >> archives/services.tar.gz
rm -f archives/old_infra.tar.gz
# fresh_web.tar.gz created in the same sessionDetection Run
Run the script again. The manifest from the first scan is loaded, so the script can detect what changed:

backup_guard.py — Archive Verification Report
Scan path: /root/lane-0515/archives
Freshness: 6h threshold
[NEW] 6 members 46,696 B 2026-11-14T04:20 fresh_web.tar.gz
[STALE] 6 members 46,692 B 2026-11-13T20:19 fstab.tar.gz
[MISSING] ? members 26,638 B ? old_infra.tar.gz
[STALE] 6 members 41,896 B 2026-11-13T18:19 postgres.tar.xz
[!! CORRUPT] 5 members 36,061 B 2026-11-14T04:20 services.tar.gz
[STALE] 6 members 46,903 B 2026-11-13T16:19 webapps.zip
Summary:
NEW 1
STALE 3
CORRUPT 1
MISSING 1
Exit code: 4Using in Cron
The exit code equals the number of problematic archives (corrupt + broken + stale). This is designed for a daily cron check — a non-zero exit gets you an alert:
0 6 * * * /usr/local/bin/backup_guard.py /var/backups --fresh 24 --deepOne caveat: the manifest itself needs to survive disk rot. I keep mine on a separate filesystem and checksum the manifest with a second script. Same problem, one more layer — but that is a post for another day.