Coverage for app/services/backup_verification.py: 100%

1"""Automated restore-verification for the daily backup (INFRA-06, part 3). 

2 

3After each successful scheduled backup, decrypt (if encrypted) the archive 

4that was just written, and check it is well-formed: every zip entry's CRC 

5checks out, ``openhangar.sql`` is present and looks like real pg_dump 

6output, and ``metadata.json`` (the backup's manifest — app version, Alembic 

7head, creation time) is present and parses. Failure fires a [SECURITY] 

8alert through the existing ntfy/email/webhook channels so operators find 

9out immediately, not the next time they actually need the backup. 

10 

11This only proves the archive is intact and decryptable — it does not spin 

12up a scratch database and replay the SQL. That heavier check stays a 

13documented manual quarterly procedure; see docs/backup_restore.md. 

14""" 

15 

16import logging 

17import os 

18 

19from init import _env_or_file # pyright: ignore[reportMissingImports] 

20from models import BackupRecord # pyright: ignore[reportMissingImports] 

21 

22from services.backup_format import ( # pyright: ignore[reportMissingImports] 

23 BackupArchiveError, 

24 parse_backup_archive, 

25) 

26 

27log = logging.getLogger("openhangar.backup") 

28 

29 

30class BackupVerificationError(Exception): 

31 """Raised when a backup archive fails integrity verification.""" 

32 

33 

34def _decrypt_if_needed(payload: bytes, filename: str) -> bytes: 

35 if not filename.endswith(".enc"): 

36 return payload 

37 key_raw = _env_or_file("BACKUP_ENCRYPTION_KEY") 

38 if not key_raw: 

39 raise BackupVerificationError( 

40 "archive is encrypted but OPENHANGAR_BACKUP_ENCRYPTION_KEY is not set" 

41 ) 

42 from config.routes import _derive_key # pyright: ignore[reportMissingImports] 

43 from cryptography.hazmat.primitives.ciphers.aead import ( 

44 AESGCM, # pyright: ignore[reportMissingImports] 

45 ) 

46 

47 key = _derive_key(key_raw) 

48 nonce, ct = payload[:12], payload[12:] 

49 try: 

50 return AESGCM(key).decrypt(nonce, ct, None) 

51 except Exception as exc: 

52 raise BackupVerificationError(f"decryption failed: {exc}") from exc 

53 

54 

55def verify_backup_record(record: BackupRecord) -> None: 

56 """Decrypt and validate *record*'s archive. 

57 

58 Raises BackupVerificationError describing the first problem found. 

59 """ 

60 if not record.path or not os.path.exists(record.path): 

61 raise BackupVerificationError(f"archive file missing: {record.path!r}") 

62 

63 with open(record.path, "rb") as fh: 

64 payload = fh.read() 

65 

66 zip_bytes = _decrypt_if_needed(payload, os.path.basename(record.path)) 

67 

68 try: 

69 parse_backup_archive(zip_bytes, require_metadata=True) 

70 except BackupArchiveError as exc: 

71 raise BackupVerificationError(str(exc)) from exc 

72 

73 

74def verify_and_alert(record: BackupRecord) -> bool: 

75 """Run verify_backup_record(); log + fire a [SECURITY] alert on failure. 

76 

77 Returns True if verification passed, False otherwise. Never raises — 

78 a verification bug must not take down the backup scheduler. 

79 """ 

80 try: 

81 verify_backup_record(record) 

82 except BackupVerificationError as exc: 

83 log.error( 

84 "[SECURITY] backup.verification_failed backup_id=%s filename=%s reason=%s", 

85 record.id, 

86 record.filename, 

87 exc, 

88 ) 

89 return False 

90 except Exception: 

91 log.exception( 

92 "[SECURITY] backup.verification_failed backup_id=%s filename=%s reason=unexpected_error", 

93 record.id, 

94 record.filename, 

95 ) 

96 return False 

97 log.info("Backup verification OK: %s", record.filename) 

98 return True