| 1 | from __future__ import annotations |
| 2 | |
| 3 | import os |
| 4 | import select |
| 5 | import shutil |
| 6 | import socket |
| 7 | import subprocess |
| 8 | import threading |
| 9 | import time |
| 10 | import uuid |
| 11 | from pathlib import Path |
| 12 | from typing import Any |
| 13 | |
| 14 | from helpers import files, virtual_desktop |
| 15 | from helpers.print_style import PrintStyle |
| 16 | |
| 17 | |
| 18 | DEFAULT_WIDTH = 1024 |
| 19 | DEFAULT_HEIGHT = 768 |
| 20 | START_TIMEOUT_SECONDS = 15.0 |
| 21 | |
| 22 | |
| 23 | def keyboard_options() -> dict[str, str]: |
| 24 | """Resolve the configured XKB keyboard layout for browser displays.""" |
| 25 | from plugins._browser.helpers.config import ( |
| 26 | KEYBOARD_LAYOUT_KEY, |
| 27 | KEYBOARD_VARIANT_KEY, |
| 28 | get_browser_config, |
| 29 | ) |
| 30 | |
| 31 | config = get_browser_config() |
| 32 | return { |
| 33 | "layout": str(config.get(KEYBOARD_LAYOUT_KEY, "") or "").strip(), |
| 34 | "variant": str(config.get(KEYBOARD_VARIANT_KEY, "") or "").strip(), |
| 35 | } |
| 36 | |
| 37 | |
| 38 | def collect_status() -> dict[str, Any]: |
| 39 | binaries = { |
| 40 | name: shutil.which(name) or "" |
| 41 | for name in ("Xvfb", "xpra", "xrandr") |
| 42 | } |
| 43 | html_root = virtual_desktop.find_xpra_html_root() |
| 44 | missing = [name for name, path in binaries.items() if not path] |
| 45 | if not html_root: |
| 46 | missing.append("xpra-html5") |
| 47 | return { |
| 48 | "available": not missing, |
| 49 | "missing": missing, |
| 50 | "binaries": binaries, |
| 51 | "xpra_html_root": str(html_root or ""), |
| 52 | } |
| 53 | |
| 54 | |
| 55 | class BrowserInteractiveView: |
| 56 | """Own one private X display and its optional Xpra viewer.""" |
| 57 | |
| 58 | def __init__(self, context_id: str) -> None: |
| 59 | self.context_id = str(context_id) |
| 60 | self.token = f"browser-{uuid.uuid4().hex}" |
| 61 | self.state_dir = Path(files.get_abs_path("tmp", "browser", "displays", self.token)) |
| 62 | self.display: int | None = None |
| 63 | self.port = 0 |
| 64 | self.width = DEFAULT_WIDTH |
| 65 | self.height = DEFAULT_HEIGHT |
| 66 | self._xvfb: subprocess.Popen[Any] | None = None |
| 67 | self._xpra: subprocess.Popen[Any] | None = None |
| 68 | self._lock = threading.RLock() |
| 69 | |
| 70 | @property |
| 71 | def display_name(self) -> str: |
| 72 | return f":{self.display}" if self.display is not None else "" |
| 73 | |
| 74 | def ensure_display(self) -> str: |
| 75 | with self._lock: |
| 76 | if self._running(self._xvfb) and self.display is not None: |
| 77 | return self.display_name |
| 78 | |
| 79 | self._stop_locked() |
| 80 | xvfb = shutil.which("Xvfb") |
| 81 | if not xvfb: |
| 82 | return "" |
| 83 | |
| 84 | self.state_dir.mkdir(parents=True, exist_ok=True) |
| 85 | self.state_dir.chmod(0o700) |
| 86 | read_fd, write_fd = os.pipe() |
| 87 | try: |
| 88 | process = subprocess.Popen( |
| 89 | [ |
| 90 | xvfb, |
| 91 | "-displayfd", |
| 92 | str(write_fd), |
| 93 | "-screen", |
| 94 | "0", |
| 95 | f"{virtual_desktop.MAX_WIDTH}x{virtual_desktop.MAX_HEIGHT}x24", |
| 96 | "+extension", |
| 97 | "GLX", |
| 98 | "+extension", |
| 99 | "RANDR", |
| 100 | "+extension", |
| 101 | "RENDER", |
| 102 | "+extension", |
| 103 | "Composite", |
| 104 | "-nolisten", |
| 105 | "tcp", |
| 106 | "-noreset", |
| 107 | "-ac", |
| 108 | ], |
| 109 | stdin=subprocess.DEVNULL, |
| 110 | stdout=subprocess.DEVNULL, |
| 111 | stderr=subprocess.DEVNULL, |
| 112 | pass_fds=(write_fd,), |
| 113 | ) |
| 114 | except OSError: |
| 115 | os.close(read_fd) |
| 116 | return "" |
| 117 | finally: |
| 118 | os.close(write_fd) |
| 119 | |
| 120 | try: |
| 121 | ready, _, _ = select.select([read_fd], [], [], 5) |
| 122 | display_number = os.read(read_fd, 32).decode().strip() if ready else "" |
| 123 | finally: |
| 124 | os.close(read_fd) |
| 125 | |
| 126 | if not display_number.isdigit() or process.poll() is not None: |
| 127 | self._terminate(process) |
| 128 | return "" |
| 129 | |
| 130 | self._xvfb = process |
| 131 | self.display = int(display_number) |
| 132 | self._apply_keyboard_layout() |
| 133 | self.resize(self.width, self.height) |
| 134 | return self.display_name |
| 135 | |
| 136 | def ensure_viewer(self, width: int = 0, height: int = 0) -> dict[str, Any]: |
| 137 | with self._lock: |
| 138 | display_name = self.ensure_display() |
| 139 | if not display_name: |
| 140 | return self._unavailable("Xvfb is unavailable.") |
| 141 | |
| 142 | status = collect_status() |
| 143 | if not status["available"]: |
| 144 | return self._unavailable( |
| 145 | f"Interactive Browser runtime needs: {', '.join(status['missing'])}." |
| 146 | ) |
| 147 | |
| 148 | self.resize(width or self.width, height or self.height) |
| 149 | if not self._running(self._xpra): |
| 150 | try: |
| 151 | self._start_xpra(str(status["binaries"]["xpra"])) |
| 152 | except Exception as exc: |
| 153 | PrintStyle.warning(f"Interactive Browser viewer failed to start: {exc}") |
| 154 | self._terminate(self._xpra) |
| 155 | self._xpra = None |
| 156 | self.port = 0 |
| 157 | virtual_desktop.unregister_session(self.token) |
| 158 | return self._unavailable(str(exc)) |
| 159 | |
| 160 | virtual_desktop.register_session( |
| 161 | token=self.token, |
| 162 | host="127.0.0.1", |
| 163 | port=self.port, |
| 164 | owner="browser", |
| 165 | title="Browser", |
| 166 | resize=self.resize, |
| 167 | ) |
| 168 | return { |
| 169 | "available": True, |
| 170 | "token": self.token, |
| 171 | "url": virtual_desktop.session_url( |
| 172 | self.token, |
| 173 | title="Browser", |
| 174 | encoding="", |
| 175 | quality=90, |
| 176 | speed=90, |
| 177 | file_transfer=False, |
| 178 | printing=False, |
| 179 | ), |
| 180 | "width": self.width, |
| 181 | "height": self.height, |
| 182 | } |
| 183 | |
| 184 | def resize(self, width: int, height: int) -> dict[str, Any]: |
| 185 | with self._lock: |
| 186 | target_width, target_height = virtual_desktop.normalize_size(width, height) |
| 187 | self.width = target_width |
| 188 | self.height = target_height |
| 189 | if self.display is None or not self._running(self._xvfb): |
| 190 | return { |
| 191 | "ok": False, |
| 192 | "error": "Browser display is unavailable.", |
| 193 | "width": target_width, |
| 194 | "height": target_height, |
| 195 | } |
| 196 | result = virtual_desktop.resize_display( |
| 197 | display=self.display, |
| 198 | width=target_width, |
| 199 | height=target_height, |
| 200 | settle_seconds=0, |
| 201 | ) |
| 202 | return result |
| 203 | |
| 204 | def close(self) -> None: |
| 205 | with self._lock: |
| 206 | self._stop_locked() |
| 207 | shutil.rmtree(self.state_dir, ignore_errors=True) |
| 208 | |
| 209 | def _apply_keyboard_layout(self) -> None: |
| 210 | """Apply the configured XKB layout to this private display.""" |
| 211 | if self.display is None: |
| 212 | return |
| 213 | options = keyboard_options() |
| 214 | if not options["layout"]: |
| 215 | return |
| 216 | setxkbmap = shutil.which("setxkbmap") |
| 217 | if not setxkbmap: |
| 218 | return |
| 219 | command = [setxkbmap, "-display", self.display_name, "-layout", options["layout"]] |
| 220 | if options["variant"]: |
| 221 | command.extend(["-variant", options["variant"]]) |
| 222 | try: |
| 223 | subprocess.run( |
| 224 | command, |
| 225 | check=False, |
| 226 | stdin=subprocess.DEVNULL, |
| 227 | stdout=subprocess.DEVNULL, |
| 228 | stderr=subprocess.DEVNULL, |
| 229 | timeout=5, |
| 230 | ) |
| 231 | except (OSError, subprocess.TimeoutExpired): |
| 232 | pass |
| 233 | |
| 234 | def _keyboard_xpra_args(self) -> list[str]: |
| 235 | options = keyboard_options() |
| 236 | if not options["layout"]: |
| 237 | return [] |
| 238 | args = [ |
| 239 | "--keyboard-sync=no", |
| 240 | "--keyboard-layout", options["layout"], |
| 241 | ] |
| 242 | if options["variant"]: |
| 243 | args.extend(["--keyboard-variant", options["variant"]]) |
| 244 | return args |
| 245 | |
| 246 | def _start_xpra(self, xpra: str) -> None: |
| 247 | self.port = self._free_port() |
| 248 | runtime_dir = self.state_dir / "runtime" |
| 249 | socket_dir = self.state_dir / "sockets" |
| 250 | runtime_dir.mkdir(parents=True, exist_ok=True) |
| 251 | socket_dir.mkdir(parents=True, exist_ok=True) |
| 252 | runtime_dir.chmod(0o700) |
| 253 | env = { |
| 254 | **os.environ, |
| 255 | "DISPLAY": self.display_name, |
| 256 | "XDG_RUNTIME_DIR": str(runtime_dir), |
| 257 | } |
| 258 | self._xpra = subprocess.Popen( |
| 259 | [ |
| 260 | xpra, |
| 261 | "shadow", |
| 262 | self.display_name, |
| 263 | "--daemon=no", |
| 264 | "--mdns=no", |
| 265 | "--html=on", |
| 266 | "--tray=no", |
| 267 | "--system-tray=no", |
| 268 | "--notifications=no", |
| 269 | "--clipboard=yes", |
| 270 | "--clipboard-direction=both", |
| 271 | "--file-transfer=no", |
| 272 | "--open-files=no", |
| 273 | "--open-url=no", |
| 274 | "--printing=no", |
| 275 | "--audio=no", |
| 276 | "--speaker=off", |
| 277 | "--microphone=off", |
| 278 | "--sharing=yes", |
| 279 | "--resize-display=yes", |
| 280 | "--encoding=auto", |
| 281 | "--quality=90", |
| 282 | "--speed=90", |
| 283 | *self._keyboard_xpra_args(), |
| 284 | f"--bind-tcp=127.0.0.1:{self.port}", |
| 285 | f"--socket-dir={socket_dir}", |
| 286 | f"--log-dir={self.state_dir}", |
| 287 | "--log-file=xpra.log", |
| 288 | ], |
| 289 | stdin=subprocess.DEVNULL, |
| 290 | stdout=subprocess.DEVNULL, |
| 291 | stderr=subprocess.DEVNULL, |
| 292 | env=env, |
| 293 | ) |
| 294 | self._wait_for_port(self._xpra, self.port) |
| 295 | |
| 296 | def _stop_locked(self) -> None: |
| 297 | virtual_desktop.unregister_session(self.token) |
| 298 | self._terminate(self._xpra) |
| 299 | self._terminate(self._xvfb) |
| 300 | self._xpra = None |
| 301 | self._xvfb = None |
| 302 | self.port = 0 |
| 303 | self.display = None |
| 304 | |
| 305 | def _unavailable(self, error: str) -> dict[str, Any]: |
| 306 | return { |
| 307 | "available": False, |
| 308 | "error": str(error or "Interactive Browser viewer is unavailable."), |
| 309 | } |
| 310 | |
| 311 | @staticmethod |
| 312 | def _running(process: subprocess.Popen[Any] | None) -> bool: |
| 313 | return bool(process and process.poll() is None) |
| 314 | |
| 315 | @staticmethod |
| 316 | def _terminate(process: subprocess.Popen[Any] | None) -> None: |
| 317 | if not process or process.poll() is not None: |
| 318 | return |
| 319 | process.terminate() |
| 320 | try: |
| 321 | process.wait(timeout=2) |
| 322 | except subprocess.TimeoutExpired: |
| 323 | process.kill() |
| 324 | process.wait(timeout=2) |
| 325 | |
| 326 | @staticmethod |
| 327 | def _free_port() -> int: |
| 328 | with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as probe: |
| 329 | probe.bind(("127.0.0.1", 0)) |
| 330 | return int(probe.getsockname()[1]) |
| 331 | |
| 332 | @staticmethod |
| 333 | def _wait_for_port( |
| 334 | process: subprocess.Popen[Any], |
| 335 | port: int, |
| 336 | timeout: float = START_TIMEOUT_SECONDS, |
| 337 | ) -> None: |
| 338 | deadline = time.monotonic() + timeout |
| 339 | while time.monotonic() < deadline: |
| 340 | if process.poll() is not None: |
| 341 | raise RuntimeError("Xpra exited before its Browser endpoint was ready.") |
| 342 | try: |
| 343 | with socket.create_connection(("127.0.0.1", port), timeout=0.2): |
| 344 | return |
| 345 | except OSError: |
| 346 | time.sleep(0.1) |
| 347 | raise TimeoutError("Timed out waiting for the interactive Browser endpoint.") |