vpn/app/core/signbox.py

from __future__ import annotations

import asyncio
import json
import os
from pathlib import Path

from app.core.errors import UnsupportedEngineError
from app.core.interface import EngineBackend
from app.models import (
    CapabilitySet,
    EngineKind,
    EngineState,
    RuntimeRequest,
    RuntimeSnapshot,
    RuntimeStats,
)


class SignBoxBackend(EngineBackend):
    """SignBox/sing-box adapter.

    Desktop proxy mode is intentionally process based. Android must use the native
    libbox AAR through `native/android`; this Python class never starts a subprocess
    on Android.
    """

    name = "signbox"
    capabilities = CapabilitySet(
        supports_vpn=False,
        supports_proxy=True,
        supports_socks=True,
        supports_http=True,
        supports_udp=True,
        supports_mobile=False,
        supports_desktop=True,
        supports_stats=True,
        supports_both=False,
    )

    def __init__(self, executable: str | None = None, state_dir: Path | None = None) -> None:
        self.executable = executable or os.getenv("SIGNBOX_BIN")
        self.capabilities = CapabilitySet(
            supports_proxy=bool(self.executable),
            supports_socks=bool(self.executable),
            supports_http=bool(self.executable),
            supports_udp=bool(self.executable),
            supports_desktop=bool(self.executable),
            supports_stats=bool(self.executable),
        )
        self.state_dir = state_dir or Path(".runtime")
        self._process: asyncio.subprocess.Process | None = None
        self._monitor_task: asyncio.Task[None] | None = None
        self._state = RuntimeSnapshot(engine=EngineKind.SIGNBOX)
        self._request: RuntimeRequest | None = None

    async def start(self, request: RuntimeRequest) -> RuntimeSnapshot:
        self._request = request
        self._state = RuntimeSnapshot(
            state=EngineState.STARTING,
            engine=EngineKind.SIGNBOX,
            mode=request.mode,
            profile_id=request.profile_id,
            revision=request.config_revision,
        )
        if not self.executable:
            self._state = self._unsupported("SIGNBOX_BIN is not configured for desktop")
            raise UnsupportedEngineError(self._state.message or "SignBox binary is missing")
        if request.mode.value != "proxy":
            self._state = self._unsupported(
                "desktop SignBox adapter currently supports proxy-only; Android uses native libbox"
            )
            raise UnsupportedEngineError(self._state.message or "unsupported desktop mode")
        if not request.engine_config:
            self._state = self._unsupported("SignBox profile config is empty")
            raise UnsupportedEngineError(self._state.message or "SignBox config is empty")
        self.state_dir.mkdir(parents=True, exist_ok=True)
        config_path = self.state_dir / f"signbox-{request.profile_id}.json"
        config_path.write_text(
            json.dumps(request.engine_config, indent=2), encoding="utf-8"
        )
        self._process = await asyncio.create_subprocess_exec(
            self.executable,
            "run",
            "-c",
            str(config_path),
            stdout=asyncio.subprocess.PIPE,
            stderr=asyncio.subprocess.PIPE,
        )
        self._monitor_task = asyncio.create_task(self._monitor(self._process))
        await asyncio.sleep(0)
        if self._process.returncode is not None:
            self._state = self._state.model_copy(
                update={"state": EngineState.ERROR, "message": "SignBox exited during startup"}
            )
            raise UnsupportedEngineError(self._state.message or "SignBox exited during startup")
        self._state = self._state.model_copy(update={"state": EngineState.RUNNING})
        return self._state

    async def stop(self) -> RuntimeSnapshot:
        if self._monitor_task is not None:
            self._monitor_task.cancel()
            self._monitor_task = None
        if self._process is not None:
            self._process.terminate()
            try:
                await asyncio.wait_for(self._process.wait(), timeout=5)
            except TimeoutError:
                self._process.kill()
                await self._process.wait()
            self._process = None
        self._state = self._state.model_copy(update={"state": EngineState.STOPPED})
        return self._state

    async def _monitor(self, process: asyncio.subprocess.Process) -> None:
        async def drain(stream: asyncio.StreamReader | None) -> None:
            if stream is None:
                return
            while await stream.readline():
                pass

        try:
            await asyncio.gather(drain(process.stdout), drain(process.stderr))
            await process.wait()
        except asyncio.CancelledError:
            raise
        if self._process is process and self._state.state not in {
            EngineState.STOPPED,
            EngineState.STOPPING,
        }:
            self._state = self._state.model_copy(
                update={"state": EngineState.ERROR, "message": "SignBox process exited unexpectedly"}
            )

    async def restart(self, request: RuntimeRequest) -> RuntimeSnapshot:
        await self.stop()
        return await self.start(request)

    async def get_state(self) -> RuntimeSnapshot:
        if self._process is not None and self._process.returncode is not None:
            self._state = self._state.model_copy(
                update={
                    "state": EngineState.ERROR,
                    "message": "SignBox process exited unexpectedly",
                }
            )
        return self._state

    async def get_stats(self) -> RuntimeStats:
        return self._state.stats

    def _unsupported(self, message: str) -> RuntimeSnapshot:
        return self._state.model_copy(update={"state": EngineState.UNSUPPORTED, "message": message})