Code reviewed and issue fixed.

app/services/log_service.py

@@ -54,8 +54,16 @@ def log_action(user_id, action, entity_type=None, entity_id=None, details=None):
     own business objects. This ensures the log entry is only persisted when
     the parent operation succeeds — an independent commit here would leave
     orphaned log entries for operations that were subsequently rolled back.
+
+    Error isolation
+    ---------------
+    On failure, only the log entry itself is expelled from the session via
+    expunge(). db.session.rollback() is intentionally NOT called here because
+    that would wipe the entire session — silently undoing the parent business
+    operation (ticket creation, user update, etc.) that triggered this log call.
     """
     ip = _get_real_ip()
+    entry = None
 
     try:
         entry = ActivityLog(
@@ -74,8 +82,14 @@ def log_action(user_id, action, entity_type=None, entity_id=None, details=None):
             f'user_id={user_id} ip={ip}'
         )
     except Exception as exc:
-        db.session.rollback()
-        logger.error(f'[ACTIVITY LOG ERROR] {exc}')
+        # Expunge only the failed log entry — do NOT roll back the full session,
+        # as that would undo the parent operation that called this function.
+        if entry is not None:
+            try:
+                db.session.expunge(entry)
+            except Exception:
+                pass
+        logger.error(f'[ACTIVITY LOG ERROR] action={action} entity={entity_type}:{entity_id} error={exc}')
 
 
 def log_ticket_history(ticket, field_name, old_value, new_value, changed_by_id):
@@ -85,8 +99,15 @@ def log_ticket_history(ticket, field_name, old_value, new_value, changed_by_id):
     ----------------
     Like log_action, this function does NOT commit — the caller is responsible
     for committing the session after all field changes have been recorded.
+
+    Error isolation
+    ---------------
+    On failure, only the failed history entry is expelled from the session.
+    db.session.rollback() is intentionally NOT called here — that would undo
+    the parent ticket update that triggered this history recording.
     """
     from app.models import TicketHistory
+    entry = None
     try:
         entry = TicketHistory(
             ticket_id  = ticket.id,
@@ -102,5 +123,11 @@ def log_ticket_history(ticket, field_name, old_value, new_value, changed_by_id):
             f'"{old_value}" -> "{new_value}" by user_id={changed_by_id}'
         )
     except Exception as exc:
-        db.session.rollback()
-        logger.error(f'[TICKET HISTORY ERROR] {exc}')
+        # Expunge only the failed history entry — do NOT roll back the full
+        # session, as that would undo the parent ticket update.
+        if entry is not None:
+            try:
+                db.session.expunge(entry)
+            except Exception:
+                pass
+        logger.error(f'[TICKET HISTORY ERROR] ticket_id={ticket.id} field={field_name} error={exc}')

app/services/notification_service.py

@@ -251,7 +251,7 @@ def notify_comment_added(comment):
             ticket_id = ticket.id,
             link      = f'/tickets/{ticket.id}#comment-{comment.id}',
         )
-        user = User.query.get(user_id)
+        user = db.session.get(User, user_id)
         if user and user.email_notif and not is_internal:
             html = render_template_string(_STATUS_UPDATE_EMAIL,
                 ticket_number = ticket.ticket_number,

app/services/validation_service.py

@@ -0,0 +1,179 @@
+"""
+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 [[]]