atlus/backend/routers/display.py

"""X11 Display — manage virtual display, launch/control GUI apps."""

import logging
import re
from typing import Optional

from fastapi import APIRouter, Depends, HTTPException
from pydantic import BaseModel

from backend.auth import get_current_user
from backend.display import display_manager, check_dependencies, _require_display_deps

router = APIRouter(prefix="/api/display", tags=["display"])
log = logging.getLogger("atlus.display")

# Safe command pattern — allow typical binary paths
_SAFE_CMD = re.compile(r"^[a-zA-Z0-9_./-]+$")


# ---------------------------------------------------------------------------
# Models
# ---------------------------------------------------------------------------

class LaunchAppRequest(BaseModel):
    command: str  # e.g. "nextcloud" or "/usr/bin/nextcloud"
    args: list[str] = []
    name: str = ""  # friendly name
    has_tray_icon: bool = False  # hint: this app has a system tray icon


class FocusWindowRequest(BaseModel):
    window_id: str


# ---------------------------------------------------------------------------
# Endpoints
# ---------------------------------------------------------------------------

@router.get("/status")
async def display_status(user: str = Depends(get_current_user)):
    """Get display session status and dependency info."""
    deps = check_dependencies()
    session = display_manager.get_session(user)
    return {
        "dependencies": deps,
        "missing": _require_display_deps(),
        "session": session.to_dict() if session else None,
    }


@router.post("/start")
async def start_display(user: str = Depends(get_current_user)):
    """Start the virtual display session."""
    missing = _require_display_deps()
    if missing:
        raise HTTPException(
            503,
            f"Missing required packages: {', '.join(missing)}. "
            f"Install with: sudo apt install {' '.join(p.lower() for p in missing)}"
        )
    session = await display_manager.ensure_started(user)
    return session.to_dict()


@router.post("/stop")
async def stop_display(user: str = Depends(get_current_user)):
    """Stop the virtual display session (kills all apps)."""
    ok = await display_manager.stop_session(user)
    if not ok:
        raise HTTPException(404, "No active display session")
    return {"ok": True}


@router.post("/apps/launch")
async def launch_app(
    req: LaunchAppRequest,
    user: str = Depends(get_current_user),
):
    """Launch a GUI application on the virtual display."""
    missing = _require_display_deps()
    if missing:
        raise HTTPException(503, f"Missing required packages: {', '.join(missing)}")

    # Validate command (prevent injection)
    if not _SAFE_CMD.match(req.command):
        raise HTTPException(400, "Invalid command — only alphanumeric, dots, dashes, slashes allowed")

    # Validate args too
    for arg in req.args:
        if arg.startswith("-"):
            continue  # flags are OK
        if not _SAFE_CMD.match(arg):
            raise HTTPException(400, f"Invalid argument: {arg}")

    session = await display_manager.ensure_started(user)
    try:
        app = await session.launch_app(
            command=req.command,
            args=req.args,
            name=req.name,
            has_tray_icon=req.has_tray_icon,
        )
    except Exception as e:
        raise HTTPException(500, f"Failed to launch app: {e}")

    return app.to_dict()


@router.get("/apps")
async def list_apps(user: str = Depends(get_current_user)):
    """List running GUI apps."""
    session = display_manager.get_session(user)
    if not session:
        return {"apps": []}
    # Prune dead apps
    dead = [aid for aid, a in session.apps.items() if not a.alive]
    for aid in dead:
        session.apps.pop(aid, None)
    return {"apps": [a.to_dict() for a in session.apps.values()]}


@router.delete("/apps/{app_id}")
async def stop_app(app_id: str, user: str = Depends(get_current_user)):
    """Stop a specific GUI app."""
    session = display_manager.get_session(user)
    if not session:
        raise HTTPException(404, "No active display session")
    ok = await session.stop_app(app_id)
    if not ok:
        raise HTTPException(404, "App not found")
    return {"ok": True}


@router.get("/windows")
async def list_windows(user: str = Depends(get_current_user)):
    """List X11 windows via wmctrl."""
    session = display_manager.get_session(user)
    if not session or not session.started:
        return {"windows": []}
    windows = await session.list_windows()
    return {"windows": windows}


@router.post("/windows/focus")
async def focus_window(
    req: FocusWindowRequest,
    user: str = Depends(get_current_user),
):
    """Focus/activate a specific window."""
    session = display_manager.get_session(user)
    if not session or not session.started:
        raise HTTPException(404, "No active display session")
    ok = await session.focus_window(req.window_id)
    if not ok:
        raise HTTPException(500, "Failed to focus window")
    return {"ok": True}


@router.get("/connect")
async def get_connection_info(user: str = Depends(get_current_user)):
    """Get WebSocket connection info for noVNC."""
    session = display_manager.get_session(user)
    if not session or not session.started:
        raise HTTPException(404, "No active display session — start one first")

    return {
        "ws_port": session.ws_port,
        "display": session.display,
        "resolution": session.resolution,
    }