yakpanel-core/YakPanel-server/backend/app/api/soft.py

"""YakPanel - App Store / Software API"""
from __future__ import annotations

import asyncio
import os
import shlex
import shutil
import subprocess
from typing import Literal

from fastapi import APIRouter, Depends, HTTPException

from app.core.utils import environment_with_system_path, exec_shell_sync
from app.api.auth import get_current_user
from app.models.user import User

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

PmKind = Literal["apt", "dnf", "yum", "microdnf", "apk", "none"]

# Per distro: apt = Debian/Ubuntu, rpm = RHEL/Fedora/Alma/Rocky, apk = Alpine (best-effort)
SOFTWARE_LIST: list[dict[str, str]] = [
    {
        "id": "nginx",
        "name": "Nginx",
        "desc": "Web server",
        "apt": "nginx",
        "rpm": "nginx",
        "apk": "nginx",
    },
    {
        "id": "mysql-server",
        "name": "MySQL Server",
        "desc": "Database server",
        "apt": "mysql-server",
        "rpm": "mysql-server",
        "apk": "mysql",
    },
    {
        "id": "mariadb-server",
        "name": "MariaDB",
        "desc": "Database server",
        "apt": "mariadb-server",
        "rpm": "mariadb-server",
        "apk": "mariadb",
    },
    {
        "id": "php",
        "name": "PHP",
        "desc": "PHP runtime",
        "apt": "php",
        "rpm": "php",
        "apk": "php",
    },
    {
        "id": "php-fpm",
        "name": "PHP-FPM",
        "desc": "PHP FastCGI",
        "apt": "php-fpm",
        "rpm": "php-fpm",
        "apk": "php-fpm",
    },
    {
        "id": "redis-server",
        "name": "Redis",
        "desc": "In-memory cache",
        "apt": "redis-server",
        "rpm": "redis",
        "apk": "redis",
    },
    {
        "id": "postgresql",
        "name": "PostgreSQL",
        "desc": "Database server",
        "apt": "postgresql",
        "rpm": "postgresql-server",
        "apk": "postgresql",
    },
    {
        "id": "mongodb",
        "name": "MongoDB",
        "desc": "NoSQL database",
        "apt": "mongodb",
        "rpm": "mongodb-server",
        "apk": "mongodb",
    },
    {
        "id": "certbot",
        "name": "Certbot",
        "desc": "Let's Encrypt SSL",
        "apt": "certbot",
        "rpm": "certbot",
        "apk": "certbot",
    },
    {
        "id": "docker",
        "name": "Docker",
        "desc": "Container runtime",
        "apt": "docker.io",
        "rpm": "docker",
        "apk": "docker",
    },
    {
        "id": "nodejs",
        "name": "Node.js",
        "desc": "JavaScript runtime",
        "apt": "nodejs",
        "rpm": "nodejs",
        "apk": "nodejs",
    },
    {
        "id": "npm",
        "name": "npm",
        "desc": "Node package manager",
        "apt": "npm",
        "rpm": "npm",
        "apk": "npm",
    },
    {
        "id": "git",
        "name": "Git",
        "desc": "Version control",
        "apt": "git",
        "rpm": "git",
        "apk": "git",
    },
    {
        "id": "python3",
        "name": "Python 3",
        "desc": "Python runtime",
        "apt": "python3",
        "rpm": "python3",
        "apk": "python3",
    },
]


def _resolve_command(name: str) -> str | None:
    """Locate an executable; systemd often provides a venv-only PATH, so scan standard dirs too."""
    std = "/usr/local/sbin:/usr/local/bin:/usr/sbin:/usr/bin:/sbin:/bin"
    merged = f"{os.environ.get('PATH', '').strip()}:{std}".strip(":")
    found = shutil.which(name, path=merged)
    if found:
        return found
    for d in ("/usr/bin", "/usr/sbin", "/bin", "/sbin", "/usr/local/bin", "/usr/local/sbin"):
        cand = os.path.join(d, name)
        if os.path.isfile(cand) and os.access(cand, os.X_OK):
            return cand
    return None


