Coverage for src/fileaudit/xml_check.py: 66%

1""" 

2License GPL3 

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

4FileAudit - XML Security Checker with DDoS Protection 

5""" 

6 

7import inspect 

8from pathlib import Path 

9from functools import wraps 

10import xml.etree.ElementTree as ET 

11from xml.parsers.expat import ExpatError 

12import urllib.request 

13import urllib.error 

14from urllib.parse import urlparse 

15import re 

16import gzip 

17import io 

18 

19# Global default fallbacks 

20DEFAULT_MAX_DEPTH = 50 

21DEFAULT_MAX_FILE_SIZE = 10 * 1024 * 1024 # 10 MB 

22DEFAULT_MAX_ATTRIBUTES = 1000 

23DEFAULT_MAX_ELEMENTS = 10000 

24DEFAULT_MAX_TEXT_LENGTH = 100000 # 100KB per text node 

25DEFAULT_MAX_NAME_LENGTH = 100 

26 

27_DOCTYPE_RE = re.compile(r"<!DOCTYPE", re.IGNORECASE) 

28 

29HEAD_TIMEOUT = 10 

30DOWNLOAD_TIMEOUT = 30 

31 

32class FileValidationError(Exception): 

33 """Custom exception for XML validation failures in FileAudit.""" 

34 

35 def __init__(self, message): 

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

37 self.original_message = str(message) 

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

39 super().__init__(full_message) 

40 

41 def __str__(self): 

42 return self.args[0] 

43 

44 

45class XMLSecurityValidator: 

46 """XML security validator with comprehensive DDoS protection.""" 

47 

48 def __init__( 

49 self, 

50 max_depth=None, 

51 max_file_size=None, 

52 max_attributes=None, 

53 max_elements=None, 

54 max_text_length=None, 

55 max_name_length=None, 

56 ): 

57 self.max_depth = max_depth or DEFAULT_MAX_DEPTH 

58 self.max_file_size = max_file_size or DEFAULT_MAX_FILE_SIZE 

59 self.max_attributes = max_attributes or DEFAULT_MAX_ATTRIBUTES 

60 self.max_elements = max_elements or DEFAULT_MAX_ELEMENTS 

61 self.max_text_length = max_text_length or DEFAULT_MAX_TEXT_LENGTH 

62 self.max_name_length = max_name_length or DEFAULT_MAX_NAME_LENGTH 

63 

64 self.reset_counters() 

65 

66 def reset_counters(self): 

67 """Reset internal counters.""" 

68 self.element_count = 0 

69 

70 def secure_parse(self, xml_content): 

71 """Parse XML and validate it against security limits.""" 

72 self.reset_counters() 

73 

74 try: 

75 parser = ET.XMLParser() 

76 root = ET.fromstring(xml_content, parser=parser) 

77 

78 self._validate_tree(root) 

79 

80 return root 

81 

82 except (ExpatError, ET.ParseError) as e: 

83 raise FileValidationError(f"XML parsing error: {e}") from e 

84 except RecursionError as e: 

85 raise FileValidationError(f"XML nesting too deep: {e}") from e 

86 

87 def _validate_tree(self, root): 

88 """Validate an XML tree without recursion.""" 

89 

90 stack = [(root, 0)] 

91 

92 while stack: 

93 element, depth = stack.pop() 

94 

95 if depth > self.max_depth: 

96 raise FileValidationError( 

97 f"XML nesting depth exceeds {self.max_depth}" 

98 ) 

99 

100 self.element_count += 1 

101 

102 if self.element_count > self.max_elements: 102 ↛ 103line 102 didn't jump to line 103 because the condition on line 102 was never true

103 raise FileValidationError( 

104 f"XML contains more than {self.max_elements} elements" 

105 ) 

106 

107 tag = str(element.tag) 

108 

109 if len(tag) > self.max_name_length: 109 ↛ 110line 109 didn't jump to line 110 because the condition on line 109 was never true

