Coverage for src/fileaudit/gz

1"""

2License GPL3

4FileAudit - GZ File Security Checker

5"""

7import gzip

8import inspect

9import os

10import stat

11from functools import wraps

12from pathlib import Path

15# Global default fallbacks

16DEFAULT_MAX_FILE_SIZE = 10 * 1024 * 1024 # 10 MB

17DEFAULT_MAX_UNCOMPRESSED_RATIO = 100 # 100:1 ratio

18DEFAULT_MAX_UNCOMPRESSED_SIZE = 100 * 1024 * 1024 # 100 MB

20# Read decompressed data in bounded chunks rather than loading the entire

21# decompressed file into memory.

22GZ_READ_CHUNK_SIZE = 1024 * 1024 # 1 MB

25class GzValidationError(Exception):

26 """Custom exception for GZip validation failures in FileAudit."""

28 def __init__(self, message):

29 self.prefix = "FileAudit Security Validation Failed -"

30 self.original_message = str(message)

31 full_message = f"{self.prefix} {self.original_message}"

32 super().__init__(full_message)

34 def __str__(self):

35 return self.args[0]

38def _validate_limit(name, value):

39 """

40 Validate a numeric security limit.

42 Args:

43 name: Name of the limit.

44 value: Limit value.

46 Raises:

47 ValueError: If the value is not a positive integer.

48 """

49 if isinstance(value, bool) or not isinstance(value, int): 49 ↛ 50line 49 didn't jump to line 50 because the condition on line 49 was never true

50 raise ValueError(f"{name} must be an integer")

52 if value <= 0: 52 ↛ exitline 52 didn't return from function '_validate_limit' because the condition on line 52 was always true

53 raise ValueError(f"{name} must be greater than 0")

56def _open_gz_file(path):

57 """

58 Open a GZip file safely and return the file object and initial stat data.

60 The file is opened before its size is checked. This avoids the common

61 stat(path) -> open(path) TOCTOU pattern.

63 O_NOFOLLOW is used where supported so the final path component cannot be

64 replaced by a symbolic link between the security check and opening.

66 Args:

67 path: Path to the GZip file.

69 Returns:

70 Tuple containing the opened binary file and its initial stat result.

72 Raises:

73 GzValidationError: If the file cannot be opened safely.

74 """

75 flags = os.O_RDONLY

77 # Windows requires O_BINARY for binary reads.

78 if hasattr(os, "O_BINARY"): 78 ↛ 79line 78 didn't jump to line 79 because the condition on line 78 was never true

79 flags |= os.O_BINARY

81 # Prevent following the final symbolic link where supported.

82 if hasattr(os, "O_NOFOLLOW"): 82 ↛ 85line 82 didn't jump to line 85 because the condition on line 82 was always true

83 flags |= os.O_NOFOLLOW

85 try:

86 fd = os.open(str(path), flags)

88 except FileNotFoundError as e:

89 raise GzValidationError(

90 f"File not found: {path}"

91 ) from e

93 except PermissionError as e:

94 raise GzValidationError(

95 f"Permission denied: {path}"

96 ) from e

98 except OSError as e:

99 raise GzValidationError(

100 f"Failed to open file '{path}': {e}"

101 ) from e

102

103 try:

104 file_stat = os.fstat(fd)

105

106 # Never process directories, devices, FIFOs, sockets, etc.

107 if not stat.S_ISREG(file_stat.st_mode):

108 raise GzValidationError(

109 f"Path is not a regular file: {path}"

110 )

111

112 file_obj = os.fdopen(fd, "rb", closefd=True)

113

114 return file_obj, file_stat

115

116 except GzValidationError:

117 os.close(fd)

118 raise

119

120 except OSError as e:

121 os.close(fd)

122 raise GzValidationError(

123 f"Failed to inspect file '{path}': {e}"

124 ) from e

125

126

127def _validate_gz_file(

128 path,

129 max_file_size,

130 max_uncompressed_ratio,

131 max_uncompressed_size

132):

