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

1import io 

2import struct 

3 

4from structlog import get_logger 

5 

6from unblob.report import EncryptionMetadataReport 

7 

8from ...extractors import Command 

9from ...file_utils import InvalidInputFormat, iterate_patterns 

10from ...models import ( 

11 File, 

12 HandlerDoc, 

13 HandlerType, 

14 HexString, 

15 Reference, 

16 StructHandler, 

17 ValidChunk, 

18) 

19 

20logger = get_logger() 

21 

22 

23class ZIPHandler(StructHandler): 

24 NAME = "zip" 

25 

26 PATTERNS = [HexString("50 4B 03 04 // Local file header only")] 

27 C_DEFINITIONS = r""" 

28 

29 typedef struct cd_file_header { 

30 uint32 magic; 

31 uint16 version_made_by; 

32 uint16 version_needed; 

33 uint16 flags; 

34 uint16 compression_method; 

35 uint16 dostime; 

36 uint16 dosdate; 

37 uint32 crc32_cs; 

38 uint32 compress_size; 

39 uint32 file_size; 

40 uint16 file_name_length; 

41 uint16 extra_field_length; 

42 uint16 file_comment_length; 

43 uint16 disk_number_start; 

44 uint16 internal_file_attr; 

45 uint32 external_file_attr; 

46 uint32 relative_offset_local_header; 

47 // char file_name[file_name_length]; 

48 // char extra_field[extra_field_length]; 

49 } partial_cd_file_header_t; 

50 

51 typedef struct end_of_central_directory 

52 { 

53 uint32 end_of_central_signature; 

54 uint16 disk_number; 

55 uint16 disk_number_with_cd; 

56 uint16 disk_entries; 

57 uint16 total_entries; 

58 uint32 central_directory_size; 

59 uint32 offset_of_cd; 

60 uint16 comment_len; 

61 char zip_file_comment[comment_len]; 

62 } end_of_central_directory_t; 

63 

64 typedef struct zip64_end_of_central_directory_locator 

65 { 

66 uint32 signature; 

67 uint32 disk_number; 

68 uint64 offset_of_cd; 

69 uint32 total_disk; 

70 } zip64_end_of_central_directory_locator_t; 

71 

72 typedef struct zip64_end_of_central_directory 

73 { 

74 uint32 signature; 

75 uint64 size_of_eocd_record; 

76 uint16 version_made_by; 

77 uint16 version_needed; 

78 uint32 disk_number; 

79 uint32 disk_number_with_cd; 

80 uint64 total_entries_disk; 

81 uint64 total_entries; 

82 uint64 size_of_cd; 

83 uint64 offset_of_cd; 

84 } zip64_end_of_central_directory_t; 

85 

86 """ 

87 HEADER_STRUCT = "end_of_central_directory_t" 

88 

89 # empty password with -p will make sure the command will not hang 

90 EXTRACTOR = Command("7z", "x", "-p", "-y", "{inpath}", "-o{outdir}") 

91 

92 DOC = HandlerDoc( 

93 name="ZIP", 

94 description="ZIP is a widely used archive file format that supports multiple compression methods, file spanning, and optional encryption. It includes metadata such as file names, sizes, and timestamps, and supports both standard and ZIP64 extensions for large files.", 

95 handler_type=HandlerType.ARCHIVE, 

96 vendor=None, 

97 references=[ 

98 Reference( 

99 title="ZIP File Format Specification", 

100 url="https://pkware.com/documents/casestudies/APPNOTE.TXT", 

101 ), 

102 Reference( 

103 title="ZIP64 Format Specification", 

104 url="https://pkware.cachefly.net/webdocs/APPNOTE/APPNOTE-6.3.1.TXT", 

105 ), 

106 ], 

107 limitations=["Does not support encrypted ZIP files."], 

108 ) 

109 

110 ENCRYPTED_FLAG = 0b0001 

111 EOCD_RECORD_HEADER = 0x6054B50 

112 ZIP64_EOCD_SIGNATURE = 0x06064B50 

113 ZIP64_EOCD_LOCATOR_HEADER = 0x07064B50 

114 

115 def has_encrypted_files( 

116 self, 

117 file: File, 

118 start_offset: int, 

119 end_of_central_directory, 

120 ) -> bool: 

121 file.seek(start_offset + end_of_central_directory.offset_of_cd, io.SEEK_SET) 

122 for _ in range(end_of_central_directory.total_entries): 

123 file_header = self.cparser_le.partial_cd_file_header_t(file) 

124 file.seek( 

125 file_header.file_name_length + file_header.extra_field_length, 

126 io.SEEK_CUR, 

127 ) 

128 if file_header.flags & self.ENCRYPTED_FLAG: 

129 return True 

130 return False 

131 

132 @staticmethod 

