1510 lines
62 KiB
Python
1510 lines
62 KiB
Python
"""Native browser runtime ownership and lifecycle management."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import fcntl
|
|
import hashlib
|
|
import http.client
|
|
import json
|
|
import logging
|
|
import os
|
|
import re
|
|
import shutil
|
|
import socket
|
|
import subprocess
|
|
import threading
|
|
import time
|
|
from collections.abc import Callable, Mapping
|
|
from contextlib import suppress
|
|
from dataclasses import dataclass, fields
|
|
from pathlib import Path
|
|
from typing import Any, Protocol
|
|
from urllib.parse import urlsplit
|
|
|
|
from .proxy import ProxyExit, ProxyRegistry
|
|
|
|
LOG = logging.getLogger("creatorhub.runtime")
|
|
# Host-managed Xvfb publishes display sockets here; this is not a gateway temp directory.
|
|
X11_SOCKET_DIR = Path(os.sep, "tmp", ".X11-unix")
|
|
|
|
RUNTIME_ID_RE = re.compile(r"^[a-f0-9]{64}$")
|
|
ALIAS_RE = re.compile(r"^[a-z0-9][a-z0-9-]{0,31}$")
|
|
NETWORK_ID_RE = re.compile(r"^native-[a-f0-9]{32}$")
|
|
PROFILE_ID_RE = re.compile(r"^[A-Za-z0-9][A-Za-z0-9._:@/-]{0,127}$")
|
|
RUNTIME_CLEANUP_SENTINEL = "runtime-not-found"
|
|
|
|
DEFAULT_CLEANUP_TIMEOUT = 30.0
|
|
DEFAULT_READY_TIMEOUT = 60.0
|
|
DEFAULT_DISPLAY_START = 100
|
|
DEFAULT_DISPLAY_END = 199
|
|
DEFAULT_CDP_PORT_START = 19000
|
|
DEFAULT_CDP_PORT_END = 19999
|
|
DEFAULT_PROXY_PORT_START = 20000
|
|
DEFAULT_PROXY_PORT_END = 20999
|
|
|
|
|
|
class BrowserRuntimeError(RuntimeError):
|
|
"""An error with a stable HTTP-facing status and optional generation ID."""
|
|
|
|
def __init__(self, message: str, status: int = 502, network_id: str = "") -> None:
|
|
super().__init__(message)
|
|
self.status = status
|
|
self.network_id = network_id
|
|
|
|
|
|
class GenerationConflict(BrowserRuntimeError):
|
|
def __init__(self, message: str = "runtime generation does not match request") -> None:
|
|
super().__init__(message, 409)
|
|
|
|
|
|
class UnmanagedRuntime(BrowserRuntimeError):
|
|
def __init__(self, message: str = "runtime is not owned by this gateway") -> None:
|
|
super().__init__(message, 409)
|
|
|
|
|
|
class RuntimeCleanupPending(BrowserRuntimeError):
|
|
def __init__(self, message: str, network_id: str = "") -> None:
|
|
super().__init__(message, 202, network_id)
|
|
|
|
|
|
class RuntimeCancelled(BrowserRuntimeError):
|
|
def __init__(self, message: str = "runtime creation was cancelled", network_id: str = "") -> None:
|
|
super().__init__(message, 409, network_id)
|
|
|
|
|
|
@dataclass(frozen=True)
|
|
class UnitStatus:
|
|
active: bool
|
|
state: str
|
|
pid: int
|
|
start_time: str
|
|
exit_code: str
|
|
|
|
|
|
class UnitManager(Protocol):
|
|
def start(
|
|
self,
|
|
unit: str,
|
|
command: list[str],
|
|
*,
|
|
environment: Mapping[str, str],
|
|
working_directory: Path,
|
|
stdout_path: Path,
|
|
limits: Mapping[str, str],
|
|
) -> UnitStatus: ...
|
|
|
|
def status(self, unit: str) -> UnitStatus: ...
|
|
|
|
def stop(self, unit: str, timeout: float) -> None: ...
|
|
|
|
def reset(self, unit: str) -> None: ...
|
|
|
|
|
|
class FileLock:
|
|
"""A process-safe exclusive lock which is released with the file descriptor."""
|
|
|
|
def __init__(self, path: Path) -> None:
|
|
self.path = path
|
|
self._fd: int | None = None
|
|
|
|
def acquire(self, blocking: bool = False) -> bool:
|
|
if self._fd is not None:
|
|
return True
|
|
self.path.parent.mkdir(parents=True, exist_ok=True, mode=0o700)
|
|
fd = os.open(self.path, os.O_RDWR | os.O_CREAT, 0o600)
|
|
operation = fcntl.LOCK_EX if blocking else fcntl.LOCK_EX | fcntl.LOCK_NB
|
|
try:
|
|
fcntl.flock(fd, operation)
|
|
except BlockingIOError:
|
|
os.close(fd)
|
|
return False
|
|
except OSError as exc:
|
|
os.close(fd)
|
|
if exc.errno not in (11, 13):
|
|
raise
|
|
return False
|
|
self._fd = fd
|
|
return True
|
|
|
|
def release(self) -> None:
|
|
fd, self._fd = self._fd, None
|
|
if fd is None:
|
|
return
|
|
try:
|
|
fcntl.flock(fd, fcntl.LOCK_UN)
|
|
finally:
|
|
os.close(fd)
|
|
|
|
def __enter__(self) -> FileLock:
|
|
if not self.acquire(blocking=True):
|
|
raise BrowserRuntimeError("resource lock could not be acquired", 409)
|
|
return self
|
|
|
|
def __exit__(self, exc_type: object, exc_value: object, traceback: object) -> None:
|
|
self.release()
|
|
|
|
|
|
@dataclass
|
|
class _Lease:
|
|
value: int
|
|
lock: FileLock
|
|
|
|
|
|
class PortAllocator:
|
|
def __init__(self, lock_dir: Path) -> None:
|
|
self.lock_dir = lock_dir
|
|
|
|
def reserve(
|
|
self,
|
|
prefix: str,
|
|
start: int,
|
|
end: int,
|
|
unavailable: Callable[[int], bool],
|
|
) -> _Lease:
|
|
for value in range(start, end + 1):
|
|
if unavailable(value):
|
|
continue
|
|
lock = FileLock(self.lock_dir / f"{prefix}-{value}.lock")
|
|
if not lock.acquire():
|
|
continue
|
|
try:
|
|
probe = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
|
|
try:
|
|
probe.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
|
|
probe.bind(("127.0.0.1", value))
|
|
finally:
|
|
probe.close()
|
|
except OSError:
|
|
lock.release()
|
|
continue
|
|
return _Lease(value, lock)
|
|
raise BrowserRuntimeError(f"no free {prefix} port is available", 503)
|
|
|
|
def reserve_existing(self, prefix: str, value: int) -> _Lease:
|
|
if not 1 <= value <= 65535:
|
|
raise BrowserRuntimeError(f"invalid {prefix} port", 409)
|
|
lock = FileLock(self.lock_dir / f"{prefix}-{value}.lock")
|
|
if not lock.acquire():
|
|
raise BrowserRuntimeError(f"{prefix} port is already reserved", 409)
|
|
return _Lease(value, lock)
|
|
|
|
|
|
class DisplayAllocator:
|
|
def __init__(self, lock_dir: Path) -> None:
|
|
self.lock_dir = lock_dir
|
|
|
|
def reserve(
|
|
self,
|
|
start: int,
|
|
end: int,
|
|
unavailable: Callable[[int], bool] | None = None,
|
|
) -> _Lease:
|
|
unavailable = unavailable or (lambda _value: False)
|
|
for value in range(start, end + 1):
|
|
if unavailable(value) or (X11_SOCKET_DIR / f"X{value}").exists():
|
|
continue
|
|
lock = FileLock(self.lock_dir / f"display-{value}.lock")
|
|
if lock.acquire():
|
|
if not (X11_SOCKET_DIR / f"X{value}").exists():
|
|
return _Lease(value, lock)
|
|
lock.release()
|
|
raise BrowserRuntimeError("no free Xvfb display is available", 503)
|
|
|
|
def reserve_existing(self, value: int) -> _Lease:
|
|
if value < 1:
|
|
raise BrowserRuntimeError("runtime display is invalid", 409)
|
|
lock = FileLock(self.lock_dir / f"display-{value}.lock")
|
|
if not lock.acquire():
|
|
raise BrowserRuntimeError("runtime display is already reserved", 409)
|
|
return _Lease(value, lock)
|
|
|
|
|
|
class SystemdUnitManager:
|
|
"""Starts only explicitly supplied commands in per-runtime user units."""
|
|
|
|
def __init__(
|
|
self,
|
|
systemd_run: str | None = None,
|
|
systemctl: str | None = None,
|
|
runner: Callable[..., subprocess.CompletedProcess[str]] | None = None,
|
|
) -> None:
|
|
self.systemd_run = systemd_run or shutil.which("systemd-run") or ""
|
|
self.systemctl = systemctl or shutil.which("systemctl") or ""
|
|
self.runner = runner or subprocess.run
|
|
|
|
def _execute(self, args: list[str], timeout: float) -> subprocess.CompletedProcess[str]:
|
|
if not self.systemd_run or not self.systemctl:
|
|
raise BrowserRuntimeError("systemd user units are unavailable", 503)
|
|
try:
|
|
return self.runner(
|
|
args,
|
|
check=False,
|
|
capture_output=True,
|
|
text=True,
|
|
timeout=timeout,
|
|
)
|
|
except (OSError, subprocess.TimeoutExpired) as exc:
|
|
raise BrowserRuntimeError("systemd operation failed", 503) from exc
|
|
|
|
def start(
|
|
self,
|
|
unit: str,
|
|
command: list[str],
|
|
*,
|
|
environment: Mapping[str, str],
|
|
working_directory: Path,
|
|
stdout_path: Path,
|
|
limits: Mapping[str, str],
|
|
) -> UnitStatus:
|
|
stdout_path.parent.mkdir(parents=True, exist_ok=True, mode=0o700)
|
|
args = [
|
|
self.systemd_run,
|
|
"--user",
|
|
"--unit",
|
|
unit,
|
|
"--collect",
|
|
"--no-block",
|
|
"--property=Type=exec",
|
|
"--property=KillMode=mixed",
|
|
"--property=NoNewPrivileges=yes",
|
|
f"--property=StandardOutput=append:{stdout_path}",
|
|
f"--property=StandardError=append:{stdout_path}",
|
|
"--working-directory",
|
|
str(working_directory),
|
|
]
|
|
for key, value in sorted(environment.items()):
|
|
args.extend(["--setenv", f"{key}={value}"])
|
|
for key, value in sorted(limits.items()):
|
|
args.append(f"--property={key}={value}")
|
|
args.extend(["--", *command])
|
|
result = self._execute(args, 30.0)
|
|
if result.returncode != 0:
|
|
LOG.error(
|
|
"native runtime unit start rejected",
|
|
extra={"unit": unit, "returncode": result.returncode},
|
|
)
|
|
raise BrowserRuntimeError("native runtime unit could not start", 503)
|
|
deadline = time.monotonic() + 10.0
|
|
while time.monotonic() < deadline:
|
|
status = self.status(unit)
|
|
if status.pid or status.active:
|
|
return status
|
|
time.sleep(0.05)
|
|
raise BrowserRuntimeError("native runtime unit did not become active", 503)
|
|
|
|
def status(self, unit: str) -> UnitStatus:
|
|
if not self.systemctl:
|
|
raise BrowserRuntimeError("systemd user units are unavailable", 503)
|
|
result = self._execute(
|
|
[
|
|
self.systemctl,
|
|
"--user",
|
|
"show",
|
|
unit,
|
|
"--property=ActiveState",
|
|
"--property=SubState",
|
|
"--property=MainPID",
|
|
"--property=ExecMainStartTimestampMonotonic",
|
|
"--property=ExecMainStatus",
|
|
],
|
|
10.0,
|
|
)
|
|
if result.returncode != 0:
|
|
return UnitStatus(False, "not-found", 0, "", "")
|
|
values: dict[str, str] = {}
|
|
for line in result.stdout.splitlines():
|
|
if "=" in line:
|
|
key, value = line.split("=", 1)
|
|
values[key] = value
|
|
try:
|
|
pid = int(values.get("MainPID", "0"))
|
|
except ValueError:
|
|
pid = 0
|
|
active_state = values.get("ActiveState", "inactive")
|
|
return UnitStatus(
|
|
active_state in {"active", "activating", "deactivating"},
|
|
values.get("SubState", active_state),
|
|
pid,
|
|
values.get("ExecMainStartTimestampMonotonic", ""),
|
|
values.get("ExecMainStatus", ""),
|
|
)
|
|
|
|
def stop(self, unit: str, timeout: float) -> None:
|
|
result = self._execute(
|
|
[self.systemctl, "--user", "stop", unit], max(timeout, 1.0)
|
|
)
|
|
if result.returncode == 0:
|
|
return
|
|
kill = self._execute(
|
|
[
|
|
self.systemctl,
|
|
"--user",
|
|
"kill",
|
|
"--kill-who=all",
|
|
"--signal=TERM",
|
|
unit,
|
|
],
|
|
10.0,
|
|
)
|
|
if kill.returncode != 0:
|
|
raise BrowserRuntimeError("native runtime unit could not stop", 503)
|
|
result = self._execute(
|
|
[self.systemctl, "--user", "stop", unit], max(timeout, 1.0)
|
|
)
|
|
if result.returncode != 0:
|
|
raise BrowserRuntimeError("native runtime unit could not stop", 503)
|
|
|
|
def reset(self, unit: str) -> None:
|
|
if not self.systemctl:
|
|
return
|
|
result = self._execute([self.systemctl, "--user", "reset-failed", unit], 10.0)
|
|
# `--collect` removes a transient unit before reset-failed runs; systemctl
|
|
# returns 1 for that already-cleaned unit. This is the idempotent success
|
|
# case, not an ignored cleanup error.
|
|
if result.returncode not in (0, 1, 5):
|
|
raise BrowserRuntimeError("native runtime unit cleanup failed", 503)
|
|
|
|
|
|
@dataclass
|
|
class RuntimeRecord:
|
|
alias: str
|
|
name: str
|
|
runtime_id: str
|
|
network_id: str
|
|
binding_version: int
|
|
profile_id: str
|
|
profile_dir: str
|
|
browser_path: str
|
|
command: list[str]
|
|
network_exit_id: str = ""
|
|
proxy_port: int = 0
|
|
display: int = 0
|
|
cdp_port: int = 0
|
|
node_id: str = ""
|
|
owner: str = ""
|
|
state: str = "created"
|
|
cleanup_state: str = "none"
|
|
cleanup_error: str = ""
|
|
browser_unit: str = ""
|
|
xvfb_unit: str = ""
|
|
browser_pid: int = 0
|
|
xvfb_pid: int = 0
|
|
browser_start_time: str = ""
|
|
xvfb_start_time: str = ""
|
|
created_at: float = 0.0
|
|
updated_at: float = 0.0
|
|
|
|
def to_dict(self) -> dict[str, Any]:
|
|
return {field.name: getattr(self, field.name) for field in fields(self)}
|
|
|
|
@classmethod
|
|
def from_dict(cls, value: object) -> RuntimeRecord:
|
|
if not isinstance(value, dict):
|
|
raise BrowserRuntimeError("runtime metadata is invalid", 500)
|
|
required = {
|
|
"alias",
|
|
"name",
|
|
"runtime_id",
|
|
"network_id",
|
|
"binding_version",
|
|
"profile_id",
|
|
"profile_dir",
|
|
"browser_path",
|
|
"command",
|
|
"state",
|
|
"cleanup_state",
|
|
}
|
|
if not required.issubset(value):
|
|
raise BrowserRuntimeError("runtime metadata is incomplete", 500)
|
|
command = value["command"]
|
|
if not isinstance(command, list) or not all(isinstance(item, str) for item in command):
|
|
raise BrowserRuntimeError("runtime command metadata is invalid", 500)
|
|
try:
|
|
record = cls(
|
|
alias=value["alias"],
|
|
name=value["name"],
|
|
runtime_id=value["runtime_id"],
|
|
network_id=value["network_id"],
|
|
binding_version=value["binding_version"],
|
|
profile_id=value["profile_id"],
|
|
profile_dir=value["profile_dir"],
|
|
browser_path=value["browser_path"],
|
|
command=command,
|
|
network_exit_id=value.get("network_exit_id", ""),
|
|
proxy_port=value.get("proxy_port", 0),
|
|
display=value.get("display", 0),
|
|
cdp_port=value.get("cdp_port", 0),
|
|
node_id=value.get("node_id", ""),
|
|
owner=value.get("owner", ""),
|
|
state=value["state"],
|
|
cleanup_state=value["cleanup_state"],
|
|
cleanup_error=value.get("cleanup_error", ""),
|
|
browser_unit=value.get("browser_unit", ""),
|
|
xvfb_unit=value.get("xvfb_unit", ""),
|
|
browser_pid=value.get("browser_pid", 0),
|
|
xvfb_pid=value.get("xvfb_pid", 0),
|
|
browser_start_time=value.get("browser_start_time", ""),
|
|
xvfb_start_time=value.get("xvfb_start_time", ""),
|
|
created_at=value.get("created_at", 0.0),
|
|
updated_at=value.get("updated_at", 0.0),
|
|
)
|
|
except (KeyError, TypeError, ValueError) as exc:
|
|
raise BrowserRuntimeError("runtime metadata is invalid", 500) from exc
|
|
if (
|
|
not isinstance(record.alias, str)
|
|
or not ALIAS_RE.fullmatch(record.alias)
|
|
or not isinstance(record.name, str)
|
|
or not isinstance(record.runtime_id, str)
|
|
or not RUNTIME_ID_RE.fullmatch(record.runtime_id)
|
|
or not isinstance(record.network_id, str)
|
|
or not NETWORK_ID_RE.fullmatch(record.network_id)
|
|
or type(record.binding_version) is not int
|
|
or record.binding_version < 1
|
|
or not isinstance(record.profile_id, str)
|
|
or not PROFILE_ID_RE.fullmatch(record.profile_id)
|
|
or not isinstance(record.browser_path, str)
|
|
or not os.path.isabs(record.browser_path)
|
|
or not isinstance(record.state, str)
|
|
or not isinstance(record.cleanup_state, str)
|
|
):
|
|
raise BrowserRuntimeError("runtime metadata is invalid", 500)
|
|
return record
|
|
|
|
|
|
class NativeRuntimeManager:
|
|
def __init__(
|
|
self,
|
|
*,
|
|
state_dir: str | Path,
|
|
profile_root: str | Path,
|
|
node_id: str,
|
|
browser_path: str,
|
|
unit_manager: UnitManager | None = None,
|
|
proxy_registry: ProxyRegistry | None = None,
|
|
cleanup_timeout: float = DEFAULT_CLEANUP_TIMEOUT,
|
|
ready_timeout: float = DEFAULT_READY_TIMEOUT,
|
|
display_start: int = DEFAULT_DISPLAY_START,
|
|
display_end: int = DEFAULT_DISPLAY_END,
|
|
external_display: int | None = None,
|
|
cdp_port_start: int = DEFAULT_CDP_PORT_START,
|
|
cdp_port_end: int = DEFAULT_CDP_PORT_END,
|
|
proxy_port_start: int = DEFAULT_PROXY_PORT_START,
|
|
proxy_port_end: int = DEFAULT_PROXY_PORT_END,
|
|
min_free_bytes: int = 20 * 1024**3,
|
|
log_max_bytes: int = 1 * 1024**3,
|
|
profile_cache_max_bytes: int = 20 * 1024**3,
|
|
clock: Callable[[], float] = time.time,
|
|
recover: bool = True,
|
|
) -> None:
|
|
self.state_dir = Path(state_dir).expanduser().resolve()
|
|
self.profile_root = Path(profile_root).expanduser().resolve()
|
|
self.node_id = node_id
|
|
self.browser_path = str(Path(browser_path).expanduser().resolve())
|
|
self.cleanup_timeout = cleanup_timeout
|
|
self.ready_timeout = ready_timeout
|
|
self.display_start = display_start
|
|
self.display_end = display_end
|
|
self.external_display = external_display
|
|
if external_display is not None and external_display < 1:
|
|
raise BrowserRuntimeError("external Xvfb display is invalid", 500)
|
|
self.cdp_port_start = cdp_port_start
|
|
self.cdp_port_end = cdp_port_end
|
|
self.proxy_port_start = proxy_port_start
|
|
self.proxy_port_end = proxy_port_end
|
|
self.min_free_bytes = min_free_bytes
|
|
self.log_max_bytes = log_max_bytes
|
|
self.profile_cache_max_bytes = profile_cache_max_bytes
|
|
if min_free_bytes < 0 or log_max_bytes < 0 or profile_cache_max_bytes < 0:
|
|
raise BrowserRuntimeError("runtime resource limits are invalid", 500)
|
|
self.clock = clock
|
|
self.unit_manager = unit_manager or SystemdUnitManager()
|
|
self.xvfb_path = shutil.which("Xvfb") or ""
|
|
self.proxies = proxy_registry or ProxyRegistry()
|
|
self.port_allocator = PortAllocator(self.state_dir / "locks")
|
|
self.display_allocator = DisplayAllocator(self.state_dir / "locks")
|
|
self._lock = threading.RLock()
|
|
self._alias_locks: dict[str, FileLock] = {}
|
|
self._cancel_events: dict[str, threading.Event] = {}
|
|
self._profile_locks: dict[str, FileLock] = {}
|
|
self._display_leases: dict[str, _Lease] = {}
|
|
self._port_leases: dict[tuple[str, str], _Lease] = {}
|
|
self._proxy_undo: dict[str, Callable[[], None]] = {}
|
|
self._mkdirs()
|
|
if recover:
|
|
self._recover()
|
|
|
|
def _check_capacity(self) -> None:
|
|
for path in (self.state_dir, self.profile_root):
|
|
usage = shutil.disk_usage(path)
|
|
if usage.free < self.min_free_bytes:
|
|
raise BrowserRuntimeError("insufficient free disk space for browser runtime", 507)
|
|
|
|
def _mkdirs(self) -> None:
|
|
for path in (
|
|
self.state_dir,
|
|
self.state_dir / "runtimes",
|
|
self.state_dir / "locks",
|
|
self.state_dir / "logs",
|
|
self.profile_root,
|
|
):
|
|
path.mkdir(parents=True, exist_ok=True, mode=0o700)
|
|
if path.stat().st_uid != os.getuid() or path.stat().st_mode & 0o077:
|
|
raise BrowserRuntimeError(f"runtime directory ownership is invalid: {path}", 500)
|
|
|
|
def close(self) -> None:
|
|
with self._lock:
|
|
for lock in list(self._alias_locks.values()):
|
|
lock.release()
|
|
self._alias_locks.clear()
|
|
for lock in list(self._profile_locks.values()):
|
|
lock.release()
|
|
self._profile_locks.clear()
|
|
for lease in list(self._display_leases.values()):
|
|
lease.lock.release()
|
|
self._display_leases.clear()
|
|
for lease in list(self._port_leases.values()):
|
|
lease.lock.release()
|
|
self._port_leases.clear()
|
|
self._proxy_undo.clear()
|
|
self.proxies.close()
|
|
|
|
def alias_lock(self, alias: str):
|
|
return _ManagedAliasLock(self, alias)
|
|
|
|
def create(self, value: dict[str, Any]) -> dict[str, Any]:
|
|
validate_runtime_input(value)
|
|
alias = value["alias"]
|
|
with self.alias_lock(alias):
|
|
existing = self._find_alias(alias, include_released=False)
|
|
if existing is not None:
|
|
if existing.cleanup_state == "pending":
|
|
raise RuntimeCleanupPending(
|
|
"runtime cleanup is pending", existing.network_id
|
|
)
|
|
raise BrowserRuntimeError("browser alias is already in use", 409)
|
|
runtime_id = __import__("secrets").token_hex(32)
|
|
network_id = "native-" + __import__("secrets").token_hex(16)
|
|
profile_id = value.get("profile_id", alias)
|
|
profile_dir = self._profile_dir(profile_id)
|
|
browser_path = self.browser_path
|
|
try:
|
|
available = os.path.isfile(browser_path) and os.access(
|
|
browser_path, os.X_OK
|
|
)
|
|
except (OSError, TypeError) as exc:
|
|
raise BrowserRuntimeError("browser executable is unavailable", 422) from exc
|
|
if not available:
|
|
raise BrowserRuntimeError("browser executable is unavailable", 422)
|
|
self._check_capacity()
|
|
now = self.clock()
|
|
record = RuntimeRecord(
|
|
alias=alias,
|
|
name=value["name"],
|
|
runtime_id=runtime_id,
|
|
network_id=network_id,
|
|
binding_version=value["binding_version"],
|
|
profile_id=profile_id,
|
|
profile_dir=str(profile_dir),
|
|
browser_path=browser_path,
|
|
command=list(value["cmd"]),
|
|
network_exit_id=value.get("network_exit_id", ""),
|
|
node_id=self.node_id,
|
|
owner=self.node_id,
|
|
state="stopped" if value.get("stopped", False) else "starting",
|
|
cleanup_state="none",
|
|
created_at=now,
|
|
updated_at=now,
|
|
)
|
|
self._prepare_profile(record)
|
|
self._write(record)
|
|
if record.state == "stopped":
|
|
return self._public(record, False)
|
|
self._cancel_events[record.runtime_id] = threading.Event()
|
|
try:
|
|
self._start_locked(record, value["network_exit"])
|
|
return self._public(record, True)
|
|
except BrowserRuntimeError:
|
|
raise
|
|
except Exception as exc:
|
|
self._mark_cleanup_pending(record, str(exc))
|
|
raise BrowserRuntimeError("native browser runtime creation failed", 502, record.network_id) from exc
|
|
finally:
|
|
self._cancel_events.pop(record.runtime_id, None)
|
|
|
|
def change_state(self, alias: str, action: str, generation: Mapping[str, Any]) -> None:
|
|
if action == "cancel":
|
|
self.cancel(alias, generation)
|
|
return
|
|
if action not in {"start", "stop"}:
|
|
raise BrowserRuntimeError("invalid runtime action", 400)
|
|
with self.alias_lock(alias):
|
|
record = self._require_record(alias, generation)
|
|
if action == "start":
|
|
if record.state == "running":
|
|
if not self._ready(record):
|
|
raise BrowserRuntimeError("browser runtime is not ready", 503, record.network_id)
|
|
return
|
|
if record.state != "stopped":
|
|
raise BrowserRuntimeError("runtime cannot be started in its current state", 409, record.network_id)
|
|
self._start_locked(record, ProxyExit("", "", 0))
|
|
return
|
|
if record.state == "stopped":
|
|
return
|
|
if record.state not in {"running", "degraded", "starting"}:
|
|
if record.cleanup_state == "pending":
|
|
raise RuntimeCleanupPending("runtime cleanup is pending", record.network_id)
|
|
raise BrowserRuntimeError("runtime cannot be stopped in its current state", 409, record.network_id)
|
|
self._stop_locked(record, released=False)
|
|
|
|
def remove(self, alias: str, generation: Mapping[str, Any], purge_profile: bool = False) -> None:
|
|
with self.alias_lock(alias):
|
|
record = self._find_record_for_generation(alias, generation)
|
|
if record is None:
|
|
if generation.get("runtime_id") == RUNTIME_CLEANUP_SENTINEL:
|
|
return
|
|
raise FileNotFoundError(alias)
|
|
self._fence(record, generation)
|
|
if record.state == "released" and record.cleanup_state == "cleaned":
|
|
if purge_profile:
|
|
self._purge_profile(record)
|
|
return
|
|
self._stop_locked(record, released=True)
|
|
if purge_profile:
|
|
self._purge_profile(record)
|
|
|
|
def restore_proxy(self, alias: str, generation: Mapping[str, Any], exit: ProxyExit) -> None:
|
|
with self.alias_lock(alias):
|
|
record = self._require_record(alias, generation, require_active=False)
|
|
if record.network_exit_id and record.network_exit_id != generation.get("network_exit_id", record.network_exit_id):
|
|
raise GenerationConflict("proxy generation does not match runtime")
|
|
if exit == ProxyExit("", "", 0):
|
|
self.proxies.remove(
|
|
alias, record.binding_version, record.runtime_id, record.network_id
|
|
)
|
|
record.proxy_port = 0
|
|
self._write(record)
|
|
return
|
|
if record.state not in {"running", "degraded"}:
|
|
raise BrowserRuntimeError("proxy can only be restored for a running runtime", 409, record.network_id)
|
|
port = record.proxy_port
|
|
if port < 1:
|
|
lease = self._reserve_port(record, "proxy", self.proxy_port_start, self.proxy_port_end)
|
|
port = lease.value
|
|
self._port_leases[(record.runtime_id, "proxy")] = lease
|
|
if (record.runtime_id, "proxy") not in self._port_leases:
|
|
self._port_leases[(record.runtime_id, "proxy")] = self.port_allocator.reserve_existing("proxy", port)
|
|
url, undo = self.proxies.configure(
|
|
alias,
|
|
record.binding_version,
|
|
"127.0.0.1",
|
|
port,
|
|
exit,
|
|
record.network_id,
|
|
)
|
|
if not self.proxies.bind(
|
|
alias, record.binding_version, url, record.runtime_id, record.network_id
|
|
):
|
|
undo()
|
|
raise GenerationConflict("proxy generation changed")
|
|
self._proxy_undo[record.runtime_id] = undo
|
|
record.proxy_port = urlsplit(url).port or 0
|
|
record.network_exit_id = generation.get("network_exit_id", record.network_exit_id)
|
|
self._write(record)
|
|
|
|
def cancel(self, alias: str, generation: Mapping[str, Any]) -> None:
|
|
with self._lock:
|
|
record = self._require_record(alias, generation, require_active=False)
|
|
if record.state != "starting":
|
|
raise BrowserRuntimeError("runtime is not starting", 409, record.network_id)
|
|
event = self._cancel_events.get(record.runtime_id)
|
|
if event is not None:
|
|
event.set()
|
|
return
|
|
with self.alias_lock(alias):
|
|
record = self._require_record(alias, generation, require_active=False)
|
|
self._stop_locked(record, released=False)
|
|
|
|
def endpoint(self, alias: str) -> str:
|
|
with self._lock:
|
|
record = self._find_alias(alias, include_released=False)
|
|
if record is None or record.state not in {"running", "degraded"}:
|
|
raise FileNotFoundError(alias)
|
|
self._refresh(record)
|
|
if record.state not in {"running", "degraded"} or not record.cdp_port:
|
|
raise BrowserRuntimeError("browser runtime is unavailable", 503, record.network_id)
|
|
return f"http://127.0.0.1:{record.cdp_port}"
|
|
|
|
def require_generation(self, alias: str, generation: Mapping[str, Any]) -> RuntimeRecord:
|
|
with self._lock:
|
|
record = self._require_record(alias, generation)
|
|
self._refresh(record)
|
|
if record.state != "running" or not self._ready(record):
|
|
raise BrowserRuntimeError("browser runtime is not ready", 503, record.network_id)
|
|
return record
|
|
|
|
def list_public(self) -> list[dict[str, Any]]:
|
|
with self._lock:
|
|
records = self._records()
|
|
result = []
|
|
for record in records:
|
|
if record.state not in {"released"}:
|
|
self._refresh(record)
|
|
result.append(self._public(record, self._ready(record) if record.state in {"running", "degraded"} else False))
|
|
return result
|
|
|
|
def retry_cleanup(self, alias: str, generation: Mapping[str, Any]) -> None:
|
|
with self.alias_lock(alias):
|
|
record = self._require_record(alias, generation, require_active=False)
|
|
if record.cleanup_state != "pending":
|
|
return
|
|
self._stop_locked(record, released=record.state != "stopped")
|
|
|
|
def _start_locked(self, record: RuntimeRecord, proxy_exit: ProxyExit) -> None:
|
|
self._check_cancel(record)
|
|
self._prepare_profile(record)
|
|
self._check_profile_size(record)
|
|
if self._profile_in_use(record):
|
|
raise BrowserRuntimeError("Profile is already in use", 409, record.network_id)
|
|
profile_lock = FileLock(Path(record.profile_dir) / ".creatorhub-profile.lock")
|
|
if not profile_lock.acquire():
|
|
raise BrowserRuntimeError("Profile is already in use", 409, record.network_id)
|
|
self._profile_locks[record.runtime_id] = profile_lock
|
|
display_lease: _Lease | None = None
|
|
cdp_lease: _Lease | None = None
|
|
proxy_lease: _Lease | None = None
|
|
try:
|
|
display_lease = self._reserve_display(record)
|
|
self._display_leases[record.runtime_id] = display_lease
|
|
self._check_cancel(record)
|
|
cdp_lease = self._reserve_port(record, "cdp", self.cdp_port_start, self.cdp_port_end)
|
|
self._port_leases[(record.runtime_id, "cdp")] = cdp_lease
|
|
record.display = display_lease.value
|
|
record.cdp_port = cdp_lease.value
|
|
record.state = "starting"
|
|
record.cleanup_state = "none"
|
|
record.cleanup_error = ""
|
|
record.browser_unit = self._unit_name("browser", record)
|
|
record.xvfb_unit = (
|
|
self._unit_name("xvfb", record)
|
|
if self.external_display is None
|
|
else ""
|
|
)
|
|
runtime_dir = self._runtime_dir(record)
|
|
runtime_dir.mkdir(parents=True, exist_ok=True, mode=0o700)
|
|
(runtime_dir / "tmp").mkdir(parents=True, exist_ok=True, mode=0o700)
|
|
self._write(record)
|
|
|
|
proxy_url = ""
|
|
if proxy_exit != ProxyExit("", "", 0):
|
|
proxy_lease = self._reserve_port(record, "proxy", self.proxy_port_start, self.proxy_port_end)
|
|
self._port_leases[(record.runtime_id, "proxy")] = proxy_lease
|
|
proxy_url, undo = self.proxies.configure(
|
|
record.alias,
|
|
record.binding_version,
|
|
"127.0.0.1",
|
|
proxy_lease.value,
|
|
proxy_exit,
|
|
record.network_id,
|
|
)
|
|
self._proxy_undo[record.runtime_id] = undo
|
|
record.proxy_port = urlsplit(proxy_url).port or 0
|
|
self._write(record)
|
|
|
|
if self.external_display is None:
|
|
xvfb_log = self._log_path(record, "xvfb")
|
|
xvfb_command = self._locked_command(
|
|
self._display_lock_path(record.display),
|
|
[
|
|
self.xvfb_path or "Xvfb",
|
|
f":{record.display}",
|
|
"-screen",
|
|
"0",
|
|
"1280x720x24",
|
|
"-nolisten",
|
|
"tcp",
|
|
],
|
|
)
|
|
xvfb_status = self.unit_manager.start(
|
|
record.xvfb_unit,
|
|
xvfb_command,
|
|
environment={"DISPLAY": f":{record.display}"},
|
|
working_directory=runtime_dir,
|
|
stdout_path=xvfb_log,
|
|
limits={"MemoryMax": "256M", "TasksMax": "128", "TimeoutStopSec": "10s"},
|
|
)
|
|
record.xvfb_pid = xvfb_status.pid
|
|
record.xvfb_start_time = xvfb_status.start_time
|
|
else:
|
|
record.xvfb_pid = 0
|
|
record.xvfb_start_time = "external"
|
|
self._release_display_lease(record)
|
|
self._write(record)
|
|
self._wait_for_display(record)
|
|
self._check_cancel(record)
|
|
|
|
browser_command = self._browser_command(record, proxy_url)
|
|
browser_log = self._log_path(record, "browser")
|
|
browser_status = self.unit_manager.start(
|
|
record.browser_unit,
|
|
self._locked_command(
|
|
Path(record.profile_dir) / ".creatorhub-profile.lock", browser_command
|
|
),
|
|
environment={"DISPLAY": f":{record.display}"},
|
|
working_directory=runtime_dir,
|
|
stdout_path=browser_log,
|
|
limits={"MemoryMax": "1G", "TasksMax": "512", "CPUQuota": "200%", "TimeoutStopSec": "10s"},
|
|
)
|
|
self._release_startup_locks(record)
|
|
record.browser_pid = browser_status.pid
|
|
record.browser_start_time = browser_status.start_time
|
|
self._write(record)
|
|
self._wait_for_cdp(record)
|
|
self._check_cancel(record)
|
|
if proxy_url and not self.proxies.bind(
|
|
record.alias,
|
|
record.binding_version,
|
|
proxy_url,
|
|
record.runtime_id,
|
|
record.network_id,
|
|
):
|
|
raise GenerationConflict("proxy generation changed")
|
|
record.state = "running"
|
|
record.cleanup_state = "none"
|
|
record.updated_at = self.clock()
|
|
self._write(record)
|
|
# Keep the Profile lock while the unit owns the browser. The unit
|
|
# also takes this lock through `flock`, so a gateway restart cannot
|
|
# create a second browser for the same Profile.
|
|
except Exception as exc:
|
|
self._cleanup_after_failure(record, exc)
|
|
if isinstance(exc, BrowserRuntimeError):
|
|
raise
|
|
raise BrowserRuntimeError("native browser runtime start failed", 502, record.network_id) from exc
|
|
|
|
def _stop_units(self, record: RuntimeRecord) -> list[str]:
|
|
errors: list[str] = []
|
|
for unit in (record.browser_unit, record.xvfb_unit):
|
|
if not unit:
|
|
continue
|
|
try:
|
|
status = self.unit_manager.status(unit)
|
|
if status.active or status.pid:
|
|
self.unit_manager.stop(unit, self.cleanup_timeout)
|
|
after = self.unit_manager.status(unit)
|
|
if after.active or after.pid:
|
|
errors.append(f"{unit}: unit remains active after stop")
|
|
self.unit_manager.reset(unit)
|
|
except Exception as exc:
|
|
errors.append(f"{unit}: {exc}")
|
|
return errors
|
|
|
|
def _stop_locked(self, record: RuntimeRecord, released: bool) -> None:
|
|
record.state = "stopping"
|
|
self._write(record)
|
|
errors = self._stop_units(record)
|
|
try:
|
|
if not self.proxies.remove(
|
|
record.alias,
|
|
record.binding_version,
|
|
record.runtime_id,
|
|
record.network_id,
|
|
):
|
|
errors.append("proxy generation does not match runtime")
|
|
except Exception as exc:
|
|
errors.append(f"proxy: {exc}")
|
|
if errors:
|
|
self._mark_cleanup_pending(record, "; ".join(errors))
|
|
raise RuntimeCleanupPending("runtime cleanup is pending", record.network_id)
|
|
self._release_runtime_leases(record)
|
|
self._release_startup_locks(record)
|
|
self._remove_runtime_tmp(record)
|
|
record.state = "released" if released else "stopped"
|
|
record.cleanup_state = "cleaned"
|
|
record.cleanup_error = ""
|
|
record.browser_pid = record.xvfb_pid = 0
|
|
record.display = record.cdp_port = record.proxy_port = 0
|
|
record.updated_at = self.clock()
|
|
self._write(record)
|
|
|
|
def _cleanup_after_failure(self, record: RuntimeRecord, error: Exception) -> None:
|
|
errors = [str(error), *self._stop_units(record)]
|
|
if len(errors) == 1:
|
|
try:
|
|
if not self.proxies.remove(
|
|
record.alias,
|
|
record.binding_version,
|
|
record.runtime_id,
|
|
record.network_id,
|
|
):
|
|
errors.append("proxy generation does not match runtime")
|
|
except Exception as exc:
|
|
errors.append(f"proxy: {exc}")
|
|
if len(errors) == 1:
|
|
self._release_runtime_leases(record)
|
|
self._remove_runtime_tmp(record)
|
|
self._release_startup_locks(record)
|
|
self._mark_cleanup_pending(record, "; ".join(item for item in errors if item))
|
|
|
|
def _mark_cleanup_pending(self, record: RuntimeRecord, error: str) -> None:
|
|
record.state = "failed"
|
|
record.cleanup_state = "pending"
|
|
record.cleanup_error = error[:4096]
|
|
record.updated_at = self.clock()
|
|
self._write(record)
|
|
LOG.error(
|
|
"native runtime cleanup pending",
|
|
extra={
|
|
"alias": record.alias,
|
|
"runtime_id": record.runtime_id,
|
|
"binding_version": record.binding_version,
|
|
"network_id": record.network_id,
|
|
"error": record.cleanup_error,
|
|
},
|
|
)
|
|
|
|
def _recover(self) -> None:
|
|
with self._lock:
|
|
for record in self._records():
|
|
if record.state not in {"running", "starting", "stopping", "degraded"}:
|
|
continue
|
|
browser = self.unit_manager.status(record.browser_unit) if record.browser_unit else UnitStatus(False, "missing", 0, "", "")
|
|
xvfb = self._display_status(record)
|
|
if browser.active and xvfb.active:
|
|
record.browser_pid = browser.pid
|
|
record.xvfb_pid = xvfb.pid
|
|
record.browser_start_time = browser.start_time
|
|
record.xvfb_start_time = xvfb.start_time
|
|
record.state = "running"
|
|
record.cleanup_state = "none"
|
|
record.cleanup_error = ""
|
|
self._write(record)
|
|
else:
|
|
self._mark_cleanup_pending(
|
|
record,
|
|
"runtime unit disappeared during gateway restart",
|
|
)
|
|
|
|
def _refresh(self, record: RuntimeRecord) -> None:
|
|
if record.state not in {"running", "degraded", "starting"}:
|
|
return
|
|
browser = self.unit_manager.status(record.browser_unit) if record.browser_unit else UnitStatus(False, "missing", 0, "", "")
|
|
xvfb = self._display_status(record)
|
|
if not browser.active or not xvfb.active:
|
|
record.state = "degraded"
|
|
record.cleanup_state = "pending"
|
|
record.cleanup_error = "runtime unit is not active"
|
|
record.updated_at = self.clock()
|
|
self._write(record)
|
|
return
|
|
record.browser_pid = browser.pid
|
|
record.xvfb_pid = xvfb.pid
|
|
if record.state == "starting":
|
|
record.state = "degraded"
|
|
self._write(record)
|
|
|
|
def _ready(self, record: RuntimeRecord) -> bool:
|
|
if record.cdp_port < 1:
|
|
return False
|
|
try:
|
|
version = self._get_json(record.cdp_port, "/json/version")
|
|
targets = self._get_json(record.cdp_port, "/json/list")
|
|
except (OSError, ValueError, BrowserRuntimeError):
|
|
return False
|
|
return isinstance(version, dict) and isinstance(targets, list) and any(
|
|
isinstance(target, dict) and target.get("type") == "page" for target in targets
|
|
)
|
|
|
|
def _display_status(self, record: RuntimeRecord) -> UnitStatus:
|
|
if record.xvfb_unit:
|
|
return self.unit_manager.status(record.xvfb_unit)
|
|
available = self._display_available(record.display)
|
|
return UnitStatus(available, "external" if available else "missing", 0, "", "")
|
|
|
|
def _display_available(self, display: int) -> bool:
|
|
return display > 0 and (X11_SOCKET_DIR / f"X{display}").exists()
|
|
|
|
def _wait_for_display(self, record: RuntimeRecord) -> None:
|
|
deadline = time.monotonic() + self.ready_timeout
|
|
while time.monotonic() < deadline:
|
|
self._check_cancel(record)
|
|
if self._display_available(record.display):
|
|
return
|
|
status = self._display_status(record)
|
|
if not status.active and status.pid == 0:
|
|
raise BrowserRuntimeError("Xvfb unit exited before readiness", 503, record.network_id)
|
|
time.sleep(0.05)
|
|
raise BrowserRuntimeError("Xvfb did not become ready", 503, record.network_id)
|
|
|
|
def _wait_for_cdp(self, record: RuntimeRecord) -> None:
|
|
deadline = time.monotonic() + self.ready_timeout
|
|
while time.monotonic() < deadline:
|
|
self._check_cancel(record)
|
|
if self._ready(record):
|
|
return
|
|
status = self.unit_manager.status(record.browser_unit)
|
|
if not status.active and status.pid == 0:
|
|
raise BrowserRuntimeError("browser unit exited before CDP readiness", 503, record.network_id)
|
|
time.sleep(0.1)
|
|
raise BrowserRuntimeError("browser CDP did not become ready", 503, record.network_id)
|
|
|
|
def _get_json(self, port: int, path: str) -> object:
|
|
connection = http.client.HTTPConnection("127.0.0.1", port, timeout=1.0)
|
|
try:
|
|
connection.request("GET", path, headers={"Accept": "application/json"})
|
|
response = connection.getresponse()
|
|
body = response.read(64 * 1024 + 1)
|
|
except (OSError, http.client.HTTPException) as exc:
|
|
raise BrowserRuntimeError("browser CDP probe failed", 503) from exc
|
|
finally:
|
|
connection.close()
|
|
if response.status != 200 or len(body) > 64 * 1024:
|
|
raise BrowserRuntimeError("browser CDP probe failed", 503)
|
|
try:
|
|
return json.loads(body)
|
|
except json.JSONDecodeError as exc:
|
|
raise BrowserRuntimeError("browser CDP response is invalid", 503) from exc
|
|
|
|
def _check_cancel(self, record: RuntimeRecord) -> None:
|
|
event = self._cancel_events.get(record.runtime_id)
|
|
if event is not None and event.is_set():
|
|
raise RuntimeCancelled(network_id=record.network_id)
|
|
|
|
def _check_profile_size(self, record: RuntimeRecord) -> None:
|
|
if self.profile_cache_max_bytes == 0:
|
|
return
|
|
total = 0
|
|
root = Path(record.profile_dir)
|
|
for path in root.rglob("*"):
|
|
if path.is_symlink() or not path.is_file():
|
|
continue
|
|
try:
|
|
total += path.stat().st_size
|
|
except OSError as exc:
|
|
raise BrowserRuntimeError("Profile size could not be checked", 503, record.network_id) from exc
|
|
if total > self.profile_cache_max_bytes:
|
|
raise BrowserRuntimeError("Profile cache exceeds configured limit", 507, record.network_id)
|
|
|
|
def _browser_command(self, record: RuntimeRecord, proxy_url: str) -> list[str]:
|
|
command = list(record.command)
|
|
if not command or command[-1] != "about:blank":
|
|
raise BrowserRuntimeError("browser command must end with about:blank", 400, record.network_id)
|
|
arguments = command[:-1]
|
|
arguments.extend(
|
|
[
|
|
"--remote-debugging-address=127.0.0.1",
|
|
f"--remote-debugging-port={record.cdp_port}",
|
|
"--remote-allow-origins=http://127.0.0.1",
|
|
"--disable-gpu",
|
|
"--disable-gpu-compositing",
|
|
f"--user-data-dir={record.profile_dir}",
|
|
"--no-first-run",
|
|
"--no-default-browser-check",
|
|
]
|
|
)
|
|
if proxy_url:
|
|
arguments.extend([f"--proxy-server={proxy_url}", "--disable-non-proxied-udp"])
|
|
arguments.append("about:blank")
|
|
return [record.browser_path, *arguments]
|
|
|
|
def _locked_command(self, lock_path: Path, command: list[str]) -> list[str]:
|
|
flock = shutil.which("flock")
|
|
if not flock:
|
|
raise BrowserRuntimeError("flock is unavailable for runtime ownership", 503)
|
|
return [flock, str(lock_path), *command]
|
|
|
|
def _profile_dir(self, profile_id: str) -> Path:
|
|
if not isinstance(profile_id, str) or not PROFILE_ID_RE.fullmatch(profile_id):
|
|
raise BrowserRuntimeError("profile_id is invalid", 400)
|
|
digest = hashlib.sha256(profile_id.encode()).hexdigest()[:32]
|
|
path = (self.profile_root / digest).resolve()
|
|
try:
|
|
path.relative_to(self.profile_root)
|
|
except ValueError as exc:
|
|
raise BrowserRuntimeError("profile path escaped configured root", 500) from exc
|
|
return path
|
|
|
|
def _prepare_profile(self, record: RuntimeRecord) -> None:
|
|
path = Path(record.profile_dir)
|
|
path.mkdir(parents=True, exist_ok=True, mode=0o700)
|
|
stat = path.stat()
|
|
if stat.st_uid != os.getuid() or stat.st_mode & 0o077:
|
|
raise BrowserRuntimeError("Profile ownership or permissions are invalid", 409, record.network_id)
|
|
|
|
def _runtime_dir(self, record: RuntimeRecord) -> Path:
|
|
return self.state_dir / "runtimes" / record.runtime_id
|
|
|
|
def _log_path(self, record: RuntimeRecord, component: str) -> Path:
|
|
path = self.state_dir / "logs" / record.runtime_id / f"{component}.log"
|
|
path.parent.mkdir(parents=True, exist_ok=True, mode=0o700)
|
|
if self.log_max_bytes and path.exists() and path.stat().st_size > self.log_max_bytes:
|
|
raise BrowserRuntimeError("runtime log exceeds configured limit", 507, record.network_id)
|
|
return path
|
|
|
|
def _display_lock_path(self, display: int) -> Path:
|
|
return self.state_dir / "locks" / f"display-{display}.lock"
|
|
|
|
def _record_path(self, record: RuntimeRecord) -> Path:
|
|
return self._runtime_dir(record) / "runtime.json"
|
|
|
|
def _write(self, record: RuntimeRecord) -> None:
|
|
path = self._record_path(record)
|
|
path.parent.mkdir(parents=True, exist_ok=True, mode=0o700)
|
|
temporary = path.with_name(f".{path.name}.{os.getpid()}.tmp")
|
|
encoded = json.dumps(record.to_dict(), ensure_ascii=False, separators=(",", ":"))
|
|
fd = os.open(temporary, os.O_WRONLY | os.O_CREAT | os.O_TRUNC, 0o600)
|
|
try:
|
|
with os.fdopen(fd, "w", encoding="utf-8") as output:
|
|
output.write(encoded)
|
|
output.flush()
|
|
os.fsync(output.fileno())
|
|
os.replace(temporary, path)
|
|
finally:
|
|
with suppress(FileNotFoundError):
|
|
temporary.unlink()
|
|
|
|
def _records(self) -> list[RuntimeRecord]:
|
|
result = []
|
|
for path in sorted((self.state_dir / "runtimes").glob("*/runtime.json")):
|
|
try:
|
|
result.append(RuntimeRecord.from_dict(json.loads(path.read_text(encoding="utf-8"))))
|
|
except OSError as exc:
|
|
raise BrowserRuntimeError("runtime metadata cannot be read", 500) from exc
|
|
except json.JSONDecodeError as exc:
|
|
raise BrowserRuntimeError("runtime metadata is invalid", 500) from exc
|
|
return sorted(result, key=lambda item: (item.created_at, item.alias))
|
|
|
|
def _find_alias(self, alias: str, include_released: bool) -> RuntimeRecord | None:
|
|
if not ALIAS_RE.fullmatch(alias):
|
|
raise BrowserRuntimeError("runtime alias is invalid", 400)
|
|
matches = [record for record in self._records() if record.alias == alias]
|
|
active = [record for record in matches if include_released or record.state != "released"]
|
|
if len(active) > 1:
|
|
if include_released:
|
|
return max(active, key=lambda item: (item.created_at, item.runtime_id))
|
|
raise UnmanagedRuntime("multiple active runtime generations use the same alias")
|
|
return active[0] if active else None
|
|
|
|
def _find_record_for_generation(
|
|
self, alias: str, generation: Mapping[str, Any]
|
|
) -> RuntimeRecord | None:
|
|
if not ALIAS_RE.fullmatch(alias):
|
|
raise BrowserRuntimeError("runtime alias is invalid", 400)
|
|
matches = [record for record in self._records() if record.alias == alias]
|
|
runtime_id = generation.get("runtime_id", "")
|
|
if isinstance(runtime_id, str) and runtime_id not in {"", RUNTIME_CLEANUP_SENTINEL}:
|
|
exact = [record for record in matches if record.runtime_id == runtime_id]
|
|
if exact:
|
|
return max(exact, key=lambda item: (item.created_at, item.runtime_id))
|
|
nonreleased = [record for record in matches if record.state != "released"]
|
|
if nonreleased:
|
|
return max(nonreleased, key=lambda item: (item.created_at, item.runtime_id))
|
|
return max(matches, key=lambda item: (item.created_at, item.runtime_id)) if matches else None
|
|
|
|
def _require_record(
|
|
self,
|
|
alias: str,
|
|
generation: Mapping[str, Any],
|
|
require_active: bool = True,
|
|
) -> RuntimeRecord:
|
|
record = self._find_alias(alias, include_released=True)
|
|
if record is None:
|
|
raise FileNotFoundError(alias)
|
|
self._fence(record, generation)
|
|
if require_active and record.state == "released":
|
|
raise FileNotFoundError(alias)
|
|
return record
|
|
|
|
def _fence(self, record: RuntimeRecord, generation: Mapping[str, Any]) -> None:
|
|
if (
|
|
generation.get("runtime_id") != record.runtime_id
|
|
or generation.get("binding_version") != record.binding_version
|
|
or generation.get("network_id") != record.network_id
|
|
):
|
|
raise GenerationConflict()
|
|
|
|
def _public(self, record: RuntimeRecord, ready: bool) -> dict[str, Any]:
|
|
endpoint = f"http://127.0.0.1:{record.cdp_port}" if ready and record.cdp_port else ""
|
|
proxy_ready = bool(
|
|
ready
|
|
and (
|
|
not record.network_exit_id
|
|
or self.proxies.ready(
|
|
record.alias,
|
|
record.proxy_port,
|
|
record.runtime_id,
|
|
record.network_id,
|
|
)
|
|
)
|
|
)
|
|
return {
|
|
"id": record.runtime_id,
|
|
"runtime_id": record.runtime_id,
|
|
"alias": record.alias,
|
|
"name": record.name,
|
|
"state": record.state,
|
|
"status": "ready" if ready else record.cleanup_error or record.state,
|
|
"ready": ready,
|
|
"endpoint": endpoint,
|
|
"binding_version": record.binding_version,
|
|
"network_exit_id": record.network_exit_id,
|
|
"network_id": record.network_id,
|
|
"node_id": record.node_id,
|
|
"owner": record.owner,
|
|
"proxy_ready": proxy_ready,
|
|
"cleanup_state": record.cleanup_state,
|
|
"cleanup_error": record.cleanup_error,
|
|
"display": record.display,
|
|
"cdp_port": record.cdp_port,
|
|
"proxy_port": record.proxy_port,
|
|
"display_mode": "external" if not record.xvfb_unit and record.display else "managed",
|
|
}
|
|
|
|
def _unit_name(self, component: str, record: RuntimeRecord) -> str:
|
|
return f"creatorhub-{record.alias}-{record.runtime_id[:16]}-{component}.service"
|
|
|
|
def _reserve_display(self, record: RuntimeRecord) -> _Lease:
|
|
if self.external_display is not None:
|
|
if any(
|
|
item.runtime_id != record.runtime_id
|
|
and item.state != "released"
|
|
and item.display == self.external_display
|
|
for item in self._records()
|
|
):
|
|
raise BrowserRuntimeError("external Xvfb display is already in use", 409, record.network_id)
|
|
if not self._display_available(self.external_display):
|
|
raise BrowserRuntimeError("external Xvfb display is unavailable", 503, record.network_id)
|
|
return self.display_allocator.reserve_existing(self.external_display)
|
|
|
|
def unavailable(value: int) -> bool:
|
|
return any(
|
|
item.runtime_id != record.runtime_id
|
|
and item.state != "released"
|
|
and item.display == value
|
|
for item in self._records()
|
|
)
|
|
|
|
return self.display_allocator.reserve(
|
|
self.display_start,
|
|
self.display_end,
|
|
unavailable,
|
|
)
|
|
|
|
def _reserve_port(self, record: RuntimeRecord, prefix: str, start: int, end: int) -> _Lease:
|
|
def unavailable(value: int) -> bool:
|
|
for item in self._records():
|
|
if item.runtime_id == record.runtime_id:
|
|
continue
|
|
if item.state in {"released"}:
|
|
continue
|
|
if value in {item.cdp_port, item.proxy_port}:
|
|
return True
|
|
return False
|
|
|
|
return self.port_allocator.reserve(prefix, start, end, unavailable)
|
|
|
|
def _release_startup_locks(self, record: RuntimeRecord) -> None:
|
|
lock = self._profile_locks.pop(record.runtime_id, None)
|
|
if lock:
|
|
lock.release()
|
|
|
|
def _release_display_lease(self, record: RuntimeRecord) -> None:
|
|
lease = self._display_leases.pop(record.runtime_id, None)
|
|
if lease:
|
|
lease.lock.release()
|
|
|
|
def _profile_in_use(self, record: RuntimeRecord) -> bool:
|
|
return any(
|
|
item.runtime_id != record.runtime_id
|
|
and item.state != "released"
|
|
and item.profile_dir == record.profile_dir
|
|
for item in self._records()
|
|
)
|
|
|
|
def _release_runtime_leases(self, record: RuntimeRecord) -> None:
|
|
display = self._display_leases.pop(record.runtime_id, None)
|
|
if display:
|
|
display.lock.release()
|
|
for key in ((record.runtime_id, "cdp"), (record.runtime_id, "proxy")):
|
|
lease = self._port_leases.pop(key, None)
|
|
if lease:
|
|
lease.lock.release()
|
|
undo = self._proxy_undo.pop(record.runtime_id, None)
|
|
if undo:
|
|
with suppress(Exception):
|
|
undo()
|
|
|
|
def _remove_runtime_tmp(self, record: RuntimeRecord) -> None:
|
|
temporary = self._runtime_dir(record) / "tmp"
|
|
if not temporary.exists():
|
|
return
|
|
for path in sorted(temporary.rglob("*"), reverse=True):
|
|
if path.is_file() or path.is_symlink():
|
|
path.unlink()
|
|
elif path.is_dir():
|
|
path.rmdir()
|
|
with suppress(OSError):
|
|
temporary.rmdir()
|
|
|
|
def _purge_profile(self, record: RuntimeRecord) -> None:
|
|
path = Path(record.profile_dir).resolve()
|
|
try:
|
|
path.relative_to(self.profile_root)
|
|
except ValueError as exc:
|
|
raise BrowserRuntimeError("Profile path does not belong to gateway", 409) from exc
|
|
if path == self.profile_root or not path.exists():
|
|
return
|
|
lock = FileLock(path / ".creatorhub-profile.lock")
|
|
if not lock.acquire():
|
|
raise RuntimeCleanupPending("Profile is still in use", record.network_id)
|
|
try:
|
|
for child in sorted(path.rglob("*"), reverse=True):
|
|
if child.is_file() or child.is_symlink():
|
|
child.unlink()
|
|
elif child.is_dir():
|
|
child.rmdir()
|
|
path.rmdir()
|
|
finally:
|
|
lock.release()
|
|
|
|
|
|
def validate_runtime_input(value: Mapping[str, Any]) -> None:
|
|
if not isinstance(value, Mapping):
|
|
raise BrowserRuntimeError("request body must be one JSON object", 400)
|
|
allowed = {
|
|
"alias",
|
|
"name",
|
|
"profile_id",
|
|
"cmd",
|
|
"binding_version",
|
|
"network_exit_id",
|
|
"network_exit",
|
|
"stopped",
|
|
}
|
|
if set(value) - allowed:
|
|
raise BrowserRuntimeError("body contains unknown fields", 400)
|
|
alias = value.get("alias", "")
|
|
name = value.get("name", "")
|
|
profile_id = value.get("profile_id", alias)
|
|
command = value.get("cmd")
|
|
binding = value.get("binding_version")
|
|
exit_id = value.get("network_exit_id", "")
|
|
stopped = value.get("stopped", False)
|
|
if not isinstance(alias, str) or not ALIAS_RE.fullmatch(alias):
|
|
raise BrowserRuntimeError("alias must match [a-z0-9][a-z0-9-]{0,31}", 400)
|
|
if not isinstance(name, str) or not 1 <= len(name) <= 64 or has_control(name):
|
|
raise BrowserRuntimeError("name must be 1..64 visible characters", 400)
|
|
if not isinstance(profile_id, str) or not PROFILE_ID_RE.fullmatch(profile_id):
|
|
raise BrowserRuntimeError("profile_id is invalid", 400)
|
|
if type(binding) is not int or binding < 1:
|
|
raise BrowserRuntimeError("binding_version is invalid", 400)
|
|
if not isinstance(command, list) or not 1 <= len(command) <= 64 or command[-1] != "about:blank":
|
|
raise BrowserRuntimeError("cmd must contain 1..64 arguments and end with about:blank", 400)
|
|
total = 0
|
|
reserved = (
|
|
"--user-data-dir",
|
|
"--remote-debugging-address",
|
|
"--remote-debugging-port",
|
|
"--remote-allow-origins",
|
|
"--remote-debugging-pipe",
|
|
"--proxy-server",
|
|
"--display",
|
|
"--headless",
|
|
)
|
|
for item in command:
|
|
if not isinstance(item, str) or not item or has_control(item) or item == "--no-sandbox":
|
|
raise BrowserRuntimeError("cmd arguments are invalid", 400)
|
|
if any(item == flag or item.startswith(flag + "=") for flag in reserved):
|
|
raise BrowserRuntimeError("cmd contains a gateway-managed flag", 400)
|
|
if item == "--disable-non-proxied-udp":
|
|
raise BrowserRuntimeError("cmd contains a gateway-managed flag", 400)
|
|
total += len(item)
|
|
if total > 4096:
|
|
raise BrowserRuntimeError("cmd arguments exceed 4096 characters", 400)
|
|
if not isinstance(exit_id, str):
|
|
raise BrowserRuntimeError("network_exit_id must be a string", 400)
|
|
if exit_id and not re.fullmatch(r"[A-Za-z0-9][A-Za-z0-9._/-]{0,127}", exit_id):
|
|
raise BrowserRuntimeError("network_exit_id is invalid", 400)
|
|
exit_value = parse_proxy_exit(value.get("network_exit", {}))
|
|
direct = not exit_id and exit_value == ProxyExit("", "", 0)
|
|
if bool(exit_id) != (exit_value != ProxyExit("", "", 0)):
|
|
raise BrowserRuntimeError("network_exit_id and network_exit must identify one binding", 400)
|
|
if type(stopped) is not bool:
|
|
raise BrowserRuntimeError("stopped must be boolean", 400)
|
|
if stopped and not direct:
|
|
raise BrowserRuntimeError("stopped browsers must use direct networking", 400)
|
|
if not direct:
|
|
validate_proxy_exit(exit_value)
|
|
if isinstance(value, dict):
|
|
value["profile_id"] = profile_id
|
|
value["network_exit"] = exit_value
|
|
value["network_exit_id"] = exit_id
|
|
|
|
|
|
def parse_proxy_exit(value: object) -> ProxyExit:
|
|
if not isinstance(value, dict):
|
|
raise BrowserRuntimeError("network_exit must be an object", 400)
|
|
allowed = {"protocol", "host", "port", "username", "password"}
|
|
if set(value) - allowed:
|
|
raise BrowserRuntimeError("network_exit contains unknown fields", 400)
|
|
try:
|
|
result = ProxyExit(
|
|
value.get("protocol", ""),
|
|
value.get("host", ""),
|
|
value.get("port", 0),
|
|
value.get("username", ""),
|
|
value.get("password", ""),
|
|
)
|
|
except (TypeError, ValueError) as exc:
|
|
raise BrowserRuntimeError("network_exit is invalid", 400) from exc
|
|
if (
|
|
not all(isinstance(item, str) for item in (result.protocol, result.host, result.username, result.credential))
|
|
or type(result.port) is not int
|
|
):
|
|
raise BrowserRuntimeError("network_exit is invalid", 400)
|
|
return result
|
|
|
|
|
|
def validate_proxy_exit(exit: ProxyExit) -> None:
|
|
if (
|
|
exit.protocol not in {"http", "https", "socks4", "socks5"}
|
|
or not exit.host
|
|
or len(exit.host) > 253
|
|
or any(char in exit.host for char in "@/[]?# \t\r\n")
|
|
or not 1 <= exit.port <= 65535
|
|
or (not exit.username and exit.credential)
|
|
or len(exit.username) > 255
|
|
or len(exit.credential) > 255
|
|
or has_control(exit.username)
|
|
or has_control(exit.credential)
|
|
):
|
|
raise BrowserRuntimeError("network_exit must contain a valid proxy endpoint", 400)
|
|
|
|
|
|
def has_control(value: str) -> bool:
|
|
return any(ord(char) < 0x20 or ord(char) == 0x7F for char in value)
|
|
|
|
|
|
class _ManagedAliasLock:
|
|
def __init__(self, manager: NativeRuntimeManager, alias: str) -> None:
|
|
self.manager = manager
|
|
self.alias = alias
|
|
self.lock: FileLock | None = None
|
|
|
|
def __enter__(self) -> _ManagedAliasLock:
|
|
if not ALIAS_RE.fullmatch(self.alias):
|
|
raise BrowserRuntimeError("runtime alias is invalid", 400)
|
|
with self.manager._lock:
|
|
lock = FileLock(self.manager.state_dir / "locks" / f"alias-{self.alias}.lock")
|
|
if not lock.acquire():
|
|
raise BrowserRuntimeError("browser alias is busy", 409)
|
|
self.manager._alias_locks[self.alias] = lock
|
|
self.lock = lock
|
|
return self
|
|
|
|
def __exit__(self, exc_type: object, exc_value: object, traceback: object) -> None:
|
|
with self.manager._lock:
|
|
lock = self.manager._alias_locks.pop(self.alias, None)
|
|
if lock:
|
|
lock.release()
|
|
self.lock = None
|