133 """

134 Internal validation function for GZip files.

135

136 Performs all security checks on a GZip file.

137

138 Security checks:

139

140 - File exists and is a regular file

141 - Symbolic links are rejected where O_NOFOLLOW is supported

142 - Compressed file size limits

143 - Streaming uncompressed size limits

144 - GZip decompression ratio limits

145 - GZip format/CRC/trailer validation

146 - Detection of compressed file size changes during validation

147

148 Args:

149 path: Path to the GZip file.

150 max_file_size: Maximum compressed file size.

151 max_uncompressed_ratio: Maximum decompression ratio.

152 max_uncompressed_size: Maximum uncompressed size.

153

154 Raises:

155 GzValidationError: If any security check fails.

156 """

157 path = Path(path)

158

159 # On platforms without O_NOFOLLOW, explicitly reject symbolic links before

160 # opening. On platforms with O_NOFOLLOW, the open operation itself

161 # provides the stronger race-resistant protection.

162 if not hasattr(os, "O_NOFOLLOW"): 162 ↛ 163line 162 didn't jump to line 163 because the condition on line 162 was never true

163 try:

164 if path.is_symlink():

165 raise GzValidationError(

166 f"Rejected {path}: Symbolic links are not allowed"

167 )

168 except OSError as e:

169 raise GzValidationError(

170 f"Failed to inspect path '{path}': {e}"

171 ) from e

172

173 # Open the actual file that will be validated.

174 raw_file, initial_stat = _open_gz_file(path)

175

176 try:

177 # ---------------------------------------------------------------

178 # 1. Compressed file size limit

179 # ---------------------------------------------------------------

180

181 compressed_size = initial_stat.st_size

182

183 if compressed_size == 0:

184 raise GzValidationError(

185 f"Rejected {path}: GZip file is empty"

186 )

187

188 if compressed_size > max_file_size:

189 raise GzValidationError(

190 f"Rejected {path}: File size ({compressed_size} bytes) "

191 f"exceeds maximum of {max_file_size} bytes"

192 )

193

194 # ---------------------------------------------------------------

195 # 2. Streaming GZip decompression

196 # ---------------------------------------------------------------

197

198 total_uncompressed_size = 0

199

200 try:

201 # GzipFile performs GZip header, DEFLATE, CRC and trailer

202 # validation. It also supports concatenated GZip members.

203 #

204 # Importantly, fileobj is the already-opened descriptor. The

205 # pathname is NOT reopened after the security checks.

206 with gzip.GzipFile(

207 fileobj=raw_file,

208 mode="rb"

209 ) as gz:

210

211 while True:

212 chunk = gz.read(GZ_READ_CHUNK_SIZE)

213

214 if not chunk:

215 break

216

217 total_uncompressed_size += len(chunk)

218

219 # Check during decompression so a malicious archive

220 # cannot expand completely before being rejected.

221 if total_uncompressed_size > max_uncompressed_size:

222 raise GzValidationError(

223 f"Rejected {path}: Uncompressed size "

224 f"({total_uncompressed_size} bytes) exceeds "

225 f"maximum of {max_uncompressed_size} bytes"

226 )

227

228 except GzValidationError:

229 raise

230

231 except gzip.BadGzipFile as e:

232 raise GzValidationError(

233 f"Invalid GZip format: {e}"

234 ) from e

235

236 except (EOFError, OSError) as e:

237 raise GzValidationError(

238 f"GZip decompression failed: {e}"

239 ) from e

240

241 # ---------------------------------------------------------------

242 # 3. Detect file modification during validation

243 # ---------------------------------------------------------------

244

245 try:

246 final_stat = os.fstat(raw_file.fileno())

247

248 except OSError as e:

249 raise GzValidationError(

250 f"Failed to re-check file '{path}': {e}"

251 ) from e

252

253 if final_stat.st_size != compressed_size:

254 raise GzValidationError(

255 f"Rejected {path}: File changed while being validated "

256 f"(size changed from {compressed_size} to "

257 f"{final_stat.st_size} bytes)"

258 )

259

260 # ---------------------------------------------------------------

261 # 4. GZip decompression ratio limit

262 # ---------------------------------------------------------------

263

264 if compressed_size > 0: 264 ↛ 276line 264 didn't jump to line 276 because the condition on line 264 was always true

265 ratio = total_uncompressed_size / compressed_size

266

267 if ratio > max_uncompressed_ratio:

268 raise GzValidationError(

269 f"Rejected {path}: Decompression ratio "

270 f"({ratio:.2f}x) exceeds maximum of "

271 f"{max_uncompressed_ratio}x "

272 f"(GZip bomb protection)"

273 )

274

275 finally:

276 raw_file.close()

277