133 def is_zip64_eocd(end_of_central_directory): 

134 # see https://pkware.cachefly.net/webdocs/APPNOTE/APPNOTE-6.3.1.TXT section J 

135 return ( 

136 end_of_central_directory.disk_number == 0xFFFF 

137 or end_of_central_directory.disk_number_with_cd == 0xFFFF 

138 or end_of_central_directory.disk_entries == 0xFFFF 

139 or end_of_central_directory.total_entries == 0xFFFF 

140 or end_of_central_directory.central_directory_size == 0xFFFFFFFF 

141 or end_of_central_directory.offset_of_cd == 0xFFFFFFFF 

142 ) 

143 

144 def has_zip64_tag(self, file): 

145 # see https://pkware.cachefly.net/webdocs/casestudies/APPNOTE.TXT section 4.3.9.2 

146 file_header = self.cparser_le.partial_cd_file_header_t(file) 

147 return ( 

148 file_header.file_size == 0xFFFFFFFF 

149 or file_header.compress_size == 0xFFFFFFFF 

150 ) 

151 

152 def _parse_zip64(self, file: File, start_offset: int, offset: int): 

153 file.seek(start_offset, io.SEEK_SET) 

154 for eocd_locator_offset in iterate_patterns( 

155 file, struct.pack("<I", self.ZIP64_EOCD_LOCATOR_HEADER) 

156 ): 

157 file.seek(eocd_locator_offset, io.SEEK_SET) 

158 eocd_locator = self.cparser_le.zip64_end_of_central_directory_locator_t( 

159 file 

160 ) 

161 logger.debug("eocd_locator", eocd_locator=eocd_locator, _verbosity=3) 

162 

163 # ZIP64 EOCD locator is right before the EOCD record 

164 if eocd_locator_offset + len(eocd_locator) == offset: 

165 file.seek(start_offset + eocd_locator.offset_of_cd) 

166 zip64_eocd = self.cparser_le.zip64_end_of_central_directory_t(file) 

167 logger.debug("zip64_eocd", zip64_eocd=zip64_eocd, _verbosity=3) 

168 

169 if zip64_eocd.signature != self.ZIP64_EOCD_SIGNATURE: 

170 raise InvalidInputFormat( 

171 "Missing ZIP64 EOCD header record header in ZIP chunk." 

172 ) 

173 return zip64_eocd 

174 return None 

175 

176 def get_zip64_eocd(self, file, start_offset, offset, end_of_central_directory): 

177 # some values in the CD can be FFFF, indicating its a zip64 

178 # if the offset of the CD is 0xFFFFFFFF, its definitely one 

179 # otherwise we check every other header indicating zip64 

180 if self.is_zip64_eocd(end_of_central_directory): 

181 return self._parse_zip64(file, start_offset, offset) 

182 

183 absolute_offset_of_cd = start_offset + end_of_central_directory.offset_of_cd 

184 

185 if 0 < absolute_offset_of_cd < offset: 

186 file.seek(absolute_offset_of_cd, io.SEEK_SET) 

187 if self.has_zip64_tag(file): 

188 return self._parse_zip64(file, start_offset, offset) 

189 

190 return None 

191 

192 def calculate_chunk(self, file: File, start_offset: int) -> ValidChunk | None: 

193 has_encrypted_files = False 

194 file.seek(start_offset, io.SEEK_SET) 

195 

196 offset = None 

197 for offset in iterate_patterns( 

198 file, struct.pack("<I", self.EOCD_RECORD_HEADER) 

199 ): 

200 file.seek(offset, io.SEEK_SET) 

201 end_of_central_directory = self.parse_header(file) 

202 

203 zip64_eocd = self.get_zip64_eocd( 

204 file, start_offset, offset, end_of_central_directory 

205 ) 

206 if zip64_eocd is not None: 

207 end_of_central_directory = zip64_eocd 

208 break 

209 

210 # the EOCD offset is equal to the offset of CD + size of CD 

211 end_of_central_directory_offset = ( 

212 start_offset 

213 + end_of_central_directory.offset_of_cd 

214 + end_of_central_directory.central_directory_size 

215 ) 

216 

217 if offset == end_of_central_directory_offset: 

218 break 

219 else: 

220 raise InvalidInputFormat("Missing EOCD record header in ZIP chunk.") 

221 

222 has_encrypted_files = self.has_encrypted_files( 

223 file, start_offset, end_of_central_directory 

224 ) 

225 

226 file.seek(offset, io.SEEK_SET) 

227 self.cparser_le.end_of_central_directory_t(file) 

228 

229 return ValidChunk( 

230 start_offset=start_offset, 

231 end_offset=file.tell(), 

232 metadata_reports=[ 

233 EncryptionMetadataReport(is_encrypted=has_encrypted_files) 

234 ], 

235 )