rmi-backend/app/_archive/legacy_2026_07/ghostunnel_manager.py

"""
Ghostunnel SSL/TLS Tunnel Integration
======================================

Secure SSL/TLS proxy with mutual authentication for non-TLS services.
Features:
- TLS termination
- Client certificate verification
- Proxying to TCP services
"""

import logging
import subprocess
from dataclasses import dataclass
from datetime import datetime

logger = logging.getLogger(__name__)


@dataclass
class GhostunnelConfig:
    """Ghostunnel configuration."""

    listen: str
    target: str
    cacert: str | None = None
    cert: str | None = None
    key: str | None = None
    clientcert: str | None = None
    acceptonly: str | None = None
    status: str | None = "127.0.0.1:8080"


# ─── GHOSTUNNEL MANAGER ───────────────────────────────────────────


class GhostunnelManager:
    """Manages Ghostunnel processes."""

    def __init__(self, ghostunnel_bin: str = "ghostunnel"):
        self.ghostunnel_bin = ghostunnel_bin
        self._available = self._check_available()
        self.processes: dict[str, subprocess.Popen] = {}

    def _check_available(self) -> bool:
        """Check if ghostunnel is installed."""
        try:
            result = subprocess.run([self.ghostunnel_bin, "version"], capture_output=True, timeout=5)
            return result.returncode == 0
        except (subprocess.TimeoutExpired, FileNotFoundError):
            return False

    def start_tunnel(self, config: GhostunnelConfig) -> str:
        """
        Start a new tunnel.

        Args:
            config: Ghostunnel configuration

        Returns:
            Process ID of the tunnel
        """
        if not self._available:
            return "stub_tunnel"

        cmd = [
            self.ghostunnel_bin,
            "server",
            "--listen",
            config.listen,
            "--target",
            config.target,
        ]

        if config.cacert:
            cmd.extend(["--cacert", config.cacert])
        if config.cert:
            cmd.extend(["--cert", config.cert])
        if config.key:
            cmd.extend(["--key", config.key])
        if config.clientcert:
            cmd.extend(["--clientcert", config.clientcert])
        if config.acceptonly:
            cmd.extend(["--acceptonly", config.acceptonly])
        if config.status:
            cmd.extend(["--status", config.status])

        try:
            process = subprocess.Popen(cmd, stdout=subprocess.PIPE, stderr=subprocess.PIPE)
            self.processes[config.listen] = process
            return str(process.pid)
        except Exception as e:
            logger.error(f"Failed to start ghostunnel: {e}")
            return ""

    def stop_tunnel(self, listen_addr: str) -> bool:
        """
        Stop a tunnel by listening address.

        Args:
            listen_addr: Listening address of the tunnel

        Returns:
            Success status
        """
        process = self.processes.get(listen_addr)
        if process:
            process.terminate()
            try:
                process.wait(timeout=5)
                del self.processes[listen_addr]
                return True
            except subprocess.TimeoutExpired:
                process.kill()
                del self.processes[listen_addr]
                return True
        return False

    def stop_all(self) -> int:
        """Stop all tunnels. Returns count of stopped processes."""
        count = 0
        for addr in list(self.processes.keys()):
            if self.stop_tunnel(addr):
                count += 1
        return count

    def get_status(self) -> dict[str, str]:
        """Get status of all tunnels."""
        status = {}
        for addr, process in self.processes.items():
            status[addr] = {
                "pid": process.pid,
                "alive": process.poll() is None,
                "started_at": datetime.utcnow().isoformat(),
            }
        return status


# ─── GLOBAL SINGLETON ─────────────────────────────────────────────

_manager: GhostunnelManager | None = None


def get_ghostunnel_manager() -> GhostunnelManager:
    """Get or create Ghostunnel manager."""
    global _manager
    if _manager is None:
        _manager = GhostunnelManager()
    return _manager


def start_tunnel(listen: str, target: str, **kwargs) -> str:
    """
    Convenience function to start a tunnel.

    Args:
        listen: Listen address (host:port)
        target: Target address (host:port)
        **kwargs: Extra config options
    """
    config = GhostunnelConfig(listen=listen, target=target, **kwargs)
    manager = get_ghostunnel_manager()
    return manager.start_tunnel(config)


def stop_tunnel(listen: str) -> bool:
    """Convenience function to stop a tunnel."""
    manager = get_ghostunnel_manager()
    return manager.stop_tunnel(listen)


def stop_all_tunnels() -> int:
    """Convenience function to stop all tunnels."""
    manager = get_ghostunnel_manager()
    return manager.stop_all()