diff --git a/apps/control-station/src/components/system/ComputeContourSettingsWindow.tsx b/apps/control-station/src/components/system/ComputeContourSettingsWindow.tsx index 63f2679..7e22e43 100644 --- a/apps/control-station/src/components/system/ComputeContourSettingsWindow.tsx +++ b/apps/control-station/src/components/system/ComputeContourSettingsWindow.tsx @@ -12,10 +12,13 @@ import { } from "@nodedc/ui-react"; import { + applyComputeContourNetwork, fetchComputeContourAgentInstall, + fetchComputeContourNetwork, type ComputeContour, type ComputeContourAgentInstall, type ComputeContourDraft, + type ComputeContourNetworkStatus, type ComputeContourPlatform, } from "../../core/system/computeContours"; import { @@ -56,6 +59,7 @@ function emptyDraft(): ComputeContourDraft { ssh_port: 22, mqtt_host: "127.0.0.1", mqtt_port: 1883, + mqtt_bind_address: "127.0.0.1", telemetry_poll_interval_seconds: DEFAULT_TELEMETRY_POLL_INTERVAL_SECONDS, mqtt_publish_interval_seconds: 2, }; @@ -72,6 +76,7 @@ function draftFromContour(contour: ComputeContour | null): ComputeContourDraft { ssh_port: contour.ssh_port, mqtt_host: contour.mqtt_host, mqtt_port: contour.mqtt_port, + mqtt_bind_address: contour.mqtt_bind_address, telemetry_poll_interval_seconds: contour.telemetry_poll_interval_seconds, mqtt_publish_interval_seconds: contour.mqtt_publish_interval_seconds, }; @@ -94,6 +99,10 @@ export function ComputeContourSettingsWindow({ ); const [activeSection, setActiveSection] = useState<"connection" | "agent">("connection"); const [install, setInstall] = useState(null); + const [network, setNetwork] = useState(null); + const [networkAction, setNetworkAction] = useState< + "refresh" | "broker" | "worker" | null + >(null); const [busy, setBusy] = useState(false); const [error, setError] = useState(null); @@ -105,9 +114,30 @@ export function ComputeContourSettingsWindow({ setMqttPublishIntervalDraft(String(nextDraft.mqtt_publish_interval_seconds)); setActiveSection("connection"); setInstall(null); + setNetwork(null); + setNetworkAction(null); setError(null); }, [contour, mode, open]); + useEffect(() => { + if (!open || mode !== "edit" || !contour) return; + const controller = new AbortController(); + setNetworkAction("refresh"); + void fetchComputeContourNetwork(contour.contour_id, controller.signal) + .then((document) => { + if (!controller.signal.aborted) setNetwork(document); + }) + .catch((reason: unknown) => { + if (!controller.signal.aborted) { + setError(reason instanceof Error ? reason.message : "Сетевая проверка недоступна."); + } + }) + .finally(() => { + if (!controller.signal.aborted) setNetworkAction(null); + }); + return () => controller.abort(); + }, [contour, mode, open]); + useEffect(() => { if (!open || mode !== "edit" || !contour) return; const controller = new AbortController(); @@ -135,6 +165,8 @@ export function ComputeContourSettingsWindow({ const valid = useMemo(() => ( Boolean(draft.display_name.trim()) && Boolean(draft.expected_node_id.trim()) + && Boolean(draft.mqtt_host.trim()) + && Boolean(draft.mqtt_bind_address.trim()) && Number.isInteger(draft.ssh_port) && draft.ssh_port > 0 && Number.isInteger(draft.mqtt_port) @@ -143,6 +175,16 @@ export function ComputeContourSettingsWindow({ && mqttPublishIntervalSeconds !== null ), [draft, mqttPublishIntervalSeconds, telemetryPollIntervalSeconds]); + const networkProfileDirty = Boolean( + contour + && ( + draft.mqtt_host !== contour.mqtt_host + || draft.mqtt_port !== contour.mqtt_port + || draft.mqtt_bind_address !== contour.mqtt_bind_address + || draft.mqtt_publish_interval_seconds !== contour.mqtt_publish_interval_seconds + ), + ); + const commitTelemetryPollInterval = () => { const resolution = resolveTelemetryPollIntervalDraft( telemetryPollIntervalDraft, @@ -205,6 +247,35 @@ export function ComputeContourSettingsWindow({ } }; + const refreshNetwork = async () => { + if (!contour || networkAction) return; + setNetworkAction("refresh"); + setError(null); + try { + setNetwork(await fetchComputeContourNetwork(contour.contour_id)); + } catch (reason) { + setError(reason instanceof Error ? reason.message : "Сетевая проверка недоступна."); + } finally { + setNetworkAction(null); + } + }; + + const applyNetwork = async (target: "broker" | "worker") => { + if (!contour || networkAction || networkProfileDirty) return; + setNetworkAction(target); + setError(null); + try { + await applyComputeContourNetwork(contour.contour_id, target); + setNetwork(await fetchComputeContourNetwork(contour.contour_id)); + } catch (reason) { + setError( + reason instanceof Error ? reason.message : "Сетевой профиль не применён.", + ); + } finally { + setNetworkAction(null); + } + }; + return (
{activeSection === "connection" ? ( + <> setDraft((current) => ({ ...current, @@ -306,6 +379,15 @@ export function ComputeContourSettingsWindow({ mqtt_port: Number(event.currentTarget.value), }))} /> + setDraft((current) => ({ + ...current, + mqtt_bind_address: event.currentTarget.value, + }))} + /> + + {network?.broker.ready ? "Broker доступен" : "Нужна проверка"} + + )} + > +
+
+
+
Endpoint Worker
+
{network + ? `${network.broker.configured_host}:${network.broker.configured_port}` + : "—"}
+
+
+
Текущий IP
+
{network?.broker.resolved_addresses.join(", ") || "—"}
+
+
+
Публикация Mac
+
{network?.broker.bind_address ?? "—"}
+
+
+
Worker 006
+
{network?.worker.reachable + ? `${network.worker.mqtt_host ?? "—"}:${network.worker.mqtt_port ?? "—"}` + : "Недоступен"}
+
+
+
+ + {network?.broker.bind_reachable + ? "Listener Mac доступен" + : "Listener Mac недоступен"} + + + {network?.worker.broker_reachable + ? "Worker видит broker" + : "Worker не видит broker"} + + + {network?.worker.matches_profile + ? "Профиль синхронизирован" + : "Профиль отличается"} + +
+
+ + + +
+ {networkProfileDirty ? ( +

Сначала сохраните изменённый сетевой профиль, затем примените его.

+ ) : null} + {error ? {error} : null} +
+ + ) : null} {activeSection === "agent" ? ( diff --git a/apps/control-station/src/core/system/computeContours.ts b/apps/control-station/src/core/system/computeContours.ts index 04b5cfe..a6af095 100644 --- a/apps/control-station/src/core/system/computeContours.ts +++ b/apps/control-station/src/core/system/computeContours.ts @@ -13,6 +13,7 @@ export interface ComputeContour { ssh_port: number; mqtt_host: string; mqtt_port: number; + mqtt_bind_address: string; telemetry_poll_interval_seconds: number; mqtt_publish_interval_seconds: number; revision: number; @@ -33,6 +34,7 @@ export interface ComputeContourDraft { ssh_port: number; mqtt_host: string; mqtt_port: number; + mqtt_bind_address: string; telemetry_poll_interval_seconds: number; mqtt_publish_interval_seconds: number; } @@ -52,6 +54,45 @@ export interface ComputeContourAgentInstall { blocked_reason: string | null; } +export interface ComputeContourNetworkStatus { + schema_version: "missioncore.compute-contour-network-status/v1"; + contour_id: string; + observed_at_utc: string; + latency_ms: number; + broker: { + configured_host: string; + configured_port: number; + bind_address: string; + resolved_addresses: string[]; + endpoint_reachable: boolean; + bind_reachable: boolean; + bind_address_owned: boolean; + ready: boolean; + }; + worker: { + checked: boolean; + reachable: boolean; + identity_matches: boolean; + node_id: string | null; + service_status: string | null; + mqtt_host: string | null; + mqtt_port: number | null; + resolved_addresses: string[]; + broker_reachable: boolean; + matches_profile: boolean; + error_code: string | null; + }; +} + +export interface ComputeContourNetworkApply { + schema_version: "missioncore.compute-contour-network-apply/v1"; + contour_id: string; + target: "broker" | "worker"; + changed: boolean; + applied_at_utc: string; + ready: boolean; +} + function isRecord(value: unknown): value is Record { return Boolean(value) && typeof value === "object" && !Array.isArray(value); } @@ -66,7 +107,11 @@ async function requestJson(url: string, init: RequestInit): Promise { }, }); if (!response.ok) { - throw new Error(`Контуры вычисления: HTTP ${response.status}.`); + const document = await response.json().catch(() => null); + const detail = isRecord(document) && typeof document.detail === "string" + ? document.detail + : null; + throw new Error(detail ?? `Контуры вычисления: HTTP ${response.status}.`); } return response.json(); } @@ -100,6 +145,7 @@ export async function createComputeContour( ssh_port: draft.ssh_port, mqtt_host: draft.mqtt_host, mqtt_port: draft.mqtt_port, + mqtt_bind_address: draft.mqtt_bind_address, telemetry_poll_interval_seconds: draft.telemetry_poll_interval_seconds, mqtt_publish_interval_seconds: draft.mqtt_publish_interval_seconds, }), @@ -152,3 +198,45 @@ export async function fetchComputeContourAgentInstall( } return document as unknown as ComputeContourAgentInstall; } + +export async function fetchComputeContourNetwork( + contourId: string, + signal?: AbortSignal, +): Promise { + const document = await requestJson( + `/api/v1/system/contours/${encodeURIComponent(contourId)}/network`, + { + method: "GET", + signal, + }, + ); + if ( + !isRecord(document) + || document.schema_version !== "missioncore.compute-contour-network-status/v1" + ) { + throw new Error("Сетевой статус контура не соответствует контракту."); + } + return document as unknown as ComputeContourNetworkStatus; +} + +export async function applyComputeContourNetwork( + contourId: string, + target: "broker" | "worker", + signal?: AbortSignal, +): Promise { + const document = await requestJson( + `/api/v1/system/contours/${encodeURIComponent(contourId)}/network/${target}`, + { + method: "POST", + signal, + }, + ); + if ( + !isRecord(document) + || document.schema_version !== "missioncore.compute-contour-network-apply/v1" + || document.target !== target + ) { + throw new Error("Результат применения сетевого профиля не соответствует контракту."); + } + return document as unknown as ComputeContourNetworkApply; +} diff --git a/apps/control-station/src/styles/system-telemetry.css b/apps/control-station/src/styles/system-telemetry.css index 6e7089a..475f59d 100644 --- a/apps/control-station/src/styles/system-telemetry.css +++ b/apps/control-station/src/styles/system-telemetry.css @@ -16,19 +16,22 @@ gap: 0.75rem; } -.compute-contour-settings__install { +.compute-contour-settings__install, +.compute-contour-settings__network { display: grid; gap: 0.75rem; } -.compute-contour-settings__install dl { +.compute-contour-settings__install dl, +.compute-contour-settings__network dl { display: grid; grid-template-columns: repeat(2, minmax(0, 1fr)); gap: 0.65rem; margin: 0; } -.compute-contour-settings__install dl > div { +.compute-contour-settings__install dl > div, +.compute-contour-settings__network dl > div { display: grid; gap: 0.22rem; min-width: 0; @@ -37,12 +40,14 @@ background: var(--station-panel-soft); } -.compute-contour-settings__install dt { +.compute-contour-settings__install dt, +.compute-contour-settings__network dt { color: var(--nodedc-text-muted); font-size: 0.58rem; } -.compute-contour-settings__install dd { +.compute-contour-settings__install dd, +.compute-contour-settings__network dd { overflow: hidden; margin: 0; color: var(--nodedc-text-primary); @@ -64,6 +69,7 @@ } .compute-contour-settings__install p, +.compute-contour-settings__network p, .compute-contour-settings__empty { margin: 0; color: var(--nodedc-text-muted); @@ -71,6 +77,14 @@ line-height: 1.5; } +.compute-contour-settings__network-statuses, +.compute-contour-settings__network-actions { + display: flex; + flex-wrap: wrap; + align-items: center; + gap: 0.55rem; +} + .system-workspace__lead, .system-section-heading { display: flex; @@ -626,7 +640,8 @@ @media (max-width: 760px) { .compute-contour-settings__form, - .compute-contour-settings__install dl { + .compute-contour-settings__install dl, + .compute-contour-settings__network dl { grid-template-columns: 1fr; } diff --git a/apps/control-station/test/systemTelemetry.test.mjs b/apps/control-station/test/systemTelemetry.test.mjs index 4dd9a54..fcfc977 100644 --- a/apps/control-station/test/systemTelemetry.test.mjs +++ b/apps/control-station/test/systemTelemetry.test.mjs @@ -89,6 +89,10 @@ test("Worker 006 telemetry remains a bounded system feature slice", async () => assert.match(contourSettings, /FieldFrame label="Операционная система"/); assert.match(contourSettings, /label="Обновление интерфейса"/); assert.match(contourSettings, /label="Интервал MQTT агента"/); + assert.match(contourSettings, /label="Публикация broker"/); + assert.match(contourSettings, /MQTT · Mac ↔ Worker/); + assert.match(contourSettings, /applyComputeContourNetwork/); + assert.match(contourSettings, /Применить на Worker/); assert.match(contourSettings, /value=\{telemetryPollIntervalDraft\}/); assert.match(contourSettings, /onBlur=\{commitTelemetryPollInterval\}/); assert.match(contourSettings, /event\.key === "Escape"/); @@ -98,6 +102,11 @@ test("Worker 006 telemetry remains a bounded system feature slice", async () => ); assert.match(contourContract, /telemetry_poll_interval_seconds: number/); assert.match(contourContract, /mqtt_publish_interval_seconds: number/); + assert.match(contourContract, /mqtt_bind_address: string/); + assert.match( + contourContract, + /missioncore\.compute-contour-network-status\/v1/, + ); assert.match(pollIntervalContract, /MIN_TELEMETRY_POLL_INTERVAL_SECONDS\s*=\s*1/); assert.match(telemetryPolling, /normalizeWorkerTelemetryPollMilliseconds/); assert.match(styles, /system-telemetry\.css/); diff --git a/src/k1link/web/compute_contour_api.py b/src/k1link/web/compute_contour_api.py index fb6a8d3..8a7b0b7 100644 --- a/src/k1link/web/compute_contour_api.py +++ b/src/k1link/web/compute_contour_api.py @@ -1,5 +1,6 @@ from __future__ import annotations +import ipaddress import json import os import re @@ -14,12 +15,25 @@ from typing import Final, Literal from fastapi import APIRouter, HTTPException from pydantic import BaseModel, ConfigDict, Field, field_validator +from k1link.web.compute_contour_network import ( + ComputeContourNetworkTarget, + NetworkOperationError, + apply_broker_network, + apply_worker_network, + probe_compute_contour_network, +) + CONTOUR_SCHEMA: Final = "missioncore.compute-contour/v1" CATALOG_SCHEMA: Final = "missioncore.compute-contour-catalog/v1" INSTALL_SCHEMA: Final = "missioncore.compute-contour-agent-install/v1" CATALOG_FILE_NAME: Final = "compute-contours.json" SAFE_IDENTIFIER: Final = re.compile(r"^[a-z0-9](?:[a-z0-9-]{0,62}[a-z0-9])?$") SAFE_NODE_ID: Final = re.compile(r"^[A-Za-z0-9](?:[A-Za-z0-9_.-]{0,126}[A-Za-z0-9])?$") +SAFE_HOSTNAME: Final = re.compile( + r"^(?=.{1,253}\.?$)(?:[A-Za-z0-9](?:[A-Za-z0-9-]{0,61}" + r"[A-Za-z0-9])?\.)*[A-Za-z0-9](?:[A-Za-z0-9-]{0,61}" + r"[A-Za-z0-9])?\.?$" +) RootProvider = Callable[[], Path] TelemetryMode = Literal["agent-mqtt", "legacy-ssh"] @@ -46,6 +60,7 @@ class ComputeContour(StrictModel): ssh_port: int = Field(default=22, ge=1, le=65535) mqtt_host: str = Field(default="127.0.0.1", min_length=1, max_length=253) mqtt_port: int = Field(default=1883, ge=1, le=65535) + mqtt_bind_address: str = Field(default="127.0.0.1", min_length=1, max_length=45) telemetry_poll_interval_seconds: int = Field(default=3, ge=1, le=60) mqtt_publish_interval_seconds: int = Field(default=2, ge=1, le=60) revision: int = Field(default=0, ge=0) @@ -75,14 +90,43 @@ class ComputeContour(StrictModel): raise ValueError("node id contains unsupported characters") return normalized - @field_validator("address", "mqtt_host") + @field_validator("address") @classmethod def validate_host(cls, value: str) -> str: normalized = value.strip() + if not normalized: + return "" if any(character.isspace() or ord(character) < 32 for character in normalized): raise ValueError("host contains whitespace or control characters") + try: + ipaddress.ip_address(normalized) + except ValueError: + if SAFE_HOSTNAME.fullmatch(normalized) is None: + raise ValueError("host must be an IP address or hostname") from None + return normalized.rstrip(".") + + @field_validator("mqtt_host") + @classmethod + def validate_mqtt_host(cls, value: str) -> str: + normalized = cls.validate_host(value) + if not normalized: + raise ValueError("MQTT host is required") return normalized + @field_validator("mqtt_bind_address") + @classmethod + def validate_bind_address(cls, value: str) -> str: + normalized = value.strip() + try: + address = ipaddress.ip_address(normalized) + except ValueError: + raise ValueError("MQTT bind address must be an IP address") from None + if address.is_unspecified or address.is_multicast: + raise ValueError("MQTT bind address must target one local interface") + if not address.is_loopback and not address.is_private: + raise ValueError("MQTT without TLS may bind only to a private LAN address") + return str(address) + class ComputeContourCreate(StrictModel): display_name: str = Field(min_length=1, max_length=80) @@ -92,6 +136,7 @@ class ComputeContourCreate(StrictModel): ssh_port: int = Field(default=22, ge=1, le=65535) mqtt_host: str = Field(default="127.0.0.1", min_length=1, max_length=253) mqtt_port: int = Field(default=1883, ge=1, le=65535) + mqtt_bind_address: str = Field(default="127.0.0.1", min_length=1, max_length=45) telemetry_poll_interval_seconds: int = Field(default=3, ge=1, le=60) mqtt_publish_interval_seconds: int = Field(default=2, ge=1, le=60) @@ -105,11 +150,21 @@ class ComputeContourCreate(StrictModel): def validate_node_id(cls, value: str) -> str: return ComputeContour.validate_node_id(value) - @field_validator("address", "mqtt_host") + @field_validator("address") @classmethod def validate_host(cls, value: str) -> str: return ComputeContour.validate_host(value) + @field_validator("mqtt_host") + @classmethod + def validate_mqtt_host(cls, value: str) -> str: + return ComputeContour.validate_mqtt_host(value) + + @field_validator("mqtt_bind_address") + @classmethod + def validate_bind_address(cls, value: str) -> str: + return ComputeContour.validate_bind_address(value) + class ComputeContourPut(ComputeContourCreate): revision: int = Field(ge=0) @@ -128,6 +183,7 @@ def default_compute_contour() -> ComputeContour: ssh_port=22, mqtt_host="127.0.0.1", mqtt_port=1883, + mqtt_bind_address="127.0.0.1", telemetry_poll_interval_seconds=3, mqtt_publish_interval_seconds=2, ) @@ -158,6 +214,7 @@ class ComputeContourStore: ssh_port=request.ssh_port, mqtt_host=request.mqtt_host, mqtt_port=request.mqtt_port, + mqtt_bind_address=request.mqtt_bind_address, telemetry_poll_interval_seconds=( request.telemetry_poll_interval_seconds ), @@ -189,6 +246,7 @@ class ComputeContourStore: "ssh_port": request.ssh_port, "mqtt_host": request.mqtt_host, "mqtt_port": request.mqtt_port, + "mqtt_bind_address": request.mqtt_bind_address, "telemetry_poll_interval_seconds": ( request.telemetry_poll_interval_seconds ), @@ -305,6 +363,20 @@ def _agent_install_document(contour: ComputeContour) -> dict[str, object]: } +def _network_target(contour: ComputeContour) -> ComputeContourNetworkTarget: + return ComputeContourNetworkTarget( + contour_id=contour.contour_id, + expected_node_id=contour.expected_node_id, + platform=contour.platform, + worker_address=contour.address, + ssh_port=contour.ssh_port, + mqtt_host=contour.mqtt_host, + mqtt_port=contour.mqtt_port, + mqtt_bind_address=contour.mqtt_bind_address, + mqtt_publish_interval_seconds=contour.mqtt_publish_interval_seconds, + ) + + def build_compute_contour_router(*, root_provider: RootProvider) -> APIRouter: store = ComputeContourStore(root_provider()) router = APIRouter(prefix="/api/v1/system", tags=["system"]) @@ -339,4 +411,36 @@ def build_compute_contour_router(*, root_provider: RootProvider) -> APIRouter: except KeyError as exc: raise HTTPException(status_code=404, detail="Контур не найден.") from exc + @router.get("/contours/{contour_id}/network") + def get_contour_network(contour_id: str) -> dict[str, object]: + try: + return probe_compute_contour_network(_network_target(store.get(contour_id))) + except KeyError as exc: + raise HTTPException(status_code=404, detail="Контур не найден.") from exc + + @router.post("/contours/{contour_id}/network/broker") + def apply_contour_broker_network(contour_id: str) -> dict[str, object]: + try: + contour = store.get(contour_id) + telemetry_plane_root = ( + Path(__file__).resolve().parents[3] / "deploy" / "telemetry-plane" + ) + return apply_broker_network( + _network_target(contour), + telemetry_plane_root, + ) + except KeyError as exc: + raise HTTPException(status_code=404, detail="Контур не найден.") from exc + except NetworkOperationError as exc: + raise HTTPException(status_code=409, detail=str(exc)) from exc + + @router.post("/contours/{contour_id}/network/worker") + def apply_contour_worker_network(contour_id: str) -> dict[str, object]: + try: + return apply_worker_network(_network_target(store.get(contour_id))) + except KeyError as exc: + raise HTTPException(status_code=404, detail="Контур не найден.") from exc + except NetworkOperationError as exc: + raise HTTPException(status_code=409, detail=str(exc)) from exc + return router diff --git a/src/k1link/web/compute_contour_network.py b/src/k1link/web/compute_contour_network.py new file mode 100644 index 0000000..9c96c12 --- /dev/null +++ b/src/k1link/web/compute_contour_network.py @@ -0,0 +1,601 @@ +from __future__ import annotations + +import base64 +import contextlib +import ipaddress +import json +import os +import socket +import stat +import subprocess +import tempfile +import time +from dataclasses import dataclass +from datetime import UTC, datetime +from pathlib import Path +from typing import Any, Final + +NETWORK_STATUS_SCHEMA: Final = "missioncore.compute-contour-network-status/v1" +NETWORK_APPLY_SCHEMA: Final = "missioncore.compute-contour-network-apply/v1" +SSH_HOST_ALIAS: Final = "mission-gpu" + + +def _utc_now() -> str: + return datetime.now(UTC).isoformat().replace("+00:00", "Z") + + +@dataclass(frozen=True) +class ComputeContourNetworkTarget: + contour_id: str + expected_node_id: str + platform: str + worker_address: str + ssh_port: int + mqtt_host: str + mqtt_port: int + mqtt_bind_address: str + mqtt_publish_interval_seconds: int + + +class NetworkOperationError(RuntimeError): + pass + + +def _resolve_addresses(host: str, port: int) -> list[str]: + try: + rows = socket.getaddrinfo(host, port, type=socket.SOCK_STREAM) + except OSError: + return [] + return sorted({str(row[4][0]) for row in rows if row[4]}) + + +def _preferred_lan_addresses(addresses: list[str]) -> list[str]: + lan = [ + value + for value in addresses + if not (address := ipaddress.ip_address(value)).is_loopback + and not address.is_link_local + ] + return lan or addresses + + +def _tcp_reachable(host: str, port: int, *, timeout: float = 2.0) -> bool: + try: + rows = socket.getaddrinfo(host, port, type=socket.SOCK_STREAM) + except OSError: + return False + for family, socket_type, protocol, _canonical_name, address in rows: + connection = socket.socket(family, socket_type, protocol) + try: + connection.settimeout(timeout) + connection.connect(address) + return True + except OSError: + continue + finally: + connection.close() + return False + + +def _address_belongs_to_host(address: str) -> bool: + family = socket.AF_INET6 if ":" in address else socket.AF_INET + candidate = socket.socket(family, socket.SOCK_STREAM) + try: + candidate.bind((address, 0)) + return True + except OSError: + return False + finally: + candidate.close() + + +def _ssh_arguments(target: ComputeContourNetworkTarget) -> list[str]: + arguments = [ + "ssh", + "-o", + "BatchMode=yes", + "-o", + "StrictHostKeyChecking=yes", + "-o", + f"HostKeyAlias={SSH_HOST_ALIAS}", + "-o", + "ConnectTimeout=5", + ] + if target.worker_address: + arguments.extend(["-o", f"HostName={target.worker_address}"]) + arguments.extend( + [ + "-p", + str(target.ssh_port), + SSH_HOST_ALIAS, + "powershell.exe", + "-NoLogo", + "-NoProfile", + "-NonInteractive", + "-Command", + ( + "$encoded=[Console]::In.ReadToEnd();" + "Invoke-Expression " + "([Text.Encoding]::Unicode.GetString(" + "[Convert]::FromBase64String($encoded)))" + ), + ] + ) + return arguments + + +def _run_worker_powershell( + target: ComputeContourNetworkTarget, + script: str, + *, + timeout: float, +) -> dict[str, Any]: + try: + completed = subprocess.run( + _ssh_arguments(target), + check=False, + capture_output=True, + input=base64.b64encode(script.encode("utf-16le")), + timeout=timeout, + ) + except (OSError, subprocess.TimeoutExpired) as exc: + raise NetworkOperationError("Worker недоступен по доверенному SSH-профилю.") from exc + if completed.returncode != 0: + raise NetworkOperationError( + "Worker отклонил сетевую операцию; конфигурация не подтверждена." + ) + for encoding in ("utf-8-sig", "cp866"): + try: + document = json.loads(completed.stdout.decode(encoding).strip()) + except (UnicodeDecodeError, json.JSONDecodeError): + continue + if isinstance(document, dict): + return document + raise NetworkOperationError("Worker вернул некорректный ответ сетевой проверки.") + + +WORKER_NETWORK_STATUS_POWERSHELL: Final = r""" +$ErrorActionPreference = "Stop" +$serviceRegistryPath = "HKLM:\SYSTEM\CurrentControlSet\Services\telegraf" +$environmentRows = @((Get-ItemProperty -Path $serviceRegistryPath).Environment) +$environment = @{} +foreach ($entry in $environmentRows) { + $name, $value = $entry -split "=", 2 + if ($name) { + $environment[$name.ToUpperInvariant()] = $value + } +} +$mqttHost = [string]$environment["MISSIONCORE_MQTT_HOST"] +$mqttPort = 0 +[void][int]::TryParse( + [string]$environment["MISSIONCORE_MQTT_PORT"], + [ref]$mqttPort +) +$resolved = @() +$brokerReachable = $false +if ($mqttHost -and $mqttPort -gt 0) { + try { + $resolved = @( + [Net.Dns]::GetHostAddresses($mqttHost) | + ForEach-Object { $_.IPAddressToString } | + Sort-Object -Unique + ) + $client = [Net.Sockets.TcpClient]::new() + try { + $task = $client.ConnectAsync($mqttHost, $mqttPort) + $brokerReachable = $task.Wait(3000) -and $client.Connected + } + finally { + $client.Dispose() + } + } + catch {} +} +[ordered]@{ + schema_version = "missioncore.compute-contour-worker-network/v1" + node_id = $env:COMPUTERNAME + service_status = (Get-Service -Name "telegraf").Status.ToString() + mqtt_host = $mqttHost + mqtt_port = $mqttPort + resolved_addresses = $resolved + broker_reachable = $brokerReachable +} | ConvertTo-Json -Depth 5 -Compress +""".strip() + + +WORKER_NETWORK_APPLY_POWERSHELL: Final = r""" +$ErrorActionPreference = "Stop" +function Test-BrokerEndpoint { + param( + [Parameter(Mandatory = $true)][string]$HostName, + [Parameter(Mandatory = $true)][int]$Port, + [int]$Attempts = 3 + ) + for ($attempt = 1; $attempt -le $Attempts; $attempt++) { + $client = [Net.Sockets.TcpClient]::new() + try { + [void][Net.Dns]::GetHostAddresses($HostName) + $task = $client.ConnectAsync($HostName, $Port) + if ($task.Wait(3000) -and $client.Connected) { + return $true + } + } + catch {} + finally { + $client.Dispose() + } + if ($attempt -lt $Attempts) { + Start-Sleep -Seconds 1 + } + } + return $false +} +$payloadJson = [Text.Encoding]::UTF8.GetString( + [Convert]::FromBase64String("__PAYLOAD_BASE64__") +) +$payload = $payloadJson | ConvertFrom-Json +if ($env:COMPUTERNAME -ne $payload.expected_node_id) { + throw "Worker identity mismatch" +} +if (-not (Test-BrokerEndpoint ` + -HostName ([string]$payload.mqtt_host) ` + -Port ([int]$payload.mqtt_port) ` + -Attempts 3)) { + throw "Broker preflight failed" +} +$serviceRegistryPath = "HKLM:\SYSTEM\CurrentControlSet\Services\telegraf" +$environmentBefore = @((Get-ItemProperty -Path $serviceRegistryPath).Environment) +$replacement = @{ + "MISSIONCORE_MQTT_HOST" = [string]$payload.mqtt_host + "MISSIONCORE_MQTT_PORT" = [string]$payload.mqtt_port + "MISSIONCORE_TELEMETRY_INTERVAL" = ( + [string]$payload.telemetry_interval_seconds + "s" + ) +} +$seen = @{} +$current = @{} +$environmentCandidate = @() +foreach ($entry in $environmentBefore) { + $name, $value = $entry -split "=", 2 + $canonicalName = $name.ToUpperInvariant() + if ($name) { + $current[$canonicalName] = $value + } + if ($replacement.ContainsKey($canonicalName)) { + $environmentCandidate += "$canonicalName=$($replacement[$canonicalName])" + $seen[$canonicalName] = $true + } + else { + $environmentCandidate += $entry + } +} +foreach ($name in $replacement.Keys) { + if (-not $seen.ContainsKey($name)) { + $environmentCandidate += "$name=$($replacement[$name])" + } +} +$changed = ( + [string]$current["MISSIONCORE_MQTT_HOST"] -ne [string]$payload.mqtt_host -or + [string]$current["MISSIONCORE_MQTT_PORT"] -ne [string]$payload.mqtt_port -or + [string]$current["MISSIONCORE_TELEMETRY_INTERVAL"] -ne ( + [string]$payload.telemetry_interval_seconds + "s" + ) +) +if ($changed) { + try { + Set-ItemProperty -Path $serviceRegistryPath -Name Environment ` + -Type MultiString -Value $environmentCandidate + Restart-Service -Name "telegraf" -Force + $service = Get-Service -Name "telegraf" + $service.WaitForStatus( + [ServiceProcess.ServiceControllerStatus]::Running, + [TimeSpan]::FromSeconds(20) + ) + if (-not (Test-BrokerEndpoint ` + -HostName ([string]$payload.mqtt_host) ` + -Port ([int]$payload.mqtt_port) ` + -Attempts 5)) { + throw "Broker verification failed" + } + } + catch { + Set-ItemProperty -Path $serviceRegistryPath -Name Environment ` + -Type MultiString -Value $environmentBefore + Restart-Service -Name "telegraf" -Force -ErrorAction SilentlyContinue + throw + } +} +elseif ((Get-Service -Name "telegraf").Status -ne "Running") { + Start-Service -Name "telegraf" + $service = Get-Service -Name "telegraf" + $service.WaitForStatus( + [ServiceProcess.ServiceControllerStatus]::Running, + [TimeSpan]::FromSeconds(20) + ) +} +$resolved = @( + [Net.Dns]::GetHostAddresses([string]$payload.mqtt_host) | + ForEach-Object { $_.IPAddressToString } | + Sort-Object -Unique +) +[ordered]@{ + schema_version = "missioncore.compute-contour-worker-network/v1" + node_id = $env:COMPUTERNAME + service_status = (Get-Service -Name "telegraf").Status.ToString() + mqtt_host = [string]$payload.mqtt_host + mqtt_port = [int]$payload.mqtt_port + resolved_addresses = $resolved + broker_reachable = $true + changed = $changed +} | ConvertTo-Json -Depth 5 -Compress +""".strip() + + +def _worker_status(target: ComputeContourNetworkTarget) -> dict[str, Any]: + if target.platform != "windows": + return { + "checked": False, + "reachable": False, + "identity_matches": False, + "node_id": None, + "service_status": None, + "mqtt_host": None, + "mqtt_port": None, + "resolved_addresses": [], + "broker_reachable": False, + "matches_profile": False, + "error_code": "unsupported-platform", + } + try: + document = _run_worker_powershell( + target, + WORKER_NETWORK_STATUS_POWERSHELL, + timeout=10, + ) + except NetworkOperationError: + return { + "checked": True, + "reachable": False, + "identity_matches": False, + "node_id": None, + "service_status": None, + "mqtt_host": None, + "mqtt_port": None, + "resolved_addresses": [], + "broker_reachable": False, + "matches_profile": False, + "error_code": "worker-unreachable", + } + node_id = document.get("node_id") + mqtt_host = document.get("mqtt_host") + mqtt_port = document.get("mqtt_port") + identity_matches = node_id == target.expected_node_id + matches_profile = ( + identity_matches + and mqtt_host == target.mqtt_host + and mqtt_port == target.mqtt_port + ) + return { + "checked": True, + "reachable": True, + "identity_matches": identity_matches, + "node_id": node_id if isinstance(node_id, str) else None, + "service_status": ( + document.get("service_status") + if isinstance(document.get("service_status"), str) + else None + ), + "mqtt_host": mqtt_host if isinstance(mqtt_host, str) else None, + "mqtt_port": mqtt_port if isinstance(mqtt_port, int) else None, + "resolved_addresses": ( + document.get("resolved_addresses") + if isinstance(document.get("resolved_addresses"), list) + else [] + ), + "broker_reachable": document.get("broker_reachable") is True, + "matches_profile": matches_profile, + "error_code": None if identity_matches else "worker-identity-mismatch", + } + + +def probe_compute_contour_network( + target: ComputeContourNetworkTarget, +) -> dict[str, Any]: + started = time.perf_counter() + resolved_addresses = _resolve_addresses(target.mqtt_host, target.mqtt_port) + endpoint_reachable = _tcp_reachable(target.mqtt_host, target.mqtt_port) + bind_reachable = _tcp_reachable( + target.mqtt_bind_address, + target.mqtt_port, + ) + return { + "schema_version": NETWORK_STATUS_SCHEMA, + "contour_id": target.contour_id, + "observed_at_utc": _utc_now(), + "latency_ms": (time.perf_counter() - started) * 1000, + "broker": { + "configured_host": target.mqtt_host, + "configured_port": target.mqtt_port, + "bind_address": target.mqtt_bind_address, + "resolved_addresses": _preferred_lan_addresses(resolved_addresses), + "endpoint_reachable": endpoint_reachable, + "bind_reachable": bind_reachable, + "bind_address_owned": _address_belongs_to_host( + target.mqtt_bind_address + ), + "ready": endpoint_reachable and bind_reachable, + }, + "worker": _worker_status(target), + } + + +def _replace_environment_value(document: str, name: str, value: str) -> str: + lines = document.splitlines() + replacement = f"{name}={value}" + replaced = False + output: list[str] = [] + for line in lines: + if line.startswith(f"{name}="): + output.append(replacement) + replaced = True + else: + output.append(line) + if not replaced: + output.append(replacement) + return "\n".join(output) + "\n" + + +def _atomic_private_write(path: Path, document: str, *, mode: int) -> None: + descriptor, temporary_name = tempfile.mkstemp( + prefix=f".{path.name}.", + suffix=".tmp", + dir=path.parent, + ) + temporary = Path(temporary_name) + try: + with os.fdopen(descriptor, "w", encoding="utf-8", newline="\n") as stream: + stream.write(document) + stream.flush() + os.fsync(stream.fileno()) + os.chmod(temporary, stat.S_IMODE(mode)) + os.replace(temporary, path) + finally: + temporary.unlink(missing_ok=True) + + +def _run_compose_broker(telemetry_plane_root: Path) -> None: + environment_path = telemetry_plane_root / ".env" + compose_path = telemetry_plane_root / "compose.yaml" + common = [ + "docker", + "compose", + "--env-file", + str(environment_path), + "-f", + str(compose_path), + ] + for suffix, timeout in ( + (["config", "--quiet"], 20), + (["up", "-d", "--no-build", "broker"], 45), + ): + try: + completed = subprocess.run( + [*common, *suffix], + cwd=telemetry_plane_root, + check=False, + capture_output=True, + timeout=timeout, + ) + except (OSError, subprocess.TimeoutExpired) as exc: + raise NetworkOperationError( + "Docker не подтвердил конфигурацию MQTT broker." + ) from exc + if completed.returncode != 0: + raise NetworkOperationError( + "Docker не подтвердил конфигурацию MQTT broker." + ) + + +def apply_broker_network( + target: ComputeContourNetworkTarget, + telemetry_plane_root: Path, +) -> dict[str, Any]: + if not _address_belongs_to_host(target.mqtt_bind_address): + raise NetworkOperationError( + "Адрес публикации MQTT не принадлежит текущему Mac." + ) + environment_path = telemetry_plane_root / ".env" + compose_path = telemetry_plane_root / "compose.yaml" + if not environment_path.is_file() or not compose_path.is_file(): + raise NetworkOperationError("Локальный telemetry plane не подготовлен.") + before = environment_path.read_text(encoding="utf-8") + after = _replace_environment_value( + before, + "MISSIONCORE_MQTT_BIND_ADDRESS", + target.mqtt_bind_address, + ) + changed = after != before + mode = environment_path.stat().st_mode + try: + if changed: + _atomic_private_write(environment_path, after, mode=mode) + _run_compose_broker(telemetry_plane_root) + if not _tcp_reachable( + target.mqtt_bind_address, + target.mqtt_port, + timeout=3, + ): + raise NetworkOperationError( + "MQTT broker не открыл настроенный адрес публикации." + ) + except Exception as exc: + if changed: + _atomic_private_write(environment_path, before, mode=mode) + with contextlib.suppress(NetworkOperationError): + _run_compose_broker(telemetry_plane_root) + if isinstance(exc, NetworkOperationError): + raise + raise NetworkOperationError( + "Не удалось применить адрес публикации MQTT broker." + ) from exc + return { + "schema_version": NETWORK_APPLY_SCHEMA, + "contour_id": target.contour_id, + "target": "broker", + "changed": changed, + "applied_at_utc": _utc_now(), + "ready": True, + } + + +def apply_worker_network( + target: ComputeContourNetworkTarget, +) -> dict[str, Any]: + if target.platform != "windows": + raise NetworkOperationError( + "Автоматическое применение сейчас поддерживает Windows Worker." + ) + resolved = _resolve_addresses(target.mqtt_host, target.mqtt_port) + if not resolved or any( + not ( + (address := ipaddress.ip_address(value)).is_private + or address.is_loopback + or address.is_link_local + ) + for value in resolved + ): + raise NetworkOperationError( + "MQTT без TLS разрешён только внутри проверенной локальной сети." + ) + payload = { + "expected_node_id": target.expected_node_id, + "mqtt_host": target.mqtt_host, + "mqtt_port": target.mqtt_port, + "telemetry_interval_seconds": target.mqtt_publish_interval_seconds, + } + payload_base64 = base64.b64encode( + json.dumps(payload, separators=(",", ":")).encode("utf-8") + ).decode("ascii") + script = WORKER_NETWORK_APPLY_POWERSHELL.replace( + "__PAYLOAD_BASE64__", + payload_base64, + ) + document = _run_worker_powershell(target, script, timeout=35) + if ( + document.get("node_id") != target.expected_node_id + or document.get("mqtt_host") != target.mqtt_host + or document.get("mqtt_port") != target.mqtt_port + or document.get("service_status") != "Running" + or document.get("broker_reachable") is not True + ): + raise NetworkOperationError( + "Worker не подтвердил применённый сетевой профиль." + ) + return { + "schema_version": NETWORK_APPLY_SCHEMA, + "contour_id": target.contour_id, + "target": "worker", + "changed": document.get("changed") is True, + "applied_at_utc": _utc_now(), + "ready": True, + } diff --git a/tests/test_compute_contour_api.py b/tests/test_compute_contour_api.py index fafdec7..d14ec0b 100644 --- a/tests/test_compute_contour_api.py +++ b/tests/test_compute_contour_api.py @@ -41,6 +41,7 @@ def test_contour_store_migrates_worker_006_as_first_configuration( assert contours[0].telemetry_mode == "agent-mqtt" assert contours[0].telemetry_poll_interval_seconds == 3 assert contours[0].mqtt_publish_interval_seconds == 2 + assert contours[0].mqtt_bind_address == "127.0.0.1" assert not store.path.exists() @@ -53,6 +54,7 @@ def test_contour_store_creates_and_updates_private_catalog(tmp_path: Path) -> No platform="linux", address="192.0.2.25", mqtt_host="192.0.2.5", + mqtt_bind_address="192.168.10.5", ) ) updated = store.update( @@ -67,6 +69,7 @@ def test_contour_store_creates_and_updates_private_catalog(tmp_path: Path) -> No ssh_port=22, mqtt_host="192.0.2.5", mqtt_port=1883, + mqtt_bind_address="192.168.10.5", telemetry_poll_interval_seconds=1, mqtt_publish_interval_seconds=4, ), @@ -91,6 +94,7 @@ def test_contour_store_creates_and_updates_private_catalog(tmp_path: Path) -> No ssh_port=22, mqtt_host="127.0.0.1", mqtt_port=1883, + mqtt_bind_address="127.0.0.1", telemetry_poll_interval_seconds=3, mqtt_publish_interval_seconds=2, ), @@ -119,6 +123,12 @@ def test_contour_router_exposes_catalog_and_safe_install_contract( assert "password" not in document["agent"]["environment"] assert document["agent"]["environment"]["MISSIONCORE_TELEMETRY_INTERVAL"] == "2s" assert document["ready"] is False + network = _endpoint( + router, + "/api/v1/system/contours/{contour_id}/network", + "GET", + ) + assert callable(network) def test_contour_router_returns_404_for_unknown_contour(tmp_path: Path) -> None: diff --git a/tests/test_compute_contour_network.py b/tests/test_compute_contour_network.py new file mode 100644 index 0000000..5003e84 --- /dev/null +++ b/tests/test_compute_contour_network.py @@ -0,0 +1,137 @@ +from __future__ import annotations + +import stat +from pathlib import Path +from typing import Any + +import pytest + +from k1link.web import compute_contour_network as network +from k1link.web.compute_contour_network import ( + ComputeContourNetworkTarget, + NetworkOperationError, +) + + +def _target(**updates: object) -> ComputeContourNetworkTarget: + values: dict[str, object] = { + "contour_id": "worker-006", + "expected_node_id": "DESKTOP-OPJ8J04", + "platform": "windows", + "worker_address": "", + "ssh_port": 22, + "mqtt_host": "mission-core.local", + "mqtt_port": 1883, + "mqtt_bind_address": "192.168.68.56", + "mqtt_publish_interval_seconds": 2, + } + values.update(updates) + return ComputeContourNetworkTarget(**values) # type: ignore[arg-type] + + +def test_broker_apply_preserves_private_environment_and_changes_only_bind( + tmp_path: Path, + monkeypatch: pytest.MonkeyPatch, +) -> None: + root = tmp_path / "telemetry-plane" + root.mkdir() + environment = root / ".env" + environment.write_text( + "MISSIONCORE_MQTT_BIND_ADDRESS=192.168.1.5\n" + "MISSIONCORE_MQTT_PASSWORD=do-not-copy\n", + encoding="utf-8", + ) + environment.chmod(0o600) + (root / "compose.yaml").write_text("services: {}\n", encoding="utf-8") + compose_calls: list[Path] = [] + monkeypatch.setattr(network, "_address_belongs_to_host", lambda _value: True) + monkeypatch.setattr(network, "_tcp_reachable", lambda *_args, **_kwargs: True) + monkeypatch.setattr( + network, + "_run_compose_broker", + lambda path: compose_calls.append(path), + ) + + result = network.apply_broker_network(_target(), root) + + document = environment.read_text(encoding="utf-8") + assert "MISSIONCORE_MQTT_BIND_ADDRESS=192.168.68.56" in document + assert "MISSIONCORE_MQTT_PASSWORD=do-not-copy" in document + assert stat.S_IMODE(environment.stat().st_mode) == 0o600 + assert compose_calls == [root] + assert result["target"] == "broker" + assert result["ready"] is True + + +def test_broker_apply_rolls_back_environment_when_listener_does_not_open( + tmp_path: Path, + monkeypatch: pytest.MonkeyPatch, +) -> None: + root = tmp_path / "telemetry-plane" + root.mkdir() + environment = root / ".env" + before = ( + "MISSIONCORE_MQTT_BIND_ADDRESS=192.168.1.5\n" + "MISSIONCORE_MQTT_PASSWORD=still-private\n" + ) + environment.write_text(before, encoding="utf-8") + environment.chmod(0o600) + (root / "compose.yaml").write_text("services: {}\n", encoding="utf-8") + monkeypatch.setattr(network, "_address_belongs_to_host", lambda _value: True) + monkeypatch.setattr(network, "_tcp_reachable", lambda *_args, **_kwargs: False) + monkeypatch.setattr(network, "_run_compose_broker", lambda _path: None) + + with pytest.raises(NetworkOperationError, match="не открыл"): + network.apply_broker_network(_target(), root) + + assert environment.read_text(encoding="utf-8") == before + + +def test_worker_apply_uses_private_resolved_endpoint_and_confirms_service( + monkeypatch: pytest.MonkeyPatch, +) -> None: + captured: dict[str, Any] = {} + monkeypatch.setattr( + network, + "_resolve_addresses", + lambda _host, _port: ["192.168.68.56"], + ) + + def run( + target: ComputeContourNetworkTarget, + script: str, + *, + timeout: float, + ) -> dict[str, Any]: + captured.update({"target": target, "script": script, "timeout": timeout}) + return { + "node_id": "DESKTOP-OPJ8J04", + "service_status": "Running", + "mqtt_host": "mission-core.local", + "mqtt_port": 1883, + "resolved_addresses": ["192.168.68.56"], + "broker_reachable": True, + } + + monkeypatch.setattr(network, "_run_worker_powershell", run) + + result = network.apply_worker_network(_target()) + + assert result["target"] == "worker" + assert result["ready"] is True + assert "__PAYLOAD_BASE64__" not in captured["script"] + assert "MISSIONCORE_MQTT_PASSWORD" not in captured["script"] + assert captured["timeout"] == 35 + + +def test_worker_apply_rejects_public_mqtt_endpoint( + monkeypatch: pytest.MonkeyPatch, +) -> None: + monkeypatch.setattr( + network, + "_resolve_addresses", + lambda _host, _port: ["8.8.8.8"], + ) + + with pytest.raises(NetworkOperationError, match="локальной сети"): + network.apply_worker_network(_target(mqtt_host="broker.example.test"))