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 CD_FILE_HEADER = 0x02014B50 

112 EOCD_RECORD_HEADER = 0x6054B50 

113 ZIP64_EOCD_SIGNATURE = 0x06064B50 

114 ZIP64_EOCD_LOCATOR_HEADER = 0x07064B50 

115 

116 def has_encrypted_files( 

117 self, 

118 file: File, 

119 start_offset: int, 

120 end_of_central_directory, 

121 ) -> bool: 

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

123 for _ in range(end_of_central_directory.total_entries): 

124 file_header = self.cparser_le.partial_cd_file_header_t(file) 

125 if file_header.magic != self.CD_FILE_HEADER: 

126 raise InvalidInputFormat( 

127 "Invalid central directory file header in ZIP chunk." 

128 ) 

129 # see https://pkware.cachefly.net/webdocs/casestudies/APPNOTE.TXT section 4.3.12 

130 # the fixed part is followed by the file name, the extra field and the file comment 

131 file.seek( 

132 file_header.file_name_length 

133 + file_header.extra_field_length 

134 + file_header.file_comment_length, 

135 io.SEEK_CUR, 

136 ) 

137 if file_header.flags & self.ENCRYPTED_FLAG: 

138 return True 

139 return False 

140 

141 @staticmethod 

142 def is_zip64_eocd(end_of_central_directory): 

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

144 return ( 

145 end_of_central_directory.disk_number == 0xFFFF 

146 or end_of_central_directory.disk_number_with_cd == 0xFFFF 

147 or end_of_central_directory.disk_entries == 0xFFFF 

148 or end_of_central_directory.total_entries == 0xFFFF 

149 or end_of_central_directory.central_directory_size == 0xFFFFFFFF 

150 or end_of_central_directory.offset_of_cd == 0xFFFFFFFF 

151 ) 

152 

153 def has_zip64_tag(self, file): 

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

155 file_header = self.cparser_le.partial_cd_file_header_t(file) 

156 return ( 

157 file_header.file_size == 0xFFFFFFFF 

158 or file_header.compress_size == 0xFFFFFFFF 

159 ) 

160 

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

162 file.seek(start_offset, io.SEEK_SET) 

163 for eocd_locator_offset in iterate_patterns( 

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

165 ): 

166 file.seek(eocd_locator_offset, io.SEEK_SET) 

167 eocd_locator = self.cparser_le.zip64_end_of_central_directory_locator_t( 

168 file 

169 ) 

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

171 

172 # ZIP64 EOCD locator is right before the EOCD record 

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

174 file.seek(start_offset + eocd_locator.offset_of_cd) 

175 zip64_eocd = self.cparser_le.zip64_end_of_central_directory_t(file) 

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

177 

178 if zip64_eocd.signature != self.ZIP64_EOCD_SIGNATURE: 

179 raise InvalidInputFormat( 

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

181 ) 

182 return zip64_eocd 

183 return None 

184 

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

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

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

188 # otherwise we check every other header indicating zip64 

189 if self.is_zip64_eocd(end_of_central_directory): 

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

191 

192 absolute_offset_of_cd = start_offset + end_of_central_directory.offset_of_cd 

193 

194 if 0 < absolute_offset_of_cd < offset: 

195 file.seek(absolute_offset_of_cd, io.SEEK_SET) 

196 if self.has_zip64_tag(file): 

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

198 

199 return None 

200 

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

202 has_encrypted_files = False 

203 file.seek(start_offset, io.SEEK_SET) 

204 

205 offset = None 

206 for offset in iterate_patterns( 

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

208 ): 

209 file.seek(offset, io.SEEK_SET) 

210 end_of_central_directory = self.parse_header(file) 

211 

212 zip64_eocd = self.get_zip64_eocd( 

213 file, start_offset, offset, end_of_central_directory 

214 ) 

215 if zip64_eocd is not None: 

216 end_of_central_directory = zip64_eocd 

217 break 

218 

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

220 end_of_central_directory_offset = ( 

221 start_offset 

222 + end_of_central_directory.offset_of_cd 

223 + end_of_central_directory.central_directory_size 

224 ) 

225 

226 if offset == end_of_central_directory_offset: 

227 break 

228 else: 

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

230 

231 has_encrypted_files = self.has_encrypted_files( 

232 file, start_offset, end_of_central_directory 

233 ) 

234 

235 file.seek(offset, io.SEEK_SET) 

236 self.cparser_le.end_of_central_directory_t(file) 

237 

238 return ValidChunk( 

239 start_offset=start_offset, 

240 end_offset=file.tell(), 

241 metadata_reports=[ 

242 EncryptionMetadataReport(is_encrypted=has_encrypted_files) 

243 ], 

244 )