fix(deploy): stop when backup permission protection fails

This commit is contained in:
cesnimda
2026-11-11 18:17:21 +01:00
parent e1c8e84502
commit 1c8761b188
2 changed files with 68 additions and 2 deletions
+5 -2
View File
@@ -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})"
+63
View File
@@ -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()