Coverage for src/fileaudit/targz_check.py: 93%

1""" 

2License GPL3 

3(C) 2026 Created by Maikel Mardjan - https://nocomplexity.com/ 

4FileAudit - TAR Security Checker 

5""" 

6import gzip 

7import tarfile 

8import io 

9import urllib.request 

10import urllib.error 

11from pathlib import Path 

12from urllib.parse import urlparse 

13from functools import wraps 

14import inspect 

15 

16 

17# Global default fallbacks 

18DEFAULT_MAX_FILE_SIZE = 10 * 1024 * 1024 # 10 MB 

19DEFAULT_MAX_UNCOMPRESSED_RATIO = 100 # 100:1 ratio 

20DEFAULT_MAX_TAR_MEMBERS = 1000 

21DEFAULT_MAX_TOTAL_EXTRACTED_SIZE = 100 * 1024 * 1024 # 100 MB 

22DEFAULT_MAX_INDIVIDUAL_FILE_SIZE = 10 * 1024 * 1024 # 10 MB 

23DEFAULT_MAX_FILENAME_LENGTH = 255 

24DEFAULT_MAX_DIRECTORY_DEPTH = 50 

25 

26 

27class TarValidationError(Exception): 

28 """Custom exception for TAR validation failures in FileAudit.""" 

29 

30 def __init__(self, message): 

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

32 self.original_message = str(message) 

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

34 super().__init__(full_message) 

35 

36 def __str__(self): 

37 return self.args[0] 

38 

39 

40def _check_tar_member(member, base_path, max_individual_size, max_filename_length, max_depth): 

41 """ 

42 Validate a single TAR member for security issues. 

43  

44 Args: 

45 member: TarInfo object 

46 base_path: Path object for the extraction base directory 

47 max_individual_size: Maximum allowed size for individual files 

48 max_filename_length: Maximum allowed filename/path length 

49 max_depth: Maximum allowed directory depth 

50  

51 Raises: 

52 TarValidationError: If any security check fails 

53 """ 

54 

55 # 1. Reject symlinks, hardlinks, devices, and FIFOs 

56 if member.issym() or member.islnk(): 

57 raise TarValidationError( 

58 f"Rejected {member.name}: Symlinks and hardlinks are not allowed" 

59 ) 

60 

61 if member.isdev() or member.isfifo(): 

62 raise TarValidationError( 

63 f"Rejected {member.name}: Device files and FIFOs are not allowed" 

64 ) 

65 

66 # 2. Filename/path length limits 

67 if len(member.name) > max_filename_length: 

68 raise TarValidationError( 

69 f"Rejected {member.name}: Path length ({len(member.name)}) exceeds " 

70 f"maximum of {max_filename_length} characters" 

71 ) 

72 

73 # 3. Path traversal protection 

74 # Resolve the full path and check if it's within the extraction directory 

75 member_path = (base_path / member.name).resolve() 

76 try: 

77 # Check if the resolved path is still within the base directory 

78 member_path.relative_to(base_path.resolve()) 

79 except ValueError: 

80 raise TarValidationError( 

81 f"Rejected {member.name}: Path traversal attempt detected" 

82 ) 

83 

84 # 4. Directory depth limits 

85 # Count path components 

86 path_parts = Path(member.name).parts 

87 depth = len(path_parts) if member.name else 0 

88 if depth > max_depth: 

89 raise TarValidationError( 

90 f"Rejected {member.name}: Directory depth ({depth}) exceeds " 

91 f"maximum of {max_depth}" 

92 ) 

93 

94 # 5. Individual file size limits (for regular files only) 

95 if member.isfile() and member.size > max_individual_size: 

96 raise TarValidationError( 

97 f"Rejected {member.name}: File size ({member.size} bytes) exceeds " 

98 f"maximum individual file size of {max_individual_size} bytes" 

99 ) 

100 

101 

102def validate_tar_gz(func_or_path=None, 

103 max_file_size=None, 

104 max_uncompressed_ratio=None, 

105 max_tar_members=None, 

106 max_total_extracted_size=None, 

107 max_individual_file_size=None, 

108 max_filename_length=None, 

109 max_directory_depth=None): 

