Coverage for src/fileaudit/csv_check.py: 79%

1""" 

2License GPL3 

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

4FileAudit - CSV Security Checker 

5 

6Direct / CLI usage: 

7 validate_csv("file.csv") 

8""" 

9 

10import csv 

11import inspect 

12import io 

13import os 

14import urllib.error 

15import urllib.request 

16from functools import wraps 

17from pathlib import Path 

18from urllib.parse import urlparse 

19 

20 

21# --------------------------------------------------------------------------- 

22# Defaults 

23# --------------------------------------------------------------------------- 

24 

25DEFAULT_MAX_FILE_SIZE = 10 * 1024 * 1024 # 10 MB 

26DEFAULT_MAX_ROWS = 100_000 

27DEFAULT_MAX_COLUMNS = 1_000 

28DEFAULT_MAX_FIELD_SIZE = 1 * 1024 * 1024 # 1 MB 

29DEFAULT_MAX_TOTAL_FIELDS = 10_000_000 

30DEFAULT_MAX_ROW_SIZE = 5 * 1024 * 1024 # 5 MB 

31DEFAULT_MAX_FILENAME_LENGTH = 255 

32 

33# Spreadsheet formula injection prefixes. 

34DEFAULT_DANGEROUS_FORMULA_PREFIXES = ( 

35 "=", 

36 "+", 

37 "-", 

38 "@", 

39) 

40 

41 

42# --------------------------------------------------------------------------- 

43# Exceptions 

44# --------------------------------------------------------------------------- 

45 

46class CsvValidationError(Exception): 

47 """Custom exception for CSV validation failures in FileAudit.""" 

48 

49 def __init__(self, message): 

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

51 self.original_message = str(message) 

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

53 super().__init__(full_message) 

54 

55 def __str__(self): 

56 return self.args[0] 

57 

58 

59# --------------------------------------------------------------------------- 

60# Individual CSV checks 

61# --------------------------------------------------------------------------- 

62 

63def _check_csv_field( 

64 value, 

65 row_number, 

66 column_number, 

67 max_field_size, 

68 reject_formula_injection, 

69 reject_control_characters, 

70): 

71 """Validate one CSV field for common security issues.""" 

72 

73 if not isinstance(value, str): 73 ↛ 74line 73 didn't jump to line 74 because the condition on line 73 was never true

74 raise CsvValidationError( 

75 f"Invalid field type at row {row_number}, " 

76 f"column {column_number}" 

77 ) 

78 

79 # NUL bytes and other problematic control characters can cause problems 

80 # when CSV data is passed to downstream applications. 

81 if reject_control_characters: 

82 for char in value: 

83 codepoint = ord(char) 

84 

85 # Permit normal tab/newline/carriage return semantics inside 

86 # quoted CSV fields, but reject other C0 control characters. 

87 if codepoint < 32 and codepoint not in (9, 10, 13): 

88 raise CsvValidationError( 

89 f"Rejected field at row {row_number}, " 

90 f"column {column_number}: " 

91 f"contains control character U+{codepoint:04X}" 

92 ) 

93 

94 # NUL deserves an explicit security error. 

95 if codepoint == 0: 95 ↛ 96line 95 didn't jump to line 96 because the condition on line 95 was never true

96 raise CsvValidationError( 

97 f"Rejected field at row {row_number}, " 

98 f"column {column_number}: contains NUL byte" 

99 ) 

100 

101 if len(value.encode("utf-8")) > max_field_size: 101 ↛ 102line 101 didn't jump to line 102 because the condition on line 101 was never true

102 raise CsvValidationError( 

103 f"Rejected field at row {row_number}, " 

104 f"column {column_number}: field size exceeds " 

105 f"maximum of {max_field_size} bytes" 

106 ) 

107 

108 # CSV itself does not execute formulas. The danger occurs when the CSV 

109 # is subsequently opened in spreadsheet software. 

110 if reject_formula_injection: 

111 stripped = value.lstrip() 

112 

113 if stripped.startswith(DEFAULT_DANGEROUS_FORMULA_PREFIXES): 

114 raise CsvValidationError( 

115 f"Rejected field at row {row_number}, " 

116 f"column {column_number}: possible spreadsheet " 

117 f"formula injection" 

118 ) 

119 

120 

121def _check_csv_row( 

122 row, 

123 row_number, 

124 max_columns, 

125 max_field_size, 

126 max_row_size, 

127 reject_formula_injection, 

128 reject_control_characters, 

129): 

