zhaolei
c64e00d835
新增设备巡检模块核心功能，包括： - 巡检数据存储与调度系统 - 华为/H3C交换机及Linux服务器采集器 - 设备扩展配置与告警规则管理 - 主机详情页巡检配置入口 - 产品名称自定义设置功能 - 网络设备SSH exec通道兼容处理
249 lines
9.0 KiB
JavaScript
249 lines
9.0 KiB
JavaScript
/* eslint-disable no-undef */
|
||
|
||
const { isSshConnAlive, isTransportExecError } = require("./execConnHealth.cjs");
|
||
const { executeBoundedSshCommand } = require("../boundedSshExec.cjs");
|
||
|
||
function createExecOnSessionApi(ctx) {
|
||
with (ctx) {
|
||
const DEFAULT_LOCAL_EXEC_MAX_BUFFER = 10 * 1024 * 1024;
|
||
|
||
function normalizeExecMaxBuffer(value, fallback = DEFAULT_LOCAL_EXEC_MAX_BUFFER) {
|
||
const numeric = Number(value);
|
||
return Number.isFinite(numeric) && numeric > 0 ? Math.floor(numeric) : fallback;
|
||
}
|
||
|
||
function isExecMaxBufferError(err) {
|
||
const code = String(err?.code || "");
|
||
const message = String(err?.message || "");
|
||
return code === "ERR_CHILD_PROCESS_STDIO_MAXBUFFER" || /maxBuffer/i.test(message);
|
||
}
|
||
|
||
/** Serialize remote exec per session to avoid SSH channel storms. */
|
||
const execQueues = new Map();
|
||
|
||
function getSession(sessionId) {
|
||
return sessions?.get?.(sessionId) ?? null;
|
||
}
|
||
|
||
function enqueueExec(sessionId, task) {
|
||
let state = execQueues.get(sessionId);
|
||
if (!state) {
|
||
state = { running: false, pending: [] };
|
||
execQueues.set(sessionId, state);
|
||
}
|
||
return new Promise((resolve) => {
|
||
state.pending.push({ task, resolve });
|
||
void drainExecQueue(sessionId);
|
||
});
|
||
}
|
||
|
||
async function drainExecQueue(sessionId) {
|
||
const state = execQueues.get(sessionId);
|
||
if (!state || state.running) return;
|
||
state.running = true;
|
||
while (state.pending.length > 0) {
|
||
const job = state.pending.shift();
|
||
if (!job) continue;
|
||
try {
|
||
const result = await job.task();
|
||
job.resolve(result);
|
||
} catch (err) {
|
||
job.resolve({ success: false, error: err?.message || String(err) });
|
||
}
|
||
}
|
||
state.running = false;
|
||
if (state.pending.length === 0) {
|
||
execQueues.delete(sessionId);
|
||
}
|
||
}
|
||
|
||
async function ensureMoshCompanion(session, sessionId, event) {
|
||
if (session?.type !== "mosh" || typeof ensureMoshStatsConnection !== "function") {
|
||
return;
|
||
}
|
||
if (session.moshStatsConn && isSshConnAlive(session.moshStatsConn)) {
|
||
return;
|
||
}
|
||
if (session.moshStatsConn && !isSshConnAlive(session.moshStatsConn)) {
|
||
session.moshStatsConn = null;
|
||
}
|
||
if (!session.moshStatsConn && !session.moshStatsConnFailed) {
|
||
await ensureMoshStatsConnection(session, sessionId, event?.sender);
|
||
}
|
||
}
|
||
|
||
async function resolveExecConnection(session, sessionId, event) {
|
||
if (!session) return null;
|
||
|
||
await ensureMoshCompanion(session, sessionId, event);
|
||
|
||
const conn = session.conn || session.moshStatsConn;
|
||
if (!conn) return null;
|
||
|
||
if (!isSshConnAlive(conn)) {
|
||
if (session.moshStatsConn === conn) {
|
||
session.moshStatsConn = null;
|
||
await ensureMoshCompanion(session, sessionId, event);
|
||
return session.conn || session.moshStatsConn;
|
||
}
|
||
return null;
|
||
}
|
||
|
||
return conn;
|
||
}
|
||
|
||
function execOnConnection(conn, command, timeoutMs, execOptions = {}) {
|
||
const maxBuffer = normalizeExecMaxBuffer(execOptions.maxBuffer);
|
||
return executeBoundedSshCommand(conn, command, {
|
||
openingTimeoutMs: timeoutMs,
|
||
runTimeoutMs: timeoutMs,
|
||
maxOutputBytes: maxBuffer,
|
||
onStream(stream) {
|
||
if (typeof execOptions.stdin === "string") {
|
||
stream.write(execOptions.stdin);
|
||
stream.end();
|
||
}
|
||
},
|
||
}).then(
|
||
({ stdout, stderr, code }) => ({
|
||
success: true,
|
||
stdout,
|
||
stderr,
|
||
code: code ?? 0,
|
||
}),
|
||
(error) => ({
|
||
success: false,
|
||
error: error?.code === "SSH_EXEC_OUTPUT_LIMIT"
|
||
? "SSH command maxBuffer exceeded"
|
||
: error?.message || String(error),
|
||
stdout: "",
|
||
stderr: "",
|
||
code: 1,
|
||
}),
|
||
);
|
||
}
|
||
|
||
async function execOnSshSession(session, sessionId, command, timeoutMs, event, execOptions = {}, allowCompanionRetry = true) {
|
||
if (session?.type === "et") {
|
||
if (typeof execOnEtSession !== "function") {
|
||
return { success: false, error: "ET command executor unavailable" };
|
||
}
|
||
return execOnEtSession(session, command, timeoutMs, {
|
||
requireTrustedHost: true,
|
||
knownHosts: session.etStatsAuth?.knownHosts,
|
||
stdin: execOptions.stdin,
|
||
maxBuffer: execOptions.maxBuffer,
|
||
});
|
||
}
|
||
|
||
// [NETWORK-DEVICE GATE] 网络设备（华为 VRP、H3C Comware 等）SSH server 通常
|
||
// 不支持在同一连接上开多个 channel——exec channel 会触发整个连接被断开。
|
||
// 从 SSH banner (remoteSshVersion) 检测，如果是网络设备就直接拒绝。
|
||
const remoteVer = String(session?.remoteSshVersion || '').toLowerCase();
|
||
const isNetworkDeviceSsh = /^H3C[-_\s]/i.test(session?.remoteSshVersion || '')
|
||
|| /^Comware-/i.test(session?.remoteSshVersion || '')
|
||
|| /^HUAWEI[-_]/i.test(session?.remoteSshVersion || '')
|
||
|| /^VRP-/i.test(session?.remoteSshVersion || '')
|
||
|| (session?.remoteSshVersion === '-')
|
||
|| /^Cisco[-_]|^CiscoIOS_/i.test(session?.remoteSshVersion || '');
|
||
if (isNetworkDeviceSsh) {
|
||
return { success: false, error: "SSH exec channel not supported on network device (Huawei/H3C/VRP)" };
|
||
}
|
||
|
||
const conn = await resolveExecConnection(session, sessionId, event);
|
||
if (!conn) {
|
||
if (session?.type === "mosh" && !session.moshStatsAuth && !session.moshStatsConnFailed) {
|
||
return { success: false, pending: true, error: "Mosh handshake in progress" };
|
||
}
|
||
return { success: false, error: "Session not found or not connected" };
|
||
}
|
||
|
||
const result = await execOnConnection(conn, command, timeoutMs, execOptions);
|
||
if (
|
||
allowCompanionRetry
|
||
&& !result.success
|
||
&& session.moshStatsConn
|
||
&& isTransportExecError(result.error)
|
||
) {
|
||
session.moshStatsConn = null;
|
||
return execOnSshSession(session, sessionId, command, timeoutMs, event, execOptions, false);
|
||
}
|
||
return result;
|
||
}
|
||
|
||
async function execOnLocalMachine(command, timeoutMs, execOptions = {}) {
|
||
const { execFile } = require("node:child_process");
|
||
const platform = process.platform;
|
||
|
||
if (platform === "win32") {
|
||
return new Promise((resolve) => {
|
||
const child = execFile(
|
||
"powershell.exe",
|
||
["-NoProfile", "-NonInteractive", "-Command", command],
|
||
{ timeout: timeoutMs, maxBuffer: normalizeExecMaxBuffer(execOptions.maxBuffer) },
|
||
(err, stdout, stderr) => {
|
||
if (err && (isExecMaxBufferError(err) || !stdout)) {
|
||
resolve({ success: false, error: err.message || String(err), stdout: "", stderr: String(stderr || "") });
|
||
return;
|
||
}
|
||
resolve({ success: true, stdout: String(stdout || ""), stderr: String(stderr || ""), code: err?.code ?? 0 });
|
||
},
|
||
);
|
||
if (typeof execOptions.stdin === "string") {
|
||
child.stdin?.end(execOptions.stdin);
|
||
}
|
||
});
|
||
}
|
||
|
||
return new Promise((resolve) => {
|
||
const child = execFile(
|
||
"sh",
|
||
["-c", command],
|
||
{ timeout: timeoutMs, maxBuffer: normalizeExecMaxBuffer(execOptions.maxBuffer) },
|
||
(err, stdout, stderr) => {
|
||
if (err && (isExecMaxBufferError(err) || !stdout)) {
|
||
resolve({ success: false, error: err.message || String(err), stdout: "", stderr: String(stderr || "") });
|
||
return;
|
||
}
|
||
resolve({ success: true, stdout: String(stdout || ""), stderr: String(stderr || ""), code: err?.code ?? 0 });
|
||
},
|
||
);
|
||
if (typeof execOptions.stdin === "string") {
|
||
child.stdin?.end(execOptions.stdin);
|
||
}
|
||
});
|
||
}
|
||
|
||
async function execOnSessionInner(event, sessionId, command, timeoutMs = 8000, execOptions = {}) {
|
||
const session = getSession(sessionId);
|
||
if (!session) {
|
||
execQueues.delete(sessionId);
|
||
return { success: false, error: "Session not found" };
|
||
}
|
||
|
||
if (session.protocol === "local" || session.type === "local") {
|
||
return execOnLocalMachine(command, timeoutMs, execOptions);
|
||
}
|
||
|
||
if (session.conn || session.type === "mosh" || session.type === "et") {
|
||
return execOnSshSession(session, sessionId, command, timeoutMs, event, execOptions);
|
||
}
|
||
|
||
return { success: false, error: "Session not supported for system management" };
|
||
}
|
||
|
||
async function execOnSession(event, sessionId, command, timeoutMs = 8000, execOptions = {}) {
|
||
return enqueueExec(sessionId, () => execOnSessionInner(event, sessionId, command, timeoutMs, execOptions));
|
||
}
|
||
|
||
function isLocalSession(sessionId) {
|
||
const session = getSession(sessionId);
|
||
return !!(session?.protocol === "local" || session?.type === "local");
|
||
}
|
||
|
||
return { execOnSession, execOnLocalMachine, isLocalSession, getSession };
|
||
}
|
||
}
|
||
|
||
module.exports = { createExecOnSessionApi };
|