pdf-quiz-generator/backend/app/utils/upload_access.py

"""Local upload paths and current-reference ACLs; no external URL fetching."""
from pathlib import Path
import posixpath
import re
import socket
from urllib.parse import unquote, urlsplit

from fastapi import HTTPException
from sqlalchemy import or_
from sqlalchemy.orm import load_only

from app.config import settings
from app.models.flashcard import Flashcard, FlashcardDeck
from app.models.media import MediaAsset
from app.models.pdf_document import PDFDocument
from app.models.question import Question
from app.services.quiz_builder import bank_question_predicate, general_question_predicate
from app.utils.quiz_access import require_question_access

# Left behind by the LMS that used to live here. The code that wrote them is
# gone; the files are not, so what they may reach is still spelled out.
# They retain their legacy policy; arbitrary unknown/orphan files do not become public.
LEGACY_LMS_PREFIXES = ('course_files/', 'course_thumbnails/', 'scorm/', 'certificates/')


def local_upload_path(value):
    if not isinstance(value, str):
        raise HTTPException(400, "Invalid image path")
    path = value.removeprefix("/uploads/")
    if (not path or any(c in path for c in "\\%?#:") or any(ord(c) < 32 for c in path)
            or path.endswith(' ') or any(part in ("", ".", "..") for part in path.split("/"))):
        raise HTTPException(400, "Unsafe upload path")
    return path


def upload_file(path):
    """Reject even in-root symlink aliases before any ACL lookup."""
    try:
        root = Path(settings.UPLOAD_DIR).resolve()
        target = root
        for part in path.split("/"):
            target = target / part
            if target.is_symlink():
                raise HTTPException(404, "File not found")
        if not target.resolve().is_relative_to(root):
            raise HTTPException(404, "File not found")
        return target
    except (OSError, ValueError):
        raise HTTPException(404, "File not found")


def _authority(url):
    host = unquote(url.hostname or '').rstrip('.').encode('idna').decode('ascii').lower()
    try:
        host = socket.inet_ntoa(socket.inet_aton(host))  # Numeric IPv4 spellings, no DNS/network.
    except OSError:
        pass
    port = url.port
    return host, None if port in (80, 443) else port


def stored_upload_path(value):
    """Interpret old image strings like the browser, without rewriting stored content."""
    if not isinstance(value, str) or not value:
        return None
    # Match the existing image URL helper's prefix behavior before URL normalization.
    raw = value if re.match(r'^(https?:)?//', value, re.I) or value.startswith('/uploads/') else '/uploads/' + value
    raw = raw.replace('\\', '/').replace('\t', '').replace('\r', '').replace('\n', '')
    raw = raw.strip(''.join(map(chr, range(33))))
    if raw.startswith('//'):
        raw = '//' + raw.lstrip('/')
    try:
        url = urlsplit(raw)
        if url.netloc:
            if url.scheme and url.scheme.lower() not in ('http', 'https'):
                return None
            if _authority(url) != _authority(urlsplit(settings.APP_URL)):
                return None
        elif url.scheme:
            return None
        path = posixpath.normpath(unquote(url.path))
        if not path.startswith('/uploads/'):
            return None
        return local_upload_path(path)
    except (ValueError, UnicodeError, HTTPException):
        return None


def references(db, path):
    # ponytail: scan reference metadata, not question bodies. Add indexed canonical
    # media keys if library size makes this hot; do not cache stale access grants.
    rows = db.query(Question.id, Question.image_path, Question.explanation_image_path).filter(
        or_(Question.image_path.isnot(None), Question.explanation_image_path.isnot(None)),
    ).all()
    ids = [qid for qid, stem, explanation in rows
           if stored_upload_path(stem) == path or stored_upload_path(explanation) == path]
    if not ids:
        return []
    return db.query(Question).options(load_only(
        Question.id, Question.source_quiz_id, Question.image_path, Question.explanation_image_path,
    )).filter(Question.id.in_(ids)).all()


def card_deck_ids(db, path):
    return {deck_id for deck_id, image in db.query(Flashcard.deck_id, Flashcard.image_path).filter(
        Flashcard.image_path.isnot(None),
    ).all() if stored_upload_path(image) == path}


def document_for_file(db, path):
    for document_id, filename in db.query(PDFDocument.id, PDFDocument.filename).all():
        if stored_upload_path(filename) == path:
            return db.get(PDFDocument, document_id)
    return None