130 """Validate a single parsed CSV row.""" 

131 

132 if len(row) > max_columns: 

133 raise CsvValidationError( 

134 f"Rejected row {row_number}: contains {len(row)} columns, " 

135 f"exceeding maximum of {max_columns}" 

136 ) 

137 

138 # Calculate the approximate serialized row size. 

139 row_size = sum( 

140 len(str(value).encode("utf-8")) 

141 for value in row 

142 ) 

143 

144 if row_size > max_row_size: 

145 raise CsvValidationError( 

146 f"Rejected row {row_number}: row size ({row_size} bytes) " 

147 f"exceeds maximum of {max_row_size} bytes" 

148 ) 

149 

150 for column_number, value in enumerate(row, start=1): 

151 _check_csv_field( 

152 value, 

153 row_number, 

154 column_number, 

155 max_field_size, 

156 reject_formula_injection, 

157 reject_control_characters, 

158 ) 

159 

160 

161# --------------------------------------------------------------------------- 

162# Internal file validation 

163# --------------------------------------------------------------------------- 

164 

165def _validate_csv_file( 

166 path, 

167 max_file_size, 

168 max_rows, 

169 max_columns, 

170 max_field_size, 

171 max_total_fields, 

172 max_row_size, 

173 max_filename_length, 

174 reject_formula_injection, 

175 reject_control_characters, 

176 encoding, 

177 dialect, 

178): 

179 """ 

180 Internal CSV validation function. 

181 

182 Validates local files and HTTPS URLs without executing or importing 

183 anything contained in the CSV. 

184 

185 Security measures: 

186 - Only HTTPS is accepted for remote URLs. 

187 - File size is enforced both via Content-Length (when present) and 

188 via a hard streaming limit on the actual bytes received. 

189 - Negative / non-numeric Content-Length values are rejected. 

190 - Local files are size-checked before being fully read. 

191 - Filename length and .csv extension are validated (local paths). 

192 - Decoding is strict; unexpected UTF-8 BOMs are rejected when 

193 encoding is plain 'utf-8'. 

194 - CSV parser limits (rows, columns, field size, total fields, row size) 

195 are enforced. 

196 - Optional formula-injection and control-character rejection. 

197 """ 

198 path_str = str(path) 

199 parsed = urlparse(path_str) 

200 is_remote = bool(parsed.scheme) 

201 

202 # --------------------------------------------------------------- 

203 # Remote URL validation 

204 # --------------------------------------------------------------- 

205 if is_remote and parsed.scheme != "https": 205 ↛ 206line 205 didn't jump to line 206 because the condition on line 205 was never true

206 raise CsvValidationError( 

207 f"Unsupported URL scheme '{parsed.scheme}': " 

208 "only 'https' is allowed." 

209 ) 

210 

211 # Optional: also require a .csv suffix on remote URLs 

212 if is_remote: 

213 url_path = parsed.path or "" 

214 if not url_path.lower().endswith(".csv"): 214 ↛ 215line 214 didn't jump to line 215 because the condition on line 214 was never true

215 raise CsvValidationError( 

216 f"Remote URL does not point to a .csv file: {path_str}" 

217 ) 

218 

219 # --------------------------------------------------------------- 

220 # Obtain CSV data 

221 # --------------------------------------------------------------- 

222 try: 

223 if is_remote: 

224 request = urllib.request.Request( 

225 path_str, 

226 headers={"User-Agent": "FileAudit-CSVValidator/1.0"}, 

227 method="GET", 

228 ) 

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

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

231 

232 if content_length is not None: 

233 try: 

234 declared_size = int(content_length) 

235 if declared_size < 0: 

236 raise CsvValidationError( 

237 "Negative Content-Length is invalid" 

238 ) 

239 if declared_size > max_file_size: 

240 raise CsvValidationError( 

241 f"Remote file size ({declared_size} bytes) " 

242 f"exceeds maximum of {max_file_size} bytes" 

243 ) 

244 except ValueError: 

245 raise CsvValidationError( 

246 f"Remote server returned invalid Content-Length: " 

247 f"{content_length!r}" 

248 ) 

249 

250 # Never trust Content-Length alone – stream with a hard ceiling 

251 chunks = [] 

252 total_size = 0 

253 while True: 

254 remaining = max_file_size - total_size + 1 

255 if remaining <= 0: 

