478 lines
13 KiB
JavaScript
478 lines
13 KiB
JavaScript
import { createHash } from "node:crypto";
|
|
import { promises as fs } from "node:fs";
|
|
import os from "node:os";
|
|
import { spawn } from "node:child_process";
|
|
import process from "node:process";
|
|
import WebSocket from "ws";
|
|
|
|
const DEFAULT_VERSION = "0.1.0";
|
|
const DEFAULT_RECONNECT_DELAY_MS = 5000;
|
|
|
|
function buildBrokerHeartbeatState(isConnected) {
|
|
return isConnected
|
|
? {
|
|
status: "ONLINE",
|
|
discovery_status: "READY",
|
|
metadata: { broker_connected: true },
|
|
}
|
|
: {
|
|
status: "DEGRADED",
|
|
discovery_status: "BROKER_DISCONNECTED",
|
|
metadata: { broker_connected: false },
|
|
};
|
|
}
|
|
|
|
export async function loadConfig(configPath) {
|
|
const raw = await fs.readFile(configPath, "utf8");
|
|
return JSON.parse(raw);
|
|
}
|
|
|
|
export async function saveConfig(configPath, config) {
|
|
await fs.writeFile(configPath, JSON.stringify(config, null, 2));
|
|
}
|
|
|
|
export function normalizeBrokerWsUrl(brokerUrl) {
|
|
if (!brokerUrl) {
|
|
throw new Error("Missing broker URL");
|
|
}
|
|
if (brokerUrl.startsWith("ws://") || brokerUrl.startsWith("wss://")) {
|
|
return brokerUrl;
|
|
}
|
|
return brokerUrl.replace(/^http:/i, "ws:").replace(/^https:/i, "wss:");
|
|
}
|
|
|
|
export async function apiRequest(baseUrl, path, method = "POST", body = {}, fetchImpl = fetch) {
|
|
const response = await fetchImpl(`${String(baseUrl).replace(/\/$/, "")}${path}`, {
|
|
method,
|
|
headers: {
|
|
"content-type": "application/json",
|
|
},
|
|
body: method === "GET" ? undefined : JSON.stringify(body),
|
|
});
|
|
|
|
const json = await response.json();
|
|
if (!response.ok) {
|
|
throw new Error(json?.data?.message || json?.message || `HTTP ${response.status}`);
|
|
}
|
|
|
|
return json.data ?? json;
|
|
}
|
|
|
|
export async function claimIfNeeded(config, configPath, fetchImpl = fetch) {
|
|
if (config.gatewayId && config.agentToken) {
|
|
return config;
|
|
}
|
|
|
|
const claimed = await apiRequest(config.apiUrl, "/edge-agent/claim", "POST", {
|
|
token: config.installToken,
|
|
hostname: config.hostname || os.hostname(),
|
|
installed_version: config.installedVersion || DEFAULT_VERSION,
|
|
metadata: {
|
|
platform: process.platform,
|
|
arch: process.arch,
|
|
},
|
|
}, fetchImpl);
|
|
|
|
const nextConfig = {
|
|
...config,
|
|
gatewayId: claimed.gateway.id,
|
|
agentToken: claimed.agent_token,
|
|
brokerUrl: claimed.broker_url || config.brokerUrl,
|
|
releaseChannel: claimed.release_channel || config.releaseChannel || "stable",
|
|
};
|
|
await saveConfig(configPath, nextConfig);
|
|
return nextConfig;
|
|
}
|
|
|
|
async function fetchJson(url, fetchImpl = fetch) {
|
|
const response = await fetchImpl(url);
|
|
if (!response.ok) {
|
|
throw new Error(`HTTP ${response.status}`);
|
|
}
|
|
return response.json();
|
|
}
|
|
|
|
function expandCandidateIps(options = {}) {
|
|
if (Array.isArray(options.candidateIps) && options.candidateIps.length > 0) {
|
|
return options.candidateIps;
|
|
}
|
|
|
|
if (typeof options.subnetPrefix === "string") {
|
|
const start = Number.isInteger(options.startHost) ? options.startHost : 1;
|
|
const end = Number.isInteger(options.endHost) ? options.endHost : 20;
|
|
const ips = [];
|
|
for (let host = start; host <= end; host += 1) {
|
|
ips.push(`${options.subnetPrefix}.${host}`);
|
|
}
|
|
return ips;
|
|
}
|
|
|
|
return [];
|
|
}
|
|
|
|
export async function discoverShellyDevices(options = {}, fetchImpl = fetch) {
|
|
const candidateIps = expandCandidateIps(options);
|
|
const discovered = [];
|
|
|
|
await Promise.all(candidateIps.map(async (ip) => {
|
|
try {
|
|
const identity = await fetchJson(`http://${ip}/shelly`, fetchImpl);
|
|
discovered.push({
|
|
id: identity.mac || identity.id || ip,
|
|
device_id: identity.mac || identity.id || ip,
|
|
local_ip: ip,
|
|
model: identity.model || identity.type || "Shelly",
|
|
channel_count: Number(identity.num_outputs || identity.num_switches || 1),
|
|
online: true,
|
|
capabilities: {
|
|
generation: identity.gen || null,
|
|
},
|
|
metadata: identity,
|
|
});
|
|
} catch {
|
|
// Ignore non-responsive candidates during opportunistic discovery.
|
|
}
|
|
}));
|
|
|
|
return discovered;
|
|
}
|
|
|
|
export async function getRelayStatus(payload, fetchImpl = fetch) {
|
|
const ip = payload.localIp || payload.local_ip || payload.ip;
|
|
const channel = Number.isInteger(payload.channel) ? payload.channel : 0;
|
|
if (!ip) {
|
|
throw new Error("Missing relay local IP");
|
|
}
|
|
|
|
try {
|
|
const rpc = await fetchJson(`http://${ip}/rpc/Switch.GetStatus?id=${channel}`, fetchImpl);
|
|
return {
|
|
online: true,
|
|
on: Boolean(rpc.output),
|
|
raw: rpc,
|
|
};
|
|
} catch {
|
|
const legacy = await fetchJson(`http://${ip}/relay/${channel}`, fetchImpl);
|
|
return {
|
|
online: true,
|
|
on: Boolean(legacy.ison ?? legacy.output),
|
|
raw: legacy,
|
|
};
|
|
}
|
|
}
|
|
|
|
export async function setRelayState(payload, fetchImpl = fetch) {
|
|
const ip = payload.localIp || payload.local_ip || payload.ip;
|
|
const channel = Number.isInteger(payload.channel) ? payload.channel : 0;
|
|
const on = Boolean(payload.on);
|
|
if (!ip) {
|
|
throw new Error("Missing relay local IP");
|
|
}
|
|
|
|
try {
|
|
const rpc = await fetchJson(`http://${ip}/rpc/Switch.Set?id=${channel}&on=${on ? "true" : "false"}`, fetchImpl);
|
|
return {
|
|
online: true,
|
|
on: Boolean(rpc.output ?? on),
|
|
raw: rpc,
|
|
};
|
|
} catch {
|
|
const legacy = await fetchJson(`http://${ip}/relay/${channel}?turn=${on ? "on" : "off"}`, fetchImpl);
|
|
return {
|
|
online: true,
|
|
on: Boolean(legacy.ison ?? legacy.output ?? on),
|
|
raw: legacy,
|
|
};
|
|
}
|
|
}
|
|
|
|
export async function runUpdate(payload, fetchImpl = fetch) {
|
|
if (!payload.artifactUrl) {
|
|
return {
|
|
updated: false,
|
|
skipped: true,
|
|
reason: "No artifact URL provided",
|
|
};
|
|
}
|
|
|
|
const response = await fetchImpl(payload.artifactUrl);
|
|
if (!response.ok) {
|
|
throw new Error(`Artifact download failed: HTTP ${response.status}`);
|
|
}
|
|
const buffer = Buffer.from(await response.arrayBuffer());
|
|
const sha256 = createHash("sha256").update(buffer).digest("hex");
|
|
if (payload.sha256 && String(payload.sha256).toLowerCase() !== sha256.toLowerCase()) {
|
|
throw new Error("Artifact checksum mismatch");
|
|
}
|
|
|
|
return {
|
|
updated: true,
|
|
sha256,
|
|
bytes: buffer.length,
|
|
};
|
|
}
|
|
|
|
function defaultShellCommand() {
|
|
if (process.platform === "win32") {
|
|
return { command: process.env.ComSpec || "cmd.exe", args: [] };
|
|
}
|
|
return { command: process.env.SHELL || "/bin/sh", args: [] };
|
|
}
|
|
|
|
export function createShellBridge(sendMessage) {
|
|
const sessions = new Map();
|
|
|
|
const open = (payload = {}) => {
|
|
const sessionId = String(payload.sessionId);
|
|
const shell = payload.shellCommand ? { command: payload.shellCommand, args: payload.shellArgs || [] } : defaultShellCommand();
|
|
const proc = spawn(shell.command, shell.args, {
|
|
cwd: payload.cwd || process.cwd(),
|
|
env: process.env,
|
|
stdio: "pipe",
|
|
});
|
|
|
|
proc.stdout.on("data", (chunk) => {
|
|
sendMessage({ type: "SHELL_OUTPUT", sessionId, data: chunk.toString("utf8") });
|
|
});
|
|
proc.stderr.on("data", (chunk) => {
|
|
sendMessage({ type: "SHELL_OUTPUT", sessionId, data: chunk.toString("utf8") });
|
|
});
|
|
proc.on("close", (code) => {
|
|
sessions.delete(sessionId);
|
|
sendMessage({ type: "SHELL_EXIT", sessionId, code });
|
|
});
|
|
|
|
sessions.set(sessionId, proc);
|
|
sendMessage({ type: "SHELL_OPENED", sessionId });
|
|
};
|
|
|
|
const input = (payload = {}) => {
|
|
const proc = sessions.get(String(payload.sessionId));
|
|
if (!proc) {
|
|
return;
|
|
}
|
|
proc.stdin.write(String(payload.data || ""));
|
|
};
|
|
|
|
const close = (payload = {}) => {
|
|
const sessionId = String(payload.sessionId);
|
|
const proc = sessions.get(sessionId);
|
|
if (!proc) {
|
|
return;
|
|
}
|
|
proc.kill();
|
|
sessions.delete(sessionId);
|
|
};
|
|
|
|
return { open, input, close };
|
|
}
|
|
|
|
export async function handleAgentCommand(command, deps = {}) {
|
|
const fetchImpl = deps.fetchImpl || fetch;
|
|
switch (command.commandType) {
|
|
case "DISCOVER_SHELLY":
|
|
return { inventory: await discoverShellyDevices(command.payload || {}, fetchImpl) };
|
|
case "GET_RELAY_STATUS":
|
|
return await getRelayStatus(command.payload || {}, fetchImpl);
|
|
case "SET_RELAY_STATE":
|
|
return await setRelayState(command.payload || {}, fetchImpl);
|
|
case "RUN_UPDATE":
|
|
return await runUpdate(command.payload || {}, fetchImpl);
|
|
case "RESTART_AGENT":
|
|
return { restarted: true };
|
|
case "REBOOT_HOST":
|
|
return { rebooted: true };
|
|
default:
|
|
throw new Error(`Unsupported agent command: ${command.commandType}`);
|
|
}
|
|
}
|
|
|
|
export function buildHeartbeatPayload(config, extra = {}) {
|
|
return {
|
|
hostname: os.hostname(),
|
|
installed_version: config.installedVersion || DEFAULT_VERSION,
|
|
target_version: config.targetVersion || config.installedVersion || DEFAULT_VERSION,
|
|
status: extra.status || "ONLINE",
|
|
discovery_status: extra.discovery_status || "READY",
|
|
inventory: extra.inventory || [],
|
|
metadata: {
|
|
release_channel: config.releaseChannel || "stable",
|
|
...extra.metadata,
|
|
},
|
|
};
|
|
}
|
|
|
|
export async function sendHeartbeat(config, fetchImpl = fetch, extra = {}) {
|
|
if (!config.gatewayId || !config.agentToken) {
|
|
throw new Error("Gateway claim must complete before heartbeat");
|
|
}
|
|
return apiRequest(
|
|
config.apiUrl,
|
|
`/edge-agent/gateways/${config.gatewayId}/heartbeat`,
|
|
"POST",
|
|
{
|
|
agent_token: config.agentToken,
|
|
...buildHeartbeatPayload(config, extra),
|
|
},
|
|
fetchImpl
|
|
);
|
|
}
|
|
|
|
export function connectBroker(
|
|
config,
|
|
{
|
|
fetchImpl = fetch,
|
|
wsFactory = (url) => new WebSocket(url),
|
|
onOpen = null,
|
|
onClose = null,
|
|
onError = null,
|
|
} = {}
|
|
) {
|
|
const shell = createShellBridge((message) => {
|
|
if (socket.readyState === WebSocket.OPEN) {
|
|
socket.send(JSON.stringify(message));
|
|
}
|
|
});
|
|
const wsUrl = `${normalizeBrokerWsUrl(config.brokerUrl)}/ws/agent?gatewayId=${encodeURIComponent(config.gatewayId)}&token=${encodeURIComponent(config.agentToken)}`;
|
|
const socket = wsFactory(wsUrl);
|
|
|
|
socket.on("open", () => {
|
|
onOpen?.(socket);
|
|
});
|
|
|
|
socket.on("close", (...args) => {
|
|
onClose?.(...args);
|
|
});
|
|
|
|
socket.on("error", (error) => {
|
|
onError?.(error);
|
|
});
|
|
|
|
socket.on("message", async (raw) => {
|
|
const message = JSON.parse(raw.toString());
|
|
try {
|
|
if (message.type === "COMMAND") {
|
|
const payload = await handleAgentCommand({
|
|
commandType: message.commandType,
|
|
payload: message.payload || {},
|
|
}, { fetchImpl });
|
|
socket.send(JSON.stringify({
|
|
type: "COMMAND_RESULT",
|
|
commandId: message.commandId,
|
|
ok: true,
|
|
payload,
|
|
}));
|
|
return;
|
|
}
|
|
|
|
if (message.type === "OPEN_ROOT_SHELL") {
|
|
shell.open(message.payload || {});
|
|
} else if (message.type === "SHELL_INPUT") {
|
|
shell.input(message.payload || {});
|
|
} else if (message.type === "CLOSE_ROOT_SHELL") {
|
|
shell.close(message.payload || {});
|
|
}
|
|
} catch (error) {
|
|
socket.send(JSON.stringify({
|
|
type: "COMMAND_RESULT",
|
|
commandId: message.commandId,
|
|
ok: false,
|
|
error: error instanceof Error ? error.message : String(error),
|
|
}));
|
|
}
|
|
});
|
|
|
|
return socket;
|
|
}
|
|
|
|
export async function startAgent({ configPath, fetchImpl = fetch, wsFactory } = {}) {
|
|
if (!configPath) {
|
|
throw new Error("Missing --config path");
|
|
}
|
|
|
|
let config = await loadConfig(configPath);
|
|
config = await claimIfNeeded(config, configPath, fetchImpl);
|
|
const intervalMs = Number(config.heartbeatIntervalSeconds || 15) * 1000;
|
|
const reconnectDelayMs = Number(config.reconnectDelayMs || DEFAULT_RECONNECT_DELAY_MS);
|
|
|
|
let brokerConnected = false;
|
|
let socket = null;
|
|
let reconnectTimer = null;
|
|
let stopped = false;
|
|
|
|
const sendBrokerAwareHeartbeat = (extra = {}) =>
|
|
sendHeartbeat(config, fetchImpl, {
|
|
...buildBrokerHeartbeatState(brokerConnected),
|
|
...extra,
|
|
});
|
|
|
|
const scheduleReconnect = () => {
|
|
if (stopped || reconnectTimer) {
|
|
return;
|
|
}
|
|
|
|
reconnectTimer = setTimeout(() => {
|
|
reconnectTimer = null;
|
|
openBrokerConnection();
|
|
}, reconnectDelayMs);
|
|
};
|
|
|
|
const openBrokerConnection = () => {
|
|
if (stopped) {
|
|
return;
|
|
}
|
|
|
|
socket = connectBroker(config, {
|
|
fetchImpl,
|
|
wsFactory,
|
|
onOpen: () => {
|
|
brokerConnected = true;
|
|
sendBrokerAwareHeartbeat().catch(() => {});
|
|
},
|
|
onClose: () => {
|
|
brokerConnected = false;
|
|
sendBrokerAwareHeartbeat().catch(() => {});
|
|
scheduleReconnect();
|
|
},
|
|
onError: () => {
|
|
brokerConnected = false;
|
|
},
|
|
});
|
|
};
|
|
|
|
await sendBrokerAwareHeartbeat();
|
|
openBrokerConnection();
|
|
|
|
const timer = setInterval(() => {
|
|
sendBrokerAwareHeartbeat().catch(() => {});
|
|
}, intervalMs);
|
|
|
|
const stop = () => {
|
|
stopped = true;
|
|
clearInterval(timer);
|
|
if (reconnectTimer) {
|
|
clearTimeout(reconnectTimer);
|
|
reconnectTimer = null;
|
|
}
|
|
socket?.close?.();
|
|
};
|
|
|
|
return {
|
|
get socket() {
|
|
return socket;
|
|
},
|
|
timer,
|
|
config,
|
|
stop,
|
|
};
|
|
}
|
|
|
|
if (import.meta.url === `file://${process.argv[1]}`) {
|
|
const configFlagIndex = process.argv.indexOf("--config");
|
|
const configPath = configFlagIndex >= 0 ? process.argv[configFlagIndex + 1] : null;
|
|
startAgent({ configPath }).catch((error) => {
|
|
console.error(error instanceof Error ? error.message : String(error));
|
|
process.exitCode = 1;
|
|
});
|
|
}
|