Coverage for /pythoncovmergedfiles/medio/medio/usr/local/lib/python3.11/site-packages/unblob/report.py: 73%

1from __future__ import annotations 

2 

3import base64 

4import hashlib 

5import stat 

6import traceback 

7from enum import Enum 

8from pathlib import Path 

9from typing import TYPE_CHECKING, Annotated, Any, TypeVar 

10 

11if TYPE_CHECKING: 

12 from collections.abc import Iterable 

13 

14from pydantic import ( 

15 BaseModel, 

16 BeforeValidator, 

17 ConfigDict, 

18 Field, 

19 computed_field, 

20 field_serializer, 

21 field_validator, 

22) 

23 

24 

25def ensure_bytes(value: Any) -> bytes: 

26 if isinstance(value, bytes): 

27 return value 

28 if isinstance(value, str): 

29 return value.encode() 

30 if value is None: 

31 return b"" 

32 raise ValueError(f"Unsupported type in ensure_bytes: {type(value)}") 

33 

34 

35class Report(BaseModel): 

36 """A common base class for different reports. This will enable easy pydantic configuration of all models from a single point in the future if desired.""" 

37 

38 @computed_field 

39 @property 

40 def __typename__(self) -> str: 

41 return self.__class__.__name__ 

42 

43 

44class Severity(Enum): 

45 """Represents possible problems encountered during execution.""" 

46 

47 ERROR = "ERROR" 

48 WARNING = "WARNING" 

49 

50 

51class ErrorReport(Report): 

52 severity: Severity 

53 

54 

55class UnknownError(ErrorReport): 

56 """Describes an exception raised during file processing.""" 

57 

58 severity: Severity = Severity.ERROR 

59 exception: str | Exception 

60 

61 model_config = ConfigDict( 

62 arbitrary_types_allowed=True 

63 ) # Necessary to support Exception type 

64 

65 def model_post_init(self, _: Any) -> None: 

66 if isinstance(self.exception, Exception): 

67 self.exception = "".join( 

68 traceback.format_exception( 

69 type(self.exception), self.exception, self.exception.__traceback__ 

70 ) 

71 ) 

72 

73 """Exceptions are also formatted at construct time.""" 

74 

75 

76class CalculateChunkExceptionReport(UnknownError): 

77 """Describes an exception raised during calculate_chunk execution.""" 

78 

79 start_offset: int 

80 # Stored in `str` rather than `Handler`, because the pickle picks ups structs from `C_DEFINITIONS` 

81 handler: str 

82 

83 

84class CalculateMultiFileExceptionReport(UnknownError): 

85 """Describes an exception raised during calculate_chunk execution.""" 

86 

87 path: Path 

88 # Stored in `str` rather than `Handler`, because the pickle picks ups structs from `C_DEFINITIONS` 

89 handler: str 

90 

91 

92class ExtractCommandFailedReport(ErrorReport): 

93 """Describes an error when failed to run the extraction command.""" 

94 

95 severity: Severity = Severity.WARNING 

96 command: str 

97 stdout: Annotated[bytes, BeforeValidator(ensure_bytes)] 

98 stderr: Annotated[bytes, BeforeValidator(ensure_bytes)] 

99 exit_code: int 

100 

101 # Use base64 to encode and decode bytes data in case there are non-standard characters 

102 @field_serializer("stdout", "stderr") 

103 def encode_bytes(self, v: bytes, _): 

104 return base64.b64encode(v).decode("ascii") 

105 

106 @field_validator("stdout", "stderr", mode="before") 

107 @classmethod 

108 def decode_bytes(cls, v: Any): 

109 if isinstance(v, str): 

110 return base64.b64decode(v) 

111 return v 

112 

113 

114class OutputDirectoryExistsReport(ErrorReport): 

115 severity: Severity = Severity.ERROR 

116 path: Path 

117 

118 

119class ExtractorDependencyNotFoundReport(ErrorReport): 

120 """Describes an error when the dependency of an extractor doesn't exist.""" 

121 

122 severity: Severity = Severity.ERROR 

123 dependencies: list[str] 

124 

125 

126class ExtractorTimedOut(ErrorReport): 

127 """Describes an error when the extractor execution timed out.""" 

128 

129 severity: Severity = Severity.ERROR 

130 cmd: str 