256 raise CsvValidationError( 

257 f"Remote file exceeds maximum size of " 

258 f"{max_file_size} bytes" 

259 ) 

260 chunk = response.read(min(64 * 1024, remaining)) 

261 if not chunk: 

262 break 

263 total_size += len(chunk) 

264 if total_size > max_file_size: 

265 raise CsvValidationError( 

266 f"Remote file exceeds maximum size of " 

267 f"{max_file_size} bytes" 

268 ) 

269 chunks.append(chunk) 

270 

271 csv_data = b"".join(chunks) 

272 

273 else: 

274 local_path = Path(path) 

275 

276 if not local_path.exists(): 

277 raise CsvValidationError(f"File not found: {local_path}") 

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

279 raise CsvValidationError(f"Path is not a file: {local_path}") 

280 

281 # Filename/path sanity check 

282 if len(local_path.name) > max_filename_length: 

283 raise CsvValidationError( 

284 f"Filename length ({len(local_path.name)}) exceeds " 

285 f"maximum of {max_filename_length} characters" 

286 ) 

287 

288 file_size = local_path.stat().st_size 

289 if file_size > max_file_size: 

290 raise CsvValidationError( 

291 f"File size ({file_size} bytes) exceeds " 

292 f"maximum of {max_file_size} bytes" 

293 ) 

294 

295 with local_path.open("rb") as file: 

296 csv_data = file.read(max_file_size + 1) 

297 if len(csv_data) > max_file_size: 297 ↛ 298line 297 didn't jump to line 298 because the condition on line 297 was never true

298 raise CsvValidationError( 

299 f"File size exceeds maximum of {max_file_size} bytes" 

300 ) 

301 

302 except urllib.error.HTTPError as exc: 

303 raise CsvValidationError( 

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

305 ) from exc 

306 except urllib.error.URLError as exc: 

307 raise CsvValidationError( 

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

309 ) from exc 

310 except CsvValidationError: 

311 raise 

312 except Exception as exc: 

313 raise CsvValidationError(f"Failed to read file: {exc}") from exc 

314 

315 # --------------------------------------------------------------- 

316 # Filename / extension check (local only – remote already checked) 

317 # --------------------------------------------------------------- 

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

319 suffix = Path(path).suffix.lower() 

320 if suffix != ".csv": 320 ↛ 321line 320 didn't jump to line 321 because the condition on line 320 was never true

321 raise CsvValidationError( 

322 f"Expected a .csv file, got '{suffix or '[no extension]'}'" 

323 ) 

324 

325 # --------------------------------------------------------------- 

326 # Decode content 

327 # --------------------------------------------------------------- 

328 try: 

329 text = csv_data.decode(encoding) 

330 except UnicodeDecodeError as exc: 

331 raise CsvValidationError( 

332 f"CSV is not valid {encoding} text: {exc}" 

333 ) from exc 

334 

335 # Reject unexpected UTF-8 BOM when using plain utf-8 

336 if encoding.lower().replace("_", "-") == "utf-8": 

337 if text.startswith("\ufeff"): 

338 raise CsvValidationError( 

339 "CSV contains a UTF-8 BOM; use encoding='utf-8-sig' " 

340 "if BOMs are intentionally supported" 

341 ) 

342 

343 # Optional early rejection of null bytes (cheap extra safety) 

344 if "\0" in text: 

345 raise CsvValidationError("Null byte (\\0) found in CSV content – rejected") 

346 

347 # --------------------------------------------------------------- 

348 # CSV parser configuration 

349 # --------------------------------------------------------------- 

350 # Note: csv.field_size_limit is process-global. 

351 # Save and restore the previous value to avoid side-effects. 

352 previous_limit = csv.field_size_limit() 

353 try: 

354 csv.field_size_limit(max_field_size) 

355 

356 reader = csv.reader( 

357 io.StringIO(text, newline=""), 

358 dialect=dialect, 

359 ) 

360 

361 row_count = 0 

362 total_fields = 0 

363 

364 for row_number, row in enumerate(reader, start=1): 

365 row_count += 1 

366 if row_count > max_rows: 

367 raise CsvValidationError( 

368 f"CSV contains more than {max_rows} rows" 

369 ) 

370 

371 total_fields += len(row) 

372 if total_fields > max_total_fields: 

373 raise CsvValidationError( 

374 f"CSV contains more than {max_total_fields} fields" 

375 ) 

376 

