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

1""" 

2License GPL3 

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

4FileAudit - TAR Security Checker 

5""" 

6import tarfile 

7import urllib.request 

8import urllib.error 

9from pathlib import Path 

10from urllib.parse import urlparse 

11from functools import wraps 

12import inspect 

13import os 

14import tempfile 

15 

16 

17# Global default fallbacks 

18DEFAULT_MAX_FILE_SIZE = 10 * 1024 * 1024 # 10 MB 

19DEFAULT_MAX_TAR_MEMBERS = 1000 

20DEFAULT_MAX_TOTAL_EXTRACTED_SIZE = 100 * 1024 * 1024 # 100 MB 

21DEFAULT_MAX_INDIVIDUAL_FILE_SIZE = 10 * 1024 * 1024 # 10 MB 

22DEFAULT_MAX_FILENAME_LENGTH = 255 

23DEFAULT_MAX_DIRECTORY_DEPTH = 50 

24 

25 

26class TarValidationError(Exception): 

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

28 

29 def __init__(self, message): 

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

31 self.original_message = str(message) 

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

33 super().__init__(full_message) 

34 

35 def __str__(self): 

36 return self.args[0] 

37 

38 

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

40 """ 

41 Validate a single TAR member for security issues. 

42  

43 Args: 

44 member: TarInfo object 

45 base_path: Path object for the extraction base directory 

46 max_individual_size: Maximum allowed size for individual files 

47 max_filename_length: Maximum allowed filename/path length 

48 max_depth: Maximum allowed directory depth 

49  

50 Raises: 

51 TarValidationError: If any security check fails 

52 """ 

53 

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

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

56 raise TarValidationError( 

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

58 ) 

59 

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

61 raise TarValidationError( 

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

63 ) 

64 

65 # 2. Filename/path length limits 

66 if len(member.name) > max_filename_length: 

67 raise TarValidationError( 

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

69 f"maximum of {max_filename_length} characters" 

70 ) 

71 

72 # 3. Path traversal protection 

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

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

75 try: 

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

77 member_path.relative_to(base_path.resolve()) 

78 except ValueError: 

79 raise TarValidationError( 

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

81 ) 

82 

83 # 4. Directory depth limits 

84 # Count path components 

85 path_parts = Path(member.name).parts 

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

87 if depth > max_depth: 87 ↛ 88line 87 didn't jump to line 88 because the condition on line 87 was never true

88 raise TarValidationError( 

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

90 f"maximum of {max_depth}" 

91 ) 

92 

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

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

95 raise TarValidationError( 

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

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

98 ) 

99 

100 

101def validate_tar(func_or_path=None, 

102 max_file_size=None, 

103 max_tar_members=None, 

104 max_total_extracted_size=None, 

105 max_individual_file_size=None, 

106 max_filename_length=None, 

107 max_directory_depth=None): 

108 """ 

109 Validate TAR files via decorator or direct invocation. 

110  

111 A TAR file validator that can operate in two modes: 

112  

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

114 passed as an argument before the function body runs. 

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

116 a boolean result. 

117  

118 Security checks performed: 

119 - File size limits 

120 - Tar member count limits 

121 - Tar total extracted size limits 

122 - Tar individual file size limits 

123 - Tar path traversal protection 

124 - Reject symlinks/hardlinks/devices/FIFOs 

125 - Filename/path length limits 

126 - Directory depth limits 

127  

128 Usage: 

129 @validate_tar 

130 @validate_tar() 

131 @validate_tar("custom_arg_name", max_file_size=5000) 

132 validate_tar("path/to/file.tar", max_tar_members=100) # CLI usage 

133  

134 Args: 

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

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

137 usage: ``@validate_tar``). 

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

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

140 target argument name in decorator mode. 

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

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

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

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

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

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

147 max_directory_depth (int): Maximum directory nesting depth. 

148  

149 Returns: 

150 Union[callable, bool, function]: 

151 * In decorator mode: the wrapped function. 

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

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

154  

155 Raises: 

156 TarValidationError: If validation fails in decorator mode. 

157 """ 

158 

159 # Resolve optional limits to defaults 

160 resolved_file_size = DEFAULT_MAX_FILE_SIZE if max_file_size is None else max_file_size 

161 resolved_members = DEFAULT_MAX_TAR_MEMBERS if max_tar_members is None else max_tar_members 

162 resolved_total_size = DEFAULT_MAX_TOTAL_EXTRACTED_SIZE if max_total_extracted_size is None else max_total_extracted_size 

163 resolved_individual_size = DEFAULT_MAX_INDIVIDUAL_FILE_SIZE if max_individual_file_size is None else max_individual_file_size 

164 resolved_filename_len = DEFAULT_MAX_FILENAME_LENGTH if max_filename_length is None else max_filename_length 

165 resolved_depth = DEFAULT_MAX_DIRECTORY_DEPTH if max_directory_depth is None else max_directory_depth 

166 

167 def _looks_like_file_path(s): 

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

