A backup you've never restored is just a hope. The job ran, the log says "done", the archive exists, and none of that tells you whether your files come back. Backups fail without telling you: an empty archive, a cron job that stopped three weeks ago, an exclude rule that dropped a whole folder, a database file copied halfway through a write, a password you no longer have.
Tools like restic check and borg check verify the repository. That's valuable, but it's a different question from the one you care about: if my server died right now, would my files come back? The only way to know is to restore them, and the only way to keep knowing is to do it automatically, every week.
So let's build a small restore test. It's one Python file with no dependencies, and every time it runs it asks five questions:
Is the newest backup recent? A job that has stopped running is one of the most common failures.
Did it shrink? A sudden drop in file count or size usually means a broken exclude rule.
Are my live files actually in it? It samples files from your live data and checks that each one is in the backup.
Do they come back intact? It restores the sample into a private temp folder and compares checksums.
Does the database open? A SQLite file that exists isn't the same as a SQLite file that works.
Every check adds one line to a single report. If any line says FAIL, the script exits with a non-zero code and, if you've set --notify, sends you an alert.
flowchart TD
A["Newest backup"] --> B["Fresh?"]
B --> C["Size sane?"]
C --> D["Sample live files untouched since the backup"]
D --> E["Present in the backup?"]
E --> F["Restore to a private temp folder"]
F --> G["Checksums, canary, SQLite integrity"]
B --> R["Report"]
C --> R
E --> R
G --> R
R --> H[("SQLite history")]
R -- any FAIL --> N["Notify and exit code 1"]The backup we're testing
The test works with any backup, but we need something concrete. Here's a deliberately plain nightly job: a tar.gz of /srv/data with one small extra, a canary file that we'll come back to.
#!/bin/sh
# nightly-backup.sh - a plain tar backup that leaves a canary behind
set -eu
SRC=${SRC:-/srv/data}
DEST=${DEST:-/backups}
STAMP=$(date +%F)
date +%s > "$SRC/.backup-canary" # proves the backup contains data from *this* run
tar -czf "$DEST/data-$STAMP.tar.gz.tmp" -C "$SRC" .
mv "$DEST/data-$STAMP.tar.gz.tmp" "$DEST/data-$STAMP.tar.gz" # a half-written file never matches the glob
find "$DEST" -name 'data-*.tar.gz' -mtime +14 -deleteTwo details matter. The archive is written to a .tmp name and renamed when it's complete, so a half-written file never matches the data-*.tar.gz pattern the test looks for (the same atomic-write trick from Make Your Python Batch Script Crash-Safe: Atomic Writes, Checkpoints and a Kill Test). And -C "$SRC" . stores paths relative to /srv/data, which affects how the test maps live files to archive entries, as you'll see in a moment.
Step 1: Is there a recent backup?
The simplest check catches a very common failure. The newest archive's modification time must be within a limit (26 hours by default, which suits a daily job):
[FAIL] fresh newest backup is 72.0 h old (limit 26 h)This check is so cheap that it gets its own mode. --quick runs only the freshness check, so you can run it every day, while the full restore test runs weekly. Add --quiet and it prints nothing unless something is wrong.
Step 2: Did the backup shrink?
Every full run records the file count and archive size in a small SQLite history table. The next run compares against the last good run and fails if either number dropped by more than 20%:
def pct(new, old):
return 100 * (new - old) / old if old else 0.0
def check_size(cfg, report, db, files, nbytes):
prev = db.execute("SELECT file_count, total_bytes FROM runs WHERE status = 'ok' "
"ORDER BY ts DESC LIMIT 1").fetchone()
if files == 0:
report.add("size", False, "the backup contains no files")
return
text = f"{files:,} files"
if not prev:
report.add("size", True, text + (f", {nbytes / 1e6:,.1f} MB archive" if nbytes else "")
+ " (first run, so no baseline yet)")
return
changes = [pct(files, prev[0])]
text += f" ({changes[0]:+.1f}%)"
if nbytes:
text += f", {nbytes / 1e6:,.1f} MB archive"
if prev[1]:
changes.append(pct(nbytes, prev[1]))
text += f" ({changes[1]:+.1f}%)"
ok = min(changes) >= -cfg.shrink_pct or cfg.accept_shrink
report.add("size", ok, text + " vs the last good run" + ("" if ok else f"; limit is -{cfg.shrink_pct:g}%"))Comparing against the last good run (not the previous run) matters. If a broken exclude rule shrinks the backup, the failed run isn't recorded as good, so the alert keeps firing until you fix it. And if the shrink is legitimate, say you deleted a large folder on purpose, run once with --accept-shrink to make the new size the baseline.
Step 3: Sample from the live side
This is the key design decision. The obvious way to pick files to test is to list the backup and choose some. But a listing can only tell you what is in the backup. It can never reveal what's missing.
So we sample from your live files instead, then check that each one exists in the backup. A folder dropped by a bad exclude rule shows up immediately as live files with no counterpart in the archive.
There's one complication: live files keep changing. A file edited after the backup was made won't match its archived copy, and that isn't a failure. So the scan only considers files whose modification time is earlier than the backup time minus a grace margin (60 minutes by default, to cover files being written while the backup ran). Everything newer is counted and skipped. So are files this user can't read and files over 500 MB (--max-file-mb), and folders it can't open are counted too, so a blind spot shows up in the report instead of hiding.
def scan_live(cfg, cutoff, rng):
"""One pass over the live files. Returns (sample, stats of what was left out).
Only files untouched since the backup, readable by us and not huge can be compared
fairly. From those we keep a uniform random sample (reservoir sampling) plus the few largest.
"""
def excluded(key):
return any(fnmatch.fnmatchcase(key, pattern) for pattern in cfg.exclude)
stats = {"changed": 0, "unreadable": 0, "too_big": 0, "folders": 0}
def unreadable_folder(error):
stats["folders"] += 1
reservoir, largest, seen = [], [], 0
for source in cfg.source:
for dirpath, dirnames, filenames in os.walk(source, onerror=unreadable_folder):
dirnames[:] = [d for d in dirnames
if not excluded((live_key(os.path.join(dirpath, d), cfg.live_root) or "") + "/")]
for name in filenames:
full = os.path.join(dirpath, name)
key = live_key(full, cfg.live_root)
if key is None or excluded(key):
continue
try:
st = os.lstat(full)
except OSError:
continue
if not stat.S_ISREG(st.st_mode):
continue
if st.st_mtime >= cutoff:
stats["changed"] += 1
elif st.st_size > cfg.max_file_mb * 1e6:
stats["too_big"] += 1
elif not os.access(full, os.R_OK):
stats["unreadable"] += 1
else:
c = Candidate(full, key, st.st_size, st.st_mtime)
seen += 1
if len(reservoir) < cfg.sample:
reservoir.append(c)
elif (j := rng.randrange(seen)) < cfg.sample:
reservoir[j] = c
heapq.heappush(largest, (c.size, seen, c))
if len(largest) > cfg.largest:
heapq.heappop(largest)
picked = {c.key: c for c in reservoir + [c for _, _, c in largest]}
return list(picked.values()), statsThe scan walks the tree once and uses reservoir sampling to keep a uniform random sample without holding every filename in memory, plus the few largest files, since big files are where truncation tends to show. A different random sample each week means coverage grows over time.
Two options connect the live files to the archive:
--live-rootis what the archive's paths are relative to. Fortar -C /srv/data .that's/srv/data. For an archive made withtar czf x.tgz /srv/data, which storessrv/data/..., you leave the default/. If every file shows up as missing, the report tells you to check this option.--excludelists patterns your backup deliberately skips, such ascache/*. Keep it in sync with the backup job's own excludes. Anything missing that you haven't excluded is, by definition, a problem.
Step 4: Read the backup safely
The tar backend lists the archive's regular files, then restores the sampled ones in a single streaming pass. Two helpers keep names safe:
def normalize(name):
"""'./a/b', '/a/b' and 'a/b' are the same stored file. None for names we refuse."""
key = posixpath.normpath(name.lstrip("/"))
if key in (".", "..") or key.startswith("../"):
return None
return key
def live_key(path, root):
"""The name a live file has inside the backup (tar's -C directory is the root)."""
return normalize(os.path.relpath(path, root))
def safe_join(dest, key):
target = (Path(dest) / key).resolve()
if not target.is_relative_to(Path(dest).resolve()):
raise ValueError(f"unsafe path in backup: {key}")
return targetnormalize() makes ./a/b, /a/b and a/b the same key, and refuses any name that climbs out with ... safe_join() double-checks that a restore target stays inside the temp folder. The backend also never calls extract(): it copies file bytes itself, so a malicious or corrupt archive can't plant symlinks, odd permissions or files elsewhere on disk.
class TarBackend:
"""Plain .tar, .tar.gz, .tar.xz and .tar.bz2 archives. The newest match of the glob is tested."""
def __init__(self, pattern):
matches = [m for m in glob.glob(os.path.expanduser(pattern)) if os.path.isfile(m)]
if not matches:
raise Problem("backup", f"no backup found matching {pattern}")
self.path = max(matches, key=os.path.getmtime)
self.label = self.path
def created(self):
return os.path.getmtime(self.path)
def total_bytes(self):
return os.path.getsize(self.path)
def entries(self):
out = []
with tarfile.open(self.path, "r:*") as tar:
for m in tar:
key = normalize(m.name) if m.isreg() else None
if key:
out.append(Entry(m.name, key, m.size))
while tar.fileobj.read(CHUNK): # tarfile stops early; reading on makes gzip/xz verify their checksums
pass
return out
def restore(self, entries, dest):
"""One streaming pass. We copy bytes ourselves, so no links, modes or paths are trusted."""
wanted = {e.key for e in entries}
with tarfile.open(self.path, "r:*") as tar:
for m in tar:
key = normalize(m.name) if m.isreg() else None
if key in wanted:
target = safe_join(dest, key)
target.parent.mkdir(parents=True, exist_ok=True)
with tar.extractfile(m) as src, open(target, "wb") as out:
shutil.copyfileobj(src, out)
return {}Look at the while tar.fileobj.read(CHUNK) loop at the end of entries(). Python's tarfile stops reading at the end-of-archive marker, which means it never reaches the checksum at the very end of a gzip stream. Without that loop, a damaged .tar.gz can slip through. Reading to the true end makes gzip, xz and bzip2 verify the entire archive, which turns a plain listing into a full integrity check for compressed archives.
Step 5: Compare, carefully
For each sampled file the comparison has three possible outcomes: ok, changed (edited while we were testing, so skipped) or bad with a reason:
def compare(c, tmp, errors):
"""Compare one restored file with its live original -> (status, text)."""
if c.key in errors:
return "bad", f"RESTORE FAILED {c.key} ({errors[c.key]})"
restored = safe_join(tmp, c.key)
if not restored.is_file():
return "bad", f"MISSING {c.key} (not there after the restore)"
try:
st = os.lstat(c.path)
if st.st_size != c.size or st.st_mtime != c.mtime:
return "changed", "" # edited while we were testing
if restored.stat().st_size != c.size:
return "bad", f"SIZE DIFFERS {c.key} ({restored.stat().st_size:,} bytes back, {c.size:,} live)"
if sha256(restored) != sha256(c.path):
return "bad", f"MISMATCH {c.key}"
except OSError as e:
return "bad", f"UNREADABLE {c.key} ({e.strerror})"
return "ok", ""Size is checked first because it's free, and the checksum only runs when sizes match. The live file is re-checked just before comparing, so a file edited mid-test is skipped instead of reported as corrupt.
Step 6: The canary
Even a perfectly restored file can be old. If your backup job stopped refreshing its data, you'd restore last month's files correctly and still be in trouble. That's what the canary line in backup.sh is for. Every run writes the current Unix time into a file inside the backed-up folder, and the test restores that file from the backup and checks the timestamp:
def check_canary(cfg, path):
"""The canary holds the Unix time the backup job wrote it: proof the data is recent."""
try:
written = int(path.read_text().split()[0])
except (OSError, ValueError, IndexError):
return False, "the restored canary file has no timestamp in it"
age = (time.time() - written) / 3600
return age <= cfg.max_age_hours, f"restored canary was written {age:.1f} h ago (limit {cfg.max_age_hours:g} h)"If the restored canary says 120 hours ago, the backup job is running but not capturing fresh data, which a plain freshness check on the archive's modification time can't tell you.
Step 7: Test the database, not just the file
Copying a SQLite file while an application is writing to it can capture a torn, half-updated copy. So the test restores any database you list with --sqlite and opens it:
def check_sqlite(path):
"""A restored SQLite file must open, pass integrity_check and not be empty."""
try:
con = sqlite3.connect(path)
try:
tables = con.execute("SELECT count(*) FROM sqlite_master WHERE type = 'table'").fetchone()[0]
result = con.execute("PRAGMA integrity_check").fetchall()
finally:
con.close()
except sqlite3.Error as e:
return False, f"{type(e).__name__}: {e}"
if result != [("ok",)]:
return False, "integrity_check: " + " ".join(str(r[0]) for r in result[:3]).replace("\n", " | ")[:160]
if tables == 0:
return False, "the database is empty (no tables)"
return True, f"integrity_check ok, {tables} table{'s' if tables != 1 else ''}"The empty-database check is there on purpose: a zero-byte or schema-less database passes PRAGMA integrity_check perfectly. If your backup job copies live databases, use SQLite's online backup (the .backup command, or VACUUM INTO) rather than copying the file, and let this check confirm it worked.
For Postgres or MySQL the same idea applies: restore the dump into a scratch database and run a sanity query. This script doesn't do that for you, but the SQLite check shows where it would plug in.
Step 8: Restore safely
Restoring means copying your real data somewhere, so the script is careful about where:
It restores into a fresh folder created with
tempfile.mkdtemp(), which is private to your user (mode 0700), and never over a live path.It checks free space first and refuses to start if the sample won't fit. Files over 500 MB are never sampled (
--max-file-mb), so one giant disk image can't fill the folder. Use--tmpif/tmpis small.It deletes the folder in a
finallyblock, even if the restore fails halfway.It times the restore and records it. "We have backups" and "we can be back up in twenty minutes" are different claims, and the history table lets you spot a restore that's slowly getting slower.
Not using tar? Bring your own commands
The built-in backend reads tar archives, but the rest of the script only needs three things from a backup tool, so any tool you can drive from a shell works through --list-cmd, --restore-cmd and --time-cmd:
--list-cmdprints the stored path of every file, one per line.--restore-cmdrestores one file.{path}is its stored path and{dest}is the target folder, and the file must land at{dest}/<path without a leading slash>.--time-cmdprints when the newest backup was made, as a Unix timestamp.
Here's how that looks for restic, as three tiny scripts (they need jq, and GNU date, so Linux):
#!/bin/sh
# restic-list.sh - every file in the newest snapshot
restic ls latest --json | jq -r 'select(.type == "file") | .path'#!/bin/sh
# restic-restore.sh <stored path> <destination>
restic restore latest --target "$2" --include "$1" --quiet#!/bin/sh
# restic-time.sh - when was the newest snapshot made?
date -d "$(restic snapshots --json | jq -r 'max_by(.time).time')" +%sexport RESTIC_REPOSITORY=/mnt/backup/restic RESTIC_PASSWORD_FILE=/root/.restic-password
python3 restoretest.py \
--list-cmd ./restic-list.sh \
--restore-cmd './restic-restore.sh {path} {dest}' \
--time-cmd ./restic-time.sh \
--source /srv/dataA few cautions. These commands follow restic's documented behaviour, but run each script by hand once before trusting it, and check that a restored file really lands under the destination folder. restic treats --include as a pattern, so file names containing *, ? or [ may need escaping. And because the restore command runs once per sampled file, a big sample against a remote repository takes a while, so lower --sample if it does.
borg follows the same shape (these are borg 1.x commands, so check them against your version): list with borg list --format '{path}{NL}' REPO::ARCHIVE, restore with cd {dest} && borg extract REPO::ARCHIVE {path}, and read the time from borg list --last 1 --json REPO. Borg lists directories too, so they inflate the file count slightly. That's harmless for the shrink check, because only files are ever restored.
Break it on purpose
A monitor you haven't seen fail is as untrustworthy as a backup you haven't restored. So let's watch it fail: take a healthy demo tree, back it up, and then break the backup in the ways backups break in practice. First, the healthy run:
Restore test - 2026-11-12 07:30
Backup: /backups/data-2026-11-12.tar.gz (made 0.0 h ago)
[PASS] fresh newest backup is 0.0 h old (limit 26 h)
[PASS] size 82 files (+0.0%), 3.4 MB archive (+0.0%) vs the last good run
[PASS] coverage 22 of 22 sampled live files are in the backup
[PASS] canary restored canary was written 0.0 h ago (limit 26 h)
[PASS] sqlite app.sqlite: integrity_check ok, 1 table
[PASS] restore 22 of 22 restored files match the live copies
[INFO] skipped 0 changed since the backup, 0 unreadable, 0 over 500 MB, 0 unreadable folders
[INFO] timing restored 22 files (2.5 MB) in 0.0 s
RESULT: OKNow the breakages. Each row is a separate run against a freshly broken backup, and the middle column is the actual line the report printed:
What I broke | What the report said |
|---|---|
The backup job stopped three days ago ( |
|
One byte flipped inside an uncompressed |
|
A |
|
A |
|
An exclude rule drops the |
|
The backup captured an empty database |
|
The job forgot to refresh the canary |
|
Two of those deserve a closer look. A flipped byte in an uncompressed .tar is the nastiest case, because the archive still reads perfectly and the damage is only visible in the content. Here the sample was widened to every file with --sample 200, so the damaged one was guaranteed to be picked:
Restore test - 2026-11-12 07:30
Backup: /backups/data-plain.tar (made 0.0 h ago)
[PASS] fresh newest backup is 0.0 h old (limit 26 h)
[PASS] size 82 files, 5.5 MB archive (first run, so no baseline yet)
[PASS] coverage 82 of 82 sampled live files are in the backup
[PASS] canary restored canary was written 0.0 h ago (limit 26 h)
[PASS] sqlite app.sqlite: integrity_check ok, 1 table
[FAIL] restore 81 of 82 restored files match the live copies
MISMATCH photos/img-07.jpg
[INFO] skipped 0 changed since the backup, 0 unreadable, 0 over 500 MB, 0 unreadable folders
[INFO] timing restored 82 files (5.4 MB) in 0.0 s
RESULT: FAIL (1 problem)And here's the dropped folder. This run follows a healthy one, which gave the size check its baseline. The size check and the coverage check both fire, and the report names the missing files:
Restore test - 2026-11-12 07:30
Backup: /backups/data-2026-11-12.tar.gz (made 0.0 h ago)
[PASS] fresh newest backup is 0.0 h old (limit 26 h)
[FAIL] size 52 files (-36.6%), 0.1 MB archive (-98.4%) vs the last good run; limit is -20%
[FAIL] coverage 14 of 21 sampled live files are in the backup
NOT IN BACKUP photos/img-28.jpg
NOT IN BACKUP photos/img-25.jpg
NOT IN BACKUP photos/img-29.jpg
NOT IN BACKUP photos/img-14.jpg
NOT IN BACKUP photos/img-24.jpg
NOT IN BACKUP photos/img-21.jpg
NOT IN BACKUP photos/img-04.jpg
[PASS] canary restored canary was written 0.0 h ago (limit 26 h)
[PASS] sqlite app.sqlite: integrity_check ok, 1 table
[PASS] restore 14 of 14 restored files match the live copies
[INFO] skipped 0 changed since the backup, 0 unreadable, 0 over 500 MB, 0 unreadable folders
[INFO] timing restored 15 files (1.8 MB) in 0.0 s
RESULT: FAIL (2 problems)One more row worth showing is the non-failure. Five files touched since the backup are counted and skipped, and the run still passes:
[INFO] skipped 5 changed since the backup, 0 unreadable, 0 over 500 MB, 0 unreadable foldersReporting and scheduling
Every full run that gets as far as the restore adds a row to the history table (time, backup time, file count, size, status, restore seconds). The script exits 0 when everything passes, 1 when any check fails and 2 for a usage error, so it works anywhere an exit code does. For push alerts, --notify runs any command on failure and pipes the report to its standard input:
--notify 'curl -s -d @- https://ntfy.sh/pick-an-unguessable-topic'(ntfy topics are public by obscurity, so anyone who guesses the name can read them. mail -s "backup problem" [email protected] works too if your server can send mail.)
Two cron entries cover everything: a daily freshness check and the weekly full test. The weekly report is mailed even when everything passes, so if that Sunday email ever stops arriving, the job itself has stopped.
# every day at 07:00 server time: is there a fresh backup? (silent unless there isn't)
0 7 * * * python3 /opt/restoretest/restoretest.py --tar '/backups/data-*.tar.gz' --quick --quiet
# every Sunday at 08:00 server time: the full restore test
0 8 * * 0 python3 /opt/restoretest/restoretest.py --tar '/backups/data-*.tar.gz' --live-root /srv/data --source /srv/data --sqlite /srv/data/db/app.sqlite --canary /srv/data/.backup-canary --db /var/lib/restoretest/history.sqliteCreate the history folder once with mkdir -p /var/lib/restoretest. If your server is on UTC and you think in IST, remember that cron runs on the server's clock. I dug into that trap, including the midnight day shift, in Why Your Cron Job Fires at the Wrong Time: IST, UTC and the Midnight Day Shift. Cron mails whatever a job prints to the crontab's owner (or to MAILTO, if you set it), which is how the daily job reports a failure and the weekly job delivers its report.
The full script
Save this as restoretest.py. Python 3.9 or newer is all it needs. It sticks to the standard library and POSIX features, so it should run on Linux and macOS.
#!/usr/bin/env python3
"""restoretest.py - prove that your backups restore.
Standard library only (Python 3.9+; written for Linux and macOS). It checks that the newest backup is
fresh and has not shrunk, restores a random sample of your LIVE files from it into a
private temp folder, compares checksums, and fails loudly when anything is off.
python3 restoretest.py --tar '/backups/data-*.tar.gz' --live-root /srv/data --source /srv/data
python3 restoretest.py --tar '/backups/data-*.tar.gz' --quick --quiet # daily: silent unless stale
"""
from __future__ import annotations
import argparse
import fnmatch
import glob
import hashlib
import heapq
import os
import posixpath
import random
import shlex
import shutil
import sqlite3
import stat
import subprocess
import sys
import tarfile
import tempfile
import time
from dataclasses import dataclass
from datetime import datetime
from pathlib import Path
CHUNK = 1 << 20
class Problem(Exception):
"""An expected failure, with a message that is fit to show in the report."""
def __init__(self, name, message):
super().__init__(message)
self.name, self.message = name, message
# ---------------------------------------------------------------- 1. names
def normalize(name):
"""'./a/b', '/a/b' and 'a/b' are the same stored file. None for names we refuse."""
key = posixpath.normpath(name.lstrip("/"))
if key in (".", "..") or key.startswith("../"):
return None
return key
def live_key(path, root):
"""The name a live file has inside the backup (tar's -C directory is the root)."""
return normalize(os.path.relpath(path, root))
def safe_join(dest, key):
target = (Path(dest) / key).resolve()
if not target.is_relative_to(Path(dest).resolve()):
raise ValueError(f"unsafe path in backup: {key}")
return target
# ---------------------------------------------------------------- 2. backends
@dataclass
class Entry:
name: str # exactly as the backup lists it (handed back to restore commands)
key: str # normalized, used for matching
size: int | None = None
class TarBackend:
"""Plain .tar, .tar.gz, .tar.xz and .tar.bz2 archives. The newest match of the glob is tested."""
def __init__(self, pattern):
matches = [m for m in glob.glob(os.path.expanduser(pattern)) if os.path.isfile(m)]
if not matches:
raise Problem("backup", f"no backup found matching {pattern}")
self.path = max(matches, key=os.path.getmtime)
self.label = self.path
def created(self):
return os.path.getmtime(self.path)
def total_bytes(self):
return os.path.getsize(self.path)
def entries(self):
out = []
with tarfile.open(self.path, "r:*") as tar:
for m in tar:
key = normalize(m.name) if m.isreg() else None
if key:
out.append(Entry(m.name, key, m.size))
while tar.fileobj.read(CHUNK): # tarfile stops early; reading on makes gzip/xz verify their checksums
pass
return out
def restore(self, entries, dest):
"""One streaming pass. We copy bytes ourselves, so no links, modes or paths are trusted."""
wanted = {e.key for e in entries}
with tarfile.open(self.path, "r:*") as tar:
for m in tar:
key = normalize(m.name) if m.isreg() else None
if key in wanted:
target = safe_join(dest, key)
target.parent.mkdir(parents=True, exist_ok=True)
with tar.extractfile(m) as src, open(target, "wb") as out:
shutil.copyfileobj(src, out)
return {}
class CommandBackend:
"""Any tool you can drive from a shell: restic, borg, rsnapshot, a cloud CLI.
list-cmd prints the stored path of every FILE, one per line
restore-cmd restores ONE file: {path} is its stored path, {dest} the target folder
(the file must end up at {dest}/<path without a leading slash>)
time-cmd prints when the newest backup was made, as a Unix timestamp
"""
def __init__(self, list_cmd, restore_cmd, time_cmd):
self.list_cmd, self.restore_cmd, self.time_cmd = list_cmd, restore_cmd, time_cmd
self.label = "custom commands"
@staticmethod
def _run(cmd):
p = subprocess.run(cmd, shell=True, capture_output=True, text=True, timeout=3600)
if p.returncode != 0:
raise RuntimeError(f"exit {p.returncode} from: {cmd}\n{p.stderr.strip()[:400]}")
return p.stdout
def created(self):
try:
return float(self._run(self.time_cmd).strip())
except (RuntimeError, ValueError) as e:
raise Problem("backup", f"could not read the backup time: {e}")
def total_bytes(self):
return None
def entries(self):
out = []
for line in self._run(self.list_cmd).split("\n"):
key = normalize(line) if line and not line.endswith("/") else None
if key:
out.append(Entry(line, key))
return out
def restore(self, entries, dest):
"""Returns {key: error} for the files that failed, so one bad file doesn't hide the rest."""
errors = {}
for e in entries:
cmd = (self.restore_cmd.replace("{dest}", shlex.quote(str(dest)))
.replace("{path}", shlex.quote(e.name)))
try:
self._run(cmd)
except (RuntimeError, subprocess.SubprocessError) as err:
errors[e.key] = str(err).splitlines()[-1][:100]
return errors
# ---------------------------------------------------------------- 3. the live side
@dataclass
class Candidate:
path: str
key: str
size: int
mtime: float
def scan_live(cfg, cutoff, rng):
"""One pass over the live files. Returns (sample, stats of what was left out).
Only files untouched since the backup, readable by us and not huge can be compared
fairly. From those we keep a uniform random sample (reservoir sampling) plus the few largest.
"""
def excluded(key):
return any(fnmatch.fnmatchcase(key, pattern) for pattern in cfg.exclude)
stats = {"changed": 0, "unreadable": 0, "too_big": 0, "folders": 0}
def unreadable_folder(error):
stats["folders"] += 1
reservoir, largest, seen = [], [], 0
for source in cfg.source:
for dirpath, dirnames, filenames in os.walk(source, onerror=unreadable_folder):
dirnames[:] = [d for d in dirnames
if not excluded((live_key(os.path.join(dirpath, d), cfg.live_root) or "") + "/")]
for name in filenames:
full = os.path.join(dirpath, name)
key = live_key(full, cfg.live_root)
if key is None or excluded(key):
continue
try:
st = os.lstat(full)
except OSError:
continue
if not stat.S_ISREG(st.st_mode):
continue
if st.st_mtime >= cutoff:
stats["changed"] += 1
elif st.st_size > cfg.max_file_mb * 1e6:
stats["too_big"] += 1
elif not os.access(full, os.R_OK):
stats["unreadable"] += 1
else:
c = Candidate(full, key, st.st_size, st.st_mtime)
seen += 1
if len(reservoir) < cfg.sample:
reservoir.append(c)
elif (j := rng.randrange(seen)) < cfg.sample:
reservoir[j] = c
heapq.heappush(largest, (c.size, seen, c))
if len(largest) > cfg.largest:
heapq.heappop(largest)
picked = {c.key: c for c in reservoir + [c for _, _, c in largest]}
return list(picked.values()), stats
def sha256(path):
h = hashlib.sha256()
with open(path, "rb") as f:
for chunk in iter(lambda: f.read(CHUNK), b""):
h.update(chunk)
return h.hexdigest()
def compare(c, tmp, errors):
"""Compare one restored file with its live original -> (status, text)."""
if c.key in errors:
return "bad", f"RESTORE FAILED {c.key} ({errors[c.key]})"
restored = safe_join(tmp, c.key)
if not restored.is_file():
return "bad", f"MISSING {c.key} (not there after the restore)"
try:
st = os.lstat(c.path)
if st.st_size != c.size or st.st_mtime != c.mtime:
return "changed", "" # edited while we were testing
if restored.stat().st_size != c.size:
return "bad", f"SIZE DIFFERS {c.key} ({restored.stat().st_size:,} bytes back, {c.size:,} live)"
if sha256(restored) != sha256(c.path):
return "bad", f"MISMATCH {c.key}"
except OSError as e:
return "bad", f"UNREADABLE {c.key} ({e.strerror})"
return "ok", ""
def check_sqlite(path):
"""A restored SQLite file must open, pass integrity_check and not be empty."""
try:
con = sqlite3.connect(path)
try:
tables = con.execute("SELECT count(*) FROM sqlite_master WHERE type = 'table'").fetchone()[0]
result = con.execute("PRAGMA integrity_check").fetchall()
finally:
con.close()
except sqlite3.Error as e:
return False, f"{type(e).__name__}: {e}"
if result != [("ok",)]:
return False, "integrity_check: " + " ".join(str(r[0]) for r in result[:3]).replace("\n", " | ")[:160]
if tables == 0:
return False, "the database is empty (no tables)"
return True, f"integrity_check ok, {tables} table{'s' if tables != 1 else ''}"
def check_canary(cfg, path):
"""The canary holds the Unix time the backup job wrote it: proof the data is recent."""
try:
written = int(path.read_text().split()[0])
except (OSError, ValueError, IndexError):
return False, "the restored canary file has no timestamp in it"
age = (time.time() - written) / 3600
return age <= cfg.max_age_hours, f"restored canary was written {age:.1f} h ago (limit {cfg.max_age_hours:g} h)"
# ---------------------------------------------------------------- 4. report and history
class Report:
def __init__(self):
self.lines, self.failures, self.backup, self.stats = [], 0, "", None
def add(self, name, ok, text, details=()):
self.lines.append(f"[{'PASS' if ok else 'FAIL'}] {name:<9} {text}")
details = list(details)
self.lines += [f" {d}" for d in details[:10]]
if len(details) > 10:
self.lines.append(f" ... and {len(details) - 10} more")
self.failures += 0 if ok else 1
def info(self, name, text):
self.lines.append(f"[INFO] {name:<9} {text}")
def text(self):
result = "RESULT: OK" if not self.failures else \
f"RESULT: FAIL ({self.failures} problem{'s' if self.failures != 1 else ''})"
head = [f"Restore test - {datetime.now():%Y-%m-%d %H:%M}"]
if self.backup:
head.append(f"Backup: {self.backup}")
return "\n".join(head + [""] + self.lines + ["", result])
def open_db(path):
db = sqlite3.connect(path)
db.execute("""CREATE TABLE IF NOT EXISTS runs (
ts REAL, backup_time REAL, file_count INTEGER, total_bytes INTEGER,
status TEXT, restore_seconds REAL)""")
return db
def notify(cmd, text):
try:
subprocess.run(cmd, shell=True, input=text, text=True, timeout=60)
except (OSError, subprocess.SubprocessError) as e:
print(f"notify command failed: {e}", file=sys.stderr)
# ---------------------------------------------------------------- 5. the test
def pct(new, old):
return 100 * (new - old) / old if old else 0.0
def check_size(cfg, report, db, files, nbytes):
prev = db.execute("SELECT file_count, total_bytes FROM runs WHERE status = 'ok' "
"ORDER BY ts DESC LIMIT 1").fetchone()
if files == 0:
report.add("size", False, "the backup contains no files")
return
text = f"{files:,} files"
if not prev:
report.add("size", True, text + (f", {nbytes / 1e6:,.1f} MB archive" if nbytes else "")
+ " (first run, so no baseline yet)")
return
changes = [pct(files, prev[0])]
text += f" ({changes[0]:+.1f}%)"
if nbytes:
text += f", {nbytes / 1e6:,.1f} MB archive"
if prev[1]:
changes.append(pct(nbytes, prev[1]))
text += f" ({changes[1]:+.1f}%)"
ok = min(changes) >= -cfg.shrink_pct or cfg.accept_shrink
report.add("size", ok, text + " vs the last good run" + ("" if ok else f"; limit is -{cfg.shrink_pct:g}%"))
def run(cfg, report, db):
backend = TarBackend(cfg.tar) if cfg.tar else CommandBackend(cfg.list_cmd, cfg.restore_cmd, cfg.time_cmd)
created = backend.created()
age = (time.time() - created) / 3600
report.backup = f"{backend.label} (made {age:.1f} h ago)"
report.add("fresh", age <= cfg.max_age_hours, f"newest backup is {age:.1f} h old (limit {cfg.max_age_hours:g} h)")
if cfg.quick:
return
try:
entries = backend.entries()
except Exception as e:
raise Problem("backup", f"cannot read the backup: {type(e).__name__}: {e}")
by_key = {e.key: e for e in entries}
check_size(cfg, report, db, len(by_key), backend.total_bytes())
# Which live files should we compare? Only ones untouched since the backup (minus a grace margin).
sample, left_out = scan_live(cfg, created - cfg.grace_minutes * 60, random.Random(cfg.seed))
present = [c for c in sample if c.key in by_key]
missing = [c.key for c in sample if c.key not in by_key]
hint = ["If every file is missing, --live-root probably doesn't match how the backup stores paths."] \
if missing and not present else []
if sample:
report.add("coverage", not missing, f"{len(present)} of {len(sample)} sampled live files are in the backup",
hint + [f"NOT IN BACKUP {k}" for k in missing])
else:
report.add("coverage", False, "nothing to compare: no unchanged, readable files found under --source "
"(is --source inside --live-root?)")
canary = live_key(cfg.canary, cfg.live_root) if cfg.canary else None
databases = [(p, live_key(p, cfg.live_root)) for p in cfg.sqlite]
extras = {k for k in [canary] + [k for _, k in databases] if k in by_key}
for label, key in [("canary", canary)] + [("sqlite", k) for _, k in databases]:
if key and key not in by_key:
report.add(label, False, f"{key} is not in the backup")
wanted = [by_key[k] for k in sorted({c.key for c in present} | extras)]
need = sum(c.size for c in present) + sum(by_key[k].size or 0 for k in extras)
tmp_root = cfg.tmp or tempfile.gettempdir()
free = shutil.disk_usage(tmp_root).free
if free < need * 1.1 + 10e6:
raise Problem("restore", f"not enough free space in {tmp_root} ({need / 1e6:,.0f} MB needed, "
f"{free / 1e6:,.0f} MB free); point --tmp somewhere bigger")
tmp = tempfile.mkdtemp(prefix="restoretest-", dir=cfg.tmp) # created with mode 0700
try:
started = time.monotonic()
try:
errors = backend.restore(wanted, tmp)
except Exception as e:
raise Problem("restore", f"the restore itself failed: {type(e).__name__}: {e}")
seconds = time.monotonic() - started
if canary in extras:
ok, text = check_canary(cfg, safe_join(tmp, canary)) if canary not in errors \
else (False, "the canary file did not restore")
report.add("canary", ok, text)
for path, key in databases:
if key in extras:
ok, text = check_sqlite(safe_join(tmp, key)) if key not in errors \
else (False, "the database did not restore")
report.add("sqlite", ok, f"{os.path.basename(path)}: {text}")
results = [compare(c, tmp, errors) for c in present]
bad = [text for status, text in results if status == "bad"]
compared = sum(1 for status, _ in results if status != "changed")
if present:
report.add("restore", not bad, f"{compared - len(bad)} of {compared} restored files match the live copies", bad)
restored_bytes = sum(p.stat().st_size for p in Path(tmp).rglob("*") if p.is_file())
report.stats = dict(created=created, files=len(by_key), nbytes=backend.total_bytes(), seconds=seconds)
report.info("skipped", f"{left_out['changed']:,} changed since the backup, {left_out['unreadable']:,} unreadable, "
f"{left_out['too_big']:,} over {cfg.max_file_mb:g} MB, "
f"{left_out['folders']:,} unreadable folders")
report.info("timing", f"restored {len(wanted)} files ({restored_bytes / 1e6:,.1f} MB) in {seconds:.1f} s")
finally:
shutil.rmtree(tmp, ignore_errors=True)
def parse():
ap = argparse.ArgumentParser(description="Prove that your backups restore.")
ap.add_argument("--tar", metavar="GLOB", help="tar archives; the newest match is tested")
ap.add_argument("--list-cmd", help="command that prints the stored path of every file")
ap.add_argument("--restore-cmd", help="command that restores one file: uses {path} and {dest}")
ap.add_argument("--time-cmd", help="command that prints the newest backup's Unix timestamp")
ap.add_argument("--source", action="append", default=[], metavar="DIR",
help="live folder that should be in the backup (repeatable)")
ap.add_argument("--live-root", default="/", help="what the backup's paths are relative to (tar's -C)")
ap.add_argument("--exclude", action="append", default=[], metavar="PATTERN",
help="live paths your backup deliberately skips, e.g. '*/cache/*' (repeatable)")
ap.add_argument("--sqlite", action="append", default=[], metavar="PATH",
help="live SQLite file that must restore and pass integrity_check (repeatable)")
ap.add_argument("--canary", metavar="PATH", help="file your backup job refreshes with `date +%%s`")
ap.add_argument("--sample", type=int, default=20, help="random files to restore (default 20)")
ap.add_argument("--largest", type=int, default=2, help="plus this many of the largest (default 2)")
ap.add_argument("--max-file-mb", type=float, default=500, help="never sample live files bigger than this (default 500)")
ap.add_argument("--max-age-hours", type=float, default=26)
ap.add_argument("--shrink-pct", type=float, default=20, help="alert if the backup shrinks more than this")
ap.add_argument("--accept-shrink", action="store_true", help="treat this run's size as the new normal")
ap.add_argument("--grace-minutes", type=float, default=60, help="ignore files modified this close to the backup")
ap.add_argument("--db", default="restoretest.sqlite", help="run history")
ap.add_argument("--tmp", metavar="DIR", help="where to restore to (default: the system temp folder)")
ap.add_argument("--notify", metavar="CMD", help="on failure, run CMD with the report on stdin")
ap.add_argument("--quick", action="store_true", help="only check that the newest backup is fresh")
ap.add_argument("--quiet", action="store_true", help="print the report only when something failed")
ap.add_argument("--seed", type=int, help="fix the random sample (for testing)")
cfg = ap.parse_args()
if bool(cfg.tar) == bool(cfg.list_cmd):
ap.error("give either --tar, or --list-cmd with --restore-cmd and --time-cmd")
if cfg.list_cmd and not (cfg.restore_cmd and cfg.time_cmd):
ap.error("--list-cmd needs --restore-cmd and --time-cmd as well")
if not cfg.quick and not cfg.source:
ap.error("--source is required (unless you use --quick)")
return cfg
def main():
cfg = parse()
db = None if cfg.quick else open_db(cfg.db)
report = Report()
try:
run(cfg, report, db)
except Problem as e:
report.add(e.name, False, e.message)
except Exception as e: # a crash must never look like silence
report.add("error", False, f"{type(e).__name__}: {e}")
if db and report.stats:
s = report.stats
db.execute("INSERT INTO runs VALUES (?, ?, ?, ?, ?, ?)",
(time.time(), s["created"], s["files"], s["nbytes"],
"fail" if report.failures else "ok", s["seconds"]))
db.commit()
text = report.text()
if report.failures or not cfg.quiet:
print(text)
if report.failures:
if cfg.notify:
notify(cfg.notify, text)
sys.exit(1)
if __name__ == "__main__":
main()What it can't catch
No test proves everything. Here's what this one doesn't cover:
A sample isn't a full restore. Random sampling is excellent at catching systemic failures, such as a missing folder, a stale job or a broken archive. It's weak at catching one bad file among thousands, because each run only has a
sample / totalchance of picking it. Compressed archives get a full-stream check for free, but for repositories, run your tool's own deep verification now and then too (for examplerestic check --read-data).Content only. Ownership, permissions and extended attributes are not compared.
Whatever this user can't read. Unreadable files and folders are skipped and counted in the
skippedline, not tested. Run the test as a user that can read everything the backup covers, usually the same account the backup job uses.Hard links. tar stores repeated hard links as link entries rather than regular files, so those can show up as "not in backup". Exclude them.
Replaced files with preserved timestamps. A file swapped for different content with the same modification time looks "unchanged" and would be reported as a mismatch.
A backup on the same machine is not a backup. If the archive lives on the disk it protects, this script will happily confirm that it restores right up until the disk dies.
Keys and passwords. An encrypted backup restores only if you still have the key. Store it somewhere other than the server being backed up, and make sure the test has access to it.
Where to take it next
Prefer files that have never been verified. Keep a table of files already tested and bias the sample toward the oldest, so coverage grows systematically instead of randomly.
Restore a database dump into a scratch Postgres or MySQL and run a sanity query, the same way the SQLite check does.
Add a monthly full verification that reads and checksums the whole archive.
This is the second small "snapshot, check, alert" tool in a row, after the dead-link checker. They share a shape: remember something, compare it with reality on a schedule, and speak up only when something is wrong. A script that proves your backups work every Sunday is worth far more than a restore you attempt for the first time during an outage.
Comments(0)
Join the discussion by logging into your account.
No comments yet. Be the first to comment!