zcbot/web/routers/web_previews.py

"""Authenticated preview controls and signed static project file gateway."""
from __future__ import annotations

import mimetypes
import re
import time
from pathlib import Path, PurePosixPath
from uuid import UUID

import jwt
from fastapi import Depends, HTTPException, Request
from fastapi.responses import FileResponse, Response

from core.paths import from_db_path
from core.web_previews import (
    WebPreviewError,
    get_web_preview,
    preview_dict,
    stop_web_preview,
)

_HTML_MAX_REWRITE = 4 * 1024 * 1024
_CSS_MAX_REWRITE = 4 * 1024 * 1024
_ATTR_ROOT_URL = re.compile(r"(?i)(\b(?:src|href|poster|action)\s*=\s*['\"])/(?!/)")
_CSS_ROOT_URL = re.compile(r"(?i)(url\(\s*['\"]?)/(?!/)")
_CSS_ROOT_IMPORT = re.compile(r"(?i)(@import\s+['\"])/(?!/)")


def _preview_headers(*, html: bool = False) -> dict[str, str]:
    headers = {
        "Cache-Control": "no-cache",
        "Referrer-Policy": "no-referrer",
        "X-Content-Type-Options": "nosniff",
        "Access-Control-Allow-Origin": "*",
    }
    if html:
        # The gateway currently shares the zcbot host. Header-level sandboxing also applies
        # when somebody opens the signed URL as a top-level page, so project code never gains
        # the host page's origin or localStorage even outside the SPA iframe.
        headers["Content-Security-Policy"] = (
            "sandbox allow-downloads allow-forms allow-modals allow-scripts; "
            "frame-ancestors 'self'"
        )
    return headers


def _rewrite_html(text: str, prefix: str) -> str:
    rewritten = _ATTR_ROOT_URL.sub(lambda m: m.group(1) + prefix + "/", text)
    base = f'<base href="{prefix}/">'
    if re.search(r"(?i)<head(?:\s[^>]*)?>", rewritten):
        return re.sub(
            r"(?i)<head(?:\s[^>]*)?>",
            lambda m: m.group(0) + base,
            rewritten,
            count=1,
        )
    return base + rewritten


def _rewrite_css(text: str, prefix: str) -> str:
    text = _CSS_ROOT_URL.sub(lambda m: m.group(1) + prefix + "/", text)
    return _CSS_ROOT_IMPORT.sub(lambda m: m.group(1) + prefix + "/", text)


def _mint_preview_token(auth_cfg, preview_id: UUID, expires_at) -> str:
    now = int(time.time())
    exp = int(expires_at.timestamp())
    return jwt.encode(
        {"typ": "web_preview", "pid": str(preview_id), "iat": now, "exp": exp},
        auth_cfg.jwt_secret,
        algorithm="HS256",
    )


def _verify_preview_token(auth_cfg, preview_id: UUID, token: str) -> None:
    try:
        payload = jwt.decode(token, auth_cfg.jwt_secret, algorithms=["HS256"])
    except jwt.InvalidTokenError as exc:
        raise HTTPException(404, "preview not found") from exc
    if payload.get("typ") != "web_preview" or payload.get("pid") != str(preview_id):
        raise HTTPException(404, "preview not found")


def _preview_asset(root: Path, raw_path: str) -> Path:
    raw = str(raw_path or "").replace("\\", "/").lstrip("/")
    path = PurePosixPath(raw)
    if any(part in {"", ".", ".."} for part in path.parts):
        raise HTTPException(404, "preview file not found")
    target = root.joinpath(*path.parts).resolve()
    try:
        target.relative_to(root.resolve())
    except ValueError as exc:
        raise HTTPException(404, "preview file not found") from exc
    return target


def register_web_preview_routes(app, *, require_user, auth_cfg) -> None:
    @app.get("/v1/web-previews/{preview_id}", tags=["web-previews"])
    def web_preview_status(
        preview_id: UUID,
        user_id: UUID = Depends(require_user),  # noqa: B008
    ):
        try:
            row, _ = get_web_preview(preview_id, user_id)
        except WebPreviewError as exc:
            raise HTTPException(404, str(exc)) from exc
        return preview_dict(row)

    @app.post("/v1/web-previews/{preview_id}/launch", tags=["web-previews"])
    def launch_web_preview(
        preview_id: UUID,
        user_id: UUID = Depends(require_user),  # noqa: B008
    ):
        try:
            row, _ = get_web_preview(preview_id, user_id)
        except WebPreviewError as exc:
            raise HTTPException(404, str(exc)) from exc
        if row.status != "active":
            raise HTTPException(410, f"preview is {row.status}")
        token = _mint_preview_token(auth_cfg, preview_id, row.expires_at)
        return {
            "preview": preview_dict(row),
            "url": f"/web-preview/{preview_id}/{token}/",
        }

    @app.delete("/v1/web-previews/{preview_id}", tags=["web-previews"])
    def stop_preview(
        preview_id: UUID,
        user_id: UUID = Depends(require_user),  # noqa: B008
    ):
        try:
            return stop_web_preview(preview_id, user_id)
        except WebPreviewError as exc:
            raise HTTPException(404, str(exc)) from exc

    def serve(preview_id: UUID, token: str, asset_path: str, request: Request):
        _verify_preview_token(auth_cfg, preview_id, token)
        try:
            row, db_working_dir = get_web_preview(preview_id)
        except WebPreviewError as exc:
            raise HTTPException(404, "preview not found") from exc
        if row.status != "active":
            raise HTTPException(410, f"preview is {row.status}")

        working_dir = from_db_path(db_working_dir).resolve()
        root = _preview_asset(working_dir, row.root_path) if row.root_path != "." else working_dir
        if not root.is_dir():
            raise HTTPException(404, "preview directory not found")

        requested = asset_path or row.entry_path
        target = _preview_asset(root, requested)
        if not target.is_file():
            accepts_html = "text/html" in (request.headers.get("accept") or "")
            extensionless = not PurePosixPath(requested).suffix
            if row.spa_fallback and accepts_html and extensionless:
                target = _preview_asset(root, row.entry_path)
            if not target.is_file():
                raise HTTPException(404, "preview file not found")

        prefix = f"/web-preview/{preview_id}/{token}"
        suffix = target.suffix.lower()
        size = target.stat().st_size
        if suffix in {".html", ".htm"} and size <= _HTML_MAX_REWRITE:
            text = target.read_text(encoding="utf-8", errors="replace")
            return Response(
                _rewrite_html(text, prefix),
                media_type="text/html; charset=utf-8",
                headers=_preview_headers(html=True),
            )
        if suffix == ".css" and size <= _CSS_MAX_REWRITE:
            text = target.read_text(encoding="utf-8", errors="replace")
            return Response(
                _rewrite_css(text, prefix),
                media_type="text/css; charset=utf-8",
                headers=_preview_headers(),
            )
        media_type = mimetypes.guess_type(target.name)[0] or "application/octet-stream"
        return FileResponse(
            str(target),
            media_type=media_type,
            headers=_preview_headers(),
        )

    @app.get("/web-preview/{preview_id}/{token}/", include_in_schema=False)
    def web_preview_entry(preview_id: UUID, token: str, request: Request):
        return serve(preview_id, token, "", request)

    @app.get("/web-preview/{preview_id}/{token}/{asset_path:path}", include_in_schema=False)
    def web_preview_asset(preview_id: UUID, token: str, asset_path: str, request: Request):
        return serve(preview_id, token, asset_path, request)