119 lines
4.3 KiB
Python
119 lines
4.3 KiB
Python
"""Control-socket client for FIPS daemons.
|
|
|
|
Standalone port of the patterns in
|
|
``fips/testing/chaos/sim/control.py`` — no Docker dependency. Speaks the
|
|
line-delimited JSON protocol defined in ``fips/src/control/protocol.rs``:
|
|
|
|
request: {"command": "show_status"}\n
|
|
response: {"status":"ok","data":{...}}\n
|
|
|
|
The control socket is a Unix domain socket (default ``/run/fips/control.sock``).
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import json
|
|
import logging
|
|
import socket
|
|
from typing import Any, Optional
|
|
|
|
log = logging.getLogger(__name__)
|
|
|
|
DEFAULT_SOCKET = "/run/fips/control.sock"
|
|
# Match the daemon's I/O timeout (control/mod.rs:29).
|
|
DEFAULT_TIMEOUT = 10.0
|
|
|
|
|
|
class ControlError(Exception):
|
|
"""Raised when a control-socket operation fails."""
|
|
|
|
|
|
class ControlClient:
|
|
"""A thin client over a FIPS daemon's Unix control socket."""
|
|
|
|
def __init__(self, socket_path: str = DEFAULT_SOCKET, timeout: float = DEFAULT_TIMEOUT):
|
|
self.socket_path = socket_path
|
|
self.timeout = timeout
|
|
|
|
# -- low-level ---------------------------------------------------------
|
|
|
|
def _send(self, payload: dict[str, Any]) -> dict[str, Any]:
|
|
"""Send one JSON request, read one JSON response line, return it."""
|
|
data = (json.dumps(payload) + "\n").encode()
|
|
try:
|
|
with socket.socket(socket.AF_UNIX, socket.SOCK_STREAM) as s:
|
|
s.settimeout(self.timeout)
|
|
s.connect(self.socket_path)
|
|
s.sendall(data)
|
|
s.shutdown(socket.SHUT_WR)
|
|
chunks: list[bytes] = []
|
|
while True:
|
|
chunk = s.recv(65536)
|
|
if not chunk:
|
|
break
|
|
chunks.append(chunk)
|
|
raw = b"".join(chunks).decode().strip()
|
|
except (OSError, socket.timeout) as e:
|
|
raise ControlError(f"control socket {self.socket_path}: {e}") from e
|
|
|
|
if not raw:
|
|
raise ControlError(f"empty response from {self.socket_path}")
|
|
|
|
try:
|
|
resp = json.loads(raw)
|
|
except json.JSONDecodeError as e:
|
|
raise ControlError(f"invalid JSON from {self.socket_path}: {e}") from e
|
|
|
|
return resp
|
|
|
|
# -- query / command ---------------------------------------------------
|
|
|
|
def query(self, command: str) -> dict[str, Any]:
|
|
"""Send a read-only query (e.g. ``show_status``). Returns the ``data`` dict."""
|
|
resp = self._send({"command": command})
|
|
if resp.get("status") != "ok":
|
|
raise ControlError(f"{command}: {resp.get('message', 'unknown error')}")
|
|
return resp.get("data", {}) or {}
|
|
|
|
def command(self, command: str, params: dict[str, Any]) -> dict[str, Any]:
|
|
"""Send a mutating command (e.g. ``connect``) with params. Returns ``data``."""
|
|
resp = self._send({"command": command, "params": params})
|
|
if resp.get("status") != "ok":
|
|
raise ControlError(f"{command}: {resp.get('message', 'unknown error')}")
|
|
return resp.get("data", {}) or {}
|
|
|
|
# -- convenience wrappers ---------------------------------------------
|
|
|
|
def show_status(self) -> dict[str, Any]:
|
|
return self.query("show_status")
|
|
|
|
def show_peers(self) -> dict[str, Any]:
|
|
return self.query("show_peers")
|
|
|
|
def show_tree(self) -> dict[str, Any]:
|
|
return self.query("show_tree")
|
|
|
|
def show_bloom(self) -> dict[str, Any]:
|
|
return self.query("show_bloom")
|
|
|
|
def show_routing(self) -> dict[str, Any]:
|
|
return self.query("show_routing")
|
|
|
|
def show_identity_cache(self) -> dict[str, Any]:
|
|
return self.query("show_identity_cache")
|
|
|
|
def connect(self, npub: str, address: str, transport: str) -> dict[str, Any]:
|
|
return self.command("connect", {"npub": npub, "address": address, "transport": transport})
|
|
|
|
def disconnect(self, npub: str) -> dict[str, Any]:
|
|
return self.command("disconnect", {"npub": npub})
|
|
|
|
|
|
def try_query(socket_path: str, command: str, timeout: float = DEFAULT_TIMEOUT) -> Optional[dict[str, Any]]:
|
|
"""Like ``ControlClient.query`` but returns ``None`` on failure instead of raising."""
|
|
try:
|
|
return ControlClient(socket_path, timeout).query(command)
|
|
except ControlError as e:
|
|
log.debug("query %s on %s failed: %s", command, socket_path, e)
|
|
return None
|