ServerKit/backend/app/utils/system.py

"""Centralized system utilities for subprocess handling.

Provides consistent privilege escalation, distro detection, package management,
and systemd service control so individual services don't need to reinvent these.
"""

import os
import shutil
import subprocess
from typing import List, Optional, Union


def _needs_sudo() -> bool:
    """Return True if the current process should prepend sudo to commands.

    Returns False when:
    - Running on Windows (no sudo concept; dev environment)
    - Already running as root (e.g. inside Docker)
    - ``sudo`` is not installed (minimal containers)
    """
    if os.name == 'nt':
        return False
    if os.geteuid() == 0:
        return False
    if not shutil.which('sudo'):
        return False
    return True


def privileged_cmd(cmd: Union[List[str], str], *, user: Optional[str] = None) -> Union[List[str], str]:
    """Return *cmd* with ``sudo`` prepended when necessary.

    Use this when you need the command list for ``Popen`` or other non-``run``
    callers.  For simple ``subprocess.run`` calls prefer :func:`run_privileged`.

    Pass *user* to run the command as a specific user (``sudo -u <user>``).

    ``sudo -n`` (non-interactive) is always used: nothing here runs attached to a
    human terminal, and callers capture output, so a password prompt would be
    invisible AND unanswerable — sudo would simply block forever, hanging the
    caller (this hung backend startup on a non-root host, via the metadata
    guard's iptables probe). Failing immediately with a non-zero exit is the
    only useful outcome, and every caller already handles that.
    """
    if isinstance(cmd, str):
        if _needs_sudo() and not cmd.lstrip().startswith('sudo '):
            if user:
                return f'sudo -n -u {user} {cmd}'
            return f'sudo -n {cmd}'
        return cmd

    cmd = list(cmd)

    # Give argv[0] an absolute path ONLY when $PATH cannot resolve it. subprocess
    # searches $PATH and nothing else, so a tool that exists but sits outside it
    # raises FileNotFoundError — which callers wrapping the probe in a broad
    # `except` then report as "not installed". This is how a successfully
    # installed ufw (in /usr/sbin, absent from the unit's PATH) surfaced in the
    # panel as "No Firewall Installed". Left untouched when $PATH already works,
    # so a resolvable command is passed through exactly as the caller wrote it.
    if cmd and cmd[0] != 'sudo' and not os.path.isabs(cmd[0]) and not shutil.which(cmd[0]):
        resolved = resolve_command(cmd[0])
        if resolved:
            cmd[0] = resolved

    if _needs_sudo() and cmd[0] != 'sudo':
        if user:
            return ['sudo', '-n', '-u', user] + cmd
        return ['sudo', '-n'] + cmd
    return cmd


# Ceiling for a privileged command that does not name its own. Generous enough
# for the slow-but-legitimate work that runs through here (package installs,
# image pulls), while guaranteeing no single call can wedge a worker forever.
# Anything genuinely longer must say so explicitly with `timeout=`.
DEFAULT_PRIVILEGED_TIMEOUT = 300

# Read-only status probes (is this package installed, is this unit active).
# They answer immediately or not at all, so they get a much tighter ceiling.
PROBE_TIMEOUT = 30


def run_privileged(cmd: Union[List[str], str], *, user: Optional[str] = None, **kwargs) -> subprocess.CompletedProcess:
    """Run a command with sudo if the current process is not root.

    Prepends ``sudo`` only when needed (not root, not Windows, sudo exists).
    Pass *user* to run the command as a specific user (``sudo -u <user>``).
    Defaults to ``capture_output=True, text=True`` but callers can override.

    A default ``timeout`` is applied when the caller does not give one: with
    output captured and no terminal, a command that never returns takes its
    caller with it silently and forever. ``TimeoutExpired`` is a far better
    outcome than a wedged request or a boot that never finishes — pass
    ``timeout=None`` to opt out deliberately.

    Returns the raw ``CompletedProcess`` so services keep their existing
    error-handling patterns.
    """
    cmd = privileged_cmd(cmd, user=user)
    kwargs.setdefault('capture_output', True)
    kwargs.setdefault('text', True)
    kwargs.setdefault('timeout', DEFAULT_PRIVILEGED_TIMEOUT)
    return subprocess.run(cmd, **kwargs)