377 _check_csv_row( 

378 row=row, 

379 row_number=row_number, 

380 max_columns=max_columns, 

381 max_field_size=max_field_size, 

382 max_row_size=max_row_size, 

383 reject_formula_injection=reject_formula_injection, 

384 reject_control_characters=reject_control_characters, 

385 ) 

386 

387 except csv.Error as exc: 

388 raise CsvValidationError(f"Invalid CSV format: {exc}") from exc 

389 except CsvValidationError: 

390 raise 

391 except Exception as exc: 

392 raise CsvValidationError(f"CSV validation failed: {exc}") from exc 

393 finally: 

394 # Restore previous global limit 

395 csv.field_size_limit(previous_limit) 

396 

397 

398# --------------------------------------------------------------------------- 

399# Public API 

400# --------------------------------------------------------------------------- 

401 

402def validate_csv( 

403 func_or_path=None, 

404 max_file_size=None, 

405 max_rows=None, 

406 max_columns=None, 

407 max_field_size=None, 

408 max_total_fields=None, 

409 max_row_size=None, 

410 max_filename_length=None, 

411 reject_formula_injection=True, 

412 reject_control_characters=True, 

413 encoding="utf-8", 

414 dialect="excel", 

415): 

416 """ 

417 Validate CSV files via decorator or direct invocation. 

418 

419 Two modes are supported: 

420 

421 1. Decorator mode: 

422 

423 @validate_csv 

424 def process_csv(csv_path): 

425 ... 

426 

427 @validate_csv() 

428 def process_csv(csv_path): 

429 ... 

430 

431 @validate_csv("input_file") 

432 def process_csv(input_file): 

433 ... 

434 

435 2. Direct / CLI mode: 

436 

437 validate_csv("data.csv") 

438 

439 Direct invocation returns True when validation succeeds and False 

440 when validation fails. Validation failures are printed to stdout. 

441 

442 Security checks performed: 

443 

444 - Maximum file size 

445 - HTTPS-only remote files 

446 - Maximum number of rows 

447 - Maximum number of columns 

448 - Maximum individual field size 

449 - Maximum total number of fields 

450 - Maximum row size 

451 - Filename length 

452 - CSV syntax validation 

453 - Encoding validation 

454 - NUL/control-character rejection 

455 - Spreadsheet formula-injection protection 

456 

457 Args: 

458 func_or_path: 

459 Callable for bare decorator usage, str/Path for direct 

460 validation, str for the decorated argument name, or None. 

461 

462 max_file_size: 

463 Maximum CSV file size in bytes. 

464 

465 max_rows: 

466 Maximum number of CSV rows. 

467 

468 max_columns: 

469 Maximum number of columns in a row. 

470 

471 max_field_size: 

472 Maximum size of an individual field in bytes. 

473 

474 max_total_fields: 

475 Maximum total number of fields in the CSV. 

476 

477 max_row_size: 

478 Maximum approximate serialized row size in bytes. 

479 

480 max_filename_length: 

481 Maximum filename length. 

482 

483 reject_formula_injection: 

484 Reject fields beginning with =, +, -, or @ after whitespace. 

485 

486 reject_control_characters: 

487 Reject dangerous control characters. 

488 

489 encoding: 

490 Text encoding used to decode the CSV. 

491 

492 dialect: 

493 CSV dialect passed to csv.reader. 

494 

495 Returns: 

496 In decorator mode: 

497 The wrapped function. 

498 

499 In direct mode: 

500 True if validation succeeds, False otherwise. 

501 

502 Raises: 

503 CsvValidationError: 

504 If validation fails in decorator mode. 

505 """ 

506 

507 # --------------------------------------------------------------- 

508 # Resolve defaults 

509 # --------------------------------------------------------------- 

510 

511 resolved_file_size = ( 

512 DEFAULT_MAX_FILE_SIZE 

513 if max_file_size is None 

514 else max_file_size 

515 ) 

516 

517 resolved_rows = ( 

518 DEFAULT_MAX_ROWS 

519 if max_rows is None 

520 else max_rows 

521 ) 

522 

523 resolved_columns = ( 

524 DEFAULT_MAX_COLUMNS 

525 if max_columns is None 

526 else max_columns 

527 ) 

528 

529 resolved_field_size = ( 

530 DEFAULT_MAX_FIELD_SIZE 

531 if max_field_size is None 

532 else max_field_size 

533 ) 

534 

