""" Shared validation helpers used across multiple route modules. Centralising these here prevents logic divergence that arises from copy-pasting validation code into auth.py, admin.py, and future modules. """ import logging logger = logging.getLogger(__name__) # ─── Password Validation ────────────────────────────────────────────────────── def validate_password(password: str, confirm: str) -> str | None: """Validate a new password and its confirmation field. Returns an error message string if validation fails, or None if the password is acceptable. Callers should flash the returned message with the 'danger' category and return early. Rules ----- - Password and confirmation must match. - Minimum length: 8 characters. Parameters ---------- password : str – the candidate password (plain text) confirm : str – the confirmation field value """ if password != confirm: return 'Passwords do not match.' if len(password) < 8: return 'Password must be at least 8 characters.' return None # ─── File MIME-Type Validation ──────────────────────────────────────────────── # Magic-byte signatures for every extension in ALLOWED_EXT / _KB_ALLOWED_EXT. # Format: extension → list of (offset, bytes) tuples that must ALL be present. # Using raw magic bytes avoids a dependency on python-magic / libmagic while # still catching the "renamed shell.php → shell.pdf" class of attack. # # References: https://en.wikipedia.org/wiki/List_of_file_signatures _MAGIC: dict[str, list[tuple[int, bytes]]] = { 'png' : [(0, b'\x89PNG\r\n\x1a\n')], 'jpg' : [(0, b'\xff\xd8\xff')], 'jpeg': [(0, b'\xff\xd8\xff')], 'gif' : [(0, b'GIF87a'), (0, b'GIF89a')], # either signature is valid 'webp': [(0, b'RIFF'), (8, b'WEBP')], 'pdf' : [(0, b'%PDF')], 'zip' : [(0, b'PK\x03\x04')], # Office Open XML (.docx, .xlsx, .pptx) are ZIP archives internally 'docx': [(0, b'PK\x03\x04')], 'xlsx': [(0, b'PK\x03\x04')], 'pptx': [(0, b'PK\x03\x04')], # Legacy OLE2 compound document (.doc, .xls, .ppt) 'doc' : [(0, b'\xd0\xcf\x11\xe0\xa1\xb1\x1a\xe1')], 'xls' : [(0, b'\xd0\xcf\x11\xe0\xa1\xb1\x1a\xe1')], 'ppt' : [(0, b'\xd0\xcf\x11\xe0\xa1\xb1\x1a\xe1')], # Plain-text formats — no reliable magic bytes; skip byte check 'txt' : [], 'csv' : [], 'log' : [], } # Maximum number of bytes to read for magic-byte inspection. # 12 bytes covers all signatures above (WEBP needs offset 8 + 4 = 12). _MAGIC_READ_BYTES = 12 # Per-file size cap (bytes). Mirrors MAX_CONTENT_LENGTH so a single # large file cannot saturate the total-request budget on its own. _MAX_FILE_BYTES = 16 * 1024 * 1024 # 16 MB def validate_file(file, allowed_extensions: set[str]) -> str | None: """Validate an uploaded file object by extension, size, and magic bytes. Returns an error message string if validation fails, or None on success. The file stream seek position is reset to 0 after inspection so callers can still read or save the file normally. Parameters ---------- file : werkzeug FileStorage object allowed_extensions : set of lowercase extension strings (without leading dot) Checks performed ---------------- 1. Extension is in the allowed set. 2. File does not exceed the per-file size cap (_MAX_FILE_BYTES). 3. Magic bytes in the file content match the declared extension, where a known signature exists. Plain-text types (txt, csv, log) are accepted on extension alone since they have no reliable magic bytes. """ filename = file.filename or '' if '.' not in filename: return 'File has no extension.' ext = filename.rsplit('.', 1)[1].lower() # ── 1. Extension check ──────────────────────────────────────────────────── if ext not in allowed_extensions: return f'File type ".{ext}" is not permitted.' # ── 2. Size check ───────────────────────────────────────────────────────── # Read the file in chunks to measure size without loading it all into RAM, # then seek back to the start so the caller can still save it. file.stream.seek(0, 2) # seek to end file_size = file.stream.tell() # position == size file.stream.seek(0) # rewind if file_size > _MAX_FILE_BYTES: mb = _MAX_FILE_BYTES // (1024 * 1024) return f'File exceeds the {mb} MB per-file size limit.' # ── 3. Magic-byte check ─────────────────────────────────────────────────── signatures = _MAGIC.get(ext) if signatures is None: # Extension is allowed but has no entry in _MAGIC — treat as safe. # This handles future extensions added to ALLOWED_EXT without a # corresponding _MAGIC entry; log a warning for visibility. logger.warning(f'[FILE VALIDATION] No magic signature defined for ext="{ext}"; skipping byte check.') return None if not signatures: # Explicit empty list means "no magic bytes available for this type" # (txt, csv, log) — accepted on extension alone. return None header = file.stream.read(_MAGIC_READ_BYTES) file.stream.seek(0) # rewind after inspection # For extensions with multiple valid signatures (e.g. GIF87a / GIF89a) # the file is valid if ANY of the listed signatures matches. matched = False for sigs in _group_by_alternative(signatures): # Each alternative is a list of (offset, bytes) pairs that must ALL match. if all(header[offset:offset + len(magic)] == magic for offset, magic in sigs): matched = True break if not matched: logger.warning( f'[FILE VALIDATION] Magic-byte mismatch: filename="{filename}" ext="{ext}" ' f'header={header.hex()}' ) return f'File content does not match its declared type (.{ext}).' return None def _group_by_alternative( signatures: list[tuple[int, bytes]], ) -> list[list[tuple[int, bytes]]]: """Split a flat signature list into per-alternative groups. For most extensions there is a single signature, so this returns [[sig1, sig2, ...]]. For GIF (two valid headers) it returns [[gif87_sig], [gif89_sig]] so the caller can treat each inner list as a complete match candidate. The rule: each entry with offset=0 starts a new alternative group. Entries with offset>0 are appended to the current group (they are additional constraints on the same file type, e.g. WEBP needs both offset-0 'RIFF' and offset-8 'WEBP'). """ groups: list[list[tuple[int, bytes]]] = [] for offset, magic in signatures: if offset == 0: groups.append([(offset, magic)]) else: if groups: groups[-1].append((offset, magic)) else: groups.append([(offset, magic)]) return groups if groups else [[]]