250 lines
10 KiB
Python
250 lines
10 KiB
Python
"""
|
||
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
|
||
import mistune
|
||
|
||
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.
|
||
- Must contain at least one uppercase letter (A-Z).
|
||
- Must contain at least one digit (0-9).
|
||
- Must contain at least one special character (!@#$%^&* etc.).
|
||
|
||
Parameters
|
||
----------
|
||
password : str – the candidate password (plain text)
|
||
confirm : str – the confirmation field value
|
||
"""
|
||
import re
|
||
if password != confirm:
|
||
return 'Passwords do not match.'
|
||
if len(password) < 8:
|
||
return 'Password must be at least 8 characters.'
|
||
if not re.search(r'[A-Z]', password):
|
||
return 'Password must contain at least one uppercase letter.'
|
||
if not re.search(r'\d', password):
|
||
return 'Password must contain at least one number.'
|
||
if not re.search(r'[!@#$%^&*()\-_=+\[\]{};:\'",.<>?/\\|`~]', password):
|
||
return 'Password must contain at least one special character.'
|
||
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.
|
||
|
||
Grouping rule
|
||
-------------
|
||
Each entry with offset=0 starts a **new alternative** group.
|
||
Entries with offset>0 are appended to the **current** group — they
|
||
represent additional byte constraints that must ALL match alongside
|
||
the group's offset-0 anchor (e.g. WEBP requires both RIFF at offset 0
|
||
AND 'WEBP' at offset 8 within the same file).
|
||
|
||
⚠️ Constraint: no two entries in the same alternative group may share
|
||
offset=0. If a future signature needs two offset-0 checks as part of
|
||
ONE alternative (i.e. two different bytes that must both appear at the
|
||
start of the same file), this function would incorrectly split them
|
||
into separate alternatives. In that case, use a combined bytes object
|
||
covering the full header range instead of two separate entries, or
|
||
refactor _MAGIC to use a dedicated tuple type that carries an
|
||
'alternative_id' discriminator.
|
||
"""
|
||
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 [[]]
|
||
|
||
|
||
# ─── Comment Body Markdown Rendering ─────────────────────────────────────────
|
||
# Comments support a safe subset of Markdown (bold, italic, code, lists,
|
||
# blockquotes, links). We render to HTML at write time and sanitize with a
|
||
# strict allowlist so stored bodies are always safe to render with | safe.
|
||
#
|
||
# Intentionally excluded from comments (present in KB allowlist):
|
||
# img, table, div, h1-h6, figure — keep comment rendering lightweight.
|
||
|
||
import bleach as _bleach
|
||
|
||
_COMMENT_ALLOWED_TAGS = {
|
||
'p', 'br',
|
||
'strong', 'em', 'u', 's', 'code', 'pre',
|
||
'ul', 'ol', 'li',
|
||
'blockquote',
|
||
'a',
|
||
'hr',
|
||
}
|
||
_COMMENT_ALLOWED_ATTRS = {
|
||
'a': ['href', 'title', 'rel'],
|
||
}
|
||
|
||
_md = mistune.create_markdown(escape=True)
|
||
|
||
|
||
def render_comment_body(raw_text: str) -> str:
|
||
"""Convert plain-text Markdown comment to sanitized HTML.
|
||
|
||
Renders Markdown to HTML with mistune, then strips any tags/attributes
|
||
not in the comment allowlist via bleach. The result is safe to render
|
||
with Jinja2's ``| safe`` filter without further escaping.
|
||
|
||
Parameters
|
||
----------
|
||
raw_text : str – the raw plain-text comment body submitted by the user
|
||
"""
|
||
if not raw_text:
|
||
return ''
|
||
html = _md(raw_text)
|
||
cleaned = _bleach.clean(
|
||
html,
|
||
tags = _COMMENT_ALLOWED_TAGS,
|
||
attributes = _COMMENT_ALLOWED_ATTRS,
|
||
strip = True,
|
||
)
|
||
return cleaned |