#!/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. applied = False try: result = subprocess.run( # noqa: S603 ["xrandr", "-display", DISPLAY, "-s", f"{safe_w}x{safe_h}"], check=False, stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL, ) applied = result.returncode == 0 except Exception: applied = False if not applied: # Fallback to default geometry if requested mode is unsupported. try: fallback_w, fallback_h = [int(x) for x in CHROME_WINDOW_SIZE.split(",", 1)] except Exception: fallback_w, fallback_h = 1920, 1080 safe_w, safe_h = _sanitize_resolution(fallback_w, fallback_h) 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()