169 if s.startswith(("http://", "https://", "ftp://", "file://")): 

170 return True 

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

172 return True 

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

174 return True 

175 if "." in s and not s.startswith("."): 

176 return True 

177 return False 

178 

179 is_decorator_mode = False 

180 

181 if func_or_path is None: 

182 is_decorator_mode = True 

183 elif callable(func_or_path): 

184 is_decorator_mode = True 

185 elif isinstance(func_or_path, str): 

186 is_decorator_mode = not _looks_like_file_path(func_or_path) 

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

188 is_decorator_mode = False 

189 

190 # Direct call / CLI mode 

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

192 try: 

193 _validate_tar_file( 

194 func_or_path, 

195 resolved_file_size, 

196 resolved_members, 

197 resolved_total_size, 

198 resolved_individual_size, 

199 resolved_filename_len, 

200 resolved_depth 

201 ) 

202 return True 

203 except Exception as e: 

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

205 return False 

206 

207 # Decorator mode 

208 def decorator(f): 

209 sig = inspect.signature(f) 

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

211 

212 if not params: 

213 raise TarValidationError( 

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

215 ) 

216 

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

218 

219 @wraps(f) 

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

221 try: 

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

223 bound_args.apply_defaults() 

224 except TypeError as e: 

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

226 

227 p = bound_args.arguments.get(target_arg) 

228 

229 if p is None: 

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

231 

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

233 _validate_tar_file( 

234 p, 

235 resolved_file_size, 

236 resolved_members, 

237 resolved_total_size, 

238 resolved_individual_size, 

239 resolved_filename_len, 

240 resolved_depth 

241 ) 

242 else: 

243 raise TarValidationError( 

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

245 ) 

246 

247 return f(*args, **kwargs) 

248 return wrapper 

249 

250 if callable(func_or_path): 

251 return decorator(func_or_path) 

252 

253 return decorator 

254 

255 

256def _validate_tar_file(path, max_file_size, max_tar_members, 

257 max_total_extracted_size, max_individual_file_size, 

258 max_filename_length, max_directory_depth): 

259 """ 

260 Internal validation function for TAR files. 

261  

262 Performs all security checks on a TAR file. 

263 """ 

264 path_str = str(path) 

265 parsed = urlparse(path_str) 

266 is_remote = bool(parsed.scheme) 

267 

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

269 raise TarValidationError( 

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

271 ) 

272 

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

274 try: 

275 if is_remote: 

276 req = urllib.request.Request(path_str) 

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

278 tar_data = response.read() 

279 if len(tar_data) > max_file_size: 

280 raise TarValidationError( 

281 f"Remote file size ({len(tar_data)} bytes) exceeds " 

282 f"maximum of {max_file_size} bytes" 

283 ) 

284 else: 

285 local_path = Path(path) 

286 if not local_path.exists(): 

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

288 if not local_path.is_file(): 288 ↛ 289line 288 didn't jump to line 289 because the condition on line 288 was never true

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

290 

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

292 raise TarValidationError( 

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

294 f"maximum of {max_file_size} bytes" 

295 ) 

296 

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

298 tar_data = f.read() 

299 

300 except urllib.error.HTTPError as e: 

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

302 except urllib.error.URLError as e: 

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

304 except Exception as e: 

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

306 

307 # Parse the TAR archive 

308 try: 

309 # Use a temporary file for large archives to avoid memory issues 

310 with tempfile.NamedTemporaryFile(delete=False) as tmp_file: 

311 tmp_file.write(tar_data) 

312 tmp_path = tmp_file.name 

313 

314 try: 

315 with tarfile.open(tmp_path, mode='r') as tar: 

316 members = tar.getmembers() 

317 

318 # Check member count 

319 if len(members) > max_tar_members: 

320 raise TarValidationError( 

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

322 f"maximum of {max_tar_members}" 

323 ) 

324 

325 # Create a temporary base path for path traversal checks 

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

327 

328 # Check total extracted size 

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

330 if total_size > max_total_extracted_size: 

331 raise TarValidationError( 

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

333 f"maximum of {max_total_extracted_size} bytes" 

334 ) 

335 

336 # Validate each member 

337 for member in members: 

338 _check_tar_member( 

339 member, 

340 temp_base, 

341 max_individual_file_size, 

342 max_filename_length, 

343 max_directory_depth 

344 ) 

345 

346 except tarfile.TarError as e: 

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

348 except Exception as e: 

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

350 finally: 

351 try: 

352 os.unlink(tmp_path) 

353 except FileNotFoundError: 

354 pass # Already removed — no data leakage. 

355 except OSError: 

356 # Permission denied or similar: the temp file may still exist. 

357 # If we're already handling an exception, don't mask it. 

358 # Otherwise, raise so the caller knows cleanup failed. 

359 if sys.exc_info()[0] is None: 

360 raise 

361 

362 

363 except Exception as e: 

364 raise TarValidationError(f"Failed to process TAR archive: {e}") from e 

365 

366