left4me/l4d2web/routes/files_routes.py

"""Routes for the overlay 'Files' section.

Two GETs, both gated to the overlay's owner or any admin (mirrors the
overlay detail page rule):

- `GET /overlays/<id>/files?path=<rel>` — HTML fragment listing one
  directory level. Used both for the initial server-rendered root and
  for HTMX swaps when a folder expands.
- `GET /overlays/<id>/files/download?path=<rel>` — streams a single file.
  Symlinks resolving anywhere under `LEFT4ME_ROOT` are allowed (so
  workshop addons stream from the shared cache); anything escaping it
  is refused.
"""
from __future__ import annotations

import os

from flask import Blueprint, Response, render_template, request, send_file
from sqlalchemy import select

from l4d2web.auth import current_user, require_login
from l4d2web.db import session_scope
from l4d2web.models import Overlay, Server
from l4d2web.services.overlay_files import (
    list_directory,
    safe_resolve_for_download,
    safe_resolve_for_listing,
    safe_resolve_for_server_download,
    safe_resolve_for_server_listing,
)


bp = Blueprint("files", __name__)


def _load_overlay_for_user(overlay_id: int, user) -> Overlay | Response:
    with session_scope() as db:
        overlay = db.scalar(select(Overlay).where(Overlay.id == overlay_id))
        if overlay is None:
            return Response(status=404)
        if not user.admin and overlay.user_id is not None and overlay.user_id != user.id:
            return Response(status=403)
        # Detach by expunging — caller only reads scalar columns we already
        # populated, so lazy loads aren't a concern.
        db.expunge(overlay)
        return overlay


@bp.get("/overlays/<int:overlay_id>/files")
@require_login
def overlay_files_fragment(overlay_id: int):
    user = current_user()
    assert user is not None
    sub_path = request.args.get("path", "")

    result = _load_overlay_for_user(overlay_id, user)
    if isinstance(result, Response):
        return result
    overlay = result

    try:
        target = safe_resolve_for_listing(overlay.path, sub_path)
        overlay_root = safe_resolve_for_listing(overlay.path, "")
    except ValueError:
        return Response("invalid path", status=400)

    if not target.is_dir():
        return Response(status=404)

    entries, truncated_count = list_directory(target, overlay_root)
    return render_template(
        "_overlay_file_tree.html",
        overlay=overlay,
        entries=entries,
        truncated=truncated_count > 0,
        truncated_count=truncated_count,
        files_base_url=f"/overlays/{overlay_id}",
        download_supported=True,
    )


@bp.get("/overlays/<int:overlay_id>/files/download")
@require_login
def overlay_files_download(overlay_id: int):
    user = current_user()
    assert user is not None
    sub_path = request.args.get("path", "")

    result = _load_overlay_for_user(overlay_id, user)
    if isinstance(result, Response):
        return result
    overlay = result

    try:
        real = safe_resolve_for_download(overlay.path, sub_path)
    except ValueError:
        return Response("invalid path", status=400)

    if not real.exists():
        return Response(status=404)
    if real.is_dir():
        return Response("not a file", status=400)

    return send_file(
        str(real), as_attachment=True, download_name=os.path.basename(real)
    )


@bp.get("/servers/<int:server_id>/files")
@require_login
def server_files_fragment(server_id: int):
    """Listing-only file tree for a server's runtime merged view. No download
    counterpart — runtime holds large game binaries the user doesn't need to
    pull down through the web app.
    """
    user = current_user()
    assert user is not None
    sub_path = request.args.get("path", "")

    with session_scope() as db:
        server = db.scalar(select(Server).where(Server.id == server_id, Server.user_id == user.id))
        if server is None:
            return Response(status=404)

    try:
        target = safe_resolve_for_server_listing(server_id, sub_path)
        merged_root = safe_resolve_for_server_listing(server_id, "")
    except ValueError:
        return Response("invalid path", status=400)

    if target is None or merged_root is None:
        return Response(status=404)
    if not target.is_dir():
        return Response(status=404)

    entries, truncated_count = list_directory(target, merged_root)
    return render_template(
        "_overlay_file_tree.html",
        entries=entries,
        truncated=truncated_count > 0,
        truncated_count=truncated_count,
        files_base_url=f"/servers/{server_id}",
        download_supported=True,
    )


@bp.get("/servers/<int:server_id>/files/download")
@require_login
def server_files_download(server_id: int):
    user = current_user()
    assert user is not None
    sub_path = request.args.get("path", "")

    with session_scope() as db:
        server = db.scalar(select(Server).where(Server.id == server_id, Server.user_id == user.id))
        if server is None:
            return Response(status=404)

    try:
        real = safe_resolve_for_server_download(server_id, sub_path)
    except ValueError:
        return Response("invalid path", status=400)

    if real is None or not real.exists():
        return Response(status=404)
    if real.is_dir():
        return Response("not a file", status=400)

    return send_file(
        str(real), as_attachment=True, download_name=os.path.basename(real)
    )