Check database dump privileges before stopping applications

tests/test_backup.py

@@ -22,6 +22,12 @@ 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)
@@ -440,6 +446,62 @@ QUIESCE_SERVICES=(nc-cron)
         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')