110 """ 

111 Validate TAR.GZ files via decorator or direct invocation. 

112  

113 A TAR.GZ file validator that can operate in two modes: 

114  

115 1. **Decorator mode** — wraps a function to validate a TAR.GZ file path 

116 passed as an argument before the function body runs. 

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

118 a boolean result. 

119  

120 Security checks performed: 

121 - File size limits 

122 - GZip decompression ratio limits 

123 - Tar member count limits 

124 - Tar total extracted size limits 

125 - Tar individual file size limits 

126 - Tar path traversal protection 

127 - Reject symlinks/hardlinks/devices/FIFOs 

128 - Filename/path length limits 

129 - Directory depth limits 

130  

131 Usage: 

132 @validate_tar_gz 

133 @validate_tar_gz() 

134 @validate_tar_gz("custom_arg_name", max_file_size=5000) 

135 validate_tar_gz("path/to/file.tar.gz", max_tar_members=100) # CLI usage 

136  

137 Args: 

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

139 * If a **callable**: the function to decorate (bare decorator 

140 usage: ``@validate_tar_gz``). 

141 * If a **str** or **Path**: the file path to validate (direct call). 

142 * If a **str** that is a valid Python identifier: treated as the 

143 target argument name in decorator mode. 

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

145 max_file_size (int): Maximum allowed compressed file size in bytes. 

146 max_uncompressed_ratio (int): Maximum GZip decompression ratio. 

147 max_tar_members (int): Maximum number of files/directories in TAR. 

148 max_total_extracted_size (int): Maximum total extracted size in bytes. 

149 max_individual_file_size (int): Maximum size per extracted file. 

150 max_filename_length (int): Maximum filename/path length. 

151 max_directory_depth (int): Maximum directory nesting depth. 

152  

153 Returns: 

154 Union[callable, bool, function]: 

155 * In decorator mode: the wrapped function. 

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

157 ``False`` if it fails (errors are printed to stdout). 

158  

159 Raises: 

160 TarValidationError: If validation fails in decorator mode. 

161 """ 

162 

163 # Resolve optional limits to defaults 

164 resolved_file_size = DEFAULT_MAX_FILE_SIZE if max_file_size is None else max_file_size 

165 resolved_ratio = DEFAULT_MAX_UNCOMPRESSED_RATIO if max_uncompressed_ratio is None else max_uncompressed_ratio 

166 resolved_members = DEFAULT_MAX_TAR_MEMBERS if max_tar_members is None else max_tar_members 

167 resolved_total_size = DEFAULT_MAX_TOTAL_EXTRACTED_SIZE if max_total_extracted_size is None else max_total_extracted_size 

168 resolved_individual_size = DEFAULT_MAX_INDIVIDUAL_FILE_SIZE if max_individual_file_size is None else max_individual_file_size 

169 resolved_filename_len = DEFAULT_MAX_FILENAME_LENGTH if max_filename_length is None else max_filename_length 

170 resolved_depth = DEFAULT_MAX_DIRECTORY_DEPTH if max_directory_depth is None else max_directory_depth 

171 

172 def _looks_like_file_path(s): 

173 """Heuristic: does this string look like a file path or URL?""" 

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

175 return True 

176 if s.startswith(("/", "\\")): 

177 return True 

178 if "/" in s or "\\" in s: 178 ↛ 179line 178 didn't jump to line 179 because the condition on line 178 was never true

179 return True 

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

181 return True 

182 return False 

183 

184 is_decorator_mode = False 

185 

186 if func_or_path is None: 

187 is_decorator_mode = True 

188 elif callable(func_or_path): 

189 is_decorator_mode = True 

190 elif isinstance(func_or_path, str): 

191 is_decorator_mode = not _looks_like_file_path(func_or_path) 

192 elif isinstance(func_or_path, Path): 192 ↛ 196line 192 didn't jump to line 196 because the condition on line 192 was always true

193 is_decorator_mode = False 

194 

195 # Direct call / CLI mode 

196 if not is_decorator_mode and isinstance(func_or_path, (str, Path)): 

197 try: 

198 _validate_tar_gz_file( 

199 func_or_path, 

200 resolved_file_size, 

201 resolved_ratio, 

202 resolved_members, 

203 resolved_total_size, 

204 resolved_individual_size, 

205 resolved_filename_len, 

206 resolved_depth 

207 ) 

208 return True 

209 except Exception as e: 

210 print(f"Exception: {e}") 

211 return False 

212 

213 # Decorator mode 

214 def decorator(f): 

215 sig = inspect.signature(f) 

216 params = list(sig.parameters.keys()) 

217 

218 if not params: 

219 raise TarValidationError( 

220 f"Decorator applied to '{f.__name__}', but it has no arguments." 

221 ) 

222 

223 target_arg = func_or_path if isinstance(func_or_path, str) and func_or_path in params else params[0] 

224 

225 @wraps(f) 

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

227 try: 

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

229 bound_args.apply_defaults() 

230 except TypeError as e: 

231 raise TarValidationError(f"Invalid function call signature: {e}") 

232 

233 p = bound_args.arguments.get(target_arg) 

234 

235 if p is None: 235 ↛ 236line 235 didn't jump to line 236 because the condition on line 235 was never true

236 raise TarValidationError(f"Missing required argument: {target_arg}") 

237 

238 if isinstance(p, (str, Path)): 

239 _validate_tar_gz_file( 

240 p, 

241 resolved_file_size, 

242 resolved_ratio, 

243 resolved_members, 

244 resolved_total_size, 

245 resolved_individual_size, 

246 resolved_filename_len, 

247 resolved_depth 

248 ) 

249 else: 

