mirror of
https://github.com/agent0ai/agent-zero.git
synced 2026-08-30 02:33:53 +00:00
bb2a2ecf16
The private Browser Xvfb always started with the US XKB layout, so non-US keyboards could not type their printed characters in the interactive Browser window: German Mac users could not type "@" via Option+L, and umlauts, brackets, and AltGr/Option symbols landed on wrong keys or vanished. Add keyboard_layout / keyboard_variant Browser settings (validated as safe XKB tokens, empty keeps US behavior). Browser startup applies the layout to the private display with setxkbmap and pins it on the Xpra shadow server (--keyboard-sync=no, --keyboard-layout, --keyboard-variant) so connecting clients cannot override the server layout. Layout changes flow through browser_runtime_config and restart internal runtimes. The fallback canvas input path also stopped discarding Alt/AltGr combinations that produce printable characters, and the settings UI offers free-text layout/variant fields linking to the official xkeyboard-config list.
347 lines
12 KiB
Python
347 lines
12 KiB
Python
from __future__ import annotations
|
|
|
|
import os
|
|
import select
|
|
import shutil
|
|
import socket
|
|
import subprocess
|
|
import threading
|
|
import time
|
|
import uuid
|
|
from pathlib import Path
|
|
from typing import Any
|
|
|
|
from helpers import files, virtual_desktop
|
|
from helpers.print_style import PrintStyle
|
|
|
|
|
|
DEFAULT_WIDTH = 1024
|
|
DEFAULT_HEIGHT = 768
|
|
START_TIMEOUT_SECONDS = 15.0
|
|
|
|
|
|
def keyboard_options() -> dict[str, str]:
|
|
"""Resolve the configured XKB keyboard layout for browser displays."""
|
|
from plugins._browser.helpers.config import (
|
|
KEYBOARD_LAYOUT_KEY,
|
|
KEYBOARD_VARIANT_KEY,
|
|
get_browser_config,
|
|
)
|
|
|
|
config = get_browser_config()
|
|
return {
|
|
"layout": str(config.get(KEYBOARD_LAYOUT_KEY, "") or "").strip(),
|
|
"variant": str(config.get(KEYBOARD_VARIANT_KEY, "") or "").strip(),
|
|
}
|
|
|
|
|
|
def collect_status() -> dict[str, Any]:
|
|
binaries = {
|
|
name: shutil.which(name) or ""
|
|
for name in ("Xvfb", "xpra", "xrandr")
|
|
}
|
|
html_root = virtual_desktop.find_xpra_html_root()
|
|
missing = [name for name, path in binaries.items() if not path]
|
|
if not html_root:
|
|
missing.append("xpra-html5")
|
|
return {
|
|
"available": not missing,
|
|
"missing": missing,
|
|
"binaries": binaries,
|
|
"xpra_html_root": str(html_root or ""),
|
|
}
|
|
|
|
|
|
class BrowserInteractiveView:
|
|
"""Own one private X display and its optional Xpra viewer."""
|
|
|
|
def __init__(self, context_id: str) -> None:
|
|
self.context_id = str(context_id)
|
|
self.token = f"browser-{uuid.uuid4().hex}"
|
|
self.state_dir = Path(files.get_abs_path("tmp", "browser", "displays", self.token))
|
|
self.display: int | None = None
|
|
self.port = 0
|
|
self.width = DEFAULT_WIDTH
|
|
self.height = DEFAULT_HEIGHT
|
|
self._xvfb: subprocess.Popen[Any] | None = None
|
|
self._xpra: subprocess.Popen[Any] | None = None
|
|
self._lock = threading.RLock()
|
|
|
|
@property
|
|
def display_name(self) -> str:
|
|
return f":{self.display}" if self.display is not None else ""
|
|
|
|
def ensure_display(self) -> str:
|
|
with self._lock:
|
|
if self._running(self._xvfb) and self.display is not None:
|
|
return self.display_name
|
|
|
|
self._stop_locked()
|
|
xvfb = shutil.which("Xvfb")
|
|
if not xvfb:
|
|
return ""
|
|
|
|
self.state_dir.mkdir(parents=True, exist_ok=True)
|
|
self.state_dir.chmod(0o700)
|
|
read_fd, write_fd = os.pipe()
|
|
try:
|
|
process = subprocess.Popen(
|
|
[
|
|
xvfb,
|
|
"-displayfd",
|
|
str(write_fd),
|
|
"-screen",
|
|
"0",
|
|
f"{virtual_desktop.MAX_WIDTH}x{virtual_desktop.MAX_HEIGHT}x24",
|
|
"+extension",
|
|
"GLX",
|
|
"+extension",
|
|
"RANDR",
|
|
"+extension",
|
|
"RENDER",
|
|
"+extension",
|
|
"Composite",
|
|
"-nolisten",
|
|
"tcp",
|
|
"-noreset",
|
|
"-ac",
|
|
],
|
|
stdin=subprocess.DEVNULL,
|
|
stdout=subprocess.DEVNULL,
|
|
stderr=subprocess.DEVNULL,
|
|
pass_fds=(write_fd,),
|
|
)
|
|
except OSError:
|
|
os.close(read_fd)
|
|
return ""
|
|
finally:
|
|
os.close(write_fd)
|
|
|
|
try:
|
|
ready, _, _ = select.select([read_fd], [], [], 5)
|
|
display_number = os.read(read_fd, 32).decode().strip() if ready else ""
|
|
finally:
|
|
os.close(read_fd)
|
|
|
|
if not display_number.isdigit() or process.poll() is not None:
|
|
self._terminate(process)
|
|
return ""
|
|
|
|
self._xvfb = process
|
|
self.display = int(display_number)
|
|
self._apply_keyboard_layout()
|
|
self.resize(self.width, self.height)
|
|
return self.display_name
|
|
|
|
def ensure_viewer(self, width: int = 0, height: int = 0) -> dict[str, Any]:
|
|
with self._lock:
|
|
display_name = self.ensure_display()
|
|
if not display_name:
|
|
return self._unavailable("Xvfb is unavailable.")
|
|
|
|
status = collect_status()
|
|
if not status["available"]:
|
|
return self._unavailable(
|
|
f"Interactive Browser runtime needs: {', '.join(status['missing'])}."
|
|
)
|
|
|
|
self.resize(width or self.width, height or self.height)
|
|
if not self._running(self._xpra):
|
|
try:
|
|
self._start_xpra(str(status["binaries"]["xpra"]))
|
|
except Exception as exc:
|
|
PrintStyle.warning(f"Interactive Browser viewer failed to start: {exc}")
|
|
self._terminate(self._xpra)
|
|
self._xpra = None
|
|
self.port = 0
|
|
virtual_desktop.unregister_session(self.token)
|
|
return self._unavailable(str(exc))
|
|
|
|
virtual_desktop.register_session(
|
|
token=self.token,
|
|
host="127.0.0.1",
|
|
port=self.port,
|
|
owner="browser",
|
|
title="Browser",
|
|
resize=self.resize,
|
|
)
|
|
return {
|
|
"available": True,
|
|
"token": self.token,
|
|
"url": virtual_desktop.session_url(
|
|
self.token,
|
|
title="Browser",
|
|
encoding="",
|
|
quality=90,
|
|
speed=90,
|
|
file_transfer=False,
|
|
printing=False,
|
|
),
|
|
"width": self.width,
|
|
"height": self.height,
|
|
}
|
|
|
|
def resize(self, width: int, height: int) -> dict[str, Any]:
|
|
with self._lock:
|
|
target_width, target_height = virtual_desktop.normalize_size(width, height)
|
|
self.width = target_width
|
|
self.height = target_height
|
|
if self.display is None or not self._running(self._xvfb):
|
|
return {
|
|
"ok": False,
|
|
"error": "Browser display is unavailable.",
|
|
"width": target_width,
|
|
"height": target_height,
|
|
}
|
|
result = virtual_desktop.resize_display(
|
|
display=self.display,
|
|
width=target_width,
|
|
height=target_height,
|
|
settle_seconds=0,
|
|
)
|
|
return result
|
|
|
|
def close(self) -> None:
|
|
with self._lock:
|
|
self._stop_locked()
|
|
shutil.rmtree(self.state_dir, ignore_errors=True)
|
|
|
|
def _apply_keyboard_layout(self) -> None:
|
|
"""Apply the configured XKB layout to this private display."""
|
|
if self.display is None:
|
|
return
|
|
options = keyboard_options()
|
|
if not options["layout"]:
|
|
return
|
|
setxkbmap = shutil.which("setxkbmap")
|
|
if not setxkbmap:
|
|
return
|
|
command = [setxkbmap, "-display", self.display_name, "-layout", options["layout"]]
|
|
if options["variant"]:
|
|
command.extend(["-variant", options["variant"]])
|
|
try:
|
|
subprocess.run(
|
|
command,
|
|
check=False,
|
|
stdin=subprocess.DEVNULL,
|
|
stdout=subprocess.DEVNULL,
|
|
stderr=subprocess.DEVNULL,
|
|
timeout=5,
|
|
)
|
|
except (OSError, subprocess.TimeoutExpired):
|
|
pass
|
|
|
|
def _keyboard_xpra_args(self) -> list[str]:
|
|
options = keyboard_options()
|
|
if not options["layout"]:
|
|
return []
|
|
args = [
|
|
"--keyboard-sync=no",
|
|
"--keyboard-layout", options["layout"],
|
|
]
|
|
if options["variant"]:
|
|
args.extend(["--keyboard-variant", options["variant"]])
|
|
return args
|
|
|
|
def _start_xpra(self, xpra: str) -> None:
|
|
self.port = self._free_port()
|
|
runtime_dir = self.state_dir / "runtime"
|
|
socket_dir = self.state_dir / "sockets"
|
|
runtime_dir.mkdir(parents=True, exist_ok=True)
|
|
socket_dir.mkdir(parents=True, exist_ok=True)
|
|
runtime_dir.chmod(0o700)
|
|
env = {
|
|
**os.environ,
|
|
"DISPLAY": self.display_name,
|
|
"XDG_RUNTIME_DIR": str(runtime_dir),
|
|
}
|
|
self._xpra = subprocess.Popen(
|
|
[
|
|
xpra,
|
|
"shadow",
|
|
self.display_name,
|
|
"--daemon=no",
|
|
"--mdns=no",
|
|
"--html=on",
|
|
"--tray=no",
|
|
"--system-tray=no",
|
|
"--notifications=no",
|
|
"--clipboard=yes",
|
|
"--clipboard-direction=both",
|
|
"--file-transfer=no",
|
|
"--open-files=no",
|
|
"--open-url=no",
|
|
"--printing=no",
|
|
"--audio=no",
|
|
"--speaker=off",
|
|
"--microphone=off",
|
|
"--sharing=yes",
|
|
"--resize-display=yes",
|
|
"--encoding=auto",
|
|
"--quality=90",
|
|
"--speed=90",
|
|
*self._keyboard_xpra_args(),
|
|
f"--bind-tcp=127.0.0.1:{self.port}",
|
|
f"--socket-dir={socket_dir}",
|
|
f"--log-dir={self.state_dir}",
|
|
"--log-file=xpra.log",
|
|
],
|
|
stdin=subprocess.DEVNULL,
|
|
stdout=subprocess.DEVNULL,
|
|
stderr=subprocess.DEVNULL,
|
|
env=env,
|
|
)
|
|
self._wait_for_port(self._xpra, self.port)
|
|
|
|
def _stop_locked(self) -> None:
|
|
virtual_desktop.unregister_session(self.token)
|
|
self._terminate(self._xpra)
|
|
self._terminate(self._xvfb)
|
|
self._xpra = None
|
|
self._xvfb = None
|
|
self.port = 0
|
|
self.display = None
|
|
|
|
def _unavailable(self, error: str) -> dict[str, Any]:
|
|
return {
|
|
"available": False,
|
|
"error": str(error or "Interactive Browser viewer is unavailable."),
|
|
}
|
|
|
|
@staticmethod
|
|
def _running(process: subprocess.Popen[Any] | None) -> bool:
|
|
return bool(process and process.poll() is None)
|
|
|
|
@staticmethod
|
|
def _terminate(process: subprocess.Popen[Any] | None) -> None:
|
|
if not process or process.poll() is not None:
|
|
return
|
|
process.terminate()
|
|
try:
|
|
process.wait(timeout=2)
|
|
except subprocess.TimeoutExpired:
|
|
process.kill()
|
|
process.wait(timeout=2)
|
|
|
|
@staticmethod
|
|
def _free_port() -> int:
|
|
with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as probe:
|
|
probe.bind(("127.0.0.1", 0))
|
|
return int(probe.getsockname()[1])
|
|
|
|
@staticmethod
|
|
def _wait_for_port(
|
|
process: subprocess.Popen[Any],
|
|
port: int,
|
|
timeout: float = START_TIMEOUT_SECONDS,
|
|
) -> None:
|
|
deadline = time.monotonic() + timeout
|
|
while time.monotonic() < deadline:
|
|
if process.poll() is not None:
|
|
raise RuntimeError("Xpra exited before its Browser endpoint was ready.")
|
|
try:
|
|
with socket.create_connection(("127.0.0.1", port), timeout=0.2):
|
|
return
|
|
except OSError:
|
|
time.sleep(0.1)
|
|
raise TimeoutError("Timed out waiting for the interactive Browser endpoint.")
|