131 timeout: float 

132 

133 

134class MaliciousSymlinkRemoved(ErrorReport): 

135 """Describes an error when malicious symlinks have been removed from disk.""" 

136 

137 severity: Severity = Severity.WARNING 

138 link: str 

139 target: str 

140 

141 

142class MultiFileCollisionReport(ErrorReport): 

143 """Describes an error when MultiFiles collide on the same file.""" 

144 

145 severity: Severity = Severity.ERROR 

146 paths: set[Path] 

147 handler: str 

148 

149 

150class StatReport(Report): 

151 path: Path 

152 size: int 

153 is_dir: bool 

154 is_file: bool 

155 is_link: bool 

156 link_target: Path | None 

157 

158 @classmethod 

159 def from_path(cls, path: Path): 

160 st = path.lstat() 

161 mode = st.st_mode 

162 try: 

163 link_target = Path.readlink(path) 

164 except OSError: 

165 link_target = None 

166 

167 return cls( 

168 path=path, 

169 size=st.st_size, 

170 is_dir=stat.S_ISDIR(mode), 

171 is_file=stat.S_ISREG(mode), 

172 is_link=stat.S_ISLNK(mode), 

173 link_target=link_target, 

174 ) 

175 

176 

177class HashReport(Report): 

178 md5: str 

179 sha1: str 

180 sha256: str 

181 

182 @classmethod 

183 def from_path(cls, path: Path): 

184 chunk_size = 1024 * 64 

185 md5 = hashlib.md5(usedforsecurity=False) 

186 sha1 = hashlib.sha1(usedforsecurity=False) 

187 sha256 = hashlib.sha256() 

188 

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

190 while chunk := f.read(chunk_size): 

191 md5.update(chunk) 

192 sha1.update(chunk) 

193 sha256.update(chunk) 

194 

195 return cls( 

196 md5=md5.hexdigest(), 

197 sha1=sha1.hexdigest(), 

198 sha256=sha256.hexdigest(), 

199 ) 

200 

201 

202class FileMagicReport(Report): 

203 magic: str 

204 mime_type: str 

205 

206 

207class RandomnessMeasurements(BaseModel): 

208 percentages: list[float] 

209 block_size: int 

210 mean: float 

211 

212 @property 

213 def highest(self): 

214 return max(self.percentages) 

215 

216 @property 

217 def lowest(self): 

218 return min(self.percentages) 

219 

220 

221class RandomnessReport(Report): 

222 shannon: RandomnessMeasurements 

223 chi_square: RandomnessMeasurements 

224 

225 

226class MetadataReport(Report): 

227 pass 

228 

229 

230class EncryptionMetadataReport(MetadataReport): 

231 is_encrypted: bool 

232 

233 

234class HandledBlobReport(Report): 

235 id: str 

236 handler_name: str 

237 extraction_reports: list[Report] 

238 metadata_reports: list[MetadataReport] = Field(default_factory=list) 

239 

240 @field_validator("extraction_reports", "metadata_reports", mode="before") 

241 @classmethod 

242 def validate_reports(cls, value: Any) -> list[Report]: 

243 return validate_report_list(value) 

244 

245 

246class ChunkReport(HandledBlobReport): 

247 start_offset: int 

248 end_offset: int 

249 size: int 

250 

251 

252class UnknownChunkReport(Report): 

253 id: str 

254 start_offset: int 

255 end_offset: int 

256 size: int 

257 randomness: RandomnessReport | None 

258 

259 @field_validator("randomness", mode="before") 

260 @classmethod 

261 def validate_randomness(cls, value: Any) -> RandomnessReport | None: 

262 if value is None: 

263 return None 

264 parsed = parse_report(value) 

265 if not isinstance(parsed, RandomnessReport): 

266 raise TypeError("Randomness must be a RandomnessReport.") 

267 return parsed 

268 

269 

270class CarveDirectoryReport(Report): 

271 carve_dir: Path 

272 

273 

274class MultiFileReport(HandledBlobReport): 

275 name: str 

276 paths: list[Path] 

277 

278 

279class ExtractedFileDeletedReport(Report): 

280 path: Path 

281 handler_name: str 

282 

283 

284class ExtractionProblem(Report): 

