Coverage for /pythoncovmergedfiles/medio/medio/usr/local/lib/python3.11/site-packages/unblob/handlers/compression/gzip.py: 68%

1"""Handler for gzip compression format. 

2 

3It is based on standard documented at 

4https://datatracker.ietf.org/doc/html/rfc1952. 

5 

6The handler will create valid chunks for each gzip compressed stream 

7instead of concatenating sequential streams into an overall 

8ValidChunk. 

9 

10We monkey patched Python builtin gzip's _GzipReader read() function to 

11stop reading as soon as it reach the EOF marker of the current gzip 

12stream. This is a requirement for unblob given that streams can be 

13malformed and followed by garbage/random content that triggers 

14BadGzipFile errors when gzip library tries to read the next stream 

15header. 

16""" 

17 

18import gzip 

19import io 

20import struct 

21import zlib 

22from pathlib import Path 

23 

24from structlog import get_logger 

25 

26from unblob.extractors import Command 

27from unblob.extractors.command import MultiFileCommand 

28from unblob.models import Extractor 

29 

30from ...file_utils import InvalidInputFormat 

31from ...models import ( 

32 DirectoryExtractor, 

33 DirectoryHandler, 

34 ExtractResult, 

35 File, 

36 Glob, 

37 Handler, 

38 HandlerDoc, 

39 HandlerType, 

40 HexString, 

41 MultiFile, 

42 Reference, 

43 ValidChunk, 

44) 

45from ._gzip_reader import SingleMemberGzipReader 

46 

47logger = get_logger() 

48 

49GZIP2_CRC_LEN = 4 

50GZIP2_SIZE_LEN = 4 

51GZIP2_FOOTER_LEN = GZIP2_CRC_LEN + GZIP2_SIZE_LEN 

52 

53# Members decompressing up to this size must have a matching CRC32 and size 

54# in their trailer. Small members are where false positives live, e.g. x86 

55# code that happens to look like a gzip header followed by a tiny deflate 

56# stream. Larger members are accepted with a bad trailer, as 7z still 

57# extracts their content. 

58GZIP2_STRICT_TRAILER_MAX_SIZE = 64 * 1024 

59 

60FLAG_EXTRA = 4 

61FLAG_NAME = 8 

62 

63 

64def get_gzip_embedded_name(path: Path) -> str: 

65 name = b"" 

66 with path.open("rb") as file: 

67 # skip magic bytes and method 

68 file.read(2) 

69 (_method, flag, _last_mtime) = struct.unpack("<BBIxx", file.read(8)) 

70 

71 if flag & FLAG_EXTRA: 

72 # Read & discard the extra field, if present 

73 [extra_len] = struct.unpack("<H", file.read(2)) 

74 file.seek(extra_len, io.SEEK_CUR) 

75 

76 if flag & FLAG_NAME: 

77 # Read and discard a null-terminated string containing the filename 

78 while True: 

79 s = file.read(1) 

80 if not s or s == b"\000": 

81 break 

82 name += s 

83 

84 # return a valid, safe name without directories! 

85 try: 

86 return Path(name.decode("utf-8")).name 

87 except UnicodeDecodeError: 

88 return "" 

89 

90 

91class GZIPExtractor(Extractor): 

92 def get_dependencies(self) -> list[str]: 

93 return ["7z"] 

94 

95 def extract(self, inpath: Path, outdir: Path) -> ExtractResult | None: 

96 name = get_gzip_embedded_name(inpath) or "gzip.uncompressed" 

97 extractor = Command("7z", "x", "-y", "{inpath}", "-so", stdout=name) 

98 return extractor.extract(inpath, outdir) 

99 

100 

101class MultiGZIPExtractor(DirectoryExtractor): 

102 def get_dependencies(self) -> list[str]: 

103 return ["7z"] 

104 

105 def extract(self, paths: list[Path], outdir: Path) -> ExtractResult | None: 

106 name = get_gzip_embedded_name(paths[0]) or "gzip.uncompressed" 

107 extractor = MultiFileCommand( 

108 "7z", "x", "-p", "-y", "{inpath}", "-so", stdout=name 

109 ) 

110 return extractor.extract(paths, outdir) 

111 

112 

113class GZIPHandler(Handler): 

114 NAME = "gzip" 

115 

116 EXTRACTOR = GZIPExtractor() 

117 

