WordPress backups with a restore drill

File: backup.py. Save every line under this filename.

"""Use a frozen staging site; destination must be outside the public site tree."""
import argparse
import hashlib
import json
import os
from pathlib import Path
import subprocess
import tarfile


def digest(path):
    with path.open('rb') as stream:
        return hashlib.file_digest(stream, 'sha256').hexdigest()


def create(site, destination):
    site, destination = Path(site).resolve(), Path(destination).resolve()
    if destination == site or site in destination.parents:
        raise ValueError('Backup must be outside the public site directory')
    destination.mkdir(parents=True, exist_ok=False, mode=0o700)
    # This does not stop external writers. Freeze cron, queues and imports first.
    subprocess.run(['wp', f'--path={site}', 'db', 'export', str(destination / 'database.sql'), '--single-transaction'], check=True)
    with tarfile.open(destination / 'site.tar.gz', 'w:gz', dereference=False) as archive:
        archive.add(site, arcname='site')
    files = ['database.sql', 'site.tar.gz']
    manifest = {name: digest(destination / name) for name in files}
    (destination / 'manifest.json').write_text(json.dumps(manifest, indent=2) + '\n', encoding='utf-8')
    for name in [*files, 'manifest.json']:
        os.chmod(destination / name, 0o600)


if __name__ == '__main__':
    parser = argparse.ArgumentParser()
    parser.add_argument('site')
    parser.add_argument('destination')
    parser.add_argument('--writers-frozen', action='store_true', required=True)
    args = parser.parse_args()
    create(args.site, args.destination)

This file belongs to the complete example above; execution order, prerequisites and limits are explained in the article.

File: verify_backup.py. Save every line under this filename.

import json
from pathlib import Path
import sys
from backup import digest


def verify(folder):
    folder = Path(folder)
    manifest = json.loads((folder / 'manifest.json').read_text(encoding='utf-8'))
    if set(manifest) != {'database.sql', 'site.tar.gz'}:
        raise ValueError('Unexpected backup contents')
    for name, expected in manifest.items():
        if (folder / name).stat().st_size == 0 or digest(folder / name) != expected:
            raise ValueError(f'Invalid backup file: {name}')
    return True


if __name__ == '__main__':
    verify(Path(sys.argv[1]))
    print('PASS: byte integrity only; a staging restore is still required.')

This file belongs to the complete example above; execution order, prerequisites and limits are explained in the article.

File: restore_plan.py. Save every line under this filename.

"""Print a plan for a NEW isolated staging site. Executes no shell commands."""
import argparse
from pathlib import Path
import shlex
from verify_backup import verify


if __name__ == '__main__':
    parser = argparse.ArgumentParser()
    parser.add_argument('backup')
    parser.add_argument('empty_staging_directory')
    args = parser.parse_args()
    source, target = Path(args.backup).resolve(), Path(args.empty_staging_directory).resolve()
    verify(source)
    if target.exists() or source == target or source in target.parents or target in source.parents:
        raise ValueError('Choose a new staging directory separate from the backup')
    q = lambda p: shlex.quote(str(p))
    print('Use an isolated host with mail, payments and outbound jobs disabled.')
    print('Review archive entries and symlinks before extracting trusted backup:')
    print(f'tar -tzf {q(source / "site.tar.gz")}')
    print(f'mkdir -p {q(target)}')
    print(f'tar -xzf {q(source / "site.tar.gz")} -C {q(target)}')
    print('Edit site/wp-config.php to use a NEW empty staging database, then:')
    print(f'wp --path={q(target / "site")} db import {q(source / "database.sql")}')
    print('Check login, Persian text, uploads, record counts and isolated test orders.')

This file belongs to the complete example above; execution order, prerequisites and limits are explained in the article.

File: test_backups.py. Save every line under this filename.

import json
from pathlib import Path
import tempfile
import unittest
from backup import create, digest
from verify_backup import verify


class BackupTests(unittest.TestCase):
    def test_corruption_is_detected(self):
        with tempfile.TemporaryDirectory() as directory:
            root = Path(directory)
            for name in ('database.sql', 'site.tar.gz'):
                (root / name).write_bytes(b'test fixture')
            manifest = {name: digest(root / name) for name in ('database.sql', 'site.tar.gz')}
            (root / 'manifest.json').write_text(json.dumps(manifest), encoding='utf-8')
            self.assertTrue(verify(root))
            (root / 'database.sql').write_bytes(b'corrupted')
            with self.assertRaises(ValueError):
                verify(root)

    def test_backup_cannot_be_public(self):
        with tempfile.TemporaryDirectory() as directory:
            with self.assertRaises(ValueError):
                create(directory, Path(directory) / 'public-backup')


if __name__ == '__main__':
    unittest.main()

This file belongs to the complete example above; execution order, prerequisites and limits are explained in the article.