110 raise FileValidationError( 

111 f"Element name too long: {tag[:50]}" 

112 ) 

113 

114 if len(element.attrib) > self.max_attributes: 114 ↛ 115line 114 didn't jump to line 115 because the condition on line 114 was never true

115 raise FileValidationError( 

116 f"Too many attributes on element '{tag}'" 

117 ) 

118 

119 for name, value in element.attrib.items(): 

120 if len(str(name)) > self.max_name_length: 120 ↛ 121line 120 didn't jump to line 121 because the condition on line 120 was never true

121 raise FileValidationError("Attribute name too long") 

122 

123 if len(str(value)) > self.max_text_length: 123 ↛ 124line 123 didn't jump to line 124 because the condition on line 123 was never true

124 raise FileValidationError("Attribute value too long") 

125 

126 if element.text is not None: 

127 if len(str(element.text)) > self.max_text_length: 127 ↛ 128line 127 didn't jump to line 128 because the condition on line 127 was never true

128 raise FileValidationError("Text node exceeds limit") 

129 

130 if element.tail is not None: 

131 if len(str(element.tail)) > self.max_text_length: 131 ↛ 132line 131 didn't jump to line 132 because the condition on line 131 was never true

132 raise FileValidationError("Tail text exceeds limit") 

133 

134 for child in reversed(element): 

135 stack.append((child, depth + 1)) 

136 

137def _reject_doctype(xml_content: str) -> None: 

138 """Reject XML containing a DOCTYPE declaration.""" 

139 if _DOCTYPE_RE.search(xml_content): 139 ↛ 140line 139 didn't jump to line 140 because the condition on line 139 was never true

140 raise FileValidationError( 

141 "DOCTYPE declarations are not allowed." 

142 ) 

143 

144 

145def _read_gzip_stream(stream, max_file_size): 

146 """ 

147 Read and decompress a gzip stream while limiting the decompressed size. 

148 """ 

149 chunks = [] 

150 total = 0 

151 

152 while True: 

153 chunk = stream.read(8192) 

154 if not chunk: 

155 break 

156 

157 total += len(chunk) 

158 

159 if total > max_file_size: 

160 raise FileValidationError( 

161 "Decompressed file exceeds maximum allowed size." 

162 ) 

163 

164 chunks.append(chunk) 

165 

166 return b"".join(chunks).decode("utf-8", errors="strict") 

167 

168def _validate_xml_file( 

169 path, 

170 max_depth, 

171 max_file_size, 

172 max_attributes, 

173 max_elements, 

174 max_text_length, 

175 max_name_length, 

176): 

177 """Secure XML validation with DDoS protection.""" 

178 

179 path_str = str(path) 

180 parsed = urlparse(path_str) 

181 is_remote = bool(parsed.scheme) 

182 

183 if is_remote and parsed.scheme.lower() != "https": 183 ↛ 184line 183 didn't jump to line 184 because the condition on line 183 was never true

184 raise FileValidationError( 

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

186 ) 

187 

188 if not is_remote: 188 ↛ 216line 188 didn't jump to line 216 because the condition on line 188 was always true

189 local_path = Path(path_str) 

190 

191 if not local_path.exists(): 

192 raise FileValidationError(f"File not found: {local_path}") 

193 

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

195 raise FileValidationError(f"Path is not a file: {local_path}") 

196 

197 try: 

198 file_size = local_path.stat().st_size 

199 

200 if file_size > max_file_size: 

201 raise FileValidationError( 

202 f"File size ({file_size} bytes) exceeds maximum " 

203 f"limit of {max_file_size} bytes." 

204 ) 

205 

206 except OSError as e: 

207 raise FileValidationError( 

208 f"Could not read file metadata: {e}" 

209 ) from e 

210 

211 else: 

212 # 

213 # HEAD is only an optimization. 