118 PATTERNS = [ 

119 HexString( 

120 """ 

121 // ID1 

122 1F 

123 // ID2 

124 8B 

125 // compression method (0x8 = DEFLATE) 

126 08 

127 // flags, 00011111 (0x1f) is the highest since the first 3 bits are reserved 

128 ( 

129 00 | 01 | 02 | 03 | 04 | 05 | 06 | 07 | 08 | 09 | 

130 0A | 0B | 0C | 0D | 0E | 0F | 10 | 11 | 12 | 13 | 

131 14 | 15 | 16 | 17 | 18 | 19 | 1A | 1B | 1C | 1D | 1E 

132 ) 

133 // unix time (uint32) + eXtra FLags (2 or 4 per RFC1952 2.3.1) 

134 // we accept any value because the RFC is not followed by some samples 

135 [5] 

136 // Operating System (0-13, or 255 per RFC1952 2.3.1) 

137 ( 

138 00 | 01 | 02 | 03 | 04 | 05 | 06 | 07 | 08 | 09 | 0A | 0B | 0C | 0D | FF 

139 ) 

140 """ 

141 ) 

142 ] 

143 

144 DOC = HandlerDoc( 

145 name="GZIP", 

146 description="GZIP is a compressed file format that uses the DEFLATE algorithm and includes metadata such as original file name and modification time. It is commonly used for efficient file storage and transfer.", 

147 handler_type=HandlerType.COMPRESSION, 

148 vendor=None, 

149 references=[ 

150 Reference( 

151 title="GZIP File Format Specification", 

152 url="https://datatracker.ietf.org/doc/html/rfc1952", 

153 ), 

154 Reference( 

155 title="GZIP Wikipedia", 

156 url="https://en.wikipedia.org/wiki/Gzip", 

157 ), 

158 ], 

159 limitations=[], 

160 ) 

161 

162 def _read_member_end_offset(self, file: File) -> int | None: 

163 fp = SingleMemberGzipReader(file) 

164 if not fp.read_header(): 

165 return None 

166 

167 try: 

168 fp.read_until_eof() 

169 except (gzip.BadGzipFile, zlib.error) as e: 

170 raise InvalidInputFormat from e 

171 

172 file.seek(-len(fp.unused_data), io.SEEK_CUR) 

173 footer = file.read(GZIP2_FOOTER_LEN) 

174 if len(footer) < GZIP2_FOOTER_LEN: 

175 raise InvalidInputFormat("Truncated gzip trailer") 

176 

177 crc, size = struct.unpack("<II", footer) 

178 if fp.size <= GZIP2_STRICT_TRAILER_MAX_SIZE and ( 

179 crc != fp.crc or size != fp.size & 0xFFFFFFFF 

180 ): 

181 raise InvalidInputFormat("gzip trailer does not match content") 

182 

183 return file.tell() 

184 

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

186 end_offset = self._read_member_end_offset(file) 

187 if end_offset is None: 

188 return None 

189 

190 while end_offset < file.size(): 

191 try: 

192 next_end_offset = self._read_member_end_offset(file) 

193 except (EOFError, gzip.BadGzipFile, InvalidInputFormat): 

194 # Stop at the last valid member if a later member-like sequence is malformed or truncated. 

195 break 

196 

197 if next_end_offset is None: 

198 break 

199 

200 end_offset = next_end_offset 

201 

202 return ValidChunk( 

203 start_offset=start_offset, 

204 end_offset=end_offset, 

205 ) 

206 

207 

208class MultiVolumeGzipHandler(DirectoryHandler): 

209 NAME = "multi-gzip" 

210 EXTRACTOR = MultiGZIPExtractor() 

211 

212 PATTERN = Glob("*.gz.*") 

213 

214 DOC = HandlerDoc( 

215 name="GZIP (multi-volume)", 

216 description="GZIP is a compressed file format that uses the DEFLATE algorithm and includes metadata such as original file name and modification time. It is commonly used for efficient file storage and transfer.", 

217 handler_type=HandlerType.COMPRESSION, 

218 vendor=None, 

219 references=[ 

220 Reference( 

221 title="GZIP File Format Specification", 

222 url="https://datatracker.ietf.org/doc/html/rfc1952", 

223 ), 

224 Reference( 

225 title="GZIP Wikipedia", 

226 url="https://en.wikipedia.org/wiki/Gzip", 

227 ), 

228 ], 

229 limitations=[], 

230 ) 

231 

232 def is_valid_gzip(self, path: Path) -> bool: 

233 try: 

234 file = File.from_path(path) 

235 except ValueError: 

236 return False 

237 

238 with file as f: 

239 try: 

240 fp = SingleMemberGzipReader(f) 

241 if not fp.read_header(): 

242 return False 

243 except gzip.BadGzipFile: 

244 return False 

245 return True 

246 

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

248 paths = sorted( 

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

250 ) 

251 

252 # we 'discard' paths that are not the first in the ordered list, 

253 # otherwise we will end up with colliding reports, one for every 

254 # path in the list. 

255 if not paths or file != paths[0]: 

256 return None 

257 

258 if self.is_valid_gzip(file): 

259 files_size = sum(path.stat().st_size for path in paths) 

260 logger.debug( 

261 "Multi-volume files", paths=paths, files_size=files_size, _verbosity=2 

262 ) 

263 

264 return MultiFile( 

265 name=paths[0].stem, 

266 paths=paths, 

267 ) 

268 return None