From 4f0cf12713077bf483e07667e77af44c9b70c2ef Mon Sep 17 00:00:00 2001 From: Claude Date: Fri, 25 Sep 2026 13:24:24 +0000 Subject: [PATCH] feat(cluster): signed command channel to nodes (Phase 4, COMMANDS flow) CommandBus queues typed, panel-signed (`cmd`) commands per node in cluster_commands: FIFO by seq, dedupe_key for desired state, expiry, pruned by cron:cluster. New ops: `commands` (long-poll after the node's high-water, held up to 20 s; frozen while quarantined) and `ack` (own commands only; raises cluster_nodes.cmd_seq). For nodes with the COMMANDS flow on (toggled on Cluster Nodes), ClusterRoute sends NodeRpc::request() as a signed node.rpc and waits for its ack, and NodeRpc::broadcast() / SignalDispatcher::kill() queue commands; other nodes and every LB keep the legacy transport. On the node, `console.php cluster:exec` checks a command again against the panel key the agent pinned and runs node.rpc through the legacy /api handlers (InternalApiController::runCommand, NodeRpc::ACTIONS only) or conn.kill_worker. Root actions stay on the signals table for now. Co-Authored-By: Claude Opus 5.5 Claude-Session: https://claude.ai/code/session_01DcNdxHieZuUh3EnMUKt7zf --- docs/adr/0004-cluster-api.md | 30 ++- src/Cli/Commands/ClusterExecCommand.php | 105 +++++++++ src/Cli/CronJobs/ClusterCronJob.php | 2 + src/Core/Cluster/NodeRpc.php | 20 ++ src/Core/Cluster/SignalDispatcher.php | 9 + src/Core/Localization/lang/ar.ini | 4 + src/Core/Localization/lang/bg.ini | 4 + src/Core/Localization/lang/de.ini | 4 + src/Core/Localization/lang/en.ini | 4 + src/Core/Localization/lang/es.ini | 4 + src/Core/Localization/lang/fr.ini | 4 + src/Core/Localization/lang/pt.ini | 4 + src/Core/Localization/lang/ru.ini | 4 + src/Domain/Cluster/ClusterAdmin.php | 8 +- src/Domain/Cluster/ClusterApi.php | 38 +++- src/Domain/Cluster/ClusterRoute.php | 96 +++++++++ src/Domain/Cluster/CommandBus.php | 199 ++++++++++++++++++ .../Controllers/Api/InternalApiController.php | 13 ++ src/Public/Views/admin/cluster_nodes.php | 9 +- tests/Unit/ClusterApiTest.php | 71 ++++++- tests/Unit/ClusterExecCommandTest.php | 47 +++++ 21 files changed, 671 insertions(+), 8 deletions(-) create mode 100644 src/Cli/Commands/ClusterExecCommand.php create mode 100644 src/Domain/Cluster/ClusterRoute.php create mode 100644 src/Domain/Cluster/CommandBus.php create mode 100644 tests/Unit/ClusterExecCommandTest.php diff --git a/docs/adr/0004-cluster-api.md b/docs/adr/0004-cluster-api.md index 09f19029..98861d7e 100644 --- a/docs/adr/0004-cluster-api.md +++ b/docs/adr/0004-cluster-api.md @@ -1,6 +1,6 @@ # ADR 0004 — Cluster API between MAIN and load balancers: the panel's contract -- **Status:** Accepted. Phase 0 (seams), Phase 1 (crypto contract, schema, settings) and Phase 2's API, Go agent and SSH enrolment of new LBs (below) are implemented. Enrolling existing LBs over SSH (`server:enrol`), `token_rekey` and enrolment by code are too. The admin page *Servers → Cluster Nodes* and `cron:cluster` are too. Phase 3 (authoritative telemetry, the 1 s liveness loop, MAIN endpoint changes) is too. Phases 4–11 are not. +- **Status:** Accepted. Phase 0 (seams), Phase 1 (crypto contract, schema, settings) and Phase 2's API, Go agent and SSH enrolment of new LBs (below) are implemented. Enrolling existing LBs over SSH (`server:enrol`), `token_rekey` and enrolment by code are too. The admin page *Servers → Cluster Nodes* and `cron:cluster` are too. Phase 3 (authoritative telemetry, the 1 s liveness loop, MAIN endpoint changes) is too. Phase 4 has its command channel (RPCs and viewer kills); root commands are not in yet. Phases 5–11 are not. - **Date:** 2026-09-25 - **Plan:** `docs/superpowers/specs/2026-09-21-main-lb-api-communication-design.md` (MAIN ↔ LB API communication, revision 3 plus corrections). - **Extension side:** `xcvm_core` ADR-002, "Cluster API: the extension's half of MAIN ↔ LB communication", cluster API version 1. @@ -88,6 +88,8 @@ The plan numbers its Phase 1 migrations 026–032. 026 and 027 were already take | `token_rekey` | POST | node signature (epoch 0, no MAC), body SEALed to the panel box key, a challenge | `active`, once a minute | panel-signed (`pre`): a new epoch's sealed token | | `hello` | POST | session | `active`, `quarantined` | BOX: state, mode, flows, proto, policy | | `heartbeat` | POST | session | `active`, `quarantined` | BOX: state, mode, flows, `pending` | +| `commands` | POST | session | `active` | BOX: signed commands after `after_seq`, held up to `wait_ms` (≤ 20 s) | +| `ack` | POST | session | `active`, `quarantined` | BOX: `ok` (own commands only) | A session request is checked in this order. Nothing is written, not even the nonce, before the MAC and, for token operations, the node signature have verified: @@ -269,6 +271,32 @@ This was checked with nginx 1.24: - `nginx -t` passes with and without the file. - On the old port, only `/cluster/v1/` reaches PHP. +### Commands (Phase 4, first increment) + +`CommandBus` queues MAIN → node commands in `cluster_commands`, FIFO per node by `seq`. Each command is a typed JSON record signed by the panel with tag `cmd`: + +```text +{"v":1, "type", "exp", "iat", "cmd_id", "seq", "node_uuid", "gen", "dedupe_key", "args"} +``` + +The extension derives the class from `type`. Kills and stops are restrictive and sign without a licence; the rest need it. A `dedupe_key` replaces a not-yet-acked command for the same desired state. Commands expire (`conn.*` 5 min, default 10 min), and `cron:cluster` prunes them. + +The flow, for a node whose COMMANDS flow is on (toggled per node on the Cluster Nodes page): + +1. **Poll.** The agent holds a `commands` long-poll. Each poll holds a PHP worker on MAIN; the plan's bus replaces that later. +2. **Checks.** The agent checks each command: the panel signature under its pinned key, its own uuid and generation, `seq` above its persisted high-water `cmd_seq`, and `exp` on MAIN's clock. +3. **Run.** It runs the command through `console.php cluster:exec`. `cluster:exec` checks the signature again and runs `node.rpc{action}` with the legacy `/api` handlers (`InternalApiController::runCommand`, actions limited to `NodeRpc::ACTIONS`) and `conn.kill_worker`. +4. **Ack.** The agent `ack`s the command with the result (≤ 64 KB) and raises its high-water. MAIN raises `cluster_nodes.cmd_seq`. +5. **Refusals.** A refused command is acked as refused, with the reason. + +On MAIN, `Domain\Cluster\ClusterRoute` sits behind the Phase 0 seams: + +- `NodeRpc::request()` becomes a signed `node.rpc`, and MAIN waits for its ack up to the caller's timeout. +- `NodeRpc::broadcast()` and `SignalDispatcher::kill()` queue without waiting. +- Nodes without the flow, and every LB, keep the legacy transport. + +Root actions (`NodeActions`) stay on the signals table until `cluster:root` and the root-owned panel-key pin exist. + ### Extension updates `console.php xcvm_core` rolls back an update whose cluster API falls outside the panel's range when the installed one was inside it. `console.php xcvm_core status` reports what is loaded. Installing an exact pinned version needs versioned paths in the binaries repo; that prerequisite is still open. diff --git a/src/Cli/Commands/ClusterExecCommand.php b/src/Cli/Commands/ClusterExecCommand.php new file mode 100644 index 00000000..bd27ae16 --- /dev/null +++ b/src/Cli/Commands/ClusterExecCommand.php @@ -0,0 +1,105 @@ +", "sig": ""}`, on + * stdin; this checks it again against the panel key the agent pinned + * (config/cluster/agent.json), runs it and writes its result to stdout for + * the agent's ack. Exit 0 means done. + * + * - `node.rpc {action, …}` — one of NodeRpc::ACTIONS, through the handlers of + * the legacy /api (InternalApiController::runCommand); + * - `conn.kill_worker {pid, rtmp}` — a viewer's PHP worker (this user's + * processes only) or an RTMP client. + * + * Runs as xc_vm; root actions are not taken here. + * + * Usage: `console.php cluster:exec < command.json` + * + * @package XC_VM_CLI_Commands + */ +class ClusterExecCommand implements CommandInterface { + /** Commands older than their exp by more than this (LB clock) are refused. */ + public const SKEW = 300; + + public function getName(): string { + return 'cluster:exec'; + } + + public function getDescription(): string { + return 'Run one signed MAIN command (from xc_agent) on this node'; + } + + public function execute(array $rArgs): int { + $rIn = json_decode((string) stream_get_contents(STDIN), true); + $rState = json_decode((string) @file_get_contents(CONFIG_PATH . 'cluster/agent.json'), true); + $rCmd = self::verify(is_array($rIn) ? $rIn : [], is_array($rState) ? $rState : [], time()); + if (is_string($rCmd)) { + fwrite(STDERR, 'cluster:exec: ' . $rCmd . "\n"); + return 2; + } + return self::run($rCmd); + } + + /** + * Check a command against the agent's pinned panel key and identity. + * + * @param array $rIn {doc, sig} + * @param array $rState the agent's state (base64 fields) + * @return array|string the command, or why it is refused + */ + public static function verify(array $rIn, array $rState, int $rNow): array|string { + $rPub = base64_decode((string) ($rState['panel_sign_pub'] ?? ''), true); + $rDoc = is_string($rIn['doc'] ?? null) ? $rIn['doc'] : ''; + $rSig = Enc::b64urlDecode((string) ($rIn['sig'] ?? '')); + if ($rPub === false || !PanelSig::verify($rPub, 'cmd', $rDoc, (string) $rSig)) { + return 'bad signature'; + } + $rCmd = json_decode($rDoc, true); + if (!is_array($rCmd) || ($rCmd['node_uuid'] ?? null) !== ($rState['node_uuid'] ?? '') || (int) ($rCmd['exp'] ?? 0) + self::SKEW < $rNow) { + return 'not for this node, or expired'; + } + return $rCmd; + } + + /** @param array $rCmd */ + public static function run(array $rCmd): int { + $rArgs = is_array($rCmd['args'] ?? null) ? $rCmd['args'] : []; + switch ($rCmd['type'] ?? '') { + case 'node.rpc': + if (!in_array($rArgs['action'] ?? null, NodeRpc::ACTIONS, true)) { + fwrite(STDERR, "cluster:exec: unknown action\n"); + return 2; + } + (new InternalApiController())->runCommand($rArgs); + return 0; + + case 'conn.kill_worker': + $rPID = (int) ($rArgs['pid'] ?? 0); + if ($rPID <= 0) { + return 2; + } + if (!empty($rArgs['rtmp'])) { + $rUrl = (string) (ServerRepository::getAll()[SERVER_ID]['rtmp_mport_url'] ?? ''); + if ($rUrl !== '') { + @file_get_contents($rUrl . 'control/drop/client?clientid=' . $rPID, false, stream_context_create(['http' => ['timeout' => 2]])); + } + } elseif (file_exists('/proc/' . $rPID)) { + posix_kill($rPID, 9); + } + echo json_encode(['result' => true]); + return 0; + } + fwrite(STDERR, "cluster:exec: unknown command type\n"); + return 2; + } +} diff --git a/src/Cli/CronJobs/ClusterCronJob.php b/src/Cli/CronJobs/ClusterCronJob.php index 05b8ce9e..74e51381 100644 --- a/src/Cli/CronJobs/ClusterCronJob.php +++ b/src/Cli/CronJobs/ClusterCronJob.php @@ -9,6 +9,7 @@ use XcVm\Core\Cluster\NodeRole; use XcVm\Core\Config\SettingsManager; use XcVm\Domain\Cluster\ClusterAudit; use XcVm\Domain\Cluster\ClusterEndpoint; +use XcVm\Domain\Cluster\CommandBus; use XcVm\Domain\Cluster\EnrolCodeService; use XcVm\Domain\Cluster\LivenessService; use XcVm\Domain\Cluster\NonceStore; @@ -60,6 +61,7 @@ class ClusterCronJob implements CommandInterface { 'epochs' => static fn() => TokenService::prune(), 'nonces' => static fn() => NonceStore::purge(), 'enrol_codes' => static fn() => EnrolCodeService::prune(), + 'commands' => static fn() => CommandBus::prune(), // MAIN's old HTTP ports past their 7 days: release them in nginx. 'endpoint' => static function () { if (ClusterEndpoint::prune(SettingsManager::getAll()) && defined('SERVER_ID')) { diff --git a/src/Core/Cluster/NodeRpc.php b/src/Core/Cluster/NodeRpc.php index 328742ae..fccd4373 100644 --- a/src/Core/Cluster/NodeRpc.php +++ b/src/Core/Cluster/NodeRpc.php @@ -3,6 +3,7 @@ namespace XcVm\Core\Cluster; use XcVm\Core\Http\ApiClient; +use XcVm\Domain\Cluster\ClusterRoute; /** * Node RPC @@ -42,6 +43,13 @@ final class NodeRpc { if (self::$rTransport !== null) { return (self::$rTransport)('request', [$rServerID], $rData, $rTimeout); } + // A node with the COMMANDS flow gets a signed node.rpc command (MAIN only). + if (class_exists(ClusterRoute::class)) { + [$rRouted, $rResult] = ClusterRoute::rpc($rServerID, $rData, $rTimeout); + if ($rRouted) { + return $rResult; + } + } return ApiClient::systemRequest($rServerID, $rData, $rTimeout); } @@ -58,6 +66,18 @@ final class NodeRpc { if (self::$rTransport !== null) { return (self::$rTransport)('broadcast', array_map('intval', $rServerIDs), $rData, 0); } + if (class_exists(ClusterRoute::class)) { + $rLegacy = []; + foreach ($rServerIDs as $rServerID) { + if (!ClusterRoute::send((int) $rServerID, $rData)[0]) { + $rLegacy[] = $rServerID; + } + } + if ($rLegacy === []) { + return ['result' => true]; + } + $rServerIDs = $rLegacy; + } return ApiClient::asyncRequest($rServerIDs, $rData); } diff --git a/src/Core/Cluster/SignalDispatcher.php b/src/Core/Cluster/SignalDispatcher.php index e97ed900..89580f46 100644 --- a/src/Core/Cluster/SignalDispatcher.php +++ b/src/Core/Cluster/SignalDispatcher.php @@ -2,6 +2,8 @@ namespace XcVm\Core\Cluster; +use XcVm\Domain\Cluster\ClusterRoute; + /** * Signal Dispatcher * @@ -36,6 +38,13 @@ final class SignalDispatcher { if ($rPID <= 0) { return false; } + // A node with the COMMANDS flow gets a signed conn.kill_worker (MAIN only). + if (class_exists(ClusterRoute::class)) { + [$rRouted, $rQueued] = ClusterRoute::kill($rServerID, $rPID, $rRTMP); + if ($rRouted) { + return $rQueued; + } + } $rRow = ['pid' => $rPID, 'server_id' => $rServerID]; if ($rRTMP) { $rRow['rtmp'] = 1; diff --git a/src/Core/Localization/lang/ar.ini b/src/Core/Localization/lang/ar.ini index 4791028a..0757a51a 100644 --- a/src/Core/Localization/lang/ar.ini +++ b/src/Core/Localization/lang/ar.ini @@ -1882,6 +1882,10 @@ cluster_telemetry_help = "تدفق القياس عن بُعد: عند تفعيل cluster_telemetry_on_done = "تم تفعيل تدفق القياس: وكيل العقدة يرسل الإحصاءات الآن." cluster_telemetry_off_done = "تم تعطيل تدفق القياس: يعود watchdog العقدة إلى الإرسال." cluster_fleet_silence = "صمتت معظم العقد في وقت واحد: يشك MAIN في شبكته أو في مجمع الواجهة، ويحتفظ بحالة كل عقدة دون وضعها غير متصلة حتى ينتهي الصمت." +cluster_commands_flow = "الأوامر" +cluster_commands_help = "تدفق الأوامر: عند تفعيله يرسل MAIN إلى هذه العقدة أوامر موقّعة (استدعاءات مثل تشغيل/إيقاف البث والفحص والمسح وقطع المشاهدين) عبر واجهة العنقود بدلًا من /api وجدول الإشارات." +cluster_commands_on_done = "تم تفعيل تدفق الأوامر: يرسل MAIN الآن أوامر موقّعة إلى هذه العقدة." +cluster_commands_off_done = "تم تعطيل تدفق الأوامر: يعود MAIN إلى استخدام /api وإشارات العقدة." restreamer_bypass_proxy = "وكيل تجاوز Restreamer" channel_sorting_type = "نوع فرز القناة" sort_vod_by_date = "فرز VOD حسب التاريخ" diff --git a/src/Core/Localization/lang/bg.ini b/src/Core/Localization/lang/bg.ini index 6dac7b52..aea8867a 100644 --- a/src/Core/Localization/lang/bg.ini +++ b/src/Core/Localization/lang/bg.ini @@ -1604,6 +1604,10 @@ cluster_telemetry_help = "Поток на телеметрията: при вк cluster_telemetry_on_done = "Потокът на телеметрията е включен: агентът на възела вече отчита статистиките." cluster_telemetry_off_done = "Потокът на телеметрията е изключен: watchdog на възела отново отчита." cluster_fleet_silence = "Повечето възли замлъкнаха едновременно: MAIN подозира собствената си мрежа или API пула и запазва състоянието на всеки възел, без да го маркира офлайн, докато тишината отмине." +cluster_commands_flow = "Команди" +cluster_commands_help = "Поток на командите: при включване MAIN изпраща на възела подписани команди (RPC като старт/стоп на потоци, проби и сканиране, и прекъсване на зрители) през API на клъстера вместо през неговия /api и таблицата със сигнали." +cluster_commands_on_done = "Потокът на командите е включен: MAIN вече изпраща подписани команди до възела." +cluster_commands_off_done = "Потокът на командите е изключен: MAIN отново използва /api и сигналите на възела." fanout_debug = "Дебъг дневник Fanout" fanout_debug_off = "Изкл." fanout_debug_all = "Всички категории" diff --git a/src/Core/Localization/lang/de.ini b/src/Core/Localization/lang/de.ini index daf6a633..499dd090 100644 --- a/src/Core/Localization/lang/de.ini +++ b/src/Core/Localization/lang/de.ini @@ -1603,6 +1603,10 @@ cluster_telemetry_help = "Telemetrie-Flow: an, MAIN bezieht die Statistiken dies cluster_telemetry_on_done = "Telemetrie-Flow an: Der Agent des Knotens meldet jetzt seine Statistiken." cluster_telemetry_off_done = "Telemetrie-Flow aus: Der Watchdog des Knotens meldet wieder." cluster_fleet_silence = "Die meisten Knoten sind gleichzeitig verstummt: MAIN vermutet sein eigenes Netz oder den API-Pool und hält den Zustand jedes Knotens, statt ihn offline zu setzen, bis die Stille endet." +cluster_commands_flow = "Befehle" +cluster_commands_help = "Befehls-Flow: an, sendet MAIN diesem Knoten signierte Befehle (RPCs wie Stream-Start/Stopp, Probes und Scans sowie Zuschauer-Kills) über die Cluster-API statt über seine /api und die Signaltabelle." +cluster_commands_on_done = "Befehls-Flow an: MAIN sendet diesem Knoten jetzt signierte Befehle." +cluster_commands_off_done = "Befehls-Flow aus: MAIN nutzt wieder die /api und die Signale des Knotens." fanout_debug = "Fanout-Debug-Protokoll" fanout_debug_off = "Aus" fanout_debug_all = "Alle Kategorien" diff --git a/src/Core/Localization/lang/en.ini b/src/Core/Localization/lang/en.ini index 5a8e5e4a..09221c67 100644 --- a/src/Core/Localization/lang/en.ini +++ b/src/Core/Localization/lang/en.ini @@ -1893,6 +1893,10 @@ cluster_telemetry_help = "Telemetry flow: when on, MAIN takes this node's stats cluster_telemetry_on_done = "Telemetry flow on: the node's agent now reports its stats." cluster_telemetry_off_done = "Telemetry flow off: the node's watchdog reports again." cluster_fleet_silence = "Most nodes went silent at once: MAIN suspects its own network or API pool, and holds every node's state instead of marking it offline until the silence clears." +cluster_commands_flow = "Commands" +cluster_commands_help = "Commands flow: when on, MAIN sends this node signed commands (RPCs such as stream start/stop, probes and scans, and viewer kills) over the cluster API instead of its /api and the signals table." +cluster_commands_on_done = "Commands flow on: MAIN now sends this node signed commands." +cluster_commands_off_done = "Commands flow off: MAIN uses the node's legacy /api and signals again." fanout_debug = "Fanout Debug Log" fanout_debug_off = "Off" fanout_debug_all = "All categories" diff --git a/src/Core/Localization/lang/es.ini b/src/Core/Localization/lang/es.ini index ca24591f..f770f6c5 100644 --- a/src/Core/Localization/lang/es.ini +++ b/src/Core/Localization/lang/es.ini @@ -1610,6 +1610,10 @@ cluster_telemetry_help = "Flujo de telemetría: activado, MAIN toma las estadís cluster_telemetry_on_done = "Flujo de telemetría activado: el agente del nodo informa ahora sus estadísticas." cluster_telemetry_off_done = "Flujo de telemetría desactivado: el watchdog del nodo vuelve a informar." cluster_fleet_silence = "La mayoría de los nodos quedaron en silencio a la vez: MAIN sospecha de su propia red o del pool de la API y mantiene el estado de cada nodo, sin marcarlos offline, hasta que pase el silencio." +cluster_commands_flow = "Comandos" +cluster_commands_help = "Flujo de comandos: activado, MAIN envía a este nodo comandos firmados (RPC como iniciar/detener streams, sondeos y escaneos, y expulsión de espectadores) por la API del clúster, en lugar de su /api y la tabla de señales." +cluster_commands_on_done = "Flujo de comandos activado: MAIN envía ahora comandos firmados a este nodo." +cluster_commands_off_done = "Flujo de comandos desactivado: MAIN vuelve a usar la /api y las señales del nodo." fanout_debug = "Registro de depuración Fanout" fanout_debug_off = "Desactivado" fanout_debug_all = "Todas las categorías" diff --git a/src/Core/Localization/lang/fr.ini b/src/Core/Localization/lang/fr.ini index 7899f197..116ba2a9 100644 --- a/src/Core/Localization/lang/fr.ini +++ b/src/Core/Localization/lang/fr.ini @@ -1603,6 +1603,10 @@ cluster_telemetry_help = "Flux de télémétrie : activé, MAIN reçoit les stat cluster_telemetry_on_done = "Flux de télémétrie activé : l'agent du nœud rapporte désormais ses statistiques." cluster_telemetry_off_done = "Flux de télémétrie désactivé : le watchdog du nœud rapporte de nouveau." cluster_fleet_silence = "La plupart des nœuds se sont tus en même temps : MAIN soupçonne son propre réseau ou le pool de l'API et conserve l'état de chaque nœud, sans les marquer hors ligne, jusqu'à la fin du silence." +cluster_commands_flow = "Commandes" +cluster_commands_help = "Flux de commandes : activé, MAIN envoie à ce nœud des commandes signées (RPC comme démarrer/arrêter un flux, sondes et scans, et déconnexion de spectateurs) par l'API du cluster, au lieu de sa /api et de la table des signaux." +cluster_commands_on_done = "Flux de commandes activé : MAIN envoie désormais des commandes signées à ce nœud." +cluster_commands_off_done = "Flux de commandes désactivé : MAIN utilise de nouveau la /api et les signaux du nœud." fanout_debug = "Journal de débogage Fanout" fanout_debug_off = "Désactivé" fanout_debug_all = "Toutes les catégories" diff --git a/src/Core/Localization/lang/pt.ini b/src/Core/Localization/lang/pt.ini index bb7a3034..6936bf14 100644 --- a/src/Core/Localization/lang/pt.ini +++ b/src/Core/Localization/lang/pt.ini @@ -1603,6 +1603,10 @@ cluster_telemetry_help = "Fluxo de telemetria: ligado, o MAIN recebe as estatís cluster_telemetry_on_done = "Fluxo de telemetria ligado: o agente do nó passa a reportar as estatísticas." cluster_telemetry_off_done = "Fluxo de telemetria desligado: o watchdog do nó volta a reportar." cluster_fleet_silence = "A maioria dos nós ficou em silêncio ao mesmo tempo: o MAIN suspeita da sua própria rede ou do pool da API e mantém o estado de cada nó, sem os marcar como offline, até o silêncio passar." +cluster_commands_flow = "Comandos" +cluster_commands_help = "Fluxo de comandos: ligado, o MAIN envia a este nó comandos assinados (RPCs como iniciar/parar streams, probes e scans, e kills de espetadores) pela API do cluster, em vez da /api e da tabela de sinais." +cluster_commands_on_done = "Fluxo de comandos ligado: o MAIN passa a enviar comandos assinados a este nó." +cluster_commands_off_done = "Fluxo de comandos desligado: o MAIN volta a usar a /api e os sinais do nó." fanout_debug = "Registo de depuração Fanout" fanout_debug_off = "Desligado" fanout_debug_all = "Todas as categorias" diff --git a/src/Core/Localization/lang/ru.ini b/src/Core/Localization/lang/ru.ini index a83649aa..09382af2 100644 --- a/src/Core/Localization/lang/ru.ini +++ b/src/Core/Localization/lang/ru.ini @@ -1603,6 +1603,10 @@ cluster_telemetry_help = "Поток телеметрии: при включен cluster_telemetry_on_done = "Поток телеметрии включён: статистику узла теперь передаёт агент." cluster_telemetry_off_done = "Поток телеметрии выключен: статистику снова передаёт watchdog узла." cluster_fleet_silence = "Большинство узлов замолчали одновременно: MAIN подозревает свою сеть или пул API и сохраняет состояние каждого узла, не помечая их offline, пока тишина не закончится." +cluster_commands_flow = "Команды" +cluster_commands_help = "Поток команд: при включении MAIN отправляет узлу подписанные команды (RPC: запуск/остановка потоков, пробы и сканирование, отключение зрителей) через API кластера вместо его /api и таблицы сигналов." +cluster_commands_on_done = "Поток команд включён: MAIN теперь отправляет узлу подписанные команды." +cluster_commands_off_done = "Поток команд выключен: MAIN снова использует /api и сигналы узла." fanout_debug = "Отладочный лог Fanout" fanout_debug_off = "Выкл." fanout_debug_all = "Все категории" diff --git a/src/Domain/Cluster/ClusterAdmin.php b/src/Domain/Cluster/ClusterAdmin.php index 88943e1f..3264e32d 100644 --- a/src/Domain/Cluster/ClusterAdmin.php +++ b/src/Domain/Cluster/ClusterAdmin.php @@ -104,14 +104,18 @@ final class ClusterAdmin { case 'telemetry_on': case 'telemetry_off': + case 'commands_on': + case 'commands_off': $rNode = NodeRegistry::byServer($rServerID); if ($rNode === null || !in_array($rNode['state'], ['active', 'quarantined'], true)) { return ['type' => 'info', 'message' => 'cluster_not_enrolled']; } - $rFlows = $rAction === 'telemetry_on' ? ((int) $rNode['flows'] | NodeRegistry::FLOW_TELEMETRY) : ((int) $rNode['flows'] & ~NodeRegistry::FLOW_TELEMETRY); + [$rName, $rSwitch] = explode('_', $rAction); + $rBit = $rName === 'telemetry' ? NodeRegistry::FLOW_TELEMETRY : NodeRegistry::FLOW_COMMANDS; + $rFlows = $rSwitch === 'on' ? ((int) $rNode['flows'] | $rBit) : ((int) $rNode['flows'] & ~$rBit); NodeRegistry::update($rServerID, ['flows' => $rFlows]); ClusterAudit::log('node.flows', $rServerID, ['flows' => $rFlows, 'was' => (int) $rNode['flows']], $rUserID === null ? 'admin' : 'admin:' . $rUserID); - return ['type' => 'success', 'message' => $rAction === 'telemetry_on' ? 'cluster_telemetry_on_done' : 'cluster_telemetry_off_done']; + return ['type' => 'success', 'message' => 'cluster_' . $rName . '_' . $rSwitch . '_done']; case 'revoke': return NodeRegistry::revoke($rServerID, $rCrypto, $rUserID === null ? 'admin' : 'admin:' . $rUserID) diff --git a/src/Domain/Cluster/ClusterApi.php b/src/Domain/Cluster/ClusterApi.php index 55126e26..5d37fa28 100644 --- a/src/Domain/Cluster/ClusterApi.php +++ b/src/Domain/Cluster/ClusterApi.php @@ -11,7 +11,8 @@ use XcVm\Core\Cluster\Crypto\SessionKeys; /** * MAIN's `/cluster/v1/` API (Phase 2: health, challenge, enrol_complete, - * enrol_code, enrol_code_status, token_refresh, token_rekey, hello, heartbeat). Transport-free: handle() takes the request + * enrol_code, enrol_code_status, token_refresh, token_rekey, hello, heartbeat; + * Phase 4: commands, ack). Transport-free: handle() takes the request * as an array and returns status, headers and body, so it is tested without * a web server; Public/cluster/index.php is the HTTP shell around it. * @@ -40,6 +41,8 @@ final class ClusterApi { 'enrol_code' => ['POST', true, null], 'enrol_code_status' => ['POST', false, null], 'hello' => ['POST', false, ['active', 'quarantined']], + 'commands' => ['POST', false, ['active']], + 'ack' => ['POST', false, ['active', 'quarantined']], 'heartbeat' => ['POST', false, ['active', 'quarantined']], ]; @@ -139,6 +142,8 @@ final class ClusterApi { 'token_refresh' => self::tokenRefresh($rCrypto, $rNode, $rKeys, $rCtx, $rH, $rPayload), 'hello' => self::hello($rNode, $rKeys, $rCtx, $rH, $rPayload, $rSettings, $rMain), 'heartbeat' => self::heartbeat($rNode, $rKeys, $rCtx, $rH, $rPayload, $rSettings), + 'commands' => self::commands($rNode, $rKeys, $rCtx, $rPayload), + 'ack' => self::ack($rCrypto, $rNode, $rKeys, $rCtx, $rH, $rPayload), }; } @@ -439,6 +444,37 @@ final class ClusterApi { ]); } + /** Longest a `commands` long-poll is held (ms). */ + public const COMMANDS_WAIT_MAX_MS = 20000; + + /** + * `commands`: the node's queued commands after its high-water, panel-signed + * (`cmd`) each. Held up to `wait_ms` while there are none, so a command + * reaches the node within a poll step of being queued. + */ + private static function commands(array $rNode, SessionKeys $rKeys, string $rCtx, array $rP): array { + $rAfter = max(0, (int) ($rP['after_seq'] ?? 0)); + $rWait = max(0, min(self::COMMANDS_WAIT_MAX_MS, (int) ($rP['wait_ms'] ?? 0))); + $rDeadline = microtime(true) + $rWait / 1000; + while (true) { + $rCommands = CommandBus::pending((int) $rNode['server_id'], $rAfter); + if ($rCommands !== [] || microtime(true) >= $rDeadline) { + break; + } + usleep(250000); + } + return ClusterReply::boxed($rKeys, $rCtx, ['commands' => $rCommands, 'main_time_ms' => ClusterClock::nowMs()]); + } + + /** `ack`: a command's outcome, accepted only for this node's own commands. */ + private static function ack(ClusterCrypto $rCrypto, array $rNode, SessionKeys $rKeys, string $rCtx, array $rH, array $rP): array { + $rCmdID = is_string($rP['cmd_id'] ?? null) && preg_match('/^[0-9a-f]{32}$/', (string) $rP['cmd_id']) ? (string) $rP['cmd_id'] : null; + if ($rCmdID === null || !CommandBus::ack((int) $rNode['server_id'], $rCmdID, !empty($rP['ok']), is_string($rP['result'] ?? null) ? (string) $rP['result'] : '')) { + return DenialFactory::deny($rCrypto, 400, 'BAD_REQUEST', $rH['node'], $rH['nonce']); + } + return ClusterReply::boxed($rKeys, $rCtx, ['ok' => true, 'main_time_ms' => ClusterClock::nowMs()]); + } + /** Map an extension refusal to a signed denial. */ private static function refusal(ClusterCrypto $rCrypto, string $rReason, array $rNode, array $rH): array { return match (true) { diff --git a/src/Domain/Cluster/ClusterRoute.php b/src/Domain/Cluster/ClusterRoute.php new file mode 100644 index 00000000..b30c7a84 --- /dev/null +++ b/src/Domain/Cluster/ClusterRoute.php @@ -0,0 +1,96 @@ + $rData NodeRpc payload (action + args) + * @return array{0: bool, 1: mixed} the node's raw result, or null on failure/timeout + */ + public static function rpc(int $rServerID, array $rData, int $rTimeout): array { + $rCrypto = self::target($rServerID); + if ($rCrypto === null) { + return [false, null]; + } + try { + $rCmdID = CommandBus::enqueue($rCrypto, $rServerID, 'node.rpc', $rData); + } catch (\Throwable) { + return [true, null]; + } + $rOutcome = CommandBus::await($rCmdID, max(1, $rTimeout)); + return [true, $rOutcome !== null && $rOutcome[0] ? $rOutcome[1] : null]; + } + + /** + * An RPC sent without waiting (NodeRpc::broadcast). + * + * @param array $rData + * @return array{0: bool, 1: bool} queued? + */ + public static function send(int $rServerID, array $rData): array { + $rCrypto = self::target($rServerID); + if ($rCrypto === null) { + return [false, false]; + } + try { + CommandBus::enqueue($rCrypto, $rServerID, 'node.rpc', $rData); + return [true, true]; + } catch (\Throwable) { + return [true, false]; + } + } + + /** + * Kill a viewer's worker on its node (SignalDispatcher::kill): restrictive, + * so it is signed even without a licence. + * + * @return array{0: bool, 1: bool} + */ + public static function kill(int $rServerID, int $rPID, bool $rRTMP): array { + $rCrypto = self::target($rServerID); + if ($rCrypto === null) { + return [false, false]; + } + try { + CommandBus::enqueue($rCrypto, $rServerID, 'conn.kill_worker', ['pid' => $rPID, 'rtmp' => $rRTMP]); + return [true, true]; + } catch (\Throwable) { + return [true, false]; + } + } + + /** Tests: supply the extension handle. Null restores the factory. */ + public static function useCrypto(?callable $rFactory): void { + self::$rCrypto = $rFactory; + } + + /** The extension, when this node takes commands; null for the legacy path. */ + private static function target(int $rServerID): ?ClusterCrypto { + try { + if (empty(SettingsManager::get('cluster_api_enabled')) || !CommandBus::accepts(NodeRegistry::byServer($rServerID))) { + return null; + } + return self::$rCrypto !== null ? (self::$rCrypto)() : ClusterCryptoFactory::create(); + } catch (\Throwable) { + return null; // no cluster tables, no extension: the legacy path + } + } +} diff --git a/src/Domain/Cluster/CommandBus.php b/src/Domain/Cluster/CommandBus.php new file mode 100644 index 00000000..7ac0e49e --- /dev/null +++ b/src/Domain/Cluster/CommandBus.php @@ -0,0 +1,199 @@ + 300, 'node.root' => 86400, 'default' => 600]; + + /** Types this increment carries. */ + public const TYPES = ['node.rpc', 'conn.kill_worker']; + + /** Types that are restrictive (always signable); the extension decides, this is informational. */ + public const RESTRICTIVE = ['conn.drop', 'conn.drop_line', 'conn.kill_worker', 'stream.stop', 'vod.stop', 'token.rotate_now', 'node.quarantine', 'node.fence', 'resync', 'config.changed']; + + /** Largest result stored from an ack. */ + public const MAX_RESULT = 65536; + + /** + * Is this node taking commands over the cluster API? + * + * @param array|null $rNode cluster_nodes row + */ + public static function accepts(?array $rNode): bool { + return $rNode !== null && $rNode['state'] === 'active' && (int) $rNode['mode'] >= 1 && ((int) $rNode['flows'] & NodeRegistry::FLOW_COMMANDS) !== 0; + } + + /** + * Sign and queue a command. Returns its cmd_id. + * + * @param array $rArgs + */ + public static function enqueue(ClusterCrypto $rCrypto, int $rServerID, string $rType, array $rArgs, ?string $rDedupeKey = null, ?int $rTtl = null): string { + if (!in_array($rType, self::TYPES, true)) { + throw new \InvalidArgumentException('Unknown command type: ' . $rType); + } + $rNode = NodeRegistry::byServer($rServerID); + if ($rNode === null) { + throw new \InvalidArgumentException('Server ' . $rServerID . ' is not an enrolled node'); + } + $rNow = ClusterClock::now(); + $rTtl ??= self::ttl($rType); + $rCmdID = bin2hex(random_bytes(16)); + for ($rAttempt = 0; ; $rAttempt++) { + self::db()->query('SELECT MAX(`seq`) AS `seq` FROM `cluster_commands` WHERE `server_id` = ?;', $rServerID); + $rSeq = max((int) (self::db()->get_row()['seq'] ?? 0), (int) $rNode['cmd_seq']) + 1; + $rDoc = (string) json_encode([ + 'v' => 1, 'type' => $rType, 'exp' => $rNow + $rTtl, 'iat' => $rNow, 'cmd_id' => $rCmdID, 'seq' => $rSeq, + 'node_uuid' => (string) $rNode['node_uuid'], 'gen' => (int) $rNode['gen'], 'dedupe_key' => $rDedupeKey, 'args' => (object) $rArgs, + ], JSON_UNESCAPED_SLASHES); + $rSig = $rCrypto->sign('cmd', $rDoc); + if ($rDedupeKey !== null) { + // A newer desired state supersedes a command not yet acked. + self::db()->query("DELETE FROM `cluster_commands` WHERE `server_id` = ? AND `dedupe_key` = ? AND `state` <> 'acked';", $rServerID, $rDedupeKey); + } + try { + self::db()->query( + 'INSERT INTO `cluster_commands` (`server_id`, `seq`, `cmd_id`, `type`, `action`, `class`, `dedupe_key`, `payload`, `sig`, `state`, `created_at`, `exp`) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?);', + $rServerID, + $rSeq, + $rCmdID, + $rType, + isset($rArgs['action']) ? substr((string) $rArgs['action'], 0, 32) : null, + in_array($rType, self::RESTRICTIVE, true) ? 'R' : 'G', + $rDedupeKey, + $rDoc, + $rSig, + 'queued', + $rNow, + $rNow + $rTtl + ); + return $rCmdID; + } catch (\Throwable $rE) { + if ($rAttempt >= 3) { + throw $rE; // a concurrent enqueue took this seq three times + } + } + } + } + + /** + * Commands for a node after its high-water, oldest first; marks them delivered. + * + * @return list + */ + public static function pending(int $rServerID, int $rAfterSeq, int $rLimit = 50): array { + self::db()->query( + "SELECT `id`, `seq`, `payload`, `sig` FROM `cluster_commands` WHERE `server_id` = ? AND `seq` > ? AND `state` IN ('queued', 'delivered') AND `exp` > ? ORDER BY `seq` ASC LIMIT " . max(1, min(200, $rLimit)) . ';', + $rServerID, + $rAfterSeq, + ClusterClock::now() + ); + $rRows = self::db()->get_rows(); + $rOut = []; + $rIDs = []; + foreach ($rRows as $rRow) { + $rOut[] = ['doc' => (string) $rRow['payload'], 'sig' => Enc::b64url((string) $rRow['sig']), 'seq' => (int) $rRow['seq']]; + $rIDs[] = (int) $rRow['id']; + } + if ($rIDs !== []) { + self::db()->query("UPDATE `cluster_commands` SET `state` = 'delivered', `delivered_at` = ? WHERE `state` = 'queued' AND `id` IN (" . implode(',', $rIDs) . ');', ClusterClock::now()); + } + return $rOut; + } + + /** + * A node's acknowledgement: only for its own commands. Raises its + * high-water (cmd_seq) so a restored queue is not replayed. + */ + public static function ack(int $rServerID, string $rCmdID, bool $rOk, string $rResult): bool { + self::db()->query('SELECT `seq`, `state` FROM `cluster_commands` WHERE `server_id` = ? AND `cmd_id` = ?;', $rServerID, $rCmdID); + $rRow = self::db()->num_rows() > 0 ? self::db()->get_row() : null; + if ($rRow === null) { + return false; + } + if (in_array($rRow['state'], ['acked', 'failed'], true)) { + return true; // a repeated ack + } + self::db()->query( + 'UPDATE `cluster_commands` SET `state` = ?, `acked_at` = ?, `result` = ? WHERE `server_id` = ? AND `cmd_id` = ?;', + $rOk ? 'acked' : 'failed', + ClusterClock::now(), + substr($rResult, 0, self::MAX_RESULT), + $rServerID, + $rCmdID + ); + self::db()->query('UPDATE `cluster_nodes` SET `cmd_seq` = ? WHERE `server_id` = ? AND `cmd_seq` < ?;', (int) $rRow['seq'], $rServerID, (int) $rRow['seq']); + return true; + } + + /** + * A command's outcome once acked: [ok, result], or null while pending. + * + * @return array{0: bool, 1: string}|null + */ + public static function result(string $rCmdID): ?array { + self::db()->query('SELECT `state`, `result` FROM `cluster_commands` WHERE `cmd_id` = ?;', $rCmdID); + $rRow = self::db()->num_rows() > 0 ? self::db()->get_row() : null; + if ($rRow === null || !in_array($rRow['state'], ['acked', 'failed'], true)) { + return null; + } + return [$rRow['state'] === 'acked', (string) $rRow['result']]; + } + + /** + * Wait for a command's outcome (the admin side of an RPC). + * + * @return array{0: bool, 1: string}|null null on timeout + */ + public static function await(string $rCmdID, float $rTimeout, int $rPollMs = 100): ?array { + $rDeadline = microtime(true) + $rTimeout; + do { + $rResult = self::result($rCmdID); + if ($rResult !== null) { + return $rResult; + } + usleep($rPollMs * 1000); + } while (microtime(true) < $rDeadline); + return self::result($rCmdID); + } + + /** Drop expired commands and outcomes older than a day. */ + public static function prune(): void { + $rNow = ClusterClock::now(); + self::db()->query("DELETE FROM `cluster_commands` WHERE (`exp` <= ? AND `state` IN ('queued', 'delivered')) OR (`acked_at` IS NOT NULL AND `acked_at` < ?);", $rNow, $rNow - 86400); + } + + private static function ttl(string $rType): int { + foreach (self::TTL as $rPrefix => $rSeconds) { + if ($rPrefix !== 'default' && str_starts_with($rType, $rPrefix)) { + return $rSeconds; + } + } + return self::TTL['default']; + } +} diff --git a/src/Public/Controllers/Api/InternalApiController.php b/src/Public/Controllers/Api/InternalApiController.php index 9914978c..9d25a673 100644 --- a/src/Public/Controllers/Api/InternalApiController.php +++ b/src/Public/Controllers/Api/InternalApiController.php @@ -62,6 +62,19 @@ class InternalApiController { $this->dispatch($rAction, $rRequest, $rSettings); } + /** + * Run one system action for a signed `node.rpc` command (cluster:exec on + * the node, Phase 4): the same handlers as the legacy /api, without its + * password and source-IP checks, which the command's panel signature + * replaces. Output is the action's response, as /api would send it. + * + * @param array $rRequest action + arguments + */ + public function runCommand(array $rRequest): void { + $this->deny = false; + $this->dispatch((string) ($rRequest['action'] ?? ''), $rRequest, SettingsManager::getAll()); + } + private function dispatch($rAction, $rRequest, $rSettings) { switch ($rAction) { case 'view_log': diff --git a/src/Public/Views/admin/cluster_nodes.php b/src/Public/Views/admin/cluster_nodes.php index e7b564b3..590c404c 100644 --- a/src/Public/Views/admin/cluster_nodes.php +++ b/src/Public/Views/admin/cluster_nodes.php @@ -83,6 +83,7 @@ $rWhen = static fn(?int $rTs): string => $rTs ? gmdate('Y-m-d H:i:s', $rTs) . ' + @@ -92,7 +93,7 @@ $rWhen = static fn(?int $rTs): string => $rTs ? gmdate('Y-m-d H:i:s', $rTs) . ' - + @@ -107,17 +108,19 @@ $rWhen = static fn(?int $rTs): string => $rTs ? gmdate('Y-m-d H:i:s', $rTs) . ' + 1, 'commands' => 2] as $rFlowName => $rFlowBit): ?> - +
- +
— + (gen ) diff --git a/tests/Unit/ClusterApiTest.php b/tests/Unit/ClusterApiTest.php index 2bc09af7..f7236503 100644 --- a/tests/Unit/ClusterApiTest.php +++ b/tests/Unit/ClusterApiTest.php @@ -53,7 +53,7 @@ final class ClusterApiTest extends TestCase { protected function setUp(): void { $this->rDb = new TestDb(); - foreach (['029_create_cluster_nodes', '032_create_cluster_audit'] as $rName) { + foreach (['029_create_cluster_nodes', '030_create_cluster_commands', '032_create_cluster_audit'] as $rName) { $this->rDb->exec($this->ddl((string) file_get_contents(dirname(__DIR__, 2) . '/src/migrations/database/up/' . $rName . '.sql'))); } $this->rDb->exec('ALTER TABLE `cluster_node_epochs` ADD COLUMN `agent_eph_pub` binary(32) DEFAULT NULL'); @@ -675,4 +675,73 @@ final class ClusterApiTest extends TestCase { [$rRes, $rReq] = $this->rekey($this->challenge(), random_bytes(32)); $this->denial($rRes, 403, 'NODE_REVOKED', $rReq); } + + public function testCommandsAreDeliveredSignedAndAcked(): void { + $rKeys = $this->active(); + $rCmd = \XcVm\Domain\Cluster\CommandBus::enqueue($this->rCrypto, self::SID, 'node.rpc', ['action' => 'get_pids']); + [$rRes, $rCtx] = $this->call('commands', ['after_seq' => 0, 'wait_ms' => 0], 1, $rKeys); + $rOut = $this->reply($rRes, $rCtx, $rKeys); + $this->assertCount(1, $rOut['commands']); + $rOne = $rOut['commands'][0]; + $this->assertTrue(PanelSig::verify($this->rCrypto->info()['panel_sign_pub'], 'cmd', $rOne['doc'], (string) Enc::b64urlDecode($rOne['sig'])), 'panel-signed, tag cmd'); + $rDoc = json_decode($rOne['doc'], true); + $this->assertSame(['node.rpc', 1, $this->rUuid, 1, $rCmd, ['action' => 'get_pids']], [$rDoc['type'], $rDoc['seq'], $rDoc['node_uuid'], $rDoc['gen'], $rDoc['cmd_id'], $rDoc['args']]); + + // Nothing after the high-water. + [$rRes, $rCtx] = $this->call('commands', ['after_seq' => 1, 'wait_ms' => 0], 1, $rKeys); + $this->assertSame([], $this->reply($rRes, $rCtx, $rKeys)['commands']); + + [$rRes, $rCtx] = $this->call('ack', ['cmd_id' => $rCmd, 'ok' => true, 'result' => '[1,2]'], 1, $rKeys); + $this->assertTrue($this->reply($rRes, $rCtx, $rKeys)['ok']); + $this->assertSame([true, '[1,2]'], \XcVm\Domain\Cluster\CommandBus::result($rCmd)); + $this->assertSame(1, (int) NodeRegistry::byServer(self::SID)['cmd_seq'], 'the high-water rises with the ack'); + [$rRes, $rCtx] = $this->call('commands', ['after_seq' => 0, 'wait_ms' => 0], 1, $rKeys); + $this->assertSame([], $this->reply($rRes, $rCtx, $rKeys)['commands'], 'an acked command is not delivered again'); + + // An ack for a command that is not this node's is refused. + [$rRes, , $rReq] = $this->call('ack', ['cmd_id' => str_repeat('ab', 16), 'ok' => true], 1, $rKeys); + $this->denial($rRes, 400, 'BAD_REQUEST', $rReq); + } + + public function testQuarantineFreezesCommands(): void { + $rKeys = $this->active(); + NodeRegistry::update(self::SID, ['state' => 'quarantined']); + [$rRes, , $rReq] = $this->call('commands', ['after_seq' => 0], 1, $rKeys); + $this->denial($rRes, 409, 'NOT_ACTIVE', $rReq); + } + + public function testCommandsExpireAndDedupe(): void { + $this->active(); + $rBus = \XcVm\Domain\Cluster\CommandBus::class; + $rFirst = $rBus::enqueue($this->rCrypto, self::SID, 'node.rpc', ['action' => 'stream', 'function' => 'start', 'stream_ids' => [1]], 'stream:1'); + $rSecond = $rBus::enqueue($this->rCrypto, self::SID, 'node.rpc', ['action' => 'stream', 'function' => 'stop', 'stream_ids' => [1]], 'stream:1'); + $rPending = $rBus::pending(self::SID, 0); + $this->assertCount(1, $rPending, 'a newer desired state replaces the unacked one'); + $this->assertSame($rSecond, json_decode($rPending[0]['doc'], true)['cmd_id']); + $this->assertNotSame($rFirst, $rSecond); + ClusterClock::fix($this->rT0 + 601000); + $this->assertSame([], $rBus::pending(self::SID, 0), 'expired'); + $rBus::prune(); + $this->rDb->query('SELECT COUNT(*) AS `n` FROM `cluster_commands`'); + $this->assertSame(0, (int) $this->rDb->get_row()['n']); + } + + public function testRoutingFollowsTheCommandsFlow(): void { + $this->active(); + SettingsManager::set($this->rSettings); + \XcVm\Domain\Cluster\ClusterRoute::useCrypto(fn() => $this->rCrypto); + try { + $this->assertSame([false, false], \XcVm\Domain\Cluster\ClusterRoute::kill(self::SID, 123, false), 'flow off: legacy'); + NodeRegistry::update(self::SID, ['flows' => NodeRegistry::FLOW_COMMANDS]); + $this->assertSame([true, true], \XcVm\Domain\Cluster\ClusterRoute::kill(self::SID, 123, false)); + $this->assertSame([true, true], \XcVm\Domain\Cluster\ClusterRoute::send(self::SID, ['action' => 'free_temp'])); + $rStart = microtime(true); + $this->assertSame([true, null], \XcVm\Domain\Cluster\ClusterRoute::rpc(self::SID, ['action' => 'get_pids'], 1), 'no ack: a timeout, as an offline node'); + $this->assertLessThan(3, microtime(true) - $rStart); + $rTypes = array_map(static fn($rC) => json_decode($rC['doc'], true)['type'], \XcVm\Domain\Cluster\CommandBus::pending(self::SID, 0)); + $this->assertSame(['conn.kill_worker', 'node.rpc', 'node.rpc'], $rTypes); + } finally { + \XcVm\Domain\Cluster\ClusterRoute::useCrypto(null); + } + } } diff --git a/tests/Unit/ClusterExecCommandTest.php b/tests/Unit/ClusterExecCommandTest.php new file mode 100644 index 00000000..acb3f40a --- /dev/null +++ b/tests/Unit/ClusterExecCommandTest.php @@ -0,0 +1,47 @@ +rCrypto = new FakeClusterCrypto(); + $this->rState = ['node_uuid' => '0f8fad5b-d9cb-469f-a165-70867728950e', 'panel_sign_pub' => base64_encode($this->rCrypto->info()['panel_sign_pub'])]; + } + + private function command(array $rOverride = [], string $rTag = 'cmd'): array { + $rDoc = (string) json_encode($rOverride + ['v' => 1, 'type' => 'node.rpc', 'exp' => 1800000600, 'iat' => 1800000000, 'cmd_id' => str_repeat('a', 32), 'seq' => 1, 'node_uuid' => $this->rState['node_uuid'], 'gen' => 1, 'dedupe_key' => null, 'args' => ['action' => 'get_pids']]); + return ['doc' => $rDoc, 'sig' => Enc::b64url($this->rCrypto->sign($rTag, $rDoc))]; + } + + public function testAGenuineCommandVerifies(): void { + $rCmd = ClusterExecCommand::verify($this->command(), $this->rState, 1800000000); + $this->assertIsArray($rCmd); + $this->assertSame('get_pids', $rCmd['args']['action']); + } + + public function testForgedOrMisaddressedCommandsAreRefused(): void { + $rGood = $this->command(); + $this->assertSame('bad signature', ClusterExecCommand::verify(['doc' => str_replace('get_pids', 'kill_pid', $rGood['doc'])] + $rGood, $this->rState, 1800000000)); + $this->assertSame('bad signature', ClusterExecCommand::verify($this->command([], 'den'), $this->rState, 1800000000), 'another tag'); + $this->assertSame('bad signature', ClusterExecCommand::verify($rGood, ['panel_sign_pub' => base64_encode(str_repeat("\1", 32))] + $this->rState, 1800000000)); + $this->assertIsString(ClusterExecCommand::verify($this->command(['node_uuid' => '11111111-1111-4111-a111-111111111111']), $this->rState, 1800000000)); + $this->assertIsString(ClusterExecCommand::verify($rGood, $this->rState, 1800000600 + ClusterExecCommand::SKEW + 1), 'stale'); + } + + public function testUnknownTypesAndActionsDoNothing(): void { + $this->assertSame(2, ClusterExecCommand::run(['type' => 'node.root', 'args' => ['action' => 'reboot']])); + $this->assertSame(2, ClusterExecCommand::run(['type' => 'node.rpc', 'args' => ['action' => 'view_log']]), 'not in NodeRpc::ACTIONS'); + $this->assertSame(2, ClusterExecCommand::run(['type' => 'conn.kill_worker', 'args' => ['pid' => 0]])); + } +}