def run_unprivileged(cmd: Union[List[str], str], *, timeout: int = 60,
                     capture_stderr: bool = False, **kwargs) -> dict:
    """Run a command WITHOUT privilege escalation; dict with stdout/stderr/rc.

    Named for what it does not do. It was called ``run_command`` — a name that
    described nothing and sat one autocomplete away from
    :func:`run_privileged`, so ``nginx -t`` went through it and could neither
    read root-owned config nor find /usr/sbin/nginx. The caller reported every
    config as invalid. ``run_unprivileged(['nginx', '-t'])`` reads as obviously
    wrong; ``run_command(['nginx', '-t'])`` did not.

    The old name was also ambiguous across the codebase: ``PythonService`` has
    an unrelated method of the same name.

    Use :func:`run_privileged` unless the command genuinely must not gain root.
    That is the common case — this helper has a handful of callers and they are
    all reads that work fine as the invoking user.
    """
    # Same argv[0] resolution as privileged_cmd: this helper does no privilege
    # escalation, so nothing else would find a command outside $PATH. Without
    # it `run_unprivileged(['nginx', '-t'])` raised FileNotFoundError on every
    # root install (nginx is /usr/sbin/nginx). Only applied when $PATH
    # genuinely misses the command.
    if isinstance(cmd, (list, tuple)):
        cmd = list(cmd)
        if cmd and cmd[0] != 'sudo' and not os.path.isabs(cmd[0]) and not shutil.which(cmd[0]):
            resolved = resolve_command(cmd[0])
            if resolved:
                cmd[0] = resolved

    kwargs.setdefault('capture_output', True)
    kwargs.setdefault('text', True)
    result = subprocess.run(cmd, timeout=timeout, **kwargs)
    return {
        'stdout': result.stdout or '',
        'stderr': result.stderr or '',
        'returncode': result.returncode,
    }


# Searched when ``$PATH`` does not resolve a command. Ordered the way a login
# shell would. ``/usr/sbin`` and ``/sbin`` matter most: the panel's systemd unit
# ships a PATH of venv:/usr/local/bin:/usr/bin:/bin, so every sbin-resident tool
# (ufw, iptables, nft) is unreachable by bare name from the service. ``/snap/bin``
# is included because snap is the recommended install route for certbot and is
# almost never on a service's PATH.
_COMMAND_SEARCH_DIRS = ('/usr/local/sbin', '/usr/local/bin', '/usr/sbin',
                        '/usr/bin', '/sbin', '/bin', '/snap/bin')


def resolve_command(cmd: str) -> Optional[str]:
    """Absolute path to *cmd*, or None when it cannot be found anywhere.

    ``shutil.which`` first, then :data:`_COMMAND_SEARCH_DIRS`. Knowing a command
    exists is not enough to run it: ``subprocess`` resolves argv[0] through
    ``$PATH`` alone and raises ``FileNotFoundError`` when that misses, so a
    caller that only checked availability would still fail to exec.
    """
    if os.path.isabs(cmd):
        return cmd if os.path.exists(cmd) else None

    found = shutil.which(cmd)
    if found:
        return found

    for directory in _COMMAND_SEARCH_DIRS:
        candidate = os.path.join(directory, cmd)
        if os.path.exists(candidate):
            return candidate

    return None


def is_command_available(cmd: str) -> bool:
    """Check whether *cmd* is available on the system.

    Uses ``shutil.which`` first, then falls back to the same sbin/local paths
    :func:`resolve_command` searches — so "available" and "runnable" can never
    disagree.
    """
    return resolve_command(cmd) is not None


def sourced_result(lines: list, source: str, source_label: str) -> dict:
    """Standard response shape for multi-source data endpoints.

    Every fallback-chain endpoint should return this shape so the frontend
    can show a consistent source-aware banner.
    """
    return {
        'success': True,
        'lines': lines,
        'count': len(lines),
        'source': source,
        'source_label': source_label,
    }