278

279

280def validate_gz(

281 func_or_path=None,

282 max_file_size=None,

283 max_uncompressed_ratio=None,

284 max_uncompressed_size=None

285):

286 """

287 Validate GZip files via decorator or direct invocation.

288

289 A GZip file validator that can operate in two modes:

290

291 1. **Decorator mode** — wraps a function to validate a GZip file path

292 passed as an argument before the function body runs.

293

294 2. **Direct call / CLI mode** — validates a file immediately and returns

295 a boolean result.

296

297 Security checks performed:

298

299 - Compressed file size limits

300 - GZip decompression ratio limits

301 - Maximum uncompressed size

302 - Streaming decompression

303 - GZip CRC/trailer validation

304 - GZip concatenated-member validation

305 - Regular-file validation

306 - Symlink protection where supported

307 - Detection of file-size changes during validation

308

309 Usage:

310

311 @validate_gz

312 def process(path):

313 ...

314

315 @validate_gz()

316 def process(path):

317 ...

318

319 @validate_gz("custom_arg_name", max_file_size=5000)

320 def process(custom_arg_name):

321 ...

322

323 validate_gz("path/to/file.gz", max_uncompressed_size=1000000)

324

325 Args:

326 func_or_path (callable, str, pathlib.Path, or None):

327 * If a **callable**: the function to decorate

328 (bare decorator usage: ``@validate_gz``).

329

330 * If a **str** or **Path** representing a file path:

331 direct validation mode.

332

333 * If a **str** that is a valid Python identifier and does not

334 look like a file path: treated as the target argument name

335 in decorator mode.

336

337 * If **None**: returns a decorator factory.

338

339 max_file_size (int):

340 Maximum allowed compressed file size in bytes.

341

342 max_uncompressed_ratio (int):

343 Maximum GZip decompression ratio.

344

345 max_uncompressed_size (int):

346 Maximum total uncompressed size in bytes.

347

348 Returns:

349 Union[callable, bool, function]:

350

351 * In decorator mode: the wrapped function.

352 * In direct call mode: ``True`` if validation passes,

353 ``False`` if validation fails.

354

355 Raises:

356 GzValidationError:

357 If validation fails in decorator mode.

358

359 ValueError:

360 If a security limit is invalid.

361 """

362

363 # ---------------------------------------------------------------

364 # Resolve optional limits to defaults

365 # ---------------------------------------------------------------

366

367 resolved_file_size = (

368 DEFAULT_MAX_FILE_SIZE

369 if max_file_size is None

370 else max_file_size

371 )

372

373 resolved_ratio = (

374 DEFAULT_MAX_UNCOMPRESSED_RATIO

375 if max_uncompressed_ratio is None

376 else max_uncompressed_ratio

377 )

378

379 resolved_uncompressed_size = (

380 DEFAULT_MAX_UNCOMPRESSED_SIZE

381 if max_uncompressed_size is None

382 else max_uncompressed_size

383 )

384

385 # Validate configuration immediately.

386 _validate_limit(

387 "max_file_size",

388 resolved_file_size

389 )

390

391 _validate_limit(

392 "max_uncompressed_ratio",

393 resolved_ratio

394 )

395

396 _validate_limit(

397 "max_uncompressed_size",

398 resolved_uncompressed_size

399 )

400

401

402 # ---------------------------------------------------------------

403 # Determine whether a string looks like a file path

404 # ---------------------------------------------------------------

405

406 def _looks_like_file_path(value):

407 """

408 Heuristic: does this string look like a file path?

409 """

410 if value.startswith(("http://", "https://", "ftp://", "file://")): 410 ↛ 411line 410 didn't jump to line 411 because the condition on line 410 was never true

411 return True

412

413 if value.startswith(("/", "\\")): 413 ↛ 414line 413 didn't jump to line 414 because the condition on line 413 was never true

414 return True

415

416 if "/" in value or "\\" in value: 416 ↛ 417line 416 didn't jump to line 417 because the condition on line 416 was never true

417 return True

418

419 if "." in value and not value.startswith("."): 419 ↛ 420line 419 didn't jump to line 420 because the condition on line 419 was never true

420 return True

421

422 return False

423

424 # ---------------------------------------------------------------

425 # Determine operating mode

426 # ---------------------------------------------------------------

427

428 is_decorator_mode = False

429

430 if func_or_path is None:

