#!/usr/bin/env python3 """Stream-based upload validation reference script. This script demonstrates a safe, vendor-neutral approach to validating uploaded files as a stream rather than trusting filenames, extensions, or client-supplied metadata. What it does: - Reads upload content incrementally - Enforces a maximum size limit while streaming - Checks basic file signatures for common formats - Supports optional allowlist-based validation - Produces a clear validation result without loading the entire file into memory This is a reference implementation for operational use and adaptation. It does not perform malware scanning, quarantine, storage, or parsing. """ from __future__ import annotations import argparse import hashlib import json import sys from dataclasses import dataclass, asdict from pathlib import Path from typing import BinaryIO, Iterable, Optional PDF_SIGNATURE = b"%PDF" PNG_SIGNATURE = bytes([0x89, 0x50, 0x4E, 0x47, 0x0D, 0x0A, 0x1A, 0x0A]) JPEG_SIGNATURES = ( bytes([0xFF, 0xD8, 0xFF]), ) ZIP_SIGNATURE = bytes([0x50, 0x4B, 0x03, 0x04]) GIF87A_SIGNATURE = b"GIF87a" GIF89A_SIGNATURE = b"GIF89a" @dataclass class ValidationResult: is_valid: bool error: Optional[str] = None detected_type: Optional[str] = None bytes_read: int = 0 sha256: Optional[str] = None SUPPORTED_TYPES = { "pdf", "png", "jpeg", "gif", "zip", } def detect_file_type(header: bytes) -> Optional[str]: """Detect a file type using common magic-number signatures.""" if header.startswith(PDF_SIGNATURE): return "pdf" if header.startswith(PNG_SIGNATURE): return "png" if any(header.startswith(sig) for sig in JPEG_SIGNATURES): return "jpeg" if header.startswith(GIF87A_SIGNATURE) or header.startswith(GIF89A_SIGNATURE): return "gif" if header.startswith(ZIP_SIGNATURE): return "zip" return None def validate_upload_stream( stream: BinaryIO, max_bytes: int, allowed_types: Optional[Iterable[str]] = None, header_bytes: int = 16, ) -> ValidationResult: """Validate an upload stream incrementally. Args: stream: A binary file-like object. max_bytes: Maximum allowed upload size in bytes. allowed_types: Optional allowlist of detected types. header_bytes: Number of initial bytes to inspect for type detection. """ if stream is None: return ValidationResult(is_valid=False, error="Missing upload stream.") if max_bytes <= 0: return ValidationResult(is_valid=False, error="max_bytes must be greater than zero.") if header_bytes <= 0: return ValidationResult(is_valid=False, error="header_bytes must be greater than zero.") allowed = set(allowed_types or SUPPORTED_TYPES) if not allowed: return ValidationResult(is_valid=False, error="No allowed file types configured.") try: hasher = hashlib.sha256() total = 0 header = stream.read(header_bytes) if not header: return ValidationResult(is_valid=False, error="Empty file.") hasher.update(header) total += len(header) detected_type = detect_file_type(header) if detected_type is None: return ValidationResult( is_valid=False, error="Unsupported or unrecognized file signature.", bytes_read=total, ) if detected_type not in allowed: return ValidationResult( is_valid=False, error=f"File type '{detected_type}' is not allowed.", detected_type=detected_type, bytes_read=total, ) buffer_size = 8192 while True: chunk = stream.read(buffer_size) if not chunk: break total += len(chunk) if total > max_bytes: return ValidationResult( is_valid=False, error="File exceeds the maximum allowed size.", detected_type=detected_type, bytes_read=total, ) hasher.update(chunk) return ValidationResult( is_valid=True, detected_type=detected_type, bytes_read=total, sha256=hasher.hexdigest(), ) except OSError as exc: return ValidationResult(is_valid=False, error=f"I/O error while reading stream: {exc}") def validate_file_path( path: Path, max_bytes: int, allowed_types: Optional[Iterable[str]] = None, ) -> ValidationResult: """Open a file path safely and validate it as a stream.""" if not path.exists(): return ValidationResult(is_valid=False, error=f"File not found: {path}") if not path.is_file(): return ValidationResult(is_valid=False, error=f"Not a regular file: {path}") with path.open("rb") as f: return validate_upload_stream(f, max_bytes=max_bytes, allowed_types=allowed_types) def parse_allowed_types(value: Optional[str]) -> Optional[list[str]]: if value is None: return None items = [item.strip().lower() for item in value.split(",") if item.strip()] for item in items: if item not in SUPPORTED_TYPES: raise argparse.ArgumentTypeError( f"Unsupported allowed type '{item}'. Supported types: {', '.join(sorted(SUPPORTED_TYPES))}" ) return items def build_parser() -> argparse.ArgumentParser: parser = argparse.ArgumentParser( description="Validate uploads as a stream using allowlisted file signatures and size limits." ) parser.add_argument( "path", nargs="?", help="Path to the file to validate. If omitted, reads from standard input.", ) parser.add_argument( "--max-bytes", type=int, default=25 * 1024 * 1024, help="Maximum allowed file size in bytes. Default: 26214400 (25 MiB).", ) parser.add_argument( "--allowed-types", type=parse_allowed_types, default=None, help="Comma-separated allowlist of file types: pdf,png,jpeg,gif,zip", ) parser.add_argument( "--json", action="store_true", help="Output the validation result as JSON.", ) return parser def main() -> int: parser = build_parser() args = parser.parse_args() if args.path: result = validate_file_path(Path(args.path), args.max_bytes, args.allowed_types) else: if sys.stdin.buffer is None: print("Standard input is not available.", file=sys.stderr) return 2 result = validate_upload_stream(sys.stdin.buffer, args.max_bytes, args.allowed_types) if args.json: print(json.dumps(asdict(result), indent=2, sort_keys=True)) else: if result.is_valid: print(f"VALID: type={result.detected_type}, bytes={result.bytes_read}, sha256={result.sha256}") else: print(f"INVALID: {result.error}") if result.detected_type: print(f"Detected type: {result.detected_type}") if result.bytes_read: print(f"Bytes read: {result.bytes_read}") return 0 if result.is_valid else 1 if __name__ == "__main__": raise SystemExit(main())