214 # Some servers reject it, so ignore failures. 

215 # 

216 try: 

217 req = urllib.request.Request(path_str, method="HEAD") 

218 

219 with urllib.request.urlopen(req, timeout=HEAD_TIMEOUT) as response: 

220 content_length = response.headers.get("Content-Length") 

221 

222 if content_length is not None: 

223 file_size = int(content_length) 

224 

225 if file_size > max_file_size: 

226 raise FileValidationError( 

227 f"Remote file size ({file_size} bytes) exceeds " 

228 f"maximum limit of {max_file_size} bytes." 

229 ) 

230 

231 except FileValidationError: 

232 raise 

233 

234 except urllib.error.HTTPError as exc: 

235 # Only fall back to GET when the server explicitly does not 

236 # support HEAD (405 or 501). For all other HTTP errors, 

237 # fail fast to avoid unnecessary follow-up requests and 

238 # reduce the SSRF / request-amplification attack surface. 

239 if exc.code not in (405, 501): 

240 raise FileValidationError( 

241 f"Remote file unreachable (HTTP {exc.code})" 

242 ) from exc 

243 

244 except urllib.error.URLError as exc: 

245 # DNS failures, connection refused, timeouts, etc. mean the 

246 # target is unreachable. Do not proceed to GET to prevent 

247 # using this validator as an SSRF probe or DoS amplifier 

248 # against internal services. 

249 raise FileValidationError( 

250 f"Could not reach remote file: {exc.reason}" 

251 ) from exc 

252 

253 # 

254 # Read file contents 

255 # 

256 try: 

257 if is_remote: 257 ↛ 258line 257 didn't jump to line 258 because the condition on line 257 was never true

258 req = urllib.request.Request(path_str) 

259 

260 with urllib.request.urlopen(req, timeout=DOWNLOAD_TIMEOUT) as response: 

261 

262 raw_bytes = response.read(max_file_size + 1) 

263 

264 if len(raw_bytes) > max_file_size: 

265 raise FileValidationError( 

266 f"Remote file exceeds maximum size " 

267 f"({max_file_size} bytes)." 

268 ) 

269 

270 content_encoding = response.headers.get( 

271 "Content-Encoding", "" 

272 ).lower() 

273 

274 if "gzip" in content_encoding: 

275 try: 

276 with gzip.GzipFile( 

277 fileobj=io.BytesIO(raw_bytes) 

278 ) as gz: 

279 content = _read_gzip_stream( 

280 gz, 

281 max_file_size, 

282 ) 

283 

284 except ( 

285 OSError, 

286 EOFError, 

287 gzip.BadGzipFile, 

288 ) as e: 

289 raise FileValidationError( 

290 f"Gzip decompression failed: {e}" 

291 ) from e 

292 

293 else: 

294 content = raw_bytes.decode( 

295 "utf-8", 

296 errors="strict", 

297 ) 

298 

299 else: 

300 with local_path.open("rb") as f: 

301 header = f.read(2) 

302 

303 if header == b"\x1f\x8b": 303 ↛ 304line 303 didn't jump to line 304 because the condition on line 303 was never true

304 try: 

305 with gzip.open(local_path, "rb") as gz: #NOSEC --This is a security check(er)!! 

306 content = _read_gzip_stream( 

307 gz, 

308 max_file_size, 

309 ) 

310 

311 except ( 

312 OSError, 

313 EOFError, 

314 gzip.BadGzipFile, 

315 ) as e: 

316 raise FileValidationError( 

317 f"Gzip decompression failed: {e}" 

318 ) from e 

319 

320 else: 

321 content = local_path.read_text( 

322 encoding="utf-8", 

323 errors="strict", 

324 ) 

325 

326 except UnicodeDecodeError as e: 

327 raise FileValidationError( 

328 f"File is not valid UTF-8: {e}" 

329 ) from e 

330 

331 except OSError as e: 

