Coverage for /pythoncovmergedfiles/medio/medio/usr/local/lib/python3.11/site-packages/unblob/handlers/archive/par2.py: 45%

1import hashlib 

2from pathlib import Path 

3 

4from unblob.file_utils import Endian, StructParser, iterate_file 

5from unblob.models import ( 

6 DirectoryHandler, 

7 Glob, 

8 HandlerDoc, 

9 HandlerType, 

10 MultiFile, 

11 Reference, 

12) 

13 

14C_DEFINITIONS = r""" 

15 typedef struct par2_header{ 

16 char magic[8]; 

17 uint64 packet_length; 

18 char md5_hash[16]; 

19 char recovery_set_id[16]; 

20 char type[16]; 

21 } par2_header_t; 

22""" 

23 

24PAR2_MAGIC = b"PAR2\x00PKT" 

25HEADER_STRUCT = "par2_header_t" 

26HEADER_PARSER = StructParser(C_DEFINITIONS) 

27 

28 

29class MultiVolumePAR2Handler(DirectoryHandler): 

30 NAME = "multi-par2" 

31 PATTERN = Glob("*.par2") 

32 EXTRACTOR = None 

33 

34 DOC = HandlerDoc( 

35 name="PAR2 (multi-volume)", 

36 description="Parchive or PAR2, is a format for creating redundant data that helps detect and repair corrupted files. These archives typically accompany split-file sets (like multi-volume RAR or ZIP archives). Each PAR2 file is composed of multiple 'packets'.", 

37 handler_type=HandlerType.ARCHIVE, 

38 vendor=None, 

39 references=[ 

40 Reference( 

41 title="Parchive Documentation", 

42 url="https://parchive.github.io/", 

43 ), 

44 ], 

45 limitations=[], 

46 ) 

47 

48 def is_valid_header(self, file_paths: list) -> bool: 

49 for path in file_paths: 

50 with path.open("rb") as f: 

51 header = HEADER_PARSER.parse(HEADER_STRUCT, f, Endian.LITTLE) 

52 if header.magic != PAR2_MAGIC: 

53 return False 

54 

55 offset_to_recovery_id = 32 

56 packet_checksum_state = hashlib.md5(usedforsecurity=False) 

57 packet_content_length = ( 

58 header.packet_length - len(header) + offset_to_recovery_id 

59 ) 

60 for chunk in iterate_file( 

61 f, offset_to_recovery_id, packet_content_length 

62 ): 

63 packet_checksum_state.update(chunk) 

64 

65 packet_checksum = packet_checksum_state.digest() 

66 

67 if packet_checksum != header.md5_hash: 

68 return False 

69 return True 

70 

71 def calculate_multifile(self, file: Path) -> MultiFile | None: 

72 paths = sorted( 

73 [p for p in file.parent.glob(f"{file.stem}.*") if p.resolve().exists()] 

74 ) 

75 

76 if len(paths) <= 1 or not self.is_valid_header(paths): 

77 return None 

78 

79 return MultiFile( 

80 name=file.stem, 

81 paths=paths, 

82 )