Fix SELinux backup execution context and probe source metadata before downtime

tests/test_backup.py

@@ -251,6 +251,67 @@ QUIESCE_SERVICES=(nc-cron)
                 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_maintenance_and_inactive_service_preserved(self):
         state=dict(services=dict(self.initial['services'],gitea='inactive'),maintenance=True)
         self.state.write_text(json.dumps(state))
@@ -296,10 +357,12 @@ QUIESCE_SERVICES=(nc-cron)
 
     def test_term_recovers_services(self):
         wrapper=self.bin/'rsync'
-        wrapper.write_text('#!/usr/bin/env bash\necho waiting > "$TEST_EVENTS.waiting"\nsleep 2\nexec /usr/bin/rsync "$@"\n')
+        # 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(100):
+            for _ in range(200):
                 if Path(str(self.events)+'.waiting').exists(): break
                 time.sleep(.05)
             else: self.fail('Backup did not reach rsync')
@@ -612,6 +675,7 @@ from pathlib import Path
 args=sys.argv[1:]
 assert '--unit=app-backup-upgrade' in args
 assert '--property=Type=oneshot' 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':