332 raise FileValidationError( 

333 f"Failed to read file: {e}" 

334 ) from e 

335 

336 # 

337 # Reject DTD/DOCTYPE 

338 # 

339 _reject_doctype(content) 

340 

341 # 

342 # Validate XML structure 

343 # 

344 validator = XMLSecurityValidator( 

345 max_depth=max_depth, 

346 max_file_size=max_file_size, 

347 max_attributes=max_attributes, 

348 max_elements=max_elements, 

349 max_text_length=max_text_length, 

350 max_name_length=max_name_length, 

351 ) 

352 

353 try: 

354 validator.secure_parse(content) 

355 

356 except FileValidationError: 

357 raise 

358 

359 except ( 

360 ET.ParseError, 

361 ExpatError, 

362 RecursionError, 

363 ) as e: 

364 raise FileValidationError( 

365 f"XML validation failed: {e}" 

366 ) from e 

367 

368 

369def validate_xml( 

370 func_or_path=None, 

371 max_depth=None, 

372 max_file_size=None, 

373 max_attributes=None, 

374 max_elements=None, 

375 max_text_length=None, 

376 max_name_length=None, 

377): 

378 """Validate XML files via decorator or direct invocation. 

379 

380 An XML file validator that can operate in two modes: 

381 

382 1. **Decorator mode** — wraps a function to validate an XML file path 

383 passed as an argument before the function body runs. 

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

385 a boolean result. 

386 

387 Usage: 

388 @validate_xml 

389 @validate_xml() 

390 @validate_xml("custom_arg_name", max_depth=50) 

391 @validate_xml(max_file_size=5000) 

392 validate_xml("path/to/file.xml", max_depth=10) # CLI / direct call usage 

393 

394 Args: 

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

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

397 usage: ``@validate_xml``). 

398 * If a **str** or **Path** that looks like a file path or URL: 

399 the file path to validate (direct call usage). 

400 * If a **str** that is a valid Python identifier (not a path): 

401 treated as the target argument name to inspect in decorator 

402 mode (e.g., ``@validate_xml("config_path")``). 

403 * If **None**: returns a decorator factory 

404 (``@validate_xml()`` or ``@validate_xml(max_depth=50)``). 

405 max_depth (int or None): Maximum allowed XML nesting depth. 

406 Falls back to ``DEFAULT_MAX_DEPTH`` if omitted. 

407 max_file_size (int or None): Maximum allowed file size in bytes. 

408 Falls back to ``DEFAULT_MAX_FILE_SIZE`` if omitted. 

409 max_attributes (int or None): Maximum number of attributes permitted 

410 per XML element. Falls back to ``DEFAULT_MAX_ATTRIBUTES`` if omitted. 

411 max_elements (int or None): Maximum number of XML elements allowed 

412 in the document. Falls back to ``DEFAULT_MAX_ELEMENTS`` if omitted. 

413 max_text_length (int or None): Maximum permitted length of text or 

414 attribute values. Falls back to ``DEFAULT_MAX_TEXT_LENGTH`` if omitted. 

415 max_name_length (int or None): Maximum permitted length of element and 

416 attribute names. Falls back to ``DEFAULT_MAX_NAME_LENGTH`` if omitted. 

417 

418 Returns: 

419 Union[callable, bool, function]: 

420 * In decorator mode: the wrapped function. 

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

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

423 

424 Raises: 

425 FileValidationError: If validation fails in decorator mode, or if 

426 the decorated function has no arguments, the target argument is 

427 missing, or the argument type is not ``str`` or ``Path``. 

428 

429 Examples: 

430 Bare decorator (validates the first argument):: 

431 

432 @validate_xml 

433 def process_data(file_path): 

434 ... 

435 

436 Decorator with custom limits:: 

437 

438 @validate_xml(max_depth=50, max_file_size=5000) 

439 def process_data(file_path): 

440 ... 

441 

442 Decorator targeting a specific argument by name:: 

443 

444 @validate_xml("config_path", max_depth=10) 

445 def process_data(config_path, other_arg): 

446 ... 

447 

448 Direct call / CLI usage:: 

449 

450 result = validate_xml("path/to/file.xml", max_depth=10) 

451 # Returns True on success, False on failure. 

452 """ 