def owns_source(db, path, user, cards):
    if user.is_admin:
        return True
    if path.startswith(f"questions/{user.id}/"):
        return True
    # Recovery of legacy question drafts is not ownership of private card-only files.
    if user.is_moderator and not cards and path.startswith('questions/'):
        return True
    match = re.fullmatch(r"images/doc_(\d+)/.+", path)
    if match and not cards:  # Card references stay on card ACL even inside extraction dirs.
        document = db.get(PDFDocument, int(match[1]))
        return document is not None and (document.user_id == user.id or user.is_moderator)
    return False


def shown_in_a_readable_article(db, path, user) -> bool:
    """True when a published article puts this picture in front of the reader.

    A figure in the media library is not, on its own, anybody's to read: the
    library is an educator's workspace and grants decide who may manage it. But
    a drawing embedded in a published article is *the article* — refusing it
    leaves a caption above an empty box, which is exactly what every
    illustration in the library did for everyone who was not an administrator.

    Matched against the article's own prose rather than a link table, because
    an educator writes a figure in by typing a markdown image into a section,
    and there is no row anywhere that says so.
    """
    from app.models.article import Article

    if db.query(MediaAsset.id).filter(MediaAsset.path == path).first() is None:
        return False
    query = db.query(Article.id, Article.content, Article.sections).filter(
        Article.deleted_at.is_(None))
    if not user.is_moderator:
        query = query.filter(Article.status == "published")
    for _id, content, sections in query.all():
        if path in (content or ""):
            return True
        for section in sections or []:
            if path in (section.get("content") or ""):
                return True
    return False


def card_access(db, cards, user):
    return bool(cards) and db.query(FlashcardDeck.id).filter(
        FlashcardDeck.id.in_(cards), FlashcardDeck.deleted_at.is_(None),
        or_(FlashcardDeck.user_id == user.id, FlashcardDeck.is_shared == 1, user.is_admin),
    ).first() is not None


def can_read_upload(db, path, user, questions, cards, attempt_id=None):
    # Sharing a question must never share its original source PDF.
    document = document_for_file(db, path)
    if document:
        return user.is_moderator or document.user_id == user.id
    if owns_source(db, path, user, cards) or card_access(db, cards, user):
        return True
    if shown_in_a_readable_article(db, path, user):
        return True
    for question in questions:
        try:
            require_question_access(db, question, user, attempt_id,
                                    review=stored_upload_path(question.image_path) != path)
            return True
        except HTTPException:
            pass
    return False


def validate_image_attachments(db, user, values):
    """Validate every supplied image before callers mutate any ORM state."""
    result = dict(values)
    for field in ("image_path", "explanation_image_path"):
        value = result.get(field)
        if value is None or value == "":
            continue
        if not isinstance(value, str):
            raise HTTPException(400, "Invalid image path")
        if re.match(r"^https?://", value, re.I):
            try:
                url = urlsplit(value)
                if (not url.hostname or url.username or url.password or '\\' in value
                        or '%' in url.netloc or any(ord(c) < 32 for c in value)):
                    raise ValueError()
                authority = _authority(url)  # Also validates ports/hostname encoding.
            except (ValueError, UnicodeError):
                raise HTTPException(400, "Invalid image URL")
            if authority != _authority(urlsplit(settings.APP_URL)) or not url.path.startswith('/uploads/'):
                continue  # External links are never fetched or granted local-file access.
            if url.query or url.fragment:
                raise HTTPException(400, "Use the original upload URL without query parameters or fragments")
            value = url.path
        path = local_upload_path(value)
        upload_file(path)
        document = document_for_file(db, path)
        if document:
            allowed = user.is_moderator or document.user_id == user.id
        else:
            cards = card_deck_ids(db, path)
            refs = references(db, path)
            if refs:
                ids = [question.id for question in refs]
                if user.is_moderator:
                    # Legacy LMS media can never be republished into the general bank.
                    allowed = db.query(Question.id).filter(
                        Question.id.in_(ids), general_question_predicate(),
                    ).first() is not None
                else:
                    allowed = db.query(Question.id).filter(
                        Question.id.in_(ids), bank_question_predicate(user),
                    ).first() is not None
            else:
                allowed = owns_source(db, path, user, cards) or card_access(db, cards, user)
        if not allowed:
            raise HTTPException(403, "Image is not available for attachment")
        result[field] = path
    return result