Make Nextcloud NAS backups independent, verified and retryable

scripts/nextcloud-borg-runner.py

@@ -0,0 +1,392 @@
+#!/usr/bin/env python3
+"""Verified cloud staging and fail-closed Borg backup. Standard library only."""
+import datetime as dt
+import fcntl
+import hashlib
+import json
+import os
+import re
+from pathlib import Path, PurePosixPath
+import shutil
+import signal
+import subprocess
+import sys
+import tempfile
+import time
+
+
+class BackupError(RuntimeError):
+    pass
+
+
+def log(message):
+    print(f'{dt.datetime.now().astimezone().isoformat(timespec="seconds")} {message}', flush=True)
+
+
+def atomic_json(path, data):
+    descriptor, name = tempfile.mkstemp(prefix=path.name + '.', dir=path.parent)
+    tmp = Path(name)
+    try:
+        with os.fdopen(descriptor, 'w') as stream:
+            json.dump(data, stream, ensure_ascii=False, indent=2)
+            stream.flush()
+            os.fsync(stream.fileno())
+        tmp.replace(path)
+        directory = os.open(path.parent, os.O_RDONLY | os.O_DIRECTORY)
+        try:
+            os.fsync(directory)
+        finally:
+            os.close(directory)
+    finally:
+        tmp.unlink(missing_ok=True)
+
+
+def read_json(path):
+    if not path.exists():
+        return {}
+    return json.loads(path.read_text())
+
+
+class Runner:
+    def __init__(self):
+        e = os.environ
+        self.state = Path(e['STATE_DIR']).expanduser().absolute()
+        self.repo = Path(e['REPO_DIR']).absolute()
+        self.stage = Path(e['STAGING_DIR']).expanduser().absolute()
+        self.remote = e['REMOTE']
+        if not re.fullmatch(r'[A-Za-z0-9_.-]+', e['ARCHIVE_PREFIX']):
+            raise BackupError('Unsafe ARCHIVE_PREFIX')
+        self.borg = e.get('BORG_BIN', 'borg')
+        self.timeout = int(e.get('COMMAND_TIMEOUT_SECONDS', '21600'))
+        self.success_file = self.state / 'last-success.json'
+        self.attempt_file = self.state / 'last-attempt.json'
+        self.child = None
+        self.phase = 'preflight'
+        self.state.mkdir(parents=True, exist_ok=True, mode=0o700)
+        self.success = read_json(self.success_file)
+        # An old timestamp from a different repository/source must never suppress a backup.
+        if self.success and (self.success.get('repo') != str(self.repo) or
+                             self.success.get('remote') != self.remote):
+            self.success = {}
+
+    def run(self, args, *, capture=False, cwd=None, timeout=None):
+        log('Running: ' + ' '.join(map(str, args)))
+        self.child = subprocess.Popen(list(map(str, args)), cwd=cwd,
+                                      stdout=subprocess.PIPE if capture else None,
+                                      start_new_session=True, text=True)
+        try:
+            out, _ = self.child.communicate(timeout=timeout or self.timeout)
+            if self.child.returncode:
+                raise BackupError(f'{args[0]} failed (exit {self.child.returncode}); phase={self.phase}')
+            return out
+        except BaseException:
+            self.stop_child()
+            raise
+        finally:
+            self.child = None
+
+    def stop_child(self):
+        if self.child and self.child.poll() is None:
+            os.killpg(self.child.pid, signal.SIGTERM)
+            try:
+                self.child.wait(timeout=10)
+            except subprocess.TimeoutExpired:
+                os.killpg(self.child.pid, signal.SIGKILL)
+                self.child.wait()
+
+    def cloud(self, *args, capture=False):
+        return self.run(['rclone', *args, '--contimeout', '20s', '--timeout', '2m',
+                         '--retries', '3', '--low-level-retries', '10',
+                         '--log-level', 'INFO', '--stats', '1m'], capture=capture)
+
+    def nas(self):
+        # Trigger autofs, then verify the backing mount. Never mkdir or init on /net.
+        self.run(['timeout', '60', 'test', '-f', self.repo / 'config'], timeout=70)
+        mounts = json.loads(self.run(['findmnt', '--json', '--target', self.repo,
+                                    '--output', 'SOURCE,FSTYPE,MAJ:MIN'], capture=True, timeout=60))['filesystems']
+        device = self.repo.stat().st_dev
+        backing_device = f'{os.major(device)}:{os.minor(device)}'
+        mounts = [mount for mount in mounts if mount.get('maj:min') == backing_device]
+        expected = os.environ.get('EXPECTED_NAS_SOURCE', 'thor:/volume1/NetBackup')
+        fstype = os.environ.get('EXPECTED_NAS_FSTYPE', 'nfs4')
+        if len(mounts) != 1 or mounts[0]['source'] != expected or mounts[0]['fstype'] != fstype:
+            raise BackupError(f'Wrong NAS mount; expected {expected} ({fstype}), got {mounts}')
+        free = shutil.disk_usage(self.repo).free
+        if free < int(os.environ.get('MIN_FREE_REPO_GB', '5')) * 1024**3:
+            raise BackupError('Insufficient free NAS space')
+
+    def manifest(self):
+        raw = json.loads(self.cloud('lsjson', self.remote, '--recursive', '--files-only',
+                                    '--hash', '--no-mimetype', capture=True))
+        result = {}
+        for item in raw:
+            name = item['Path']
+            if (not name or name.startswith('/') or '..' in PurePosixPath(name).parts or
+                    '\x00' in name or item.get('IsDir') or item['Size'] < 0 or name in result):
+                raise BackupError('Unsafe, duplicate or unsupported cloud entry')
+            result[name] = {'size': item['Size'], 'mtime': item.get('ModTime'),
+                            'hashes': {k: v for k, v in item.get('Hashes', {}).items() if v}}
+        return result
+
+    def verify_local_hashes(self, manifest):
+        # Independent validation avoids any size-only fallback chosen inside rclone.
+        for name, item in manifest.items():
+            path = self.stage / name
+            if path.is_symlink() or not path.is_file() or path.stat().st_size != item['size']:
+                raise BackupError(f'Staging file missing, unsafe or wrong size: {name}')
+            hashes = {key.lower().replace('-', ''): value for key, value in item['hashes'].items()}
+            algorithm = next((key for key in ('sha1', 'md5') if hashes.get(key)), None)
+            if algorithm:
+                digest = hashlib.new(algorithm, usedforsecurity=False)
+                with path.open('rb') as stream:
+                    while chunk := stream.read(8 * 1024 * 1024):
+                        digest.update(chunk)
+                if digest.hexdigest().lower() != hashes[algorithm].lower():
+                    raise BackupError(f'Independent staging hash mismatch: {name}')
+        log('Independent local size/hash verification passed')
+
+    @staticmethod
+    def totals(manifest):
+        return len(manifest), sum(x['size'] for x in manifest.values())
+
+    def guard_source(self, manifest, baseline):
+        count, size = self.totals(manifest)
+        if count < int(os.environ.get('MIN_SOURCE_FILES', '1')) or size <= 0:
+            raise BackupError('Empty or implausibly small cloud source; refusing backup')
+        percent = float(os.environ.get('MAX_SOURCE_DROP_PERCENT', '20'))
+        if not 0 <= percent <= 100:
+            raise BackupError('MAX_SOURCE_DROP_PERCENT must be between 0 and 100')
+        for key, current in [('files', count), ('bytes', size)]:
+            previous = baseline.get(key, 0)
+            if previous and current < previous * (1 - percent / 100):
+                raise BackupError(f'Cloud {key} dropped by more than {percent}%: {previous} -> {current}')
+        log(f'Cloud inventory: {count} files, {size} bytes')
+        return count, size
+
+    def staging_preflight(self, manifest):
+        # Refuse symlinked, overlapping, mounted or foreign staging paths before rclone sync.
+        for other in (self.repo, self.state, Path.home() / 'Nextcloud'):
+            if self.stage == other or self.stage in other.parents or other in self.stage.parents:
+                raise BackupError(f'Unsafe staging overlap: {other}')
+        if self.stage.resolve() != self.stage:
+            raise BackupError('Staging path contains a symlink')
+        self.stage.mkdir(parents=True, exist_ok=True, mode=0o700)
+        os.chmod(self.stage, 0o700)
+        for base, dirs, files in os.walk(self.stage):
+            for name in dirs + files:
+                path = Path(base) / name
+                if path.is_symlink() or path.is_mount():
+                    raise BackupError(f'Unexpected symlink/mount in staging: {path}')
+        if self.stage.is_mount():
+            raise BackupError('Staging must be an ordinary local directory')
+        owner_file = self.stage.parent / (self.stage.name + '.owner.json')
+        expected = {'remote': self.remote, 'stage': str(self.stage)}
+        if owner_file.exists():
+            if read_json(owner_file) != expected:
+                raise BackupError('Staging belongs to another cloud source')
+        elif any(self.stage.iterdir()):
+            raise BackupError('Nonempty unowned staging directory; refusing sync')
+        else:
+            atomic_json(owner_file, expected)
+        mount = json.loads(self.run(['findmnt', '--json', '--target', self.stage,
+                                    '--output', 'FSTYPE'], capture=True))['filesystems'][0]
+        if mount['fstype'] not in ('ext4', 'btrfs', 'xfs', 'f2fs', 'zfs', 'tmpfs'):
+            raise BackupError(f'Staging must be local; found {mount["fstype"]}')
+        total = self.totals(manifest)[1]
+        # Conservative full-copy headroom also covers temporary transfer files.
+        need = total + int(os.environ.get('MIN_FREE_STAGE_GB', '5')) * 1024**3
+        if shutil.disk_usage(self.stage).free < need:
+            raise BackupError(f'Insufficient staging space: need {need} bytes free')
+
+    def baseline(self):
+        if self.success:
+            return self.success
+        data = json.loads(self.run([self.borg, 'info', '--json', '--last', '1',
+                                   '--glob-archives', os.environ['ARCHIVE_PREFIX'] + '-*',
+                                   self.repo], capture=True))
+        archives = data.get('archives', [])
+        if not archives:
+            return {}
+        stats = archives[0]['stats']
+        return {'files': stats['nfiles'], 'bytes': stats['original_size']}
+
+    def stale(self):
+        age = time.time() - self.success.get('completed_at', 0)
+        return age < 0 or age > float(os.environ.get('MAX_BACKUP_AGE_HOURS', '36')) * 3600
+
+    def notify(self, message):
+        log('WARNING: ' + message)
+        stamp = self.state / 'last-notification.json'
+        if time.time() - read_json(stamp).get('at', 0) < 6 * 3600:
+            return
+        if shutil.which('notify-send'):
+            result = subprocess.run(['notify-send', '--urgency=critical',
+                                     'Nextcloud NAS-Sicherung', message], timeout=15,
+                                    stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL)
+            if result.returncode == 0:
+                atomic_json(stamp, {'at': time.time()})
+
+    def backup(self, if_due=False):
+        with Path(os.environ['LOCK_FILE']).open('a') as lock:
+            try:
+                fcntl.flock(lock, fcntl.LOCK_EX | fcntl.LOCK_NB)
+            except BlockingIOError:
+                log('Another backup/maintenance command holds the lock; skipping')
+                return
+            age = time.time() - self.success.get('completed_at', 0)
+            if if_due and 0 <= age < float(os.environ.get('BACKUP_INTERVAL_HOURS', '24')) * 3600:
+                if self.success.get('maintenance_pending') or read_json(self.attempt_file).get('phase') == 'retention':
+                    self.nas()
+                    self.maintenance()
+                    atomic_json(self.attempt_file, {'status': 'success', 'finished_at': time.time()})
+                log('Verified backup is still current; no transfer needed')
+                return
+            attempt = {'started_at': time.time(), 'status': 'running'}
+            atomic_json(self.attempt_file, attempt)
+            try:
+                self.perform()
+            except BaseException as exc:
+                atomic_json(self.attempt_file, {**attempt, 'finished_at': time.time(),
+                                              'status': 'failed', 'phase': self.phase, 'error': str(exc)})
+                try:
+                    self.notify('Sicherung fehlgeschlagen: ' + str(exc) +
+                                (' — Altersgrenze ohne bestätigtes Backup überschritten.' if self.stale() else ''))
+                except Exception as notify_error:
+                    log(f'Notification unavailable: {notify_error}')
+                raise
+            atomic_json(self.attempt_file, {**attempt, 'finished_at': time.time(), 'status': 'success'})
+
+    def check_identity(self):
+        identity = json.loads(self.run([self.borg, 'info', '--json', self.repo], capture=True))['repository']['id']
+        bound = read_json(self.state / 'repository.json')
+        if bound and bound != {'repo': str(self.repo), 'id': identity}:
+            raise BackupError('Repository identity changed; refusing backup/retention')
+        atomic_json(self.state / 'repository.json', {'repo': str(self.repo), 'id': identity})
+
+    def perform(self):
+        self.nas()
+        self.check_identity()
+        excludes = Path(os.environ['EXCLUDE_FILE'])
+        if excludes.exists() and any(x.strip() and not x.lstrip().startswith('#')
+                                     for x in excludes.read_text().splitlines()):
+            raise BackupError('Active legacy Borg excludes found; resolve before full cloud backup')
+        baseline = self.baseline()
+        self.phase = 'cloud inventory'
+        before = self.manifest()
+        count, size = self.guard_source(before, baseline)
+        self.staging_preflight(before)
+        self.phase = 'cloud sync'
+        common = {'md5', 'sha1'}
+        unhashed = sorted(name for name, item in before.items() if item['size'] and not any(
+            key.lower().replace('-', '') in common for key in item['hashes']))
+        # rclone's raw file lists cannot encode newline names. Fall back to a full byte read.
+        full_download = any('\n' in name or '\r' in name for name in unhashed)
+        file_list = self.state / 'unhashed-files.txt'
+        if unhashed and not full_download:
+            file_list.write_text('\n'.join(unhashed) + '\n')
+            log(f'Reading {len(unhashed)} files without server hashes in full')
+            self.cloud('copy', self.remote, str(self.stage), '--ignore-times',
+                       '--files-from-raw', str(file_list), '--transfers', '4')
+        sync_flags = ['--ignore-times'] if full_download else ['--checksum']
+        self.cloud('sync', self.remote, str(self.stage), '--check-first', '--delete-after',
+                   '--max-delete', os.environ.get('MAX_STAGE_DELETES', '1000'),
+                   '--create-empty-src-dirs', '--transfers', '4', '--checkers', '8', *sync_flags)
+        self.phase = 'cloud verification'
+        self.cloud('check', self.remote, str(self.stage), *(['--download'] if full_download else []))
+        if unhashed and not full_download:
+            self.cloud('check', self.remote, str(self.stage), '--download',
+                       '--files-from-raw', str(file_list))
+        self.verify_local_hashes(before)
+        after = self.manifest()
+        if before != after:
+            raise BackupError('Cloud changed during sync/verification; retry with a stable inventory')
+        atomic_json(self.state / 'verified-inventory.json', after)
+        self.nas()
+        self.phase = 'borg create'
+        archive = os.environ['ARCHIVE_PREFIX'] + '-' + dt.datetime.now().strftime('%Y-%m-%d_%H-%M-%S')
+        target = f'{self.repo}::{archive}'
+        self.run([self.borg, 'create', '--stats', '--show-rc', '--compression',
+                  os.environ['BORG_COMPRESSION'], target, '.'], cwd=self.stage)
+        data = json.loads(self.run([self.borg, 'info', '--json', target], capture=True))
+        stats = data['archives'][0]['stats']
+        if stats['nfiles'] != count or stats['original_size'] != size:
+            raise BackupError(f'Archive inventory mismatch: {stats}; expected {count} files/{size} bytes')
+        self.phase = 'archive verification'
+        self.run([self.borg, 'check', '--archives-only', '--verify-data',
+                  '--glob-archives', archive, self.repo])
+        self.run([self.borg, 'check', '--repository-only', self.repo])
+        # Success means a complete verified archive. Retention failure must not lose this fact.
+        self.success = {'completed_at': time.time(), 'archive': archive, 'files': count,
+                        'bytes': size, 'repo': str(self.repo), 'remote': self.remote,
+                        'archive_id': data['archives'][0]['id'], 'maintenance_pending': True}
+        atomic_json(self.success_file, self.success)
+        log(f'BACKUP VERIFIED: {archive} ({count} files, {size} bytes)')
+        self.maintenance()
+        log('Backup and maintenance finished successfully')
+
+    def maintenance(self):
+        self.phase = 'retention'
+        self.check_identity()
+        if self.success.get('archive'):
+            archived = json.loads(self.run([self.borg, 'info', '--json',
+                                  f'{self.repo}::{self.success["archive"]}'], capture=True))['archives'][0]
+            if archived['id'] != self.success['archive_id']:
+                raise BackupError('Verified archive identity changed; refusing retention')
+        if os.environ.get('PRUNE_AFTER_BACKUP', 'true') == 'true':
+            self.run([self.borg, 'prune', '--list', '--show-rc', '--glob-archives',
+                      os.environ['ARCHIVE_PREFIX'] + '-*', '--keep-within',
+                      os.environ['RETENTION_WITHIN'], '--keep-last', '7', self.repo])
+            if os.environ.get('COMPACT_AFTER_PRUNE', 'true') == 'true':
+                self.run([self.borg, 'compact', self.repo])
+        if self.success:
+            self.success['maintenance_pending'] = False
+            atomic_json(self.success_file, self.success)
+        log('Retention finished successfully')
+
+
+def main():
+    os.umask(0o077)
+    runner = Runner()
+    def interrupted(signum, _frame):
+        runner.stop_child()
+        raise BackupError(f'Interrupted by signal {signum}')
+    signal.signal(signal.SIGTERM, interrupted)
+    signal.signal(signal.SIGINT, interrupted)
+    if sys.argv[1:] in (['health'], ['health', '--notify']):
+        print(json.dumps({'healthy': not runner.stale(), 'last_success': runner.success,
+                          'last_attempt': read_json(runner.attempt_file)}, ensure_ascii=False, indent=2))
+        if runner.stale() and '--notify' in sys.argv:
+            runner.notify('NAS-Sicherung überfällig: kein bestätigtes Backup innerhalb der Altersgrenze. Protokolle und health prüfen.')
+        return 2 if runner.stale() else 0
+    if sys.argv[1:] not in (['backup'], ['backup', '--if-due']):
+        raise BackupError('Usage: runner backup [--if-due] | health')
+    logs = Path(os.environ['LOG_DIR'])
+    logs.mkdir(parents=True, exist_ok=True, mode=0o700)
+    path = logs / ('backup-' + dt.datetime.now().strftime('%Y%m%d-%H%M%S') + '.log')
+    # Journal + file, without an orphaned tee process on timeout/interruption.
+    for old in logs.glob('backup-*.log'):
+        if old.is_file() and not old.is_symlink() and old.stat().st_mtime < time.time() - 90 * 86400:
+            old.unlink()
+    with path.open('a', buffering=1) as stream:
+        class Tee:
+            def write(self, data):
+                sys.__stdout__.write(data)
+                stream.write(data)
+            def flush(self):
+                sys.__stdout__.flush()
+                stream.flush()
+        sys.stdout = Tee()
+        # Child output goes to journal; lifecycle/phase/status additionally go to the run log.
+        try:
+            runner.backup('--if-due' in sys.argv)
+        finally:
+            sys.stdout = sys.__stdout__
+    return 0
+
+
+if __name__ == '__main__':
+    try:
+        sys.exit(main())
+    except (Exception, KeyboardInterrupt) as error:
+        log(f'ERROR: {error}')
+        sys.exit(1)