535 resolved_total_fields = ( 

536 DEFAULT_MAX_TOTAL_FIELDS 

537 if max_total_fields is None 

538 else max_total_fields 

539 ) 

540 

541 resolved_row_size = ( 

542 DEFAULT_MAX_ROW_SIZE 

543 if max_row_size is None 

544 else max_row_size 

545 ) 

546 

547 resolved_filename_length = ( 

548 DEFAULT_MAX_FILENAME_LENGTH 

549 if max_filename_length is None 

550 else max_filename_length 

551 ) 

552 

553 # --------------------------------------------------------------- 

554 # Determine decorator vs direct invocation 

555 # --------------------------------------------------------------- 

556 

557 def _looks_like_file_path(value): 

558 """Heuristic used to distinguish a path from an argument name.""" 

559 

560 if value.startswith( 

561 ("http://", "https://", "ftp://", "file://") 

562 ): 

563 return True 

564 

565 if value.startswith(("/", "\\")): 

566 return True 

567 

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

569 return True 

570 

571 if "." in value and not value.startswith("."): 

572 return True 

573 

574 return False 

575 

576 is_decorator_mode = False 

577 

578 if func_or_path is None: 

579 is_decorator_mode = True 

580 

581 elif callable(func_or_path): 

582 is_decorator_mode = True 

583 

584 elif isinstance(func_or_path, str): 

585 is_decorator_mode = not _looks_like_file_path(func_or_path) 

586 

587 elif isinstance(func_or_path, Path): 

588 is_decorator_mode = False 

589 

590 # --------------------------------------------------------------- 

591 # Shared validator invocation 

592 # --------------------------------------------------------------- 

593 

594 def _validate(path): 

595 _validate_csv_file( 

596 path=path, 

597 max_file_size=resolved_file_size, 

598 max_rows=resolved_rows, 

599 max_columns=resolved_columns, 

600 max_field_size=resolved_field_size, 

601 max_total_fields=resolved_total_fields, 

602 max_row_size=resolved_row_size, 

603 max_filename_length=resolved_filename_length, 

604 reject_formula_injection=reject_formula_injection, 

605 reject_control_characters=reject_control_characters, 

606 encoding=encoding, 

607 dialect=dialect, 

608 ) 

609 

610 # --------------------------------------------------------------- 

611 # Direct / CLI mode 

612 # --------------------------------------------------------------- 

613 

614 if ( 

615 not is_decorator_mode 

616 and isinstance(func_or_path, (str, Path)) 

617 ): 

618 try: 

619 _validate(func_or_path) 

620 return True 

621 

622 except Exception as exc: 

623 print(f"Exception: {exc}") 

624 return False 

625 

626 # --------------------------------------------------------------- 

627 # Decorator mode 

628 # --------------------------------------------------------------- 

629 

630 def decorator(function): 

631 signature = inspect.signature(function) 

632 params = list(signature.parameters.keys()) 

633 

634 if not params: 

635 raise CsvValidationError( 

636 f"Decorator applied to '{function.__name__}', " 

637 "but it has no arguments." 

638 ) 

639 

640 # Explicit argument name: 

641 # 

642 # @validate_csv("csv_path") 

643 # 

644 # Otherwise use the first argument. 

645 target_arg = ( 

646 func_or_path 

647 if ( 

648 isinstance(func_or_path, str) 

649 and func_or_path in params 

650 ) 

651 else params[0] 

652 ) 

653 

654 @wraps(function) 

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

656 try: 

657 bound_args = signature.bind(*args, **kwargs) 

658 bound_args.apply_defaults() 

659 

660 except TypeError as exc: 

661 raise CsvValidationError( 

662 f"Invalid function call signature: {exc}" 

663 ) from exc 

664 

665 csv_path = bound_args.arguments.get(target_arg) 

666 

667 if csv_path is None: 667 ↛ 668line 667 didn't jump to line 668 because the condition on line 667 was never true

668 raise CsvValidationError( 

669 f"Missing required argument: {target_arg}" 

670 ) 

671 

672 if not isinstance(csv_path, (str, Path)): 

673 raise CsvValidationError( 

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

675 f"got {type(csv_path).__name__}" 

676 ) 

677 

678 _validate(csv_path) 

679 

680 return function(*args, **kwargs) 

681 

682 return wrapper 

683 

684 if callable(func_or_path): 

685 return decorator(func_or_path) 

686 

687 return decorator