431 is_decorator_mode = True

432

433 elif callable(func_or_path):

434 is_decorator_mode = True

435

436 elif isinstance(func_or_path, str):

437 # FIX: A string is only an argument name if it is a valid

438 # Python identifier AND does not look like a file path.

439 is_decorator_mode = (

440 func_or_path.isidentifier()

441 and not _looks_like_file_path(func_or_path)

442 )

443

444 elif isinstance(func_or_path, Path):

445 is_decorator_mode = False

446

447 else:

448 raise TypeError(

449 "Expected callable, str, pathlib.Path, or None, "

450 f"got {type(func_or_path).__name__}"

451 )

452

453 # ---------------------------------------------------------------

454 # Direct call / CLI mode

455 # ---------------------------------------------------------------

456

457 if not is_decorator_mode and isinstance(

458 func_or_path,

459 (str, Path)

460 ):

461 try:

462 _validate_gz_file(

463 func_or_path,

464 resolved_file_size,

465 resolved_ratio,

466 resolved_uncompressed_size

467 )

468

469 return True

470

471 except GzValidationError:

472 # Direct invocation intentionally has a boolean API.

473 #

474 # Do not catch Exception here: programming/configuration errors

475 # must not be silently hidden.

476 return False

477

478 # ---------------------------------------------------------------

479 # Decorator mode

480 # ---------------------------------------------------------------

481

482 def decorator(f):

483 try:

484 sig = inspect.signature(f)

485

486 except (TypeError, ValueError) as e:

487 raise GzValidationError(

488 f"Unable to inspect function '{f.__name__}': {e}"

489 ) from e

490

491 params = list(sig.parameters.values())

492

493 if not params:

494 raise GzValidationError(

495 f"Decorator applied to '{f.__name__}', "

496 "but it has no arguments."

497 )

498

499 # -----------------------------------------------------------

500 # Determine target argument

501 # -----------------------------------------------------------

502

503 if (

504 isinstance(func_or_path, str)

505 and func_or_path in sig.parameters

506 ):

507 target_arg = func_or_path

508

509 else:

510 positional_params = [

511 p for p in params

512 if p.kind in (

513 inspect.Parameter.POSITIONAL_ONLY,

514 inspect.Parameter.POSITIONAL_OR_KEYWORD

515 )

516 ]

517

518 if not positional_params:

519 raise GzValidationError(

520 f"Decorator applied to '{f.__name__}', but it has "

521 "no positional argument available for the GZip path. "

522 "Specify the argument name explicitly."

523 )

524

525 target_arg = positional_params[0].name

526

527 target_parameter = sig.parameters[target_arg]

528

529 if target_parameter.kind in (

530 inspect.Parameter.VAR_POSITIONAL,

531 inspect.Parameter.VAR_KEYWORD

532 ):

533 raise GzValidationError(

534 f"Argument '{target_arg}' cannot contain the GZip path"

535 )

536

537 # -----------------------------------------------------------

538 # Wrapped function

539 # -----------------------------------------------------------

540

541 @wraps(f)

542 def wrapper(*args, **kwargs):

543 try:

544 bound_args = sig.bind(*args, **kwargs)

545 bound_args.apply_defaults()

546

547 except TypeError as e:

548 raise GzValidationError(

549 f"Invalid function call signature: {e}"

550 ) from e

551

552 p = bound_args.arguments.get(target_arg)

553

554 if p is None:

555 raise GzValidationError(

556 f"Missing required argument: {target_arg}"

557 )

558

559 if not isinstance(p, (str, Path, os.PathLike)):

560 raise GzValidationError(

561 f"Expected Path or str for {target_arg}, "

562 f"got {type(p).__name__}"

563 )

564

565 _validate_gz_file(

566 p,

567 resolved_file_size,

568 resolved_ratio,

569 resolved_uncompressed_size

570 )

571

572 return f(*args, **kwargs)

573

574 return wrapper

575

576 # ---------------------------------------------------------------

577 # Bare decorator usage: @validate_gz

578 # ---------------------------------------------------------------

579

580 if callable(func_or_path):

581 return decorator(func_or_path)

582

583 # ---------------------------------------------------------------

584 # Factory usage: @validate_gz(...)

585 # ---------------------------------------------------------------

586

587 return decorator

588

Coverage for src/fileaudit/gz_check.py: 86%

159 statements