"""File validation service for uploaded documents.""" import io import struct from pathlib import Path import structlog from openpyxl import load_workbook logger = structlog.get_logger(__name__) class FileValidationError(Exception): """Custom exception for file validation errors. Args: code: Error code (INVALID_EXTENSION, INVALID_FORMAT, etc.) message: Human-readable error message. """ def __init__(self, code: str, message: str): self.code = code self.message = message super().__init__(message) # Magic byte signatures for document formats XLSX_MAGIC_BYTES = b"PK" # .xlsx and .xlsm files are ZIP archives (starts with PK) XLS_MAGIC_BYTES = b"\xd0\xcf\x11\xe0" # .xls files are OLE2 format PDF_MAGIC_BYTES = b"%PDF-" # PDF files start with %PDF- # Maximum file size in bytes (50MB) MAX_FILE_SIZE_BYTES = 50 * 1024 * 1024 def validate_file_extension(filename: str) -> None: """Validate that file has a valid Excel, Word, PDF, or PowerPoint extension. Args: filename: Name of the uploaded file. Raises: FileValidationError: If extension is not .xlsx, .xls, .xlsm, .docx, .doc, .docm, .pdf, .pptx, .ppt, or .pptm """ valid_extensions = {".xlsx", ".xls", ".xlsm", ".docx", ".doc", ".docm", ".pdf", ".pptx", ".ppt", ".pptm"} file_ext = Path(filename).suffix.lower() if file_ext not in valid_extensions: logger.warning( "invalid_file_extension", filename=filename, extension=file_ext, valid_extensions=list(valid_extensions), ) raise FileValidationError( code="INVALID_EXTENSION", message=f"File extension must be .xlsx, .xls, .xlsm, .docx, .doc, .docm, .pdf, .pptx, .ppt, or .pptm, got {file_ext}", ) logger.debug("file_extension_valid", filename=filename, extension=file_ext) def validate_magic_bytes(file_content: bytes, filename: str) -> None: """Validate that file signature matches Excel, Word, PDF, or PowerPoint format. Checks magic bytes to prevent file extension spoofing. - .xlsx/.xlsm/.docx/.pptx/.pptm: ZIP format (PK signature) - .xls/.doc/.docm/.ppt: OLE2 format (D0CF11E0 signature) - .pdf: PDF format (%PDF- signature) Args: file_content: Raw bytes of the uploaded file. filename: Name of the file (for logging). Raises: FileValidationError: If file signature doesn't match expected format. """ if len(file_content) < 5: logger.warning( "file_too_small_for_magic_bytes", filename=filename, size=len(file_content), ) raise FileValidationError( code="INVALID_FORMAT", message="File is too small to be a valid document", ) # Check for ZIP format (.xlsx/.xlsm/.docx - starts with 'PK') if file_content[:2] == XLSX_MAGIC_BYTES: logger.debug("magic_bytes_valid_zip_format", filename=filename) return # Check for OLE2 format (.xls/.doc/.docm) if file_content[:4] == XLS_MAGIC_BYTES: logger.debug("magic_bytes_valid_ole2_format", filename=filename) return # Check for PDF format if file_content[:5] == PDF_MAGIC_BYTES: logger.debug("magic_bytes_valid_pdf_format", filename=filename) return # None of the formats matched actual_bytes = file_content[:5].hex() logger.warning( "invalid_magic_bytes", filename=filename, actual_bytes=actual_bytes, expected_zip=XLSX_MAGIC_BYTES.hex(), expected_ole2=XLS_MAGIC_BYTES.hex(), expected_pdf=PDF_MAGIC_BYTES.hex(), ) raise FileValidationError( code="INVALID_FORMAT", message="File signature does not match expected format (possible renamed file)", ) def validate_file_size(file_size: int, max_size: int = MAX_FILE_SIZE_BYTES) -> None: """Validate that file size does not exceed maximum. Args: file_size: Size of the file in bytes. max_size: Maximum allowed size in bytes (default: 50MB). Raises: FileValidationError: If file size exceeds limit. """ if file_size > max_size: logger.warning( "file_size_exceeded", file_size=file_size, max_size=max_size, size_mb=file_size / (1024 * 1024), ) raise FileValidationError( code="FILE_TOO_LARGE", message=f"File size {file_size / (1024 * 1024):.2f}MB exceeds maximum {max_size / (1024 * 1024)}MB", ) logger.debug("file_size_valid", file_size=file_size, max_size=max_size) def validate_file_with_openpyxl(file_content: bytes, filename: str) -> None: """Validate that file can be opened and has data. Checks for: 1. File is not corrupted (can be opened with openpyxl) 2. File has at least one sheet 3. At least one sheet has data (non-empty) Args: file_content: Raw bytes of the uploaded file. filename: Name of the file (for logging). Raises: FileValidationError: If file is corrupted or empty. """ try: # Load workbook from bytes (read-only mode for efficiency) workbook = load_workbook(io.BytesIO(file_content), read_only=True, data_only=True) except Exception as e: logger.warning( "file_corrupted", filename=filename, error=str(e), error_type=type(e).__name__, ) raise FileValidationError( code="CORRUPTED_FILE", message=f"File cannot be opened as Excel: {str(e)}", ) from e # Check if workbook has sheets sheet_names = workbook.sheetnames if not sheet_names: logger.warning("no_sheets_found", filename=filename) workbook.close() raise FileValidationError( code="EMPTY_FILE", message="Excel file has no sheets", ) # Check if at least one sheet has data has_data = False for sheet_name in sheet_names: sheet = workbook[sheet_name] # Check if sheet has any rows with data # In read-only mode, we need to iterate to check try: # Get first row to check if sheet has any content first_row = next(sheet.iter_rows(min_row=1, max_row=1), None) if first_row and any(cell.value is not None for cell in first_row): has_data = True break except StopIteration: continue workbook.close() if not has_data: logger.warning( "all_sheets_empty", filename=filename, sheet_count=len(sheet_names), ) raise FileValidationError( code="EMPTY_FILE", message="All sheets in Excel file are empty", ) logger.debug( "file_openpyxl_valid", filename=filename, sheet_count=len(sheet_names), ) def validate_uploaded_file(filename: str, file_content: bytes, max_size: int = MAX_FILE_SIZE_BYTES) -> None: """Orchestrate all file validation checks. Runs validations in fail-fast order (cheapest first): 1. Extension check 2. Size check 3. Magic bytes check 4. Corruption/empty check with openpyxl (Excel files only) Supported formats: .xlsx, .xls, .xlsm, .docx, .doc, .docm, .pdf, .pptx, .ppt, .pptm Args: filename: Name of the uploaded file. file_content: Raw bytes of the uploaded file. max_size: Maximum allowed file size in bytes (default: 50MB). Raises: FileValidationError: If any validation fails, with specific error code. """ logger.info("file_validation_started", filename=filename, file_size=len(file_content)) # Validation order: extension → size → magic bytes → corruption (Excel only) validate_file_extension(filename) validate_file_size(len(file_content), max_size) validate_magic_bytes(file_content, filename) # Only validate with openpyxl for Excel files file_ext = Path(filename).suffix.lower() if file_ext in {".xlsx", ".xls", ".xlsm"}: validate_file_with_openpyxl(file_content, filename) logger.info("file_validation_passed", filename=filename)