class PackageManager:
    """Cross-distro package management helpers.

    Detects ``apt``, ``dnf``, or ``yum`` once and caches the result.
    """

    _detected: Optional[str] = None
    _detection_done: bool = False

    @classmethod
    def detect(cls) -> Optional[str]:
        """Return ``'apt'``, ``'dnf'``, ``'yum'``, or ``None``."""
        if cls._detection_done:
            return cls._detected

        for manager in ('apt', 'dnf', 'yum'):
            if shutil.which(manager):
                cls._detected = manager
                cls._detection_done = True
                return cls._detected

        cls._detection_done = True
        return cls._detected

    @classmethod
    def is_available(cls) -> bool:
        """Return ``True`` if any supported package manager was found."""
        return cls.detect() is not None

    @classmethod
    def is_installed(cls, package: str) -> bool:
        """Check whether *package* is installed (cross-distro).

        Uses ``dpkg -s`` on apt systems and ``rpm -q`` on dnf/yum systems.
        Catches ``FileNotFoundError`` so it works on any distro.
        """
        manager = cls.detect()

        if manager == 'apt':
            try:
                result = subprocess.run(
                    ['dpkg', '-s', package],
                    capture_output=True, text=True, timeout=PROBE_TIMEOUT,
                )
                return (
                    result.returncode == 0
                    and 'Status: install ok installed' in result.stdout
                )
            except FileNotFoundError:
                return False

        if manager in ('dnf', 'yum'):
            try:
                result = subprocess.run(
                    ['rpm', '-q', package],
                    capture_output=True, text=True, timeout=PROBE_TIMEOUT,
                )
                return result.returncode == 0
            except FileNotFoundError:
                return False

        return False

    @classmethod
    def install(cls, packages: Union[str, List[str]], timeout: int = 300) -> subprocess.CompletedProcess:
        """Install one or more packages (cross-distro).

        Raises ``RuntimeError`` when no supported package manager is found.
        """
        manager = cls.detect()
        if manager is None:
            raise RuntimeError('No supported package manager found (apt/dnf/yum)')

        if isinstance(packages, str):
            packages = [packages]

        cmd = [manager, 'install', '-y'] + packages
        return run_privileged(cmd, timeout=timeout)

    @classmethod
    def reset_cache(cls) -> None:
        """Reset the cached detection (useful in tests)."""
        cls._detected = None
        cls._detection_done = False


class ServiceControl:
    """Thin wrappers around ``systemctl`` that use :func:`run_privileged`."""

    @staticmethod
    def start(service: str, **kwargs) -> subprocess.CompletedProcess:
        return run_privileged(['systemctl', 'start', service], **kwargs)

    @staticmethod
    def stop(service: str, **kwargs) -> subprocess.CompletedProcess:
        return run_privileged(['systemctl', 'stop', service], **kwargs)

    @staticmethod
    def restart(service: str, **kwargs) -> subprocess.CompletedProcess:
        return run_privileged(['systemctl', 'restart', service], **kwargs)

    @staticmethod
    def reload(service: str, **kwargs) -> subprocess.CompletedProcess:
        return run_privileged(['systemctl', 'reload', service], **kwargs)

    @staticmethod
    def enable(service: str, **kwargs) -> subprocess.CompletedProcess:
        return run_privileged(['systemctl', 'enable', service], **kwargs)

    @staticmethod
    def disable(service: str, **kwargs) -> subprocess.CompletedProcess:
        return run_privileged(['systemctl', 'disable', service], **kwargs)

    @staticmethod
    def daemon_reload(**kwargs) -> subprocess.CompletedProcess:
        return run_privileged(['systemctl', 'daemon-reload'], **kwargs)

    @staticmethod
    def is_active(service: str) -> bool:
        """Return ``True`` when the service is active.  No sudo needed.

        Raises ``FileNotFoundError`` when ``systemctl`` itself is missing: a
        host without systemd is "could not check", not "not running" — the
        doctor converts that to a warn row, and swallowing it here made every
        such host read as a repairable failure. Callers that want the old
        leniency guard their own call (php/postfix/security services do).
        """
        result = subprocess.run(
            ['systemctl', 'is-active', service],
            capture_output=True, text=True, timeout=PROBE_TIMEOUT,
        )
        return result.stdout.strip() == 'active'

    @staticmethod
    def is_enabled(service: str) -> bool:
        """Return ``True`` when the service is enabled.  No sudo needed."""
        try:
            result = subprocess.run(
                ['systemctl', 'is-enabled', service],
                capture_output=True, text=True, timeout=PROBE_TIMEOUT,
            )
            return result.stdout.strip() == 'enabled'
        except FileNotFoundError:
            return False