"""Integration tests use real tar/rsync/rclone crypt and a real isolated MariaDB. systemd/occ are simulated; error paths inject failures around real commands. Production services never receive test commands. """ import gzip import hashlib import importlib.util import io import json import os from pathlib import Path import shutil import signal import subprocess import tarfile import tempfile import time import unittest from email import policy from email.parser import BytesParser REPO = Path(__file__).resolve().parents[1] SPEC = importlib.util.spec_from_file_location("backup_set", REPO / "lib/backup_set.py") HELPER = importlib.util.module_from_spec(SPEC) SPEC.loader.exec_module(HELPER) class ArchiveValidation(unittest.TestCase): def test_reload_parser_requires_global_privilege(self): self.assertTrue(HELPER.has_reload("GRANT SELECT, RELOAD ON *.* TO `backup`@`localhost` IDENTIFIED BY PASSWORD 'secret-hash'")) self.assertTrue(HELPER.has_reload('GRANT ALL PRIVILEGES ON *.* TO `root`@`localhost`')) self.assertFalse(HELPER.has_reload('GRANT ALL PRIVILEGES ON `wordpress`.* TO `backup`@`localhost`')) self.assertFalse(HELPER.has_reload('GRANT `RELOAD` TO `backup`@`localhost`')) def archive(self, entries): temp = tempfile.TemporaryDirectory() self.addCleanup(temp.cleanup) path = Path(temp.name) / 'test.tar.gz' with tarfile.open(path, 'w:gz') as archive: for name, kind, target in entries: info = tarfile.TarInfo(name) info.type = kind info.linkname = target archive.addfile(info) return path def test_traversal_and_link_pivot_rejected(self): for entries in [ [('files/wordpress', tarfile.DIRTYPE, ''), ('files/wordpress/../../evil', tarfile.REGTYPE, '')], [('files/wordpress', tarfile.DIRTYPE, ''), ('files/wordpress/link', tarfile.SYMTYPE, '/tmp'), ('files/wordpress/link/evil', tarfile.REGTYPE, '')], [('files/wordpress', tarfile.DIRTYPE, ''), ('files/wordpress/hard', tarfile.LNKTYPE, '/etc/shadow')], [('files/wordpress', tarfile.SYMTYPE, '/tmp')], [('files/wordpress', tarfile.DIRTYPE, ''), ('files/wordpress/dev', tarfile.CHRTYPE, '')], ]: with self.subTest(entries=entries), self.assertRaises(ValueError): HELPER.check_tar(self.archive(entries), ['files/wordpress']) def test_leaf_symlinks_allowed(self): HELPER.check_tar(self.archive([('files/wordpress', tarfile.DIRTYPE, ''), ('files/wordpress/cert', tarfile.SYMTYPE, '/etc/ssl/cert.pem')]), ['files/wordpress']) def test_truncated_compression_rejected(self): path = self.archive([('files/wordpress', tarfile.DIRTYPE, '')]) path.write_bytes(path.read_bytes()[:-5]) with self.assertRaises((ValueError, tarfile.TarError)): HELPER.check_tar(path, ['files/wordpress']) def test_writer_drain_waits_and_reports_timeout(self): with tempfile.TemporaryDirectory() as directory: writer=subprocess.Popen(['python3','-c','import time; time.sleep(1)',directory+'/cron.php']) try: started=time.monotonic() HELPER.wait_nextcloud(directory,3) self.assertGreaterEqual(time.monotonic()-started,.5) finally: writer.wait(timeout=5) writer=subprocess.Popen(['python3','-c','import time; time.sleep(5)',directory+'/occ']) try: with self.assertRaisesRegex(ValueError,'writers still active'): HELPER.wait_nextcloud(directory,1) finally: writer.terminate() writer.wait(timeout=5) def test_retention_keeps_minimum_and_ignores_v1(self): result = subprocess.run(['python3', str(REPO/'lib/backup_set.py'), 'candidates', 'appbackup', '7', '2'], input='appbackup_2020-01-01_00-00-00_aaaaaaaa\nappbackup_2020-01-02_00-00-00_bbbbbbbb\nappbackup_2020-01-03_00-00-00_cccccccc\nappbackup_2019-01-01_00-00-00\n../evil\n', text=True, capture_output=True, check=True) self.assertEqual(result.stdout.strip(), 'appbackup_2020-01-01_00-00-00_aaaaaaaa') @unittest.skipUnless(os.geteuid() == 0 and shutil.which('mariadbd'), 'run integration tests in tests/Dockerfile container') class Integration(unittest.TestCase): @classmethod def setUpClass(cls): cls.dbtemp = tempfile.TemporaryDirectory(prefix='app-backup-mariadb-') cls.dbdir = Path(cls.dbtemp.name) cls.socket = cls.dbdir/'mysql.sock' subprocess.run(['mariadb-install-db', '--no-defaults', f'--datadir={cls.dbdir}/data', '--auth-root-authentication-method=normal', '--skip-test-db'], stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL, check=True) cls.server = subprocess.Popen(['mariadbd','--no-defaults','--user=root',f'--datadir={cls.dbdir}/data', f'--socket={cls.socket}', f'--pid-file={cls.dbdir}/pid', '--skip-networking'], stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL) for _ in range(100): if subprocess.run(['mariadb','--no-defaults',f'--socket={cls.socket}','-uroot','-e','SELECT 1'], stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL).returncode == 0: break time.sleep(.1) else: raise RuntimeError('Test MariaDB did not start') @classmethod def tearDownClass(cls): cls.server.terminate() cls.server.wait(timeout=20) cls.dbtemp.cleanup() def sql(self, query): return subprocess.check_output(['mariadb','--no-defaults', f'--socket={self.socket}','-uroot','-N','-e',query], text=True) def setUp(self): self.temp = tempfile.TemporaryDirectory(prefix='app-backup-test-') self.addCleanup(self.temp.cleanup) self.base = Path(self.temp.name) self.src = self.base/'sources' self.wp = self.src/'web' self.nc = self.wp/'cloud' self.data = self.nc/'userdata' self.bin = self.base/'bin' self.bin.mkdir() for p in [self.wp,self.nc,self.data,self.src/'gitea',self.src/'gitea-etc',self.src/'mail',self.src/'postfix',self.src/'dovecot']: p.mkdir(parents=True, exist_ok=True) (p/'payload.txt').write_text(f'payload for {p.name}') (self.wp/'nested/data').mkdir(parents=True) (self.wp/'nested/data/keep.txt').write_text('nested data must survive') (self.nc/'occ').write_text('test stub') self.runtime=self.base/'runtime' self.runtime.mkdir() self.state=self.base/'state.json' self.initial=dict(services={s:'active' for s in ['httpd','php-fpm','gitea','postfix','dovecot','nc-cron']}, maintenance=False) self.state.write_text(json.dumps(self.initial)) self.events=self.base/'events' self.events.touch() mock='''#!/usr/bin/env python3 import json,os,sys from pathlib import Path name=Path(sys.argv[0]).name args=sys.argv[1:] p=Path(os.environ['TEST_STATE']) s=json.loads(p.read_text()) with open(os.environ['TEST_EVENTS'],'a') as f: f.write(name+' '+' '.join(args)+'\\n') if name=='systemctl': cmd=args[0] service=args[1] if len(args)>1 else '' runtime=Path(os.environ['TEST_RUNTIME']) unit=service if service.endswith(('.service','.timer','.socket','.path')) else service+'.service' files=sorted((runtime/'systemd/system'/f'{unit}.d').glob('*.conf')) guards=[] for file in files: for line in file.read_text().splitlines(): if line.startswith('ConditionPathExists=!'): guards.append(Path(line.split('=!',1)[1])) stamp_file=Path(str(p)+'.stamps') stamps=json.loads(stamp_file.read_text()) if stamp_file.exists() else {} if cmd=='show': if '--property=LoadState' in args: print('masked' if service in s.get('masked',[]) else ('loaded' if service in s['services'] else 'not-found')) elif '--property=Id' in args: print(unit) elif '--property=DropInPaths' in args: print(' '.join(map(str,files)) if not os.environ.get('TEST_IGNORE_GUARD') else '') elif '--property=ActiveEnterTimestampMonotonic' in args: print('ActiveEnterTimestampMonotonic='+str(stamps.get(service,1))+'\\nInactiveExitTimestampMonotonic='+str(stamps.get(service,1))) elif '--property=Environment' in args or '--property=TriggeredBy' in args: print('') else: print(s['services'].get(service,'inactive')) elif cmd=='is-active': sys.exit(0 if s['services'].get(args[-1])=='active' else 3) elif cmd in ('stop','start'): if os.environ.get('TEST_FAIL_SERVICE')==cmd+':'+service: sys.exit(1) if cmd=='start': if service in s.get('masked',[]): sys.exit(1) if any(g.exists() for g in guards): with open(os.environ['TEST_EVENTS'],'a') as f: f.write('guard blocked '+service+'\\n') sys.exit(0) # systemd skips a failed Condition without starting the unit stamps[service]=stamps.get(service,1)+1 stamp_file.write_text(json.dumps(stamps)) s['services'][service]='active' if cmd=='start' else 'inactive' elif cmd=='enable': s['services'][args[-1]]='active' elif cmd=='daemon-reload': if os.environ.get('TEST_FAIL_GUARD_RELEASE') and list(runtime.glob('app-backup-guard-*')) and not list(runtime.glob('systemd/system/*/zzzz-app-backup-*.conf')): sys.exit(1) elif name=='sudo': os.execvp(args[2],args[2:]) elif name=='php': cmd=args[1] if cmd=='status': print(json.dumps({'maintenance':s['maintenance']})) elif cmd=='maintenance:mode': s['maintenance']='--on' in args p.write_text(json.dumps(s)) ''' for name in ['systemctl','php','sudo']: path=self.bin/name path.write_text(mock) path.chmod(0o755) # Wrappers inject failures but normally delegate to the real tools. for name in ['rsync','mariadb-dump','mariadb','zstd','rclone']: real=shutil.which(name) script=f'''#!/usr/bin/env bash printf '%s\\n' '{name}' >> "$TEST_EVENTS" if [[ ${{TEST_FAIL_TOOL:-}} == {name} ]]; then exit 42; fi if [[ {name} == rclone && ${{TEST_FAIL_CHECK:-}} == true && " $* " == *' check '* ]]; then exit 43; fi exec {real} "$@" ''' path=self.bin/name path.write_text(script) path.chmod(0o755) self.cnf=self.base/'db.cnf' self.cnf.write_text(f'[client]\nuser=root\nsocket={self.socket}\n') self.cnf.chmod(0o600) self.sql('DROP DATABASE IF EXISTS backup_test; CREATE DATABASE backup_test; CREATE TABLE backup_test.t (id INT PRIMARY KEY, body TEXT) ENGINE=MyISAM; INSERT INTO backup_test.t VALUES (1,"original");') self.config=self.base/'config' self.config.write_text(f'''SERVICE_GUARD_RUNTIME_ROOT='{self.runtime}' WORKDIR='{self.base}/work' LOG_DIR='{self.base}/logs' LOCKFILE='{self.base}/lock' MIN_FREE_GB=0 COMPRESSOR=zstd ENABLE_UPLOAD=false ENABLE_MAIL_REPORT=false ENABLE_REMOTE_RETENTION=false ENABLE_WORDPRESS=true WP_DIR='{self.wp}' ENABLE_NEXTCLOUD=true NC_DIR='{self.nc}' NC_DATA_DIR='{self.data}' NC_OCC_USER=root ENABLE_GITEA=true GITEA_DATA_DIR='{self.src}/gitea' GITEA_ETC_DIR='{self.src}/gitea-etc' ENABLE_MAIL=true ENABLE_MAIL_HOME_DIRS=false MAIL_DIR='{self.src}/mail' POSTFIX_DIR='{self.src}/postfix' DOVECOT_DIR='{self.src}/dovecot' WP_DB_NAME=backup_test NC_DB_NAME=backup_test GITEA_DB_NAME=backup_test WP_DB_CNF='{self.cnf}' NC_DB_CNF='{self.cnf}' GITEA_DB_CNF='{self.cnf}' QUIESCE_SERVICES=(nc-cron) ''') self.config.chmod(0o600) self.env=dict(os.environ,TEST_RUNTIME=str(self.runtime), PATH=str(self.bin)+':'+os.environ['PATH'],TEST_STATE=str(self.state),TEST_EVENTS=str(self.events)) def run_script(self, script, *args, ok=True, **env): proc=subprocess.run(['bash',str(REPO/'usr-local-sbin'/script),'--config',str(self.config),*map(str,args)],env=dict(self.env,**env),text=True,stdout=subprocess.PIPE,stderr=subprocess.STDOUT,timeout=90) if ok: self.assertEqual(proc.returncode,0,proc.stdout) else: self.assertNotEqual(proc.returncode,0,proc.stdout) return proc def backup(self, **env): self.run_script('app-backup.sh',**env) return next((self.base/'work/archives').glob('appbackup_*')) def test_real_database_and_files_round_trip(self): run=self.backup() self.assertEqual(json.loads(self.state.read_text()),self.initial) data=HELPER.load_set(run) self.assertEqual(len(data['archives']),12) with subprocess.Popen(['zstd','-dc',str(run/'wordpress.tar.zst')],stdout=subprocess.PIPE) as p: with tarfile.open(fileobj=p.stdout,mode='r|') as t: names=[m.name for m in t] self.assertNotIn('files/wordpress/cloud',names) self.assertIn('files/wordpress/nested/data/keep.txt',names) self.sql('UPDATE backup_test.t SET body="changed"') (self.wp/'payload.txt').write_text('changed') self.events.write_text('') self.run_script('app-restore.sh','--local-run',run,'--dry-run') self.assertNotIn('systemctl stop',self.events.read_text()) self.assertEqual((self.wp/'payload.txt').read_text(),'changed') self.run_script('app-restore.sh','--local-run',run,'--apply','--yes') self.assertEqual(self.sql('SELECT body FROM backup_test.t').strip(),'original') self.assertEqual((self.wp/'payload.txt').read_text(),'payload for web') self.assertEqual(json.loads(self.state.read_text()),self.initial) def test_dump_and_copy_failures_do_not_publish(self): for tool in ['mariadb-dump','rsync','zstd']: with self.subTest(tool=tool): self.run_script('app-backup.sh',ok=False,TEST_FAIL_TOOL=tool) self.assertFalse(list((self.base/'work/archives').glob('appbackup_*'))) self.assertEqual(json.loads(self.state.read_text()),self.initial) def test_rsync_source_denial_fails_preflight_before_stopping_services(self): canonical=self.configure_upgrade() original=canonical.read_bytes() wrapper=self.bin/'rsync' wrapper.write_text('''#!/usr/bin/env bash if [[ " $* " == *" --list-only "* ]]; then echo 'rsync: simulated source opendir permission denied' >&2 exit 23 fi exec /usr/bin/rsync "$@" ''') wrapper.chmod(0o755) package=self.build_package() result=subprocess.run(['bash',str(package),'--foreground'],env=self.env,capture_output=True,text=True,timeout=60) self.assertNotEqual(result.returncode,0,result.stdout+result.stderr) self.assertIn('rsync cannot list source',result.stdout+result.stderr) self.assertEqual(canonical.read_bytes(),original) self.assertEqual(json.loads(self.state.read_text())['services']['app-backup.timer'],'active') self.assertNotIn('systemctl stop',self.events.read_text()) self.assertNotIn('maintenance:mode --on',self.events.read_text()) def test_directory_attribute_denial_is_caught_before_maintenance(self): wrapper=self.bin/'rsync' wrapper.write_text('''#!/usr/bin/env bash if [[ " $* " == *" --exclude=/* "* ]]; then echo 'rsync: simulated security.selinux relabel denial' >&2 exit 23 fi exec /usr/bin/rsync "$@" ''') wrapper.chmod(0o755) failed=self.run_script('app-backup.sh','--check',ok=False) self.assertIn('cannot preserve directory metadata',failed.stdout) self.assertNotIn('systemctl stop',self.events.read_text()) self.assertEqual(json.loads(self.state.read_text()),self.initial) self.assertFalse(list((self.base/'work/staging').iterdir())) def test_metadata_probe_copies_no_children_and_rsync_inherits_no_locks(self): wrapper=self.bin/'rsync' wrapper.write_text('''#!/usr/bin/env python3 import os,subprocess,sys from pathlib import Path for fd in (8,9): try: os.fstat(fd) except OSError: pass else: sys.exit('rsync inherited a coordinator lock descriptor') result=subprocess.run(['/usr/bin/rsync',*sys.argv[1:]]) if result.returncode: sys.exit(result.returncode) if '--exclude=/*' in sys.argv: destination=Path(sys.argv[-1]) assert not list(destination.iterdir()), 'probe copied source contents' source=Path(sys.argv[-2]) assert destination.stat().st_mode == source.stat().st_mode ''') wrapper.chmod(0o755) self.run_script('app-backup.sh','--check') self.assertFalse(list((self.base/'work/staging').iterdir())) self.assertNotIn('systemctl stop',self.events.read_text()) run=self.backup() self.run_script('app-restore.sh','--local-run',run,'--components','wordpress','--apply','--yes') def test_start_guard_blocks_monitor_during_copy_and_preserves_inactive_writer(self): state=dict(services=dict(self.initial['services'],gitea='inactive'),maintenance=False) self.state.write_text(json.dumps(state)) wrapper=self.bin/'rsync' wrapper.write_text('''#!/usr/bin/env bash if [[ " $* " != *" --list-only "* && " $* " != *" --exclude=/* "* ]]; then systemctl start httpd systemctl start php-fpm systemctl start gitea fi exec /usr/bin/rsync "$@" ''') self.backup() self.assertEqual(json.loads(self.state.read_text()),state) for unit in ['httpd','php-fpm','gitea']: self.assertIn('guard blocked '+unit,self.events.read_text()) self.assertFalse(list(self.runtime.glob('systemd/system/*/zzzz-app-backup-*.conf'))) self.assertFalse(list(self.runtime.glob('app-backup-guard-*'))) def inject_writer_restart(self, final_state): wrapper=self.bin/'rsync' wrapper.write_text('''#!/usr/bin/env python3 import json,os,subprocess,sys from pathlib import Path result=subprocess.run(['/usr/bin/rsync',*sys.argv[1:]]) if result.returncode: sys.exit(result.returncode) if '--list-only' not in sys.argv and '--exclude=/*' not in sys.argv: p=Path(os.environ['TEST_STATE']) state=json.loads(p.read_text()) state['services']['httpd']='''+repr(final_state)+''' p.write_text(json.dumps(state)) p=Path(str(p)+'.stamps') stamps=json.loads(p.read_text()) if p.exists() else {} stamps['httpd']=stamps.get('httpd',1)+1 p.write_text(json.dumps(stamps)) ''') result=self.run_script('app-backup.sh',ok=False) self.assertIn('Consistency lost:',result.stdout) self.assertFalse(list((self.base/'work/archives').glob('appbackup_*'))) self.assertEqual(json.loads(self.state.read_text()),self.initial) self.assertFalse(list(self.runtime.glob('systemd/system/*/zzzz-app-backup-*.conf'))) def test_running_writer_prevents_backup_publication(self): self.inject_writer_restart('active') def test_restarted_then_stopped_writer_prevents_backup_publication(self): self.inject_writer_restart('inactive') def test_unloaded_start_guard_aborts_before_service_stop(self): result=self.run_script('app-backup.sh',ok=False,TEST_IGNORE_GUARD='1') self.assertIn('Start guard not loaded last',result.stdout) self.assertNotIn('systemctl stop',self.events.read_text()) self.assertEqual(json.loads(self.state.read_text()),self.initial) self.assertFalse(list(self.runtime.glob('app-backup-guard-*'))) def test_existing_service_mask_is_preserved(self): state=dict(services=dict(self.initial['services'],gitea='inactive'),maintenance=False,masked=['gitea']) self.state.write_text(json.dumps(state)) self.backup() self.assertEqual(json.loads(self.state.read_text()),state) self.assertNotIn('systemctl start gitea',self.events.read_text()) def test_guard_release_failure_keeps_recovery_and_blocks_publication(self): result=self.run_script('app-backup.sh',ok=False,TEST_FAIL_GUARD_RELEASE='1') self.assertIn('Could not release writer start guards',result.stdout) self.assertFalse(list((self.base/'work/archives').glob('appbackup_*'))) self.assertTrue(json.loads(self.state.read_text())['maintenance']) self.assertTrue(all(s=='inactive' for s in json.loads(self.state.read_text())['services'].values())) recovery=json.loads(next((self.base/'work').glob('recovery-*.json')).read_text()) self.assertTrue(Path(recovery['service_guard_directory']).is_dir()) def test_maintenance_and_inactive_service_preserved(self): state=dict(services=dict(self.initial['services'],gitea='inactive'),maintenance=True) self.state.write_text(json.dumps(state)) self.backup() self.assertEqual(json.loads(self.state.read_text()),state) def test_corruption_and_missing_archive_block_restore(self): run=self.backup() path=run/'wordpress.tar.zst' content=path.read_bytes() path.write_bytes(content[:-10]) self.events.write_text('') self.run_script('app-restore.sh','--local-run',run,'--apply','--yes',ok=False) self.assertNotIn('systemctl stop',self.events.read_text()) path.unlink() self.run_script('app-restore.sh','--local-run',run,'--apply','--yes',ok=False) self.assertNotIn('systemctl stop',self.events.read_text()) def test_partial_restore_keeps_services_stopped(self): run=self.backup() self.run_script('app-restore.sh','--local-run',run,'--apply','--yes',ok=False,TEST_FAIL_TOOL='mariadb') # Connection preflight fails BEFORE mutation: service state stays unchanged. self.assertEqual(json.loads(self.state.read_text()),self.initial) wrapper=self.bin/'mariadb' wrapper.write_text('#!/usr/bin/env bash\nif [[ " $* " != *" --execute=SELECT 1 "* ]]; then exit 44; fi\nexec /usr/bin/mariadb "$@"\n') result=self.run_script('app-restore.sh','--local-run',run,'--apply','--yes',ok=False) self.assertIn('Partial restore',result.stdout) self.assertTrue(json.loads(self.state.read_text())['maintenance']) self.assertEqual(json.loads(self.state.read_text())['services']['httpd'],'inactive') recovery=json.loads(next((self.base/'work').glob('recovery-*.json')).read_text()) self.assertTrue(recovery['service_guard_files']) self.assertTrue(all(Path(p).is_file() for p in recovery['service_guard_files'])) def test_check_does_not_claim_successful_backup(self): self.run_script('app-backup.sh','--check') self.assertFalse((self.base/'work/last-successful-backup.json').exists()) self.assertNotIn('systemctl stop',self.events.read_text()) def test_service_stop_and_start_failures_are_reported(self): self.run_script('app-backup.sh',ok=False,TEST_FAIL_SERVICE='stop:gitea') self.assertEqual(json.loads(self.state.read_text()),self.initial) self.run_script('app-backup.sh',ok=False,TEST_FAIL_SERVICE='start:gitea') self.assertTrue(list((self.base/'work').glob('recovery-*.json'))) self.assertNotEqual(json.loads((self.base/'work/last-backup.json').read_text())['exit_code'],0) self.assertFalse(list((self.base/'work/archives').glob('appbackup_*'))) def test_term_recovers_services(self): wrapper=self.bin/'rsync' # Signal only during the real copy, after services have stopped. The # new access probes run earlier and must not trigger this recovery test. wrapper.write_text('#!/usr/bin/env bash\nif [[ " $* " != *" --list-only "* && " $* " != *" --exclude=/* "* ]]; then\n echo waiting > "$TEST_EVENTS.waiting"\n sleep 2\nfi\nexec /usr/bin/rsync "$@"\n') proc=subprocess.Popen(['bash',str(REPO/'usr-local-sbin/app-backup.sh'),'--config',str(self.config)],env=self.env,text=True,stdout=subprocess.PIPE,stderr=subprocess.STDOUT) try: for _ in range(200): if Path(str(self.events)+'.waiting').exists(): break time.sleep(.05) else: self.fail('Backup did not reach rsync') proc.terminate() output=proc.communicate(timeout=20)[0] self.assertEqual(proc.returncode,143,output) self.assertEqual(json.loads(self.state.read_text()),self.initial) self.assertFalse(list((self.base/'work/archives').glob('appbackup_*'))) finally: if proc.poll() is None: proc.kill(); proc.wait() def test_selected_strict_restore_preserves_excluded_nextcloud(self): run=self.backup() (self.wp/'extra.txt').write_text('delete me') (self.data/'outside-selection.txt').write_text('keep me') with self.config.open('a') as f: f.write('RESTORE_STRICT_DELETE=true\n') self.run_script('app-restore.sh','--local-run',run,'--apply','--yes','--no-db','--components','wordpress') self.assertFalse((self.wp/'extra.txt').exists()) self.assertEqual((self.data/'outside-selection.txt').read_text(),'keep me') self.assertEqual(json.loads(self.state.read_text()),self.initial) def test_local_and_remote_retention_preserve_partial_and_v1(self): self.crypt_config() current=self.backup() archives=current.parent old=[] for day in range(1,4): name=f'appbackup_2020-01-0{day}_00-00-00_aaaaaaaa' path=archives/name shutil.copytree(current,path) manifest=json.loads((path/'manifest.json').read_text()) manifest['run_id']=name (path/'manifest.json').write_text(json.dumps(manifest)) (path/'COMPLETE').write_text('APP_BACKUP_V2 '+HELPER.digest(path/'manifest.json')+'\n') subprocess.run(['/usr/bin/rclone','--config',str(self.base/'rclone.conf'),'copy',str(path),'crypt:sets/'+name],check=True,capture_output=True) old.append(path) partial=archives/'appbackup_2019-01-01_00-00-00_aaaaaaaa' partial.mkdir() v1=archives/'appbackup_2019-01-01_00-00-00' v1.mkdir() with self.config.open('a') as f: f.write('ENABLE_REMOTE_RETENTION=true\nRETENTION_KEEP_LAST=2\nLOCAL_RETENTION_DAYS=1\nREMOTE_RETENTION_DAYS=1\n') self.run_script('app-backup.sh') self.assertTrue(current.is_dir()) self.assertTrue(partial.is_dir()) self.assertTrue(v1.is_dir()) self.assertTrue(all(not p.exists() for p in old)) remote=subprocess.check_output(['/usr/bin/rclone','--config',str(self.base/'rclone.conf'),'lsf','crypt:sets','--dirs-only'],text=True) self.assertNotIn('2020-',remote) self.assertIn(current.name,remote) def test_update_preserves_configuration_schedule_and_rolls_back(self): # Container-only test of the actual absolute installation paths. config=Path('/etc/app-backup/app-backup.conf') config.parent.mkdir(exist_ok=True) original=f'LOCKFILE="{self.base}/install.lock"\n# existing custom configuration\n' config.write_text(original) config.chmod(0o600) old_script=Path('/usr/local/sbin/app-backup.sh') old_script.parent.mkdir(parents=True,exist_ok=True) old_script.write_text('#!/bin/bash\necho previous-version\n') old_script.chmod(0o750) timer=Path('/etc/systemd/system/app-backup.timer') timer.parent.mkdir(parents=True,exist_ok=True) timer_text='[Timer]\nOnCalendar=*-*-* 04:10:00\n' timer.write_text(timer_text) state=json.loads(self.state.read_text()) state['services']['app-backup.timer']='active' self.state.write_text(json.dumps(state)) proc=subprocess.run(['bash',str(REPO/'install.sh')],env=self.env,capture_output=True,text=True) self.assertEqual(proc.returncode,0,proc.stdout+proc.stderr) self.assertEqual(config.read_text(),original) self.assertEqual(timer.read_text(),timer_text) self.assertTrue(Path(str(config)+'.dist').exists()) self.assertEqual(old_script.read_text(),(REPO/'usr-local-sbin/app-backup.sh').read_text()) self.assertEqual(json.loads(self.state.read_text())['services']['app-backup.timer'],'active') snapshot=Path(proc.stdout.split('Rollback package: ')[1].splitlines()[0]) rollback=subprocess.run(['bash',str(snapshot/'rollback.sh')],env=self.env,capture_output=True,text=True) self.assertEqual(rollback.returncode,0,rollback.stdout+rollback.stderr) self.assertIn('previous-version',old_script.read_text()) self.assertEqual(config.read_text(),original) self.assertEqual(json.loads(self.state.read_text())['services']['app-backup.timer'],'inactive') def test_legacy_extraction_is_explicit_and_isolated(self): folder=self.base/'legacy' folder.mkdir() archive=folder/'appbackup_2026-02-11_02-30-00_wordpress.tar.gz' with tarfile.open(archive,'w:gz') as tar: tar.add(self.wp,arcname='files/wordpress') self.run_script('app-restore.sh','--local-run',folder,'--extract-only',ok=False) self.run_script('app-restore.sh','--local-run',folder,'--legacy-unverified','--extract-only') self.assertTrue(list((self.base/'work/restore').glob('*/extracted/*.tar.gz/files/wordpress/payload.txt'))) self.run_script('app-restore.sh','--local-run',folder,'--legacy-unverified','--apply','--yes',ok=False) self.assertNotIn('systemctl stop',self.events.read_text()) def test_workspace_overlap_does_not_change_source_permissions(self): self.wp.chmod(0o755) with self.config.open('a') as f: f.write(f'WORKDIR="{self.wp}"\n') self.run_script('app-backup.sh','--check',ok=False) self.assertEqual(self.wp.stat().st_mode & 0o777,0o755) def build_package(self): output=self.base/'package' subprocess.run(['python3',str(REPO/'tools/build-update-package.py'),'--version','integration','--output',str(output)],check=True,capture_output=True) return output/'app-backup-update-integration.run' def configure_upgrade(self): self.crypt_config() canonical=Path('/etc/app-backup/app-backup.conf') canonical.parent.mkdir(exist_ok=True) shutil.copyfile(self.config,canonical) canonical.chmod(0o600) # Old installations commonly left credential files group-readable. self.cnf.chmod(0o640) state=json.loads(self.state.read_text()) state['services']['app-backup.timer']='active' self.state.write_text(json.dumps(state)) return canonical def test_automatic_package_upgrade_and_combined_rollback(self): self.enable_report() canonical=self.configure_upgrade() original=canonical.read_bytes() package=self.build_package() result=subprocess.run(['bash',str(package),'--foreground'],env=self.env,capture_output=True,text=True,timeout=120) self.assertEqual(result.returncode,0,result.stdout+result.stderr) state=Path(result.stdout.split('Update-Sicherung und Protokoll: ')[1].splitlines()[0]) self.assertIn('SUCCESS',(state/'result.txt').read_text()) message,body=self.read_report() self.assertEqual(str(message['From']),'app-backup@example.invalid') self.assertIn('OK (Inhalt und Abschlussmarker',body) self.assertEqual(Path('/usr/local/lib/app-backup/report.py').read_bytes(),(REPO/'lib/report.py').read_bytes()) self.assertTrue((state/'verified-backup-run').is_file()) self.assertIn('ENABLE_REMOTE_RETENTION=false',canonical.read_text()) self.assertEqual(self.cnf.stat().st_mode & 0o777,0o600) self.assertEqual(json.loads(self.state.read_text())['services']['app-backup.timer'],'active') rollback=subprocess.run(['bash',str(state/'rollback-update.sh')],env=self.env,capture_output=True,text=True,timeout=30) self.assertEqual(rollback.returncode,0,rollback.stdout+rollback.stderr) self.assertEqual(canonical.read_bytes(),original) self.assertEqual(json.loads(self.state.read_text())['services']['app-backup.timer'],'inactive') def test_package_preflight_failure_preserves_live_configuration_and_timer(self): canonical=self.configure_upgrade() original=canonical.read_bytes() package=self.build_package() result=subprocess.run(['bash',str(package),'--foreground'],env=dict(self.env,TEST_FAIL_TOOL='mariadb'),capture_output=True,text=True,timeout=60) self.assertNotEqual(result.returncode,0,result.stdout+result.stderr) self.assertEqual(canonical.read_bytes(),original) self.assertEqual(self.cnf.stat().st_mode & 0o777,0o640) self.assertEqual(json.loads(self.state.read_text())['services']['app-backup.timer'],'active') self.assertNotIn('systemctl stop',self.events.read_text()) def restricted_backup_account(self): self.sql("CREATE USER 'app_backup_limited'@'localhost'; " "GRANT SELECT, SHOW VIEW, EVENT, TRIGGER ON backup_test.* TO 'app_backup_limited'@'localhost'; " "GRANT SELECT ON mysql.proc TO 'app_backup_limited'@'localhost';") self.addCleanup(self.sql, "DROP USER 'app_backup_limited'@'localhost'") self.cnf.write_text(f'[client]\nuser=app_backup_limited\nsocket={self.socket}\n') def test_missing_reload_blocks_upgrade_then_targeted_grant_allows_backup(self): canonical=self.configure_upgrade() self.restricted_backup_account() original=canonical.read_bytes() package=self.build_package() result=subprocess.run(['bash',str(package),'--foreground'],env=self.env,capture_output=True,text=True,timeout=60) self.assertNotEqual(result.returncode,0,result.stdout+result.stderr) self.assertIn('global RELOAD privilege missing',result.stdout+result.stderr) self.assertEqual(canonical.read_bytes(),original) self.assertEqual(self.cnf.stat().st_mode & 0o777,0o640) self.assertNotIn('systemctl stop',self.events.read_text()) self.assertEqual(json.loads(self.state.read_text())['services']['app-backup.timer'],'active') self.sql("GRANT RELOAD ON *.* TO 'app_backup_limited'@'localhost'") result=subprocess.run(['bash',str(package),'--foreground'],env=self.env,capture_output=True,text=True,timeout=120) self.assertEqual(result.returncode,0,result.stdout+result.stderr) state=Path(result.stdout.split('Update-Sicherung und Protokoll: ')[1].splitlines()[0]) self.assertTrue((state/'result.txt').read_text().startswith('SUCCESS')) def test_reload_from_enabled_nested_role_is_accepted_but_inactive_role_is_not(self): self.restricted_backup_account() self.sql("CREATE ROLE backup_reload_child; CREATE ROLE backup_reload_parent; " "GRANT RELOAD ON *.* TO backup_reload_child; " "GRANT backup_reload_child TO backup_reload_parent; " "GRANT backup_reload_parent TO 'app_backup_limited'@'localhost';") self.addCleanup(self.sql, 'DROP ROLE backup_reload_parent; DROP ROLE backup_reload_child') failed=self.run_script('app-backup.sh','--check',ok=False) self.assertIn('global RELOAD privilege missing',failed.stdout) self.sql("SET DEFAULT ROLE backup_reload_parent FOR 'app_backup_limited'@'localhost'") passed=self.run_script('app-backup.sh','--check') self.assertIn('confirmed through an enabled role',passed.stdout) self.assertNotIn('systemctl stop',self.events.read_text()) self.backup() def test_missing_routine_metadata_access_blocks_before_maintenance(self): self.restricted_backup_account() self.sql("GRANT RELOAD ON *.* TO 'app_backup_limited'@'localhost'; " "REVOKE SELECT ON mysql.proc FROM 'app_backup_limited'@'localhost';") failed=self.run_script('app-backup.sh','--check',ok=False) self.assertIn('SELECT on mysql.proc required',failed.stdout) self.assertNotIn('systemctl stop',self.events.read_text()) self.assertEqual(json.loads(self.state.read_text()),self.initial) self.sql("GRANT SELECT ON mysql.proc TO 'app_backup_limited'@'localhost'; " "CREATE PROCEDURE backup_test.backup_probe() SELECT 1;") run=self.backup() with subprocess.Popen(['zstd','-dc',str(run/'db-wordpress.tar.zst')],stdout=subprocess.PIPE) as proc: with tarfile.open(fileobj=proc.stdout,mode='r|') as archive: routine_dump=archive.extractfile(archive.next()).read() self.assertIn(b'PROCEDURE `backup_probe`',routine_dump) def test_missing_mail_store_requires_explicit_resolution(self): canonical=self.configure_upgrade() with canonical.open('a') as f: f.write(f'MAIL_DIR="{self.src}/missing-mail"\n') original=canonical.read_bytes() package=self.build_package() result=subprocess.run(['bash',str(package),'--foreground'],env=self.env,capture_output=True,text=True,timeout=60) self.assertNotEqual(result.returncode,0) self.assertIn('--mail-dir DIR',result.stdout+result.stderr) self.assertEqual(canonical.read_bytes(),original) self.assertFalse((self.src/'missing-mail').exists()) self.assertEqual(json.loads(self.state.read_text())['services']['app-backup.timer'],'active') self.assertNotIn('systemctl stop',self.events.read_text()) def test_package_mail_directory_override(self): canonical=self.configure_upgrade() with canonical.open('a') as f: f.write(f'MAIL_DIR="{self.src}/missing-mail"\n') package=self.build_package() result=subprocess.run(['bash',str(package),'--foreground','--mail-dir',str(self.src/'mail')],env=self.env,capture_output=True,text=True,timeout=120) self.assertEqual(result.returncode,0,result.stdout+result.stderr) self.assertIn('ENABLE_MAIL_STORE=true',canonical.read_text()) run=next((self.base/'work/archives').glob('appbackup_*')) self.assertTrue((run/'mail.tar.zst').is_file()) self.assertTrue((run/'UPLOADED').is_file()) def test_package_homes_only_keeps_mail_configs_and_service_stops(self): canonical=self.configure_upgrade() home=self.src/'johannes' for folder in ['cur','new','tmp']: (home/'Mail'/folder).mkdir(parents=True) mailbox=home/'Mail/cur/test-message:2,S' mailbox.write_bytes(b'Subject: retained mail\n\nMailbox content\n') mailbox.chmod(0o600) original=mailbox.read_bytes() with canonical.open('a') as f: f.write(f'MAIL_DIR="{self.src}/missing-mail"\nENABLE_MAIL_HOME_DIRS=true\nMAIL_HOME_DIRS=("{home}")\n') package=self.build_package() result=subprocess.run(['bash',str(package),'--foreground','--mail-homes-only'],env=self.env,capture_output=True,text=True,timeout=120) self.assertEqual(result.returncode,0,result.stdout+result.stderr) self.assertIn('ENABLE_MAIL_STORE=false',canonical.read_text()) run=next((self.base/'work/archives').glob('appbackup_*')) for name in ['mail-home-0','postfix','dovecot']: self.assertTrue((run/f'{name}.tar.zst').is_file()) self.assertFalse((run/'mail.tar.zst').exists()) self.assertTrue((run/'UPLOADED').is_file()) for unit in ['postfix','dovecot']: self.assertIn(f'systemctl stop {unit}',self.events.read_text()) self.assertEqual(json.loads(self.state.read_text())['services'][unit],'active') mailbox.write_text('changed after backup') restored=subprocess.run(['bash',str(REPO/'usr-local-sbin/app-restore.sh'),'--config',str(canonical), '--local-run',str(run),'--components','mail-home-0','--apply','--yes'], env=self.env,capture_output=True,text=True,timeout=60) self.assertEqual(restored.returncode,0,restored.stdout+restored.stderr) self.assertEqual(mailbox.read_bytes(),original) self.assertEqual(mailbox.stat().st_mode & 0o777,0o600) def test_homes_only_rejects_disabled_empty_and_missing_homes(self): canonical=self.configure_upgrade() original=canonical.read_text() package=self.build_package() for settings in ['ENABLE_MAIL_HOME_DIRS=false', 'ENABLE_MAIL_HOME_DIRS=true\nMAIL_HOME_DIRS=()', f'ENABLE_MAIL_HOME_DIRS=true\nMAIL_HOME_DIRS=("{self.src}/missing-home")']: with self.subTest(settings=settings): canonical.write_text(original+'\n'+settings+'\n') before=canonical.read_bytes() result=subprocess.run(['bash',str(package),'--foreground','--mail-homes-only'],env=self.env,capture_output=True,text=True,timeout=60) self.assertNotEqual(result.returncode,0,result.stdout+result.stderr) self.assertEqual(canonical.read_bytes(),before) self.assertEqual(json.loads(self.state.read_text())['services']['app-backup.timer'],'active') self.assertNotIn('systemctl stop',self.events.read_text()) def test_package_failed_backup_keeps_timer_stopped_and_rollback_available(self): canonical=self.configure_upgrade() original=canonical.read_bytes() package=self.build_package() result=subprocess.run(['bash',str(package),'--foreground'],env=dict(self.env,TEST_FAIL_TOOL='mariadb-dump'),capture_output=True,text=True,timeout=120) self.assertNotEqual(result.returncode,0,result.stdout+result.stderr) state=Path(result.stdout.split('Update-Sicherung und Protokoll: ')[1].splitlines()[0]) self.assertIn('FAILED',(state/'result.txt').read_text()) self.assertTrue((state/'rollback-update.sh').is_file()) self.assertEqual(json.loads(self.state.read_text())['services']['app-backup.timer'],'inactive') self.assertFalse((state/'verified-backup-run').exists()) rollback=subprocess.run(['bash',str(state/'rollback-update.sh')],env=self.env,capture_output=True,text=True,timeout=30) self.assertEqual(rollback.returncode,0,rollback.stdout+rollback.stderr) self.assertEqual(canonical.read_bytes(),original) def test_package_corruption_is_rejected_before_preparation(self): package=self.build_package() original=package.read_bytes() package.write_bytes(original[:-1]+bytes([original[-1]^1])) result=subprocess.run(['bash',str(package),'--foreground'],env=self.env,capture_output=True,text=True,timeout=20) self.assertNotEqual(result.returncode,0) self.assertIn('Paket-Pruefsumme falsch',result.stderr) self.assertNotIn('systemctl',self.events.read_text()) def test_inherited_lock_flag_cannot_bypass_another_process(self): import fcntl with (self.base/'lock').open('w') as lock: fcntl.flock(lock,fcntl.LOCK_EX | fcntl.LOCK_NB) result=self.run_script('app-backup.sh','--check',ok=False,APP_BACKUP_INHERIT_LOCK='1') self.assertIn('Inherited lock descriptor',result.stdout) def systemd_launcher(self): """Model start-job waiting, detached FDs, and ambiguous submit failures.""" launcher=self.bin/'systemd-run' launcher.write_text('''#!/usr/bin/env python3 import os,subprocess,sys from pathlib import Path args=sys.argv[1:] assert '--unit=app-backup-upgrade' in args assert '--property=Type=oneshot' in args assert '--property=KillMode=control-group' in args assert '--property=SELinuxContext=system_u:system_r:unconfined_service_t:s0' in args command=args[args.index('/bin/bash'):] state=Path(command[-1]) if os.environ.get('TEST_SYSTEMD_SUBMIT') == 'reject': sys.exit('systemd refused submission') with open(state/'test-worker.log','w') as log: worker=subprocess.Popen(command,stdin=subprocess.DEVNULL,stdout=log,stderr=subprocess.STDOUT,env=os.environ,close_fds=True,start_new_session=True) (state/'test-worker.pid').write_text(str(worker.pid)) if os.environ.get('TEST_SYSTEMD_SUBMIT') == 'accepted-error': sys.exit('simulated reply failure after systemd accepted the job') # Type=oneshot remains starting until ExecStart exits. Real systemd-run waits # for that start job unless --no-block is used; our old stub missed this. if '--no-block' not in args: try: sys.exit(worker.wait(timeout=3)) except subprocess.TimeoutExpired: worker.terminate() worker.wait(timeout=5) sys.exit('oneshot start job still waiting: launcher/worker lock deadlock') ''') launcher.chmod(0o755) def assert_group_term_recovers_services(self, tool): self.configure_upgrade() self.systemd_launcher() wrapper=self.bin/tool if tool=='rsync': match='[[ " $* " != *" --list-only "* && " $* " != *" --exclude=/* "* ]]' else: match='[[ " $* " == *" copy "* ]]' wrapper.write_text('#!/usr/bin/env bash\nif '+match+'; then\n echo waiting > "$TEST_EVENTS.waiting"\n sleep 120\nfi\nexec /usr/bin/'+tool+' "$@"\n') package=self.build_package() result=subprocess.run(['bash',str(package)],env=self.env,capture_output=True,text=True,timeout=60) self.assertEqual(result.returncode,0,result.stdout+result.stderr) state=Path(result.stdout.split('Update-Sicherung und Protokoll: ')[1].splitlines()[0]) pid=int((state/'test-worker.pid').read_text()) try: for _ in range(400): if Path(str(self.events)+'.waiting').exists(): break time.sleep(.1) else: self.fail('Worker did not reach '+tool+': '+(state/'test-worker.log').read_text()) os.killpg(pid,signal.SIGTERM) # emulate systemd KillMode=control-group for _ in range(200): status=(state/'result.txt').read_text() if status.startswith('FAILED'): break time.sleep(.1) self.assertTrue(status.startswith('FAILED'),status+(state/'test-worker.log').read_text()) actual=json.loads(self.state.read_text()) self.assertFalse(actual['maintenance']) for unit in self.initial['services']: self.assertEqual(actual['services'][unit],'active',unit) self.assertEqual(actual['services']['app-backup.timer'],'inactive') if tool=='rsync': self.assertFalse(list((self.base/'work/archives').glob('appbackup_*'))) else: self.assertFalse(list((self.base/'work/archives').glob('appbackup_*/UPLOADED'))) self.assertFalse(list(self.runtime.glob('app-backup-guard-*'))) self.assertIn('Service started: httpd',(state/'update.log').read_text()) finally: try: os.killpg(pid,signal.SIGKILL) except ProcessLookupError: pass def test_group_term_interrupts_worker_copy_and_recovers_services(self): self.assert_group_term_recovers_services('rsync') def test_group_term_interrupts_worker_upload_and_keeps_timer_stopped(self): self.assert_group_term_recovers_services('rclone') def test_detached_package_handoff_finishes_after_starter_exits(self): self.configure_upgrade() self.systemd_launcher() package=self.build_package() result=subprocess.run(['bash',str(package)],env=self.env,capture_output=True,text=True,timeout=60) self.assertEqual(result.returncode,0,result.stdout+result.stderr) state=Path(result.stdout.split('Update-Sicherung und Protokoll: ')[1].splitlines()[0]) for _ in range(120): status=(state/'result.txt').read_text() if (state/'result.txt').exists() else '' if status.startswith(('SUCCESS','FAILED')): break time.sleep(.5) self.assertTrue(status.startswith('SUCCESS'),status+(state/'test-worker.log').read_text()) self.assertIn('unabhaengig von SSH',result.stdout) def test_systemd_submission_failure_preserves_installation_and_reports_result(self): canonical=self.configure_upgrade() original=canonical.read_bytes() self.systemd_launcher() package=self.build_package() result=subprocess.run(['bash',str(package)],env=dict(self.env,TEST_SYSTEMD_SUBMIT='reject'),capture_output=True,text=True,timeout=60) self.assertNotEqual(result.returncode,0) state=Path(result.stdout.split('Update-Sicherung und Protokoll: ')[1].splitlines()[0]) self.assertIn('FAILED: systemd submission failed',(state/'result.txt').read_text()) self.assertIn(str(state/'result.txt'),result.stdout) self.assertEqual(canonical.read_bytes(),original) self.assertEqual(json.loads(self.state.read_text())['services']['app-backup.timer'],'active') self.assertNotIn('systemctl stop',self.events.read_text()) def test_failed_submission_revokes_even_an_already_queued_worker(self): canonical=self.configure_upgrade() original=canonical.read_bytes() self.systemd_launcher() package=self.build_package() result=subprocess.run(['bash',str(package)],env=dict(self.env,TEST_SYSTEMD_SUBMIT='accepted-error'),capture_output=True,text=True,timeout=60) self.assertNotEqual(result.returncode,0) state=Path(result.stdout.split('Update-Sicherung und Protokoll: ')[1].splitlines()[0]) for _ in range(100): log=(state/'test-worker.log').read_text() if 'Installation not started' in log: break time.sleep(.1) self.assertIn('submission was cancelled',log) self.assertIn('Installation not started',log) self.assertTrue((state/'result.txt').read_text().startswith('FAILED')) self.assertEqual(canonical.read_bytes(),original) self.assertEqual(json.loads(self.state.read_text())['services']['app-backup.timer'],'active') self.assertNotIn('systemctl stop',self.events.read_text()) self.assertFalse((state/'package-rollback').exists()) def test_worker_lock_timeout_is_persisted_before_installation(self): import fcntl canonical=self.configure_upgrade() original=canonical.read_bytes() self.systemd_launcher() package=self.build_package() # Obtain a complete prepared state without starting an actual worker. result=subprocess.run(['bash',str(package)],env=dict(self.env,TEST_SYSTEMD_SUBMIT='reject'),capture_output=True,text=True,timeout=60) self.assertNotEqual(result.returncode,0) state=Path(result.stdout.split('Update-Sicherung und Protokoll: ')[1].splitlines()[0]) (state/'result.txt').write_text('QUEUED\n') wrapper=self.bin/'flock' # Keep real flock contention; shorten only the 60s timeout in this test. wrapper.write_text('#!/usr/bin/env bash\nif [[ $1 == -w && $2 == 60 ]]; then shift 2; exec /usr/bin/flock -w 0 "$@"; fi\nexec /usr/bin/flock "$@"\n') wrapper.chmod(0o755) with open('/run/app-backup-upgrade.lock','a') as lock: fcntl.flock(lock,fcntl.LOCK_EX | fcntl.LOCK_NB) failed=subprocess.run(['bash',str(REPO/'update.sh'),'--worker',str(state)],env=self.env,capture_output=True,text=True,timeout=10) self.assertNotEqual(failed.returncode,0) self.assertIn('Timed out waiting for update coordinator lock',(state/'update.log').read_text()) self.assertIn('Installation not started',(state/'result.txt').read_text()) self.assertTrue((state/'result.txt').read_text().startswith('FAILED')) self.assertEqual(canonical.read_bytes(),original) self.assertEqual(json.loads(self.state.read_text())['services']['app-backup.timer'],'active') self.assertNotIn('systemctl stop',self.events.read_text()) def test_cron_inventory_requires_explicit_policy_and_sorts_timer_first(self): canonical=self.configure_upgrade() cron=Path('/etc/cron.d/app-backup-test-nextcloud') cron.parent.mkdir(exist_ok=True) cron.write_text(f'*/5 * * * * root /usr/bin/php -f {self.nc}/cron.php\n') self.addCleanup(cron.unlink) state=json.loads(self.state.read_text()) state['services']['crond.service']='active' state['services']['nc-cron.timer']='active' self.state.write_text(json.dumps(state)) package=self.build_package() result=subprocess.run(['bash',str(package),'--foreground'],env=self.env,capture_output=True,text=True,timeout=60) self.assertNotEqual(result.returncode,0,result.stdout+result.stderr) self.assertIn('noninteractive',result.stdout+result.stderr) self.assertEqual(json.loads(self.state.read_text())['services']['app-backup.timer'],'active') result=subprocess.run(['bash',str(package),'--foreground','--cron-services','nc-cron,crond.service,nc-cron.timer'],env=self.env,capture_output=True,text=True,timeout=120) self.assertEqual(result.returncode,0,result.stdout+result.stderr) text=canonical.read_text() line=[line for line in text.splitlines() if line.startswith('QUIESCE_SERVICES=')][-1] self.assertLess(line.index('nc-cron.timer'),line.index('crond.service')) self.assertEqual(json.loads(self.state.read_text())['services']['crond.service'],'active') def crypt_config(self): (self.base/'encrypted').mkdir() password=subprocess.check_output(['/usr/bin/rclone','obscure','integration-test-only'],text=True).strip() path=self.base/'rclone.conf' path.write_text(f'[crypt]\ntype = crypt\nremote = {self.base}/encrypted\npassword = {password}\n') path.chmod(0o600) with self.config.open('a') as f: f.write(f'ENABLE_UPLOAD=true\nRCLONE_CONFIG="{path}"\nRCLONE_REMOTE_BASE="crypt:sets"\nRCLONE_RETRIES=1\nRCLONE_LOW_LEVEL_RETRIES=1\n') return path def enable_report(self, behaviour='capture'): # This fake MTA captures stdin locally. Tests never send any real mail. path=self.bin/'capture-sendmail' path.write_text(f'''#!/usr/bin/env python3 import pathlib, sys, time if {behaviour!r} == 'timeout': time.sleep(10) pathlib.Path({str(self.base/'mail.eml')!r}).write_bytes(sys.stdin.buffer.read()) pathlib.Path({str(self.base/'mail-args')!r}).write_text(' '.join(sys.argv[1:])) sys.exit(42 if {behaviour!r} == 'fail' else 0) ''') path.chmod(0o755) with self.config.open('a') as f: f.write(f'''ENABLE_MAIL_REPORT=true MAIL_TO='Johannes ' MAIL_FROM='app-backup@example.invalid' MAIL_SUBJECT_PREFIX='[Backup-Test]' MAIL_INCLUDE_LOG_TAIL_LINES=200 MAIL_SENDMAIL_BIN='{path}' MAIL_SEND_TIMEOUT=1 REPORT_SECRET='config-content-must-not-enter-the-mail' ''') def read_report(self): raw=(self.base/'mail.eml').read_bytes() message=BytesParser(policy=policy.default).parsebytes(raw) return message,message.get_content() def test_report_success_has_sizes_after_cleanup_and_preserves_headers(self): self.enable_report() run=self.backup() message,body=self.read_report() self.assertIn('[Backup-Test] SUCCESS',str(message['Subject'])) self.assertEqual(str(message['From']),'app-backup@example.invalid') self.assertIn('backup@example.invalid',str(message['To'])) for value in ['Status: SUCCESS','Exit code: 0','Dauer:', 'db staged:', 'wordpress staged:','Staging total:','wordpress.tar.zst:', 'Deaktiviert (nur lokal)',f'SUCCESS: {run.name}']: self.assertIn(value,body) self.assertNotIn('config-content-must-not-enter-the-mail',body) self.assertFalse(list((self.base/'work/staging').iterdir())) self.assertEqual((self.base/'mail-args').read_text(),'-oi -t') saved=next((self.base/'logs').glob('report_*.eml')) self.assertEqual(saved.read_bytes(),(self.base/'mail.eml').read_bytes()) self.assertEqual(saved.stat().st_mode & 0o777,0o600) def test_report_failure_names_phase_and_has_final_error(self): self.enable_report() result=self.run_script('app-backup.sh',ok=False,TEST_FAIL_TOOL='mariadb-dump') message,body=self.read_report() self.assertEqual(result.returncode,42) self.assertIn('[Backup-Test] FAIL',str(message['Subject'])) self.assertIn('Exit code: 42',body) self.assertIn('Letzte Phase: DB-Dump:',body) self.assertIn('FAILED (exit 42)',body) self.assertEqual(json.loads(self.state.read_text()),self.initial) def test_report_upload_readback_failure_is_not_reported_as_ok(self): self.enable_report() self.crypt_config() self.run_script('app-backup.sh',ok=False,TEST_FAIL_CHECK='true') _,body=self.read_report() self.assertIn('Status: FAIL',body) self.assertIn('Upload Status: Nicht abgeschlossen: Rückleseprüfung',body) self.assertNotIn('Upload Status: OK',body) def test_report_mta_failure_keeps_backup_result_and_local_message(self): self.enable_report('fail') result=self.run_script('app-backup.sh') self.assertIn('WARNING: mail report failed',result.stdout) self.assertEqual(json.loads((self.base/'work/last-backup.json').read_text())['exit_code'],0) self.assertTrue(list((self.base/'logs').glob('report_*.eml'))) failed=self.run_script('app-backup.sh',ok=False,TEST_FAIL_TOOL='mariadb-dump') self.assertEqual(failed.returncode,42) self.assertIn('WARNING: mail report failed',failed.stdout) def test_report_timeout_is_bounded_and_header_injection_is_rejected(self): self.enable_report('timeout') result=self.run_script('app-backup.sh') self.assertIn('mail report failed (TimeoutExpired)',result.stdout) self.assertFalse((self.base/'mail.eml').exists()) self.enable_report() with self.config.open('a') as f: f.write("MAIL_TO=$'recipient\\nBcc: attacker@example.invalid'\n") result=self.run_script('app-backup.sh') self.assertIn('mail report failed (ValueError)',result.stdout) self.assertFalse((self.base/'mail.eml').exists()) def test_report_check_and_restore_do_not_send_backup_messages(self): self.enable_report() self.run_script('app-backup.sh','--check') self.assertFalse((self.base/'mail.eml').exists()) run=self.backup() (self.base/'mail.eml').unlink() self.run_script('app-restore.sh','--local-run',run,'--verify-only') self.assertFalse((self.base/'mail.eml').exists()) def test_real_crypt_upload_readback_remote_restore(self): self.crypt_config() # Exercise periodic stderr statistics during a real throttled crypt # transfer; they must not contaminate captured COMPLETE-marker stdout. (self.wp/'statistics-test.bin').write_bytes(os.urandom(512*1024)) with self.config.open('a') as f: f.write('RCLONE_STATS_INTERVAL="100ms"\nRCLONE_BWLIMIT="256k"\n') run=self.backup() self.assertTrue((run/'UPLOADED').is_file()) self.assertFalse(list((self.base/'encrypted').rglob('wordpress.tar.zst'))) log=next((self.base/'logs').glob('backup_*.log')).read_text() self.assertIn('NOTICE:',log) self.assertIn('Remote upload started',log) self.assertIn('Remote readback passed',log) self.run_script('app-restore.sh','--remote-run',run.name,'--verify-only') def test_failed_readback_does_not_publish_remote_marker(self): self.crypt_config() self.run_script('app-backup.sh',ok=False,TEST_FAIL_CHECK='true') run=next((self.base/'work/archives').glob('appbackup_*')) self.assertFalse((run/'UPLOADED').exists()) proc=subprocess.run(['/usr/bin/rclone','--config',str(self.base/'rclone.conf'),'cat',f'crypt:sets/{run.name}/COMPLETE'],capture_output=True) self.assertNotEqual(proc.returncode,0) def remote_failure(self, mode, **env): # Keep real crypt transfers. Inject errors at the command boundary, not # a pretend OneDrive server: provider retry behaviour is checked separately. wrapper = self.bin/'rclone' wrapper.write_text('''#!/usr/bin/env python3 import os, subprocess, sys args=sys.argv[1:] mode=os.environ.get('TEST_REMOTE_FAILURE','') def rc(arguments, **kw): return subprocess.run(['/usr/bin/rclone',*arguments], **kw) if 'copyto' in args and mode=='marker-write': sys.exit(44) if 'cat' in args and mode=='marker-read': print('incorrect completion marker') sys.exit(0) if 'cat' in args and mode=='marker-newline': data=rc(args,capture_output=True,check=True).stdout sys.stdout.buffer.write(data+b'\\n') sys.exit(0) result=rc(args) if result.returncode: sys.exit(result.returncode) if 'copy' in args: if mode=='upload': # Models an unsuccessful command even though the remote received bytes. sys.exit(44) if mode=='corrupt': config=args[args.index('--config')+1] remote=args[args.index('copy')+2]+'/wordpress.tar.zst' prefix=['--config',config] data=bytearray(rc([*prefix,'cat',remote],capture_output=True,check=True).stdout) data[len(data)//2]^=1 # Same-size plaintext corruption, re-encrypted through the real crypt backend. # The backup must read contents back, not accept a matching size alone. rc([*prefix,'rcat',remote],input=bytes(data),check=True) ''') wrapper.chmod(0o755) return self.run_script('app-backup.sh',ok=False,TEST_REMOTE_FAILURE=mode,**env) def assert_remote_failure_retained(self, result, marker_present=False): status=json.loads((self.base/'work/last-backup.json').read_text()) self.assertNotEqual(status['exit_code'],0) run=self.base/'work/archives'/status['run'] HELPER.load_set(run) # Local hashes/manifest must still be valid. self.assertFalse((run/'UPLOADED').exists()) self.assertEqual(json.loads(self.state.read_text()),self.initial) self.assertFalse(list(self.runtime.glob('app-backup-guard-*'))) self.assertNotIn('-- Local retention:',result.stdout) self.assertNotIn('-- Remote retention:',result.stdout) marker=subprocess.run(['/usr/bin/rclone','--config',str(self.base/'rclone.conf'), 'cat',f'crypt:sets/{run.name}/COMPLETE'],capture_output=True) self.assertEqual(marker.returncode==0,marker_present) return run def test_upload_failure_preserves_previous_success_and_retention_candidates(self): self.crypt_config() current=self.backup() previous=(self.base/'work/last-successful-backup.json').read_bytes() old=[] for day in range(1,4): name=f'appbackup_2020-01-0{day}_00-00-00_aaaaaaaa' path=current.parent/name shutil.copytree(current,path) manifest=json.loads((path/'manifest.json').read_text()) manifest['run_id']=name (path/'manifest.json').write_text(json.dumps(manifest)) (path/'COMPLETE').write_text('APP_BACKUP_V2 '+HELPER.digest(path/'manifest.json')+'\n') subprocess.run(['/usr/bin/rclone','--config',str(self.base/'rclone.conf'), 'copy',str(path),'crypt:sets/'+name],check=True,capture_output=True) old.append(path) with self.config.open('a') as f: f.write('ENABLE_REMOTE_RETENTION=true\nRETENTION_KEEP_LAST=2\nLOCAL_RETENTION_DAYS=1\nREMOTE_RETENTION_DAYS=1\n') result=self.remote_failure('upload') self.assert_remote_failure_retained(result) self.assertEqual(result.returncode,44) self.assertEqual((self.base/'work/last-successful-backup.json').read_bytes(),previous) remote=subprocess.check_output(['/usr/bin/rclone','--config',str(self.base/'rclone.conf'), 'lsf','crypt:sets','--dirs-only'],text=True) for path in old: self.assertTrue(path.is_dir()) self.assertIn(path.name,remote) def test_same_size_remote_corruption_blocks_completion(self): self.crypt_config() result=self.remote_failure('corrupt') self.assert_remote_failure_retained(result) self.assertFalse((self.base/'work/last-successful-backup.json').exists()) def test_environment_cannot_weaken_remote_readback(self): self.crypt_config() result=self.remote_failure('corrupt',RCLONE_SIZE_ONLY='true') self.assertIn('Unsupported rclone environment override: RCLONE_SIZE_ONLY',result.stdout) self.assertNotIn('systemctl stop',self.events.read_text()) def test_unsupported_rclone_environment_rejected_before_remote_access(self): self.crypt_config() for key,value in [('RCLONE_EXCLUDE','*'),('RCLONE_DRY_RUN','true'), ('RCLONE_IGNORE_ERRORS','true'),('RCLONE_NO_CHECK_CERTIFICATE','true')]: with self.subTest(key=key): self.events.write_text('') result=self.run_script('app-backup.sh','--check',ok=False,**{key:value}) self.assertIn('Unsupported rclone environment override: '+key,result.stdout) self.assertNotIn('rclone',self.events.read_text()) self.assertNotIn('systemctl stop',self.events.read_text()) # Supported settings remain usable when an administrator exports them. self.run_script('app-backup.sh','--check',RCLONE_RETRIES='2',RCLONE_STATS_INTERVAL='1m') def test_marker_write_failure_blocks_upload_success(self): self.crypt_config() result=self.remote_failure('marker-write') self.assert_remote_failure_retained(result) self.assertEqual(result.returncode,44) def test_marker_read_mismatch_blocks_upload_success(self): self.crypt_config() result=self.remote_failure('marker-read') # Archives passed readback and COMPLETE was written, but its final # confirmation failed. Local success and retention must remain blocked. self.assert_remote_failure_retained(result,marker_present=True) self.assertIn('Remote completion marker mismatch',result.stdout) def test_marker_extra_newline_blocks_upload_success(self): self.crypt_config() result=self.remote_failure('marker-newline') self.assert_remote_failure_retained(result,marker_present=True) self.assertIn('Remote completion marker mismatch',result.stdout) def test_unsafe_crypt_and_source_overlap_rejected(self): path=self.crypt_config() with path.open('a') as f: f.write('no_data_encryption = true\n') self.run_script('app-backup.sh','--check',ok=False) self.assertNotIn('systemctl stop',self.events.read_text()) with self.config.open('a') as f: f.write(f'ENABLE_UPLOAD=false\nWP_DIR="{self.base}"\n') self.run_script('app-backup.sh','--check',ok=False) def test_lock_excludes_concurrent_run(self): import fcntl with (self.base/'lock').open('w') as lock: fcntl.flock(lock,fcntl.LOCK_EX | fcntl.LOCK_NB) result=self.run_script('app-backup.sh','--check',ok=False) self.assertIn('Another backup/restore/update',result.stdout) def test_gzip_and_extract_only(self): with self.config.open('a') as f: f.write('COMPRESSOR=gzip\n') run=self.backup() self.run_script('app-restore.sh','--local-run',run,'--extract-only') self.assertTrue(list((self.base/'work/restore').glob('*/extracted/wordpress/files/wordpress/payload.txt'))) if __name__ == '__main__': unittest.main()