sp-hydra-veil-core/app/utils/encrypted_proxy/singbox.py

import json
import os
import subprocess
import time
from pathlib import Path

from core.Constants import Constants


class SingboxRunner:
    """
    Gestiona el ciclo de vida de sing-box via wrapper sudo.

    Flujo:
        write_config() → start() → is_running() → stop()

    El wrapper es el único binario con NOPASSWD en sudoers.
    Nunca se llama sing-box ni ip directamente con sudo desde Python.
    """

    _WRAPPER    = Constants.SINGBOX_WRAPPER
    _PID_FILE   = Path(Constants.SINGBOX_PID_FILE)
    _LOG_FILE   = Path(Constants.SINGBOX_LOG_FILE)
    _CONFIG_DIR = Path(Constants.SINGBOX_CONFIG_DIR)

    _START_TIMEOUT_S = 10.0
    _START_POLL_S    = 0.5

    # ── Público ───────────────────────────────────────────────────────────────

    def start(self, config_path: Path) -> bool:
        """
        Para cualquier instancia previa y lanza sing-box via wrapper.
        Retorna True solo cuando el proceso está confirmado vivo.
        """
        self.stop()
        self._assert_config_safe(config_path)
        self._ensure_log_file()

        try:
            subprocess.Popen(
                ['sudo', self._WRAPPER, str(config_path), 'start'],
                stdout=subprocess.DEVNULL,
                stderr=subprocess.DEVNULL,  # el wrapper escribe el log como root
                stdin=subprocess.DEVNULL,
                env={**os.environ, 'SUDO_ASKPASS': '/bin/false'},
            )
        except FileNotFoundError:
            raise RuntimeError(
                f'Wrapper no encontrado: {self._WRAPPER}\n'
                'Ejecutá el installer primero.'
            )

        # Esperar confirmación por PID file, no por tiempo fijo
        deadline = time.monotonic() + self._START_TIMEOUT_S
        while time.monotonic() < deadline:
            time.sleep(self._START_POLL_S)
            if self.is_running():
                return True

        raise RuntimeError(
            f'sing-box no levantó en {self._START_TIMEOUT_S}s.\n'
            f'Último error:\n{self.get_last_error()}'
        )

    def stop(self) -> None:
        """
        Para sing-box limpiamente via wrapper.
        Nunca lanza excepción — stop debe ser siempre seguro de llamar.
        """
        try:
            subprocess.run(
                ['sudo', self._WRAPPER, '', 'stop'],
                stdout=subprocess.DEVNULL,
                stderr=subprocess.DEVNULL,
                stdin=subprocess.DEVNULL,
                env={**os.environ, 'SUDO_ASKPASS': '/bin/false'},
                timeout=15,
            )
        except (subprocess.TimeoutExpired, FileNotFoundError, OSError):
            # Wrapper no responde — kill directo por PID file como último recurso
            self._emergency_kill()

    def is_running(self) -> bool:
        """
        Verifica por PID file + señal 0.
        PermissionError significa que el proceso VIVE (corre como root).
        """
        if not self._PID_FILE.exists():
            return False
        try:
            pid = int(self._PID_FILE.read_text().strip())
            if pid <= 0:
                raise ValueError
            os.kill(pid, 0)
            return True
        except (ValueError, ProcessLookupError):
            # PID corrupto o proceso muerto — limpiar
            self._PID_FILE.unlink(missing_ok=True)
            return False
        except PermissionError:
            # Proceso vivo pero de root — PermissionError es señal de vida
            return True

    def get_last_error(self) -> str:
        """
        Lee las últimas 80 líneas del log via sudo para evitar
        problemas de permisos (el log lo escribe root via wrapper).
        """
        try:
            result = subprocess.run(
                ['sudo', self._WRAPPER, '', 'log'],
                capture_output=True,
                text=True,
                timeout=5,
                stdin=subprocess.DEVNULL,
                env={**os.environ, 'SUDO_ASKPASS': '/bin/false'},
            )
            return result.stdout.strip()
        except Exception:
            # Fallback: leer directo si tenemos permisos
            try:
                lines = self._LOG_FILE.read_text(errors='replace').splitlines()
                return '\n'.join(lines[-80:])
            except OSError:
                return ''

    def write_config(self, config_path: Path, config: dict) -> None:
        """
        Escribe el JSON de forma atómica con permisos 0o600.
        Write en .tmp + rename — nunca deja un config a medias.
        """
        self._assert_config_safe(config_path)
        config_path.parent.mkdir(parents=True, exist_ok=True, mode=0o700)

        tmp = config_path.with_suffix('.tmp')
        try:
            with open(tmp, 'w') as f:
                json.dump(config, f, indent=2)
            os.chmod(tmp, 0o600)
            tmp.rename(config_path)
        except Exception:
            tmp.unlink(missing_ok=True)
            raise

    # ── Privado ───────────────────────────────────────────────────────────────

    def _ensure_log_file(self) -> None:
        """
        Crea el log file con permisos del usuario actual si no existe.
        Debe llamarse ANTES de que el wrapper lo tome como root.
        """
        self._LOG_FILE.parent.mkdir(parents=True, exist_ok=True)
        if not self._LOG_FILE.exists():
            self._LOG_FILE.touch(mode=0o600)

    def _assert_config_safe(self, config_path: Path) -> None:
        """
        Verifica que el config está dentro del directorio permitido.
        Lanza ValueError si no — nunca continúa con path inseguro.
        """
        try:
            config_path.resolve().relative_to(self._CONFIG_DIR.resolve())
        except ValueError:
            raise ValueError(
                f'Config fuera del directorio permitido.\n'
                f'  Config:    {config_path}\n'
                f'  Permitido: {self._CONFIG_DIR}'
            )

    def _emergency_kill(self) -> None:
        """
        Kill directo por PID file cuando el wrapper no responde.
        Último recurso — no usa sudo.
        """
        if not self._PID_FILE.exists():
            return
        try:
            pid = int(self._PID_FILE.read_text().strip())
            if pid > 0:
                os.kill(pid, 9)
        except (ValueError, ProcessLookupError, PermissionError, OSError):
            pass
        finally:
            self._PID_FILE.unlink(missing_ok=True)