285 """A non-fatal problem discovered during extraction. 

286 

287 A report like this still means, that the extraction was successful, 

288 but there were problems that got resolved. 

289 The output is expected to be complete, with the exception of 

290 the reported path. 

291 

292 Examples 

293 -------- 

294 - duplicate entries for certain archive formats (tar, zip) 

295 - unsafe symlinks pointing outside of extraction directory 

296 

297 """ 

298 

299 problem: str 

300 resolution: str 

301 path: str | None = None 

302 

303 @property 

304 def log_msg(self): 

305 return f"{self.problem} {self.resolution}" 

306 

307 def log_with(self, logger): 

308 logger.warning(self.log_msg, path=self.path) 

309 

310 

311class PathTraversalProblem(ExtractionProblem): 

312 extraction_path: str 

313 

314 def log_with(self, logger): 

315 logger.warning( 

316 self.log_msg, 

317 path=self.path, 

318 extraction_path=self.extraction_path, 

319 ) 

320 

321 

322class LinkExtractionProblem(ExtractionProblem): 

323 link_path: str 

324 

325 def log_with(self, logger): 

326 logger.warning(self.log_msg, path=self.path, link_path=self.link_path) 

327 

328 

329class SpecialFileExtractionProblem(ExtractionProblem): 

330 mode: int 

331 device: int 

332 

333 def log_with(self, logger): 

334 logger.warning(self.log_msg, path=self.path, mode=self.mode, device=self.device) 

335 

336 

337class ExtendedAttributeExtractionProblem(ExtractionProblem): 

338 attribute: str 

339 

340 def log_with(self, logger): 

341 logger.warning(self.log_msg, path=self.path, attribute=self.attribute) 

342 

343 

344BUILTIN_REPORT_TYPES: tuple[type[Report], ...] = ( 

345 ErrorReport, 

346 UnknownError, 

347 CalculateChunkExceptionReport, 

348 CalculateMultiFileExceptionReport, 

349 ExtractCommandFailedReport, 

350 OutputDirectoryExistsReport, 

351 ExtractorDependencyNotFoundReport, 

352 ExtractedFileDeletedReport, 

353 ExtractorTimedOut, 

354 MaliciousSymlinkRemoved, 

355 MultiFileCollisionReport, 

356 StatReport, 

357 HashReport, 

358 FileMagicReport, 

359 RandomnessReport, 

360 EncryptionMetadataReport, 

361 ChunkReport, 

362 UnknownChunkReport, 

363 CarveDirectoryReport, 

364 MultiFileReport, 

365 ExtractionProblem, 

366 PathTraversalProblem, 

367 LinkExtractionProblem, 

368 SpecialFileExtractionProblem, 

369) 

370 

371_REPORT_REGISTRY: dict[str, type[Report]] = {} 

372ReportType = TypeVar("ReportType", bound=type[Report]) 

373 

374 

375def register_report_type(report_type: ReportType) -> ReportType: 

376 typename = report_type.__name__ 

377 existing = _REPORT_REGISTRY.get(typename) 

378 if existing is not None and existing is not report_type: 

379 raise ValueError(f"Report type name conflict: {typename}") 

380 _REPORT_REGISTRY[typename] = report_type 

381 return report_type 

382 

383 

384def register_report_types(report_types: Iterable[type[Report]]) -> None: 

385 for report_type in report_types: 

386 register_report_type(report_type) 

387 

388 

389def get_report_type(typename: str) -> type[Report] | None: 

390 return _REPORT_REGISTRY.get(typename) 

391 

392 

393def parse_report(report: Report | dict[str, Any]) -> Report: 

394 if isinstance(report, Report): 

395 return report 

396 if not isinstance(report, dict): 

397 raise TypeError("Report data must be a mapping or Report instance.") 

398 typename = report.get("__typename__") 

399 if not typename: 

400 raise ValueError("Report data is missing __typename__.") 

401 report_type = get_report_type(typename) 

402 if report_type is None: 

403 raise ValueError(f"Unknown report type: {typename}") 

404 return report_type.model_validate(report) 

405 

406 

407def validate_report_list(value: Any) -> list[Report]: 

408 if not isinstance(value, list): 

409 raise TypeError("Report list must be a list.") 

410 return [parse_report(item) for item in value] 

411 

412 

413register_report_types(BUILTIN_REPORT_TYPES)