Stend_mont/universal-runtime/manager.py

#!/usr/bin/env python3
import json
import os
import signal
import subprocess
import threading
from http.server import BaseHTTPRequestHandler, HTTPServer

DISPLAY = os.environ.get("DISPLAY", ":1")
CHROME_WINDOW_SIZE = os.environ.get("CHROME_WINDOW_SIZE", "1920,1080")
RESOLUTION_MIN_WIDTH = int(os.environ.get("WEB_RESOLUTION_MIN_WIDTH", "1024"))
RESOLUTION_MIN_HEIGHT = int(os.environ.get("WEB_RESOLUTION_MIN_HEIGHT", "720"))
RESOLUTION_MAX_WIDTH = int(os.environ.get("WEB_RESOLUTION_MAX_WIDTH", "3840"))
RESOLUTION_MAX_HEIGHT = int(os.environ.get("WEB_RESOLUTION_MAX_HEIGHT", "2160"))

_state = {
    "proc": None,
    "mode": "idle",
    "target": "",
    "resolution": CHROME_WINDOW_SIZE,
}
_lock = threading.Lock()


def _stop_current() -> None:
    proc = _state.get("proc")
    if not proc:
        return
    try:
        os.killpg(os.getpgid(proc.pid), signal.SIGTERM)
        proc.wait(timeout=4)
    except Exception:
        try:
            os.killpg(os.getpgid(proc.pid), signal.SIGKILL)
        except Exception:
            pass
    finally:
        _state["proc"] = None


def _start_process(cmd: list[str], mode: str, target: str) -> None:
    _stop_current()
    logf = open("/tmp/session-app.log", "a", buffering=1)
    env = os.environ.copy()
    env["DISPLAY"] = DISPLAY
    proc = subprocess.Popen(  # noqa: S603
        cmd,
        stdout=logf,
        stderr=subprocess.STDOUT,
        env=env,
        start_new_session=True,
    )
    _state["proc"] = proc
    _state["mode"] = mode
    _state["target"] = target


def _sanitize_resolution(width: int | None, height: int | None) -> tuple[int, int]:
    if not width or not height:
        try:
            default_w, default_h = [int(x) for x in CHROME_WINDOW_SIZE.split(",", 1)]
            return default_w, default_h
        except Exception:
            return 1920, 1080

    safe_w = max(RESOLUTION_MIN_WIDTH, min(int(width), RESOLUTION_MAX_WIDTH))
    safe_h = max(RESOLUTION_MIN_HEIGHT, min(int(height), RESOLUTION_MAX_HEIGHT))
    return safe_w, safe_h


def apply_resolution(width: int | None, height: int | None) -> tuple[int, int]:
    safe_w, safe_h = _sanitize_resolution(width, height)
    # Best effort: Xvfb usually exposes RandR and accepts xrandr -s.
    try:
        subprocess.run(  # noqa: S603
            ["xrandr", "-display", DISPLAY, "-s", f"{safe_w}x{safe_h}"],
            check=False,
            stdout=subprocess.DEVNULL,
            stderr=subprocess.DEVNULL,
        )
    except Exception:
        pass
    _state["resolution"] = f"{safe_w},{safe_h}"
    return safe_w, safe_h


def open_web(url: str, width: int | None = None, height: int | None = None) -> None:
    safe_w, safe_h = apply_resolution(width, height)
    cmd = [
        "chromium",
        "--no-sandbox",
        "--disable-dev-shm-usage",
        "--disable-gpu",
        "--use-gl=swiftshader",
        "--kiosk",
        "--disable-translate",
        "--disable-features=TranslateUI,ExtensionsToolbarMenu",
        "--disable-pinch",
        "--overscroll-history-navigation=0",
        "--ignore-certificate-errors",
        "--allow-insecure-localhost",
        "--allow-running-insecure-content",
        f"--window-size={safe_w},{safe_h}",
        "--no-first-run",
        "--no-default-browser-check",
        url,
    ]
    _start_process(cmd, "web", url)


def open_rdp(payload: dict) -> None:
    host = (payload.get("host") or "").strip()
    if not host:
        raise ValueError("host is required")
    port = str(payload.get("port") or "3389").strip()
    user = (payload.get("user") or "").strip()
    password = (payload.get("password") or "").strip()
    domain = (payload.get("domain") or "").strip()
    security = (payload.get("security") or "").strip().lower()

    cmd = [
        "xfreerdp",
        f"/v:{host}:{port}",
        "/cert:ignore",
        "/f",
        "/dynamic-resolution",
        "/network:auto",
        "+clipboard",
    ]
    if security:
        cmd.append(f"/sec:{security}")
    if user:
        cmd.append(f"/u:{user}")
    if password:
        cmd.append(f"/p:{password}")
    if domain:
        cmd.append(f"/d:{domain}")

    safe_target = f"{host}:{port}"
    _start_process(cmd, "rdp", safe_target)


class Handler(BaseHTTPRequestHandler):
    def _read_json(self):
        length = int(self.headers.get("Content-Length", "0"))
        if length <= 0:
            return {}
        raw = self.rfile.read(length)
        return json.loads(raw.decode("utf-8"))

    def _json(self, code: int, payload: dict):
        body = json.dumps(payload).encode("utf-8")
        self.send_response(code)
        self.send_header("Content-Type", "application/json")
        self.send_header("Content-Length", str(len(body)))
        self.end_headers()
        self.wfile.write(body)

    def do_GET(self):
        if self.path == "/health":
            proc = _state.get("proc")
            running = bool(proc and proc.poll() is None)
            self._json(
                200,
                {
                    "ok": True,
                    "mode": _state.get("mode", "idle"),
                    "running": running,
                    "target": _state.get("target", ""),
                    "resolution": _state.get("resolution", CHROME_WINDOW_SIZE),
                },
            )
            return
        self._json(404, {"detail": "Not found"})

    def do_POST(self):
        try:
            data = self._read_json()
            if self.path == "/open":
                url = (data.get("url") or "").strip()
                if not (url.startswith("http://") or url.startswith("https://")):
                    self._json(400, {"detail": "Invalid URL"})
                    return
                width = data.get("width")
                height = data.get("height")
                with _lock:
                    open_web(url, width=width, height=height)
                self._json(
                    200,
                    {
                        "ok": True,
                        "mode": "web",
                        "target": url,
                        "resolution": _state.get("resolution", CHROME_WINDOW_SIZE),
                    },
                )
                return
            if self.path == "/resolution":
                width = data.get("width")
                height = data.get("height")
                with _lock:
                    safe_w, safe_h = apply_resolution(width, height)
                self._json(200, {"ok": True, "width": safe_w, "height": safe_h})
                return
            if self.path == "/rdp":
                with _lock:
                    open_rdp(data)
                self._json(200, {"ok": True, "mode": "rdp"})
                return
            self._json(404, {"detail": "Not found"})
        except Exception as exc:
            self._json(500, {"detail": str(exc)})

    def log_message(self, fmt, *args):
        return


if __name__ == "__main__":
    server = HTTPServer(("0.0.0.0", 7000), Handler)
    server.serve_forever()