def _detect_package_manager() -> PmKind:
    if _resolve_command("apt-get"):
        return "apt"
    if _resolve_command("dnf"):
        return "dnf"
    if _resolve_command("yum"):
        return "yum"
    if _resolve_command("microdnf"):
        return "microdnf"
    if _resolve_command("apk"):
        return "apk"
    return "none"


def _package_name(entry: dict[str, str], pm: PmKind) -> str:
    if pm == "apt":
        return entry["apt"]
    if pm in ("dnf", "yum", "microdnf"):
        return entry["rpm"]
    if pm == "apk":
        return entry.get("apk") or entry["apt"]
    return entry["apt"]


def _apt_env() -> dict[str, str]:
    env = environment_with_system_path()
    env.setdefault("DEBIAN_FRONTEND", "noninteractive")
    env.setdefault("APT_LISTCHANGES_FRONTEND", "none")
    return env


def _run_shell(script: str, timeout: int, env: dict[str, str] | None = None) -> None:
    run_env = environment_with_system_path(env)
    result = subprocess.run(
        script,
        shell=True,
        capture_output=True,
        text=True,
        timeout=timeout,
        env=run_env,
    )
    if result.returncode == 0:
        return
    out = (result.stdout or "").strip()
    err = (result.stderr or "").strip()
    msg = err or out or f"Command failed (exit {result.returncode})"
    raise HTTPException(status_code=500, detail=msg[:4000])


def _check_installed_apt(pkg: str) -> tuple[bool, str]:
    out, _ = exec_shell_sync(f"dpkg -l {shlex.quote(pkg)} 2>/dev/null | grep ^ii", timeout=5)
    if out.strip():
        parts = out.split()
        if len(parts) >= 3:
            return True, parts[2]
    return False, ""


def _check_installed_rpm(pkg: str) -> tuple[bool, str]:
    try:
        rpm_bin = _resolve_command("rpm") or "rpm"
        result = subprocess.run(
            [rpm_bin, "-q", "--queryformat", "%{EVR}", pkg],
            capture_output=True,
            text=True,
            timeout=5,
            env=environment_with_system_path(),
        )
        if result.returncode != 0:
            return False, ""
        ver = (result.stdout or "").strip()
        return (True, ver) if ver else (True, "")
    except (FileNotFoundError, subprocess.TimeoutExpired):
        return False, ""


def _check_installed_apk(pkg: str) -> tuple[bool, str]:
    try:
        apk_bin = _resolve_command("apk") or "apk"
        r = subprocess.run(
            [apk_bin, "info", "-e", pkg],
            capture_output=True,
            text=True,
            timeout=5,
            env=environment_with_system_path(),
        )
        if r.returncode != 0:
            return False, ""
        out = (r.stdout or "").strip()
        if not out:
            return True, ""
        first = out.split("\n")[0].strip()
        return True, first
    except (FileNotFoundError, subprocess.TimeoutExpired):
        return False, ""


def _check_installed(pm: PmKind, pkg: str) -> tuple[bool, str]:
    if pm == "apt":
        return _check_installed_apt(pkg)
    if pm in ("dnf", "yum", "microdnf"):
        return _check_installed_rpm(pkg)
    if pm == "apk":
        return _check_installed_apk(pkg)
    return False, ""


