Verify Backup Archives With a Python Script

– min read

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

  1. Python 3.6+ (stdlib only — no pip)
  2. Read access to your backup directory
  3. 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.

Baseline scan output showing five new archives
Baseline run — all archives detected as NEW
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: 0


Simulating 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 session


Detection Run

Run the script again. The manifest from the first scan is loaded, so the script can detect what changed:

Detection run showing corrupt, stale, and missing archives
Detection run — CORRUPT, STALE, and MISSING flags
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: 4


Using 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 --deep


One 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.

Mohamad ZulfahmyCybersecurity · OffSec · Entrepreneur
About me ↗
ON THIS PAGE
On this page