atlus/backend/routers/packages.py

"""Package management — apt-get / dpkg wrapper for Debian-based systems."""

import asyncio
import os
import re
import shutil

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

from backend.auth import get_current_user

router = APIRouter(prefix="/api/packages", tags=["packages"])

# ---------------------------------------------------------------------------
# Guard
# ---------------------------------------------------------------------------

def _find_apt():
    """Find apt-cache binary, checking common paths if PATH is restricted."""
    found = shutil.which("apt-cache")
    if found:
        return True
    # systemd services may have a minimal PATH — check directly
    for p in ("/usr/bin/apt-cache", "/usr/sbin/apt-cache"):
        if os.path.isfile(p) and os.access(p, os.X_OK):
            return True
    return False


_HAS_APT = _find_apt()


def _require_apt():
    if not _HAS_APT:
        raise HTTPException(503, "apt package manager not available on this system")


# ---------------------------------------------------------------------------
# Helpers
# ---------------------------------------------------------------------------

_PKG_NAME_RE = re.compile(r"^[a-z0-9][a-z0-9.+\-:]+$")


def _validate_name(name: str) -> str:
    name = name.strip()
    if not _PKG_NAME_RE.match(name):
        raise HTTPException(400, f"Invalid package name: {name}")
    return name


def _safe_env():
    """Build environment with full PATH for apt/dpkg commands."""
    env = {**os.environ, "LC_ALL": "C", "DEBIAN_FRONTEND": "noninteractive"}
    # Ensure standard paths are included (systemd may use minimal PATH)
    path = env.get("PATH", "")
    for p in ("/usr/bin", "/usr/sbin", "/bin", "/sbin"):
        if p not in path:
            path = p + ":" + path
    env["PATH"] = path
    return env


async def _apt(*args: str, timeout: float = 30) -> str:
    """Run an apt/dpkg command and return stdout."""
    _require_apt()
    cmd = list(args)
    proc = await asyncio.create_subprocess_exec(
        *cmd,
        stdout=asyncio.subprocess.PIPE,
        stderr=asyncio.subprocess.PIPE,
        env=_safe_env(),
    )
    try:
        stdout, stderr = await asyncio.wait_for(proc.communicate(), timeout=timeout)
    except asyncio.TimeoutError:
        proc.kill()
        await proc.wait()
        raise HTTPException(504, "Package operation timed out")
    if proc.returncode != 0:
        msg = stderr.decode().strip() or stdout.decode().strip()
        raise HTTPException(500, f"Package error: {msg}")
    return stdout.decode()


async def _apt_nofail(*args: str, timeout: float = 30) -> tuple[int, str, str]:
    """Run an apt/dpkg command, return (returncode, stdout, stderr) without raising."""
    _require_apt()
    cmd = list(args)
    proc = await asyncio.create_subprocess_exec(
        *cmd,
        stdout=asyncio.subprocess.PIPE,
        stderr=asyncio.subprocess.PIPE,
        env=_safe_env(),
    )
    try:
        stdout, stderr = await asyncio.wait_for(proc.communicate(), timeout=timeout)
    except asyncio.TimeoutError:
        proc.kill()
        await proc.wait()
        return 1, "", "Timed out"
    return proc.returncode, stdout.decode(), stderr.decode()


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

class PackageAction(BaseModel):
    name: str


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

@router.get("/search")
async def search_packages(
    q: str = Query(..., min_length=2, description="Search query"),
    _user: str = Depends(get_current_user),
):
    """Search for packages matching a query."""
    output = await _apt("apt-cache", "search", q)
    results = []
    for line in output.strip().splitlines():
        if not line.strip():
            continue
        parts = line.split(" - ", 1)
        if len(parts) == 2:
            results.append({"name": parts[0].strip(), "summary": parts[1].strip()})
        if len(results) >= 100:
            break
    return results


@router.get("/info/{name}")
async def package_info(
    name: str,
    _user: str = Depends(get_current_user),
):
    """Get detailed information about a package."""
    name = _validate_name(name)

    # Get package info from apt-cache
    output = await _apt("apt-cache", "show", name)

    # Parse key-value fields
    info = {}
    current_key = None
    current_val = ""
    for line in output.splitlines():
        if line.startswith(" "):
            # Continuation of previous field
            current_val += "\n" + line.strip()
        elif ": " in line:
            if current_key:
                info[current_key] = current_val
            current_key, current_val = line.split(": ", 1)
        elif line == "":
            if current_key:
                info[current_key] = current_val
                current_key = None
                current_val = ""
            break  # Only parse first stanza
    if current_key:
        info[current_key] = current_val

    # Check installed status
    rc, dpkg_out, _ = await _apt_nofail(
        "dpkg-query", "-W", "-f=${Status}\t${Version}\n", name
    )
    installed_version = None
    is_installed = False
    if rc == 0 and dpkg_out.strip():
        parts = dpkg_out.strip().split("\t")
        if len(parts) >= 2 and "install ok installed" in parts[0]:
            is_installed = True
            installed_version = parts[1]

    return {
        "name": info.get("Package", name),
        "version": info.get("Version", ""),
        "installed": is_installed,
        "installed_version": installed_version,
        "description": info.get("Description", info.get("Description-en", "")),
        "size": info.get("Size", ""),
        "installed_size": info.get("Installed-Size", ""),
        "depends": info.get("Depends", ""),
        "section": info.get("Section", ""),
        "maintainer": info.get("Maintainer", ""),
        "homepage": info.get("Homepage", ""),
        "architecture": info.get("Architecture", ""),
    }


@router.post("/install")
async def install_package(
    req: PackageAction,
    _user: str = Depends(get_current_user),
):
    """Install a package via apt-get."""
    name = _validate_name(req.name)
    output = await _apt("apt-get", "install", "-y", name, timeout=300)
    return {"success": True, "package": name, "output": output[-500:]}


@router.post("/remove")
async def remove_package(
    req: PackageAction,
    _user: str = Depends(get_current_user),
):
    """Remove a package via apt-get."""
    name = _validate_name(req.name)
    output = await _apt("apt-get", "remove", "-y", name, timeout=120)
    return {"success": True, "package": name, "output": output[-500:]}


@router.post("/update-cache")
async def update_cache(_user: str = Depends(get_current_user)):
    """Run apt-get update to refresh package lists."""
    output = await _apt("apt-get", "update", timeout=120)
    return {"success": True, "output": output[-500:]}