import ssh2 from "ssh2"; import { StringDecoder } from "node:string_decoder"; export const unescapeQuery = (value) => String(value).replace( /\\([\\spnrtvf/|])/g, (_, c) => ({ "\\": "\\", s: " ", p: "|", n: "\n", r: "\r", t: "\t", v: "\v", f: "\f", "/": "/", "|": "|", })[c], ); export const escapeQuery = (value) => String(value).replace( /[\\ |\n\r\t]/g, (c) => ({ "\\": "\\\\", " ": "\\s", "|": "\\p", "\n": "\\n", "\r": "\\r", "\t": "\\t", })[c], ); export function parseQuery(line) { return String(line) .split("|") .map((row) => Object.fromEntries( row .trim() .split(/\s+/) .filter((v) => v.includes("=")) .map((pair) => { const i = pair.indexOf("="); return [pair.slice(0, i), unescapeQuery(pair.slice(i + 1))]; }), ), ); } export function createEventListener( connection, onEvent, onState, { Client = ssh2.Client, channelId = 0 } = {}, ) { let lastError = "", decoder = new StringDecoder("utf8"), client, stream, buffer = "", active, pending = [], closed = false, retry, ready = false, connecting = false, fingerprint = connection.sshFingerprint || ""; const state = (status, error) => { if (closed) return; if (error) lastError = error; if (status === "connected") lastError = ""; onState({ status, error: error || lastError, channelId, fingerprint, updatedAt: new Date().toISOString(), }); }; const rejectPending = (error) => { if (active) { clearTimeout(active.timer); active.reject(error); active = null; } for (const entry of pending) entry.reject(error); pending = []; }; function pump() { if (active || !stream || !pending.length) return; active = pending.shift(); active.timer = setTimeout(() => { const error = new Error("SSH Query não respondeu em 10 segundos."); rejectPending(error); client?.destroy(); }, 10000); stream.write(active.command + "\n"); } function send(command) { return new Promise((resolve, reject) => { pending.push({ command, resolve, reject, rows: [] }); pump(); }); } function data(chunk) { buffer += decoder.write(chunk); if (buffer.length > 1000000) { client.destroy(); return; } let index; while ((index = buffer.indexOf("\n")) >= 0) { const line = buffer.slice(0, index).replace(/\r/g, "").trim(); buffer = buffer.slice(index + 1); if (!line) continue; if (/^notify\w+\s/.test(line)) { const split = line.indexOf(" "); for (const event of parseQuery(line.slice(split + 1))) onEvent(line.slice(0, split), { ...event, __listenerChannel: String(channelId), }); continue; } if (/^error id=/.test(line)) { const result = parseQuery(line)[0]; if (active) { const entry = active; clearTimeout(entry.timer); active = null; if (Number(result.id) === 0) entry.resolve(entry.rows.flatMap(parseQuery)); else entry.reject( new Error( `SSH Query ${result.id}: ${result.msg || "pedido recusado"}`, ), ); pump(); } continue; } if (active && line.includes("=") && !/^Welcome/i.test(line)) active.rows.push(line); } } function reconnect() { ready = false; connecting = false; if (closed) return; state("disconnected"); clearTimeout(retry); retry = setTimeout(connect, 10000); retry.unref?.(); } function connect() { if (closed || connecting) return; connecting = true; state("connecting"); client = new Client(); client .on("error", (error) => { state("error", error.message); rejectPending(error); }) .on("close", () => { stream = null; rejectPending(new Error("Ligação SSH fechada.")); reconnect(); }) .on("ready", () => client.shell(false, async (error, channel) => { if (error) { state("error", error.message); client.end(); return; } stream = channel; buffer = ""; decoder = new StringDecoder("utf8"); channel.on("data", data); channel.on("close", () => { if (!closed) client.end(); }); channel.on("error", (error) => { state("error", error.message); client.destroy(); }); try { await send(`use sid=${connection.serverId}`); await send( `clientupdate client_nickname=XanaX-Botflow-${connection.serverId}-${channelId}`, ); if (channelId) { const who = (await send("whoami"))[0]; await send( `clientmove clid=${who.client_id || who.clid} cid=${channelId}`, ); await send("servernotifyregister event=textchannel"); } else { for (const event of [ "server", "channel", "textprivate", "textserver", ]) await send( `servernotifyregister event=${event}${event === "channel" ? " id=0" : ""}`, ); } ready = true; connecting = false; state("connected"); } catch (error) { state("error", error.message); client.end(); } }), ); client.connect({ host: connection.sshHost || new URL(connection.url).hostname, port: Number(connection.sshPort || 10022), username: connection.sshUsername, password: connection.sshPassword, readyTimeout: 10000, keepaliveInterval: 15000, keepaliveCountMax: 3, hostHash: "sha256", hostVerifier: (hash) => { if (fingerprint && fingerprint !== hash) { state( "error", "A impressão digital SSH mudou. Confirma o servidor na configuração da ligação.", ); return false; } fingerprint = hash; return true; }, }); } connect(); return { close() { closed = true; clearTimeout(retry); rejectPending(new Error("Escuta de eventos terminada.")); client?.destroy(); }, get ready() { return ready; }, }; }