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

1"""Shared parsing for decrypted backup archive contents. 

2 

3Used by both the restore path (init.py's ``restore-backup`` CLI command) 

4and the automated restore-verification path (backup_verification.py) so 

5malformed-archive handling can't diverge between them. Before this was 

6extracted, ``restore_backup_command`` parsed the same zip structure inline 

7with no error handling at all (an unhandled ``zipfile.BadZipFile`` or 

8``KeyError`` on a missing ``openhangar.sql`` entry would crash the CLI 

9command with a raw traceback), while ``verify_backup_record`` already 

10handled both cases cleanly — this closes that gap rather than leaving the 

11two paths to drift. 

12""" 

13 

14from __future__ import annotations 

15 

16import json 

17import zipfile 

18from io import BytesIO 

19from typing import Any 

20 

21# pg_dump always opens a plain-text dump with this comment line. 

22_SQL_DUMP_MARKER = b"-- PostgreSQL database dump" 

23 

24 

25class BackupArchiveError(Exception): 

26 """Raised when a decrypted backup archive is malformed or incomplete.""" 

27 

28 

29def parse_backup_archive( 

30 zip_bytes: bytes, *, require_metadata: bool = False 

31) -> tuple[dict[str, Any], bytes, list[str]]: 

32 """Parse a decrypted backup zip's contents. 

33 

34 Returns ``(metadata, sql_bytes, upload_entries)``. Raises 

35 ``BackupArchiveError`` describing the first problem found — callers 

36 only ever need to handle this one exception type, never zipfile's or 

37 json's own. 

38 

39 ``metadata.json`` is optional by default (``restore_backup_command``'s 

40 original behaviour: missing manifest → ``{}``, still restorable). 

41 Pass ``require_metadata=True`` to reject a missing manifest instead 

42 (``verify_backup_record``'s stricter original behaviour, since a 

43 freshly-created backup should always have one). 

44 """ 

45 try: 

46 with zipfile.ZipFile(BytesIO(zip_bytes)) as zf: 

47 bad_entry = zf.testzip() 

48 if bad_entry is not None: 

49 raise BackupArchiveError(f"CRC check failed for {bad_entry!r}") 

50 

51 names = zf.namelist() 

52 

53 if "openhangar.sql" not in names: 

54 raise BackupArchiveError("openhangar.sql is missing from the archive") 

55 sql_bytes = zf.read("openhangar.sql") 

56 if _SQL_DUMP_MARKER not in sql_bytes[:2048]: 

57 raise BackupArchiveError( 

58 "openhangar.sql does not look like a pg_dump SQL dump" 

59 ) 

60 

61 metadata: dict[str, Any] = {} 

62 if "metadata.json" in names: 

63 try: 

64 parsed = json.loads(zf.read("metadata.json")) 

65 except (json.JSONDecodeError, UnicodeDecodeError) as exc: 

66 raise BackupArchiveError( 

67 f"metadata.json manifest is not valid JSON: {exc}" 

68 ) from exc 

69 if not isinstance(parsed, dict): 

70 raise BackupArchiveError( 

71 "metadata.json manifest is not a JSON object" 

72 ) 

73 metadata = parsed 

74 elif require_metadata: 

75 raise BackupArchiveError( 

76 "metadata.json manifest is missing from the archive" 

77 ) 

78 

79 upload_entries = [n for n in names if n.startswith("uploads/")] 

80 except zipfile.BadZipFile as exc: 

81 raise BackupArchiveError(f"not a valid zip archive: {exc}") from exc 

82 

83 return metadata, sql_bytes, upload_entries