mirror of
https://github.com/openhop-dev/openhop_repeater.git
synced 2026-10-23 21:43:02 +00:00
541 lines
21 KiB
Python
541 lines
21 KiB
Python
"""JSON-lines Unix domain socket IPC between Repeater and the plugin manager."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import hashlib
|
|
import json
|
|
import logging
|
|
import os
|
|
import socket
|
|
import tempfile
|
|
import threading
|
|
import time
|
|
from pathlib import Path
|
|
from typing import Any, Callable, Optional
|
|
|
|
from .manager import PluginManager, PluginManagerError
|
|
|
|
logger = logging.getLogger("PluginIPC")
|
|
|
|
DEFAULT_TIMEOUT = 30.0
|
|
OPERATION_TIMEOUT = 900.0 # completion budget, not a subprocess cancellation deadline
|
|
MAX_CONNECTIONS = 16
|
|
SLOW_OPS = frozenset({"install", "catalogue_install", "update", "catalogue", "check_update"})
|
|
INSTALL_OPS = frozenset({"install", "catalogue_install", "update"})
|
|
MAX_LINE_BYTES = 8 * 1024 * 1024 # install paths are short; keep headroom
|
|
MAX_AF_UNIX_PATH_BYTES = 92
|
|
|
|
|
|
def _safe_socket_path(path: Path | str) -> Path:
|
|
candidate = Path(path)
|
|
if len(str(candidate)) <= MAX_AF_UNIX_PATH_BYTES:
|
|
return candidate
|
|
digest = hashlib.sha256(str(candidate).encode("utf-8")).hexdigest()[:16]
|
|
fallback = Path(tempfile.gettempdir()) / f"openhop-plugin-{digest}.sock"
|
|
logger.warning(
|
|
"Unix socket path %s is too long for AF_UNIX; using %s instead", candidate, fallback
|
|
)
|
|
return fallback
|
|
|
|
|
|
class PluginIPCError(Exception):
|
|
def __init__(self, message: str, code: int = 500):
|
|
super().__init__(message)
|
|
self.code = code
|
|
|
|
|
|
class PluginIPCOutcomeUnknown(PluginIPCError):
|
|
"""Transport stopped waiting; the manager may still complete the operation."""
|
|
|
|
def __init__(self, *, upload_safe: bool = False):
|
|
super().__init__(
|
|
"Plugin manager response was lost or timed out; outcome is unknown. "
|
|
"The operation may still complete. Check plugin status before retrying.",
|
|
504,
|
|
)
|
|
self.outcome = "unknown"
|
|
self.upload_safe = upload_safe
|
|
|
|
|
|
class PluginManagerUnavailable(PluginIPCError):
|
|
def __init__(self, message: str = "plugin manager unavailable"):
|
|
super().__init__(message, code=503)
|
|
|
|
|
|
def _read_line(
|
|
conn: socket.socket,
|
|
max_bytes: int = MAX_LINE_BYTES,
|
|
*,
|
|
buffer: Optional[bytearray] = None,
|
|
deadline: Optional[float] = None,
|
|
) -> bytes:
|
|
buf = buffer if buffer is not None else bytearray()
|
|
if deadline is None:
|
|
deadline = time.monotonic() + (conn.gettimeout() or DEFAULT_TIMEOUT)
|
|
while True:
|
|
end = buf.find(b"\n")
|
|
if end >= 0:
|
|
if end > max_bytes:
|
|
raise PluginIPCError("IPC message too large", 413)
|
|
line = bytes(buf[:end])
|
|
del buf[: end + 1]
|
|
return line
|
|
if len(buf) > max_bytes:
|
|
raise PluginIPCError("IPC message too large", 413)
|
|
remaining = deadline - time.monotonic()
|
|
if remaining <= 0:
|
|
raise TimeoutError("IPC response deadline exceeded")
|
|
conn.settimeout(remaining)
|
|
chunk = conn.recv(min(4096, max_bytes + 1 - len(buf)))
|
|
if not chunk:
|
|
if buf:
|
|
raise PluginIPCError("incomplete IPC message", 502)
|
|
return b""
|
|
buf.extend(chunk)
|
|
|
|
|
|
def _send_msg(conn: socket.socket, payload: dict[str, Any]) -> None:
|
|
data = (json.dumps(payload, separators=(",", ":")) + "\n").encode("utf-8")
|
|
conn.sendall(data)
|
|
|
|
|
|
class PluginIPCServer:
|
|
"""Serve plugin manager operations over a Unix domain socket."""
|
|
|
|
def __init__(self, socket_path: Path | str, manager: PluginManager):
|
|
self.socket_path = _safe_socket_path(socket_path)
|
|
self.manager = manager
|
|
self._sock: Optional[socket.socket] = None
|
|
self._thread: Optional[threading.Thread] = None
|
|
self._stop = threading.Event()
|
|
self._slots = threading.BoundedSemaphore(MAX_CONNECTIONS)
|
|
self._slow_slot = threading.BoundedSemaphore(1)
|
|
self._workers: set[threading.Thread] = set()
|
|
self._workers_lock = threading.Lock()
|
|
self._handlers: dict[str, Callable[[dict[str, Any]], Any]] = {
|
|
"list": self._op_list,
|
|
"status": self._op_status,
|
|
"install": self._op_install,
|
|
"enable": self._op_enable,
|
|
"disable": self._op_disable,
|
|
"start": self._op_start,
|
|
"stop": self._op_stop,
|
|
"restart": self._op_restart,
|
|
"logs": self._op_logs,
|
|
"uninstall": self._op_uninstall,
|
|
"ui_info": self._op_ui_info,
|
|
"get_config": self._op_get_config,
|
|
"get_runtime": self._op_get_runtime,
|
|
"set_config": self._op_set_config,
|
|
"catalogue": self._op_catalogue,
|
|
"catalogue_install": self._op_catalogue_install,
|
|
"check_update": self._op_check_update,
|
|
"update": self._op_update,
|
|
"progress": self._op_progress,
|
|
"ping": self._op_ping,
|
|
}
|
|
|
|
def start(self) -> None:
|
|
self.socket_path.parent.mkdir(parents=True, exist_ok=True)
|
|
if self.socket_path.exists():
|
|
try:
|
|
self.socket_path.unlink()
|
|
except OSError as exc:
|
|
logger.warning("Could not remove stale socket %s: %s", self.socket_path, exc)
|
|
|
|
self._sock = socket.socket(socket.AF_UNIX, socket.SOCK_STREAM)
|
|
self._sock.bind(str(self.socket_path))
|
|
try:
|
|
# Repeater and the manager may be separate processes in one service
|
|
# group, so the local IPC socket is intentionally group-writable.
|
|
os.chmod(self.socket_path, 0o660) # nosec B103
|
|
except OSError as exc:
|
|
logger.debug("Could not set IPC socket permissions: %s", exc)
|
|
self._sock.listen(16)
|
|
self._sock.settimeout(1.0)
|
|
self._stop.clear()
|
|
self._thread = threading.Thread(target=self._serve_loop, name="plugin-ipc", daemon=True)
|
|
self._thread.start()
|
|
logger.info("Plugin manager IPC listening on %s", self.socket_path)
|
|
|
|
def stop(self) -> None:
|
|
self._stop.set()
|
|
sock = self._sock
|
|
self._sock = None
|
|
if sock is not None:
|
|
try:
|
|
sock.close()
|
|
except OSError:
|
|
pass
|
|
thread = self._thread
|
|
if thread and thread.is_alive():
|
|
thread.join(timeout=3)
|
|
self._thread = None
|
|
deadline = time.monotonic() + 3
|
|
with self._workers_lock:
|
|
workers = list(self._workers)
|
|
for worker in workers:
|
|
worker.join(timeout=max(0, deadline - time.monotonic()))
|
|
if self.socket_path.exists():
|
|
try:
|
|
self.socket_path.unlink()
|
|
except OSError:
|
|
pass
|
|
|
|
def _serve_loop(self) -> None:
|
|
sock = self._sock
|
|
if sock is None:
|
|
logger.error("IPC serve loop started without a socket")
|
|
return
|
|
while not self._stop.is_set():
|
|
try:
|
|
conn, _ = sock.accept()
|
|
except socket.timeout:
|
|
continue
|
|
except OSError:
|
|
if self._stop.is_set():
|
|
break
|
|
continue
|
|
if not self._slots.acquire(blocking=False):
|
|
# No unbounded executor queue; a saturated transport rejects immediately.
|
|
try:
|
|
conn.settimeout(0.1)
|
|
_send_msg(
|
|
conn, {"ok": False, "error": "plugin IPC busy; retry later", "code": 503}
|
|
)
|
|
except OSError:
|
|
pass
|
|
finally:
|
|
conn.close()
|
|
continue
|
|
worker = threading.Thread(target=self._connection_worker, args=(conn,), daemon=True)
|
|
with self._workers_lock:
|
|
self._workers.add(worker)
|
|
worker.start()
|
|
|
|
def _connection_worker(self, conn: socket.socket) -> None:
|
|
try:
|
|
self._handle_connection(conn)
|
|
except Exception as exc:
|
|
logger.debug("IPC connection error: %s", exc)
|
|
finally:
|
|
conn.close()
|
|
with self._workers_lock:
|
|
self._workers.discard(threading.current_thread())
|
|
self._slots.release()
|
|
|
|
def _handle_connection(self, conn: socket.socket) -> None:
|
|
conn.settimeout(DEFAULT_TIMEOUT)
|
|
raw = _read_line(conn)
|
|
if not raw:
|
|
return
|
|
req_id: Any = None
|
|
try:
|
|
request = json.loads(raw.decode("utf-8"))
|
|
if not isinstance(request, dict):
|
|
raise PluginIPCError("request must be a JSON object", 400)
|
|
req_id = request.get("id")
|
|
op = request.get("op")
|
|
if not isinstance(op, str) or op not in self._handlers:
|
|
raise PluginIPCError(f"unknown op: {op!r}", 400)
|
|
slow = op in SLOW_OPS
|
|
if slow and not self._slow_slot.acquire(blocking=False):
|
|
raise PluginIPCError("plugin download/install busy; retry later", 503)
|
|
try:
|
|
if op == "install":
|
|
wheel_path = request.get("wheel_path")
|
|
if not isinstance(wheel_path, str) or not wheel_path.strip():
|
|
raise PluginIPCError("wheel_path is required", 400)
|
|
with self.manager.stage_upload(wheel_path.strip()) as staged:
|
|
if request.get("completion_protocol") == 1:
|
|
_send_msg(conn, {"id": req_id, "ok": True, "processing": True})
|
|
result = self.manager.install(staged)
|
|
else:
|
|
if op in INSTALL_OPS and request.get("completion_protocol") == 1:
|
|
_send_msg(conn, {"id": req_id, "ok": True, "processing": True})
|
|
result = self._handlers[op](request)
|
|
_send_msg(conn, {"id": req_id, "ok": True, "result": result})
|
|
finally:
|
|
if slow:
|
|
self._slow_slot.release()
|
|
except PluginManagerError as exc:
|
|
_send_msg(
|
|
conn,
|
|
{"id": req_id, "ok": False, "error": str(exc), "code": int(exc.code)},
|
|
)
|
|
except PluginIPCError as exc:
|
|
_send_msg(
|
|
conn,
|
|
{"id": req_id, "ok": False, "error": str(exc), "code": int(exc.code)},
|
|
)
|
|
except OSError:
|
|
# A disconnected waiter is not an operation failure or cancellation.
|
|
raise
|
|
except Exception as exc:
|
|
logger.exception("IPC handler failure")
|
|
_send_msg(
|
|
conn,
|
|
{"id": req_id, "ok": False, "error": str(exc), "code": 500},
|
|
)
|
|
|
|
def _op_ping(self, request: dict[str, Any]) -> dict[str, Any]:
|
|
return {"pong": True}
|
|
|
|
def _op_list(self, request: dict[str, Any]) -> dict[str, Any]:
|
|
return {"plugins": self.manager.list_plugins()}
|
|
|
|
def _require_id(self, request: dict[str, Any]) -> str:
|
|
plugin_id = request.get("plugin_id") or request.get("id_name")
|
|
# Avoid clashing with request correlation "id"
|
|
if plugin_id is None:
|
|
plugin_id = request.get("plugin")
|
|
if not isinstance(plugin_id, str) or not plugin_id.strip():
|
|
raise PluginIPCError("plugin_id is required", 400)
|
|
return plugin_id.strip()
|
|
|
|
def _op_status(self, request: dict[str, Any]) -> dict[str, Any]:
|
|
return self.manager.status(self._require_id(request))
|
|
|
|
def _op_install(self, request: dict[str, Any]) -> dict[str, Any]:
|
|
wheel_path = request.get("wheel_path")
|
|
if not isinstance(wheel_path, str) or not wheel_path.strip():
|
|
raise PluginIPCError("wheel_path is required", 400)
|
|
return self.manager.install(wheel_path.strip())
|
|
|
|
def _op_enable(self, request: dict[str, Any]) -> dict[str, Any]:
|
|
return self.manager.enable(self._require_id(request))
|
|
|
|
def _op_disable(self, request: dict[str, Any]) -> dict[str, Any]:
|
|
return self.manager.disable(self._require_id(request))
|
|
|
|
def _op_start(self, request: dict[str, Any]) -> dict[str, Any]:
|
|
return self.manager.start_plugin(self._require_id(request))
|
|
|
|
def _op_stop(self, request: dict[str, Any]) -> dict[str, Any]:
|
|
return self.manager.stop_plugin(self._require_id(request))
|
|
|
|
def _op_restart(self, request: dict[str, Any]) -> dict[str, Any]:
|
|
return self.manager.restart_plugin(self._require_id(request))
|
|
|
|
def _op_logs(self, request: dict[str, Any]) -> dict[str, Any]:
|
|
tail = request.get("tail", 200)
|
|
try:
|
|
tail_i = int(tail)
|
|
except (TypeError, ValueError) as exc:
|
|
raise PluginIPCError("tail must be an integer", 400) from exc
|
|
return self.manager.logs(self._require_id(request), tail=tail_i)
|
|
|
|
def _op_uninstall(self, request: dict[str, Any]) -> dict[str, Any]:
|
|
delete_data = bool(request.get("delete_data", False))
|
|
return self.manager.uninstall(self._require_id(request), delete_data=delete_data)
|
|
|
|
def _op_ui_info(self, request: dict[str, Any]) -> dict[str, Any]:
|
|
plugin_id = self._require_id(request)
|
|
status = self.manager.status(plugin_id)
|
|
root = self.manager.ui_root_for(plugin_id)
|
|
return {
|
|
**status,
|
|
"ui_root": str(root) if root else None,
|
|
}
|
|
|
|
def _op_get_config(self, request: dict[str, Any]) -> dict[str, Any]:
|
|
return self.manager.get_config(self._require_id(request))
|
|
|
|
def _op_get_runtime(self, request: dict[str, Any]) -> dict[str, Any]:
|
|
return self.manager.get_runtime(self._require_id(request))
|
|
|
|
def _op_set_config(self, request: dict[str, Any]) -> dict[str, Any]:
|
|
config = request.get("config")
|
|
if not isinstance(config, dict):
|
|
raise PluginIPCError("config must be a JSON object", 400)
|
|
restart = bool(request.get("restart", False))
|
|
return self.manager.set_config(self._require_id(request), config, restart=restart)
|
|
|
|
def _op_catalogue(self, request: dict[str, Any]) -> dict[str, Any]:
|
|
force = bool(request.get("force_refresh") or request.get("refresh"))
|
|
return self.manager.list_catalogue(force_refresh=force)
|
|
|
|
def _op_catalogue_install(self, request: dict[str, Any]) -> dict[str, Any]:
|
|
plugin_id = self._require_id(request)
|
|
version = request.get("version")
|
|
if version is not None:
|
|
version = str(version).strip() or None
|
|
force = bool(request.get("force_refresh") or request.get("refresh"))
|
|
return self.manager.install_from_catalogue(plugin_id, version=version, force_refresh=force)
|
|
|
|
def _op_check_update(self, request: dict[str, Any]) -> dict[str, Any]:
|
|
plugin_id = self._require_id(request)
|
|
force = bool(request.get("force_refresh") or request.get("refresh"))
|
|
return self.manager.check_update(plugin_id, force_refresh=force)
|
|
|
|
def _op_progress(self, request: dict[str, Any]) -> dict[str, Any]:
|
|
plugin_id = self._require_id(request)
|
|
try:
|
|
since = int(request.get("since") or 0)
|
|
except (TypeError, ValueError):
|
|
raise PluginIPCError("since must be an integer", 400)
|
|
return self.manager.progress(plugin_id, since=since)
|
|
|
|
def _op_update(self, request: dict[str, Any]) -> dict[str, Any]:
|
|
plugin_id = self._require_id(request)
|
|
version = request.get("version")
|
|
if version is not None:
|
|
version = str(version).strip() or None
|
|
force = bool(request.get("force_refresh") or request.get("refresh"))
|
|
return self.manager.update_plugin(plugin_id, version=version, force_refresh=force)
|
|
|
|
|
|
class PluginIPCClient:
|
|
"""Client used by the Repeater HTTP layer."""
|
|
|
|
def __init__(self, socket_path: Path | str, timeout: float = DEFAULT_TIMEOUT):
|
|
self.socket_path = _safe_socket_path(socket_path)
|
|
self.timeout = timeout
|
|
self.operation_timeout = OPERATION_TIMEOUT
|
|
self._next_id = 1
|
|
self._lock = threading.Lock()
|
|
|
|
def available(self) -> bool:
|
|
return self.socket_path.exists()
|
|
|
|
def call(self, op: str, **params: Any) -> Any:
|
|
if not self.available():
|
|
raise PluginManagerUnavailable()
|
|
|
|
with self._lock:
|
|
req_id = self._next_id
|
|
self._next_id += 1
|
|
|
|
request = {"id": req_id, "op": op, **params}
|
|
if op in INSTALL_OPS:
|
|
request["completion_protocol"] = 1
|
|
try:
|
|
sock = socket.socket(socket.AF_UNIX, socket.SOCK_STREAM)
|
|
sock.settimeout(self.timeout)
|
|
sock.connect(str(self.socket_path))
|
|
except OSError as exc:
|
|
sock.close()
|
|
raise PluginManagerUnavailable(f"plugin manager unavailable: {exc}") from exc
|
|
|
|
upload_safe = False
|
|
try:
|
|
_send_msg(sock, request)
|
|
buffer = bytearray()
|
|
deadline = time.monotonic() + self.timeout
|
|
completion_deadline = time.monotonic() + self.operation_timeout
|
|
while True:
|
|
try:
|
|
raw = _read_line(sock, buffer=buffer, deadline=deadline)
|
|
except PluginIPCError as exc:
|
|
raise PluginIPCOutcomeUnknown(upload_safe=upload_safe) from exc
|
|
if not raw:
|
|
raise PluginIPCOutcomeUnknown(upload_safe=upload_safe)
|
|
try:
|
|
response = json.loads(raw.decode("utf-8"))
|
|
except (ValueError, UnicodeError) as exc:
|
|
raise PluginIPCOutcomeUnknown(upload_safe=upload_safe) from exc
|
|
if not isinstance(response, dict):
|
|
raise PluginIPCOutcomeUnknown(upload_safe=upload_safe)
|
|
if not response.get("ok"):
|
|
code = int(response.get("code") or 500)
|
|
err = str(response.get("error") or "plugin manager error")
|
|
raise PluginIPCError(err, code)
|
|
if response.get("id") != req_id:
|
|
raise PluginIPCOutcomeUnknown(upload_safe=upload_safe)
|
|
if response.get("processing") is True:
|
|
upload_safe = op == "install"
|
|
deadline = completion_deadline
|
|
continue
|
|
return response.get("result")
|
|
except OSError as exc:
|
|
raise PluginIPCOutcomeUnknown(upload_safe=upload_safe) from exc
|
|
finally:
|
|
try:
|
|
sock.close()
|
|
except OSError:
|
|
pass
|
|
|
|
def list_plugins(self) -> list[dict[str, Any]]:
|
|
result = self.call("list") or {}
|
|
return list(result.get("plugins") or [])
|
|
|
|
def status(self, plugin_id: str) -> dict[str, Any]:
|
|
return self.call("status", plugin_id=plugin_id)
|
|
|
|
def install(self, wheel_path: str) -> dict[str, Any]:
|
|
return self.call("install", wheel_path=wheel_path)
|
|
|
|
def enable(self, plugin_id: str) -> dict[str, Any]:
|
|
return self.call("enable", plugin_id=plugin_id)
|
|
|
|
def disable(self, plugin_id: str) -> dict[str, Any]:
|
|
return self.call("disable", plugin_id=plugin_id)
|
|
|
|
def start(self, plugin_id: str) -> dict[str, Any]:
|
|
return self.call("start", plugin_id=plugin_id)
|
|
|
|
def stop(self, plugin_id: str) -> dict[str, Any]:
|
|
return self.call("stop", plugin_id=plugin_id)
|
|
|
|
def restart(self, plugin_id: str) -> dict[str, Any]:
|
|
return self.call("restart", plugin_id=plugin_id)
|
|
|
|
def logs(self, plugin_id: str, tail: int = 200) -> dict[str, Any]:
|
|
return self.call("logs", plugin_id=plugin_id, tail=tail)
|
|
|
|
def uninstall(self, plugin_id: str, delete_data: bool = False) -> dict[str, Any]:
|
|
return self.call("uninstall", plugin_id=plugin_id, delete_data=delete_data)
|
|
|
|
def ui_info(self, plugin_id: str) -> dict[str, Any]:
|
|
return self.call("ui_info", plugin_id=plugin_id)
|
|
|
|
def get_config(self, plugin_id: str) -> dict[str, Any]:
|
|
return self.call("get_config", plugin_id=plugin_id)
|
|
|
|
def get_runtime(self, plugin_id: str) -> dict[str, Any]:
|
|
return self.call("get_runtime", plugin_id=plugin_id)
|
|
|
|
def set_config(
|
|
self, plugin_id: str, config: dict[str, Any], *, restart: bool = False
|
|
) -> dict[str, Any]:
|
|
return self.call("set_config", plugin_id=plugin_id, config=config, restart=restart)
|
|
|
|
def catalogue(self, *, force_refresh: bool = False) -> dict[str, Any]:
|
|
return self.call("catalogue", force_refresh=bool(force_refresh))
|
|
|
|
def catalogue_install(
|
|
self,
|
|
plugin_id: str,
|
|
*,
|
|
version: Optional[str] = None,
|
|
force_refresh: bool = False,
|
|
) -> dict[str, Any]:
|
|
payload: dict[str, Any] = {
|
|
"plugin_id": plugin_id,
|
|
"force_refresh": bool(force_refresh),
|
|
}
|
|
if version:
|
|
payload["version"] = version
|
|
return self.call("catalogue_install", **payload)
|
|
|
|
def progress(self, plugin_id: str, *, since: int = 0) -> dict[str, Any]:
|
|
return self.call("progress", plugin_id=plugin_id, since=int(since))
|
|
|
|
def check_update(self, plugin_id: str, *, force_refresh: bool = False) -> dict[str, Any]:
|
|
return self.call("check_update", plugin_id=plugin_id, force_refresh=bool(force_refresh))
|
|
|
|
def update(
|
|
self,
|
|
plugin_id: str,
|
|
*,
|
|
version: Optional[str] = None,
|
|
force_refresh: bool = False,
|
|
) -> dict[str, Any]:
|
|
payload: dict[str, Any] = {
|
|
"plugin_id": plugin_id,
|
|
"force_refresh": bool(force_refresh),
|
|
}
|
|
if version:
|
|
payload["version"] = version
|
|
return self.call("update", **payload)
|