def _install_script(pm: PmKind, pkg: str) -> tuple[str, int, dict[str, str] | None]:
    q = shlex.quote(pkg)
    if pm == "apt":
        ag = _resolve_command("apt-get")
        if not ag:
            raise HTTPException(status_code=501, detail="apt-get not found on this system.")
        exe = shlex.quote(ag)
        return f"{exe} update -qq && {exe} install -y {q}", 600, _apt_env()
    if pm == "dnf":
        exe = shlex.quote(x) if (x := _resolve_command("dnf")) else None
        if exe:
            return f"{exe} install -y {q}", 600, None
    if pm == "yum":
        exe = shlex.quote(x) if (x := _resolve_command("yum")) else None
        if exe:
            return f"{exe} install -y {q}", 600, None
    if pm == "microdnf":
        exe = shlex.quote(x) if (x := _resolve_command("microdnf")) else None
        if exe:
            return f"{exe} install -y {q}", 600, None
    if pm == "apk":
        exe = shlex.quote(x) if (x := _resolve_command("apk")) else None
        if exe:
            return f"{exe} update && {exe} add {q}", 600, None
    raise HTTPException(
        status_code=501,
        detail="No supported package manager found (need apt-get, dnf, yum, microdnf, or apk).",
    )


def _uninstall_script(pm: PmKind, pkg: str) -> tuple[str, int, dict[str, str] | None]:
    q = shlex.quote(pkg)
    if pm == "apt":
        ag = _resolve_command("apt-get")
        if ag:
            exe = shlex.quote(ag)
            return f"{exe} remove -y {q}", 180, _apt_env()
        raise HTTPException(status_code=501, detail="apt-get not found on this system.")
    if pm == "dnf":
        exe = shlex.quote(x) if (x := _resolve_command("dnf")) else None
        if exe:
            return f"{exe} remove -y {q}", 180, None
    if pm == "yum":
        exe = shlex.quote(x) if (x := _resolve_command("yum")) else None
        if exe:
            return f"{exe} remove -y {q}", 180, None
    if pm == "microdnf":
        exe = shlex.quote(x) if (x := _resolve_command("microdnf")) else None
        if exe:
            return f"{exe} remove -y {q}", 180, None
    if pm == "apk":
        exe = shlex.quote(x) if (x := _resolve_command("apk")) else None
        if exe:
            return f"{exe} del {q}", 120, None
    raise HTTPException(
        status_code=501,
        detail="No supported package manager found (need apt-get, dnf, yum, microdnf, or apk).",
    )


@router.get("/list")
async def soft_list(current_user: User = Depends(get_current_user)):
    """List software with install status for this OS."""
    pm = _detect_package_manager()
    manager_label = pm if pm != "none" else "unknown"
    result: list[dict] = []
    for s in SOFTWARE_LIST:
        pkg = _package_name(s, pm) if pm != "none" else s["apt"]
        installed, version = _check_installed(pm, pkg) if pm != "none" else (False, "")
        result.append(
            {
                "id": s["id"],
                "name": s["name"],
                "desc": s["desc"],
                "pkg": pkg,
                "installed": installed,
                "version": version if installed else "",
                "package_manager": manager_label,
            }
        )
    return {"software": result, "package_manager": manager_label}


@router.post("/install/{pkg_id}")
async def soft_install(
    pkg_id: str,
    current_user: User = Depends(get_current_user),
):
    """Install package via system package manager (root privileges required)."""
    pm = _detect_package_manager()
    entry = next((s for s in SOFTWARE_LIST if s["id"] == pkg_id), None)
    if not entry:
        raise HTTPException(status_code=404, detail="Package not found")
    pkg = _package_name(entry, pm)
    script, timeout, env = _install_script(pm, pkg)
    await asyncio.to_thread(_run_shell, script, timeout, env)
    return {"status": True, "msg": "Installed"}


@router.post("/uninstall/{pkg_id}")
async def soft_uninstall(
    pkg_id: str,
    current_user: User = Depends(get_current_user),
):
    """Uninstall package via system package manager."""
    pm = _detect_package_manager()
    entry = next((s for s in SOFTWARE_LIST if s["id"] == pkg_id), None)
    if not entry:
        raise HTTPException(status_code=404, detail="Package not found")
    pkg = _package_name(entry, pm)
    script, timeout, env = _uninstall_script(pm, pkg)
    await asyncio.to_thread(_run_shell, script, timeout, env)
    return {"status": True, "msg": "Uninstalled"}