250 raise TarValidationError( 

251 f"Expected Path or str for {target_arg}, got {type(p).__name__}" 

252 ) 

253 

254 return f(*args, **kwargs) 

255 return wrapper 

256 

257 if callable(func_or_path): 

258 return decorator(func_or_path) 

259 

260 return decorator 

261 

262 

263def _validate_tar_gz_file(path, max_file_size, max_uncompressed_ratio, 

264 max_tar_members, max_total_extracted_size, 

265 max_individual_file_size, max_filename_length, 

266 max_directory_depth): 

267 """ 

268 Internal validation function for TAR.GZ files. 

269  

270 Performs all security checks on a TAR.GZ file. 

271 """ 

272 path_str = str(path) 

273 parsed = urlparse(path_str) 

274 is_remote = bool(parsed.scheme) 

275 

276 if is_remote and parsed.scheme != "https": 

277 raise TarValidationError( 

278 f"Unsupported URL scheme '{parsed.scheme}': only 'https' is allowed." 

279 ) 

280 

281 # Get file content as bytes (either local or remote) 

282 try: 

283 if is_remote: 

284 req = urllib.request.Request(path_str) 

285 with urllib.request.urlopen(req, timeout=30) as response: 

286 compressed_data = response.read() 

287 if len(compressed_data) > max_file_size: 

288 raise TarValidationError( 

289 f"Remote file size ({len(compressed_data)} bytes) exceeds " 

290 f"maximum of {max_file_size} bytes" 

291 ) 

292 else: 

293 local_path = Path(path) 

294 if not local_path.exists(): 

295 raise TarValidationError(f"File not found: {local_path}") 

296 if not local_path.is_file(): 

297 raise TarValidationError(f"Path is not a file: {local_path}") 

298 

299 if local_path.stat().st_size > max_file_size: 

300 raise TarValidationError( 

301 f"File size ({local_path.stat().st_size} bytes) exceeds " 

302 f"maximum of {max_file_size} bytes" 

303 ) 

304 

305 with local_path.open('rb') as f: 

306 compressed_data = f.read() 

307 

308 except urllib.error.HTTPError as e: 

309 raise TarValidationError(f"Remote file unreachable (HTTP {e.code})") from e 

310 except urllib.error.URLError as e: 

311 raise TarValidationError(f"Could not reach remote file: {e.reason}") from e 

312 except Exception as e: 

313 raise TarValidationError(f"Failed to read file: {e}") from e 

314 

315 # Decompress and check ratio 

316 try: 

317 with gzip.GzipFile(fileobj=io.BytesIO(compressed_data)) as gz: 

318 decompressed_data = gz.read() 

319 original_size = len(compressed_data) 

320 decompressed_size = len(decompressed_data) 

321 

322 # Check decompression ratio 

323 if original_size > 0: 323 ↛ 337line 323 didn't jump to line 337

324 ratio = decompressed_size / original_size 

325 if ratio > max_uncompressed_ratio: 

326 raise TarValidationError( 

327 f"Decompression ratio ({ratio:.2f}x) exceeds maximum " 

328 f"of {max_uncompressed_ratio}x (zip bomb protection)" 

329 ) 

330 

331 except gzip.BadGzipFile as e: 

332 raise TarValidationError(f"Invalid GZip format: {e}") from e 

333 except Exception as e: 

334 raise TarValidationError(f"GZip decompression failed: {e}") from e 

335 

336 # Now parse the TAR archive from the decompressed data 

337 try: 

338 with tarfile.open(fileobj=io.BytesIO(decompressed_data), mode='r') as tar: 

339 members = tar.getmembers() 

340 

341 # Check member count 

342 if len(members) > max_tar_members: 342 ↛ 343line 342 didn't jump to line 343 because the condition on line 342 was never true

343 raise TarValidationError( 

344 f"Archive contains {len(members)} members, exceeding " 

345 f"maximum of {max_tar_members}" 

346 ) 

347 

348 # Create a temporary base path for path traversal checks 

349 temp_base = Path("/tmp/tar_validation_extract") 

350 

351 # Check total extracted size 

352 total_size = sum(member.size for member in members if member.isfile()) 

353 if total_size > max_total_extracted_size: 353 ↛ 354line 353 didn't jump to line 354 because the condition on line 353 was never true

354 raise TarValidationError( 

355 f"Total extracted size ({total_size} bytes) exceeds " 

356 f"maximum of {max_total_extracted_size} bytes" 

357 ) 

358 

359 # Validate each member 

360 for member in members: 

361 _check_tar_member( 

362 member, 

363 temp_base, 

364 max_individual_file_size, 

365 max_filename_length, 

366 max_directory_depth 

367 ) 

368 

369 except tarfile.TarError as e: 

370 raise TarValidationError(f"Invalid TAR archive: {e}") from e 

371 except Exception as e: 

372 raise TarValidationError(f"TAR validation failed: {e}") from e 

373