453 resolved_depth = DEFAULT_MAX_DEPTH if max_depth is None else max_depth 

454 resolved_size = ( 

455 DEFAULT_MAX_FILE_SIZE if max_file_size is None else max_file_size 

456 ) 

457 resolved_attributes = ( 

458 DEFAULT_MAX_ATTRIBUTES 

459 if max_attributes is None 

460 else max_attributes 

461 ) 

462 resolved_elements = ( 

463 DEFAULT_MAX_ELEMENTS 

464 if max_elements is None 

465 else max_elements 

466 ) 

467 resolved_text = ( 

468 DEFAULT_MAX_TEXT_LENGTH 

469 if max_text_length is None 

470 else max_text_length 

471 ) 

472 resolved_name = ( 

473 DEFAULT_MAX_NAME_LENGTH 

474 if max_name_length is None 

475 else max_name_length 

476 ) 

477 

478 def _looks_like_file_path(s: str) -> bool: 

479 p = Path(s) 

480 

481 return ( 

482 p.is_absolute() 

483 or bool(p.suffix) 

484 or "/" in s 

485 or "\\" in s 

486 or s.startswith(("http://", "https://")) 

487 ) 

488 

489 # 

490 # Direct invocation 

491 # 

492 if isinstance(func_or_path, (str, Path)) and _looks_like_file_path( 

493 str(func_or_path) 

494 ): 

495 try: 

496 _validate_xml_file( 

497 func_or_path, 

498 max_depth=resolved_depth, 

499 max_file_size=resolved_size, 

500 max_attributes=resolved_attributes, 

501 max_elements=resolved_elements, 

502 max_text_length=resolved_text, 

503 max_name_length=resolved_name, 

504 ) 

505 return True 

506 except FileValidationError: 

507 return False 

508 

509 # 

510 # Decorator 

511 # 

512 def decorator(func): 

513 sig = inspect.signature(func) 

514 params = list(sig.parameters) 

515 

516 if not params: 

517 raise FileValidationError( 

518 f"Decorator applied to '{func.__name__}', " 

519 "but the function has no parameters." 

520 ) 

521 

522 target_arg = ( 

523 func_or_path 

524 if isinstance(func_or_path, str) 

525 and func_or_path in params 

526 else params[0] 

527 ) 

528 

529 @wraps(func) 

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

531 try: 

532 bound = sig.bind(*args, **kwargs) 

533 bound.apply_defaults() 

534 except TypeError as e: 

535 raise FileValidationError( 

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

537 ) from e 

538 

539 value = bound.arguments.get(target_arg) 

540 

541 if value is None: 541 ↛ 542line 541 didn't jump to line 542 because the condition on line 541 was never true

542 raise FileValidationError( 

543 f"Missing required argument '{target_arg}'." 

544 ) 

545 

546 if not isinstance(value, (str, Path)): 

547 raise FileValidationError( 

548 f"Expected a file path (str or pathlib.Path) " 

549 f"for '{target_arg}', got " 

550 f"{type(value).__name__}." 

551 ) 

552 

553 _validate_xml_file( 

554 value, 

555 max_depth=resolved_depth, 

556 max_file_size=resolved_size, 

557 max_attributes=resolved_attributes, 

558 max_elements=resolved_elements, 

559 max_text_length=resolved_text, 

560 max_name_length=resolved_name, 

561 ) 

562 

563 return func(*args, **kwargs) 

564 

565 return wrapper 

566 

567 if callable(func_or_path): 

568 return decorator(func_or_path) 

569 

570 return decorator 

571 

572