diff --git a/deploy/deploy.sh b/deploy/deploy.sh index 32ec12ba..49012cc0 100644 --- a/deploy/deploy.sh +++ b/deploy/deploy.sh @@ -407,8 +407,11 @@ verify_backup_file_shape() { report_backup() { local target="$1" size if ! chmod 600 "$target" 2>/dev/null; then - docker run --rm -v "$BACKUP_DIR":/backup -e BACKUP_FILE="$(basename "$target")" \ - alpine:3 sh -c 'chmod 600 "/backup/$BACKUP_FILE"' + if ! docker run --rm -v "$BACKUP_DIR":/backup -e BACKUP_FILE="$(basename "$target")" \ + alpine:3 sh -c 'chmod 600 "/backup/$BACKUP_FILE"'; then + echo "Could not restrict backup permissions for ${target}. Aborting deploy." + return 1 + fi fi size="$(du -h "$target" | cut -f1)" echo "Backup verified: ${target} (${size})" diff --git a/scripts/test-deploy-backup.py b/scripts/test-deploy-backup.py index 13835b66..9e228042 100644 --- a/scripts/test-deploy-backup.py +++ b/scripts/test-deploy-backup.py @@ -294,5 +294,68 @@ class BackupRetentionTests(unittest.TestCase): self.assertIn('DEPLOY WOULD CONTINUE', result.stdout) +class BackupPermissionTests(unittest.TestCase): + def permission_gate(self, kind, host_status, fallback_status): + source = (ROOT / 'deploy/deploy.sh').read_text(encoding='utf-8') + functions = [] + for name in ['report_backup', 'verify_sql_backup', 'verify_volume_backup']: + start = source.index(name + '() {') + functions.append(source[start:source.index('\n}\n', start) + 3]) + caller = 'backup_database' if kind == 'database' else 'backup_recovery_volumes' + start = source.index('if ! ' + caller + '; then') + gate = source[start:source.index('\nfi\n', start) + 4] + with tempfile.TemporaryDirectory(prefix='backup-permission-') as directory: + target = Path(directory) / 'synthetic-backup.gz' + payload = b'synthetic permission fixture' + target.write_bytes(payload) + script = '\n'.join([ + 'set -euo pipefail', + 'BACKUP_DIR=' + shlex.quote(Path(directory).as_posix()), + 'TARGET=' + shlex.quote(target.as_posix()), + 'chmod() { echo "HOST PERMISSIONS"; return ' + str(host_status) + '; }', + 'docker() { if [[ "${@: -1}" == *chmod* ]]; then echo "FALLBACK PERMISSIONS"; return ' + str(fallback_status) + '; fi; return 0; }', + 'du() { printf "1K\\tsynthetic\\n"; }', + 'rm() { echo "UNEXPECTED REMOVAL"; return 99; }', + 'gzip() { printf "CREATE TABLE synthetic (id int);\\n-- Dump completed\\n"; }', + 'verify_backup_file_shape() { [ -s "$1" ]; }', + *functions, + 'backup_database() { verify_sql_backup "$TARGET"; }', + 'backup_recovery_volumes() { verify_volume_backup "$TARGET"; }', + gate, + 'echo "NEXT DEPLOY GATE WOULD START"', + ]) + result = subprocess.run([BASH, '--noprofile', '--norc'], input=script, text=True, + encoding='utf-8', capture_output=True, cwd=ROOT) + self.assertEqual(payload, target.read_bytes(), 'Permission failure must not remove the backup artifact') + self.assertNotIn('UNEXPECTED REMOVAL', result.stdout) + return result + + def test_host_permission_success_does_not_need_fallback(self): + for kind in ['database', 'recovery']: + with self.subTest(kind=kind): + result = self.permission_gate(kind, 0, 23) + self.assertEqual(0, result.returncode, result.stdout + result.stderr) + self.assertNotIn('FALLBACK PERMISSIONS', result.stdout) + self.assertIn('Backup verified', result.stdout) + self.assertIn('NEXT DEPLOY GATE WOULD START', result.stdout) + + def test_fallback_permission_success_preserves_verification(self): + for kind in ['database', 'recovery']: + with self.subTest(kind=kind): + result = self.permission_gate(kind, 1, 0) + self.assertEqual(0, result.returncode, result.stdout + result.stderr) + self.assertIn('FALLBACK PERMISSIONS', result.stdout) + self.assertIn('Backup verified', result.stdout) + self.assertIn('NEXT DEPLOY GATE WOULD START', result.stdout) + + def test_both_permission_failures_block_verification_and_next_gate(self): + for kind in ['database', 'recovery']: + with self.subTest(kind=kind): + result = self.permission_gate(kind, 1, 23) + self.assertNotEqual(0, result.returncode, result.stdout + result.stderr) + self.assertNotIn('Backup verified', result.stdout) + self.assertNotIn('NEXT DEPLOY GATE WOULD START', result.stdout) + + if __name__ == '__main__': unittest.main()