563 lines
28 KiB
JavaScript
563 lines
28 KiB
JavaScript
'use strict';
|
||
|
||
const fs = require('node:fs');
|
||
const path = require('node:path');
|
||
const os = require('node:os');
|
||
const net = require('node:net');
|
||
const crypto = require('node:crypto');
|
||
const { spawnSync } = require('node:child_process');
|
||
|
||
const VERSION = '0.6.8';
|
||
const CONTROLLER = String(process.env.NG_CONTROLLER || '').replace(/\/+$/, '');
|
||
const AGENT_KEY = process.env.NG_AGENT_KEY || '';
|
||
const SINGBOX_BIN = process.env.NG_SINGBOX_BIN || '/usr/local/bin/nexusgate-sing-box';
|
||
const SINGBOX_STATS_BIN = process.env.NG_SINGBOX_STATS_BIN || '/usr/local/bin/nexusgate-sing-box-stats';
|
||
const ROOT = process.env.NG_CONFIG_DIR || '/etc/nexusgate/agent';
|
||
const RESOURCE_DIR = path.join(ROOT, 'resources');
|
||
const SINGBOX_CONFIG = process.env.NG_SINGBOX_CONFIG || '/etc/nexusgate/sing-box/config.json';
|
||
const SINGBOX_LOG = process.env.NG_SINGBOX_LOG || '/var/log/nexusgate/sing-box-access.log';
|
||
const SINGBOX_CURSOR = process.env.NG_SINGBOX_CURSOR || '/etc/nexusgate/sing-box-cursor.json';
|
||
const KEY_FILE = process.env.NG_KEY_FILE || '/etc/nexusgate/keys.json';
|
||
const POLL_MS = Math.max(3, Number(process.env.NG_POLL_SECONDS || 8)) * 1000;
|
||
let lastEngineError = '';
|
||
let certificateCache = { fingerprint:'', checkedAt:0, domains:[] };
|
||
|
||
function sleep(ms) { return new Promise((resolve) => setTimeout(resolve, ms)); }
|
||
function log(...args) { console.log(new Date().toISOString(), ...args); }
|
||
function safeError(value) {
|
||
return String(value || '').replace(/\bPrivate\s*Key\s*:\s*[^\s,;]+/gi, 'PrivateKey: [REDACTED]')
|
||
.replace(/\bBearer\s+[^\s,;]+/gi, 'Bearer [REDACTED]');
|
||
}
|
||
|
||
async function request(endpoint, options = {}) {
|
||
const controller = new AbortController();
|
||
const timeout = setTimeout(() => controller.abort(), 20000);
|
||
try {
|
||
const response = await fetch(`${CONTROLLER}${endpoint}`, {
|
||
...options,
|
||
headers: { authorization: `Bearer ${AGENT_KEY}`, 'content-type': 'application/json', ...(options.headers || {}) },
|
||
signal: controller.signal
|
||
});
|
||
const body = await response.json().catch(() => ({}));
|
||
if (!response.ok) throw new Error(body.message || `Controller returned ${response.status}`);
|
||
return body;
|
||
} finally { clearTimeout(timeout); }
|
||
}
|
||
|
||
function ensureDirectories() {
|
||
fs.mkdirSync(RESOURCE_DIR, { recursive: true, mode: 0o700 });
|
||
fs.mkdirSync('/var/log/nexusgate', { recursive: true, mode: 0o750 });
|
||
}
|
||
|
||
function loadKeys() {
|
||
try { return JSON.parse(fs.readFileSync(KEY_FILE, 'utf8')); } catch (error) { if (error.code === 'ENOENT') return {}; throw error; }
|
||
}
|
||
|
||
function saveKeys(keys) {
|
||
const temp = `${KEY_FILE}.tmp`;
|
||
fs.writeFileSync(temp, `${JSON.stringify(keys, null, 2)}\n`, { mode: 0o600 });
|
||
fs.renameSync(temp, KEY_FILE); fs.chmodSync(KEY_FILE, 0o600);
|
||
}
|
||
|
||
function realityKey(keyId) {
|
||
const keys = loadKeys();
|
||
if (keys[keyId]) return keys[keyId];
|
||
const kp = crypto.generateKeyPairSync('x25519');
|
||
const priv = kp.privateKey.export({ format: 'jwk' });
|
||
const pub = kp.publicKey.export({ format: 'jwk' });
|
||
if (!priv.d || !pub.x) throw new Error('X25519 密钥生成失败');
|
||
// JWK 用 base64url，sing-box 要标准 base64，用 Buffer 转
|
||
const b64urlToB64 = (s) => Buffer.from(s, 'base64url').toString('base64');
|
||
keys[keyId] = { privateKey: b64urlToB64(priv.d), publicKey: b64urlToB64(pub.x), createdAt: new Date().toISOString() };
|
||
saveKeys(keys);
|
||
return keys[keyId];
|
||
}
|
||
|
||
function materialize(value, artifacts) {
|
||
if (Array.isArray(value)) return value.map((item) => materialize(item, artifacts));
|
||
if (value && typeof value === 'object') return Object.fromEntries(Object.entries(value).map(([key, item]) => [key, materialize(item, artifacts)]));
|
||
if (typeof value !== 'string') return value;
|
||
const match = value.match(/^\$\{REALITY_PRIVATE:(.+)}$/);
|
||
if (!match) return value;
|
||
const pair = realityKey(match[1]);
|
||
artifacts.realityPublicKey = pair.publicKey;
|
||
return pair.privateKey;
|
||
}
|
||
|
||
function readResources() {
|
||
return fs.readdirSync(RESOURCE_DIR).filter((name) => name.endsWith('.json')).sort().map((name) => JSON.parse(fs.readFileSync(path.join(RESOURCE_DIR, name), 'utf8')));
|
||
}
|
||
|
||
// 云机器的公网 IP 通常是 NAT 地址，不在本地网卡上；直接 bind 会
|
||
// "cannot assign requested address" 导致引擎起不来。非本地地址回退到对应栈的通配符。
|
||
function normalizeInboundListen(inbound) {
|
||
if (!inbound || typeof inbound !== 'object') return inbound;
|
||
const text = typeof inbound.listen === 'string' ? inbound.listen.trim() : '';
|
||
if (!text || !net.isIP(text) || text === '0.0.0.0' || text === '::') return inbound;
|
||
let local = false;
|
||
for (const list of Object.values(os.networkInterfaces())) {
|
||
for (const info of list || []) if (info && info.address === text) { local = true; break; }
|
||
if (local) break;
|
||
}
|
||
if (local) return inbound;
|
||
const fallback = net.isIPv6(text) ? '::' : '0.0.0.0';
|
||
log(`Listen ${text} 不是本机地址，回退为 ${fallback}`);
|
||
return { ...inbound, listen: fallback };
|
||
}
|
||
|
||
function combinedSingBoxConfig(resources) {
|
||
return {
|
||
log: { level: 'info', output: SINGBOX_LOG, timestamp: true },
|
||
inbounds: resources.flatMap((item) => item.inbounds || []).map(normalizeInboundListen),
|
||
outbounds: [{ type: 'direct', tag: 'ng-fallback-direct' }, ...resources.flatMap((item) => item.outbounds || [])],
|
||
route: { rules: resources.flatMap((item) => item.routingRules || []), final: 'ng-fallback-direct' },
|
||
experimental: { v2ray_api: { listen: `127.0.0.1:${process.env.NG_SINGBOX_API_PORT || 10086}`,
|
||
stats: { enabled: true, inbounds: resources.flatMap((item) => item.inbounds || []).map((item) => item.tag),
|
||
// sing-box 只对显式列出的用户维护 user>>> 计数器；AnyTLS/SS 用 name，VLESS 用 name，SOCKS5 用 username。
|
||
users: resources.flatMap((item) => item.inbounds || []).flatMap((item) => item.users || []).map((item) => item.name || item.username).filter(Boolean) } } }
|
||
};
|
||
}
|
||
|
||
function run(command, args, timeout = 30000) {
|
||
const result = spawnSync(command, args, { encoding: 'utf8', timeout });
|
||
if (result.status !== 0) throw new Error(`${command} ${args.join(' ')} failed: ${safeError(result.stderr || result.stdout || '').slice(0, 1200)}`);
|
||
return result.stdout;
|
||
}
|
||
|
||
function serviceCommand(action, service) {
|
||
if (fs.existsSync('/run/systemd/system')) return run('systemctl', [action, `${service}.service`]);
|
||
return run('rc-service', [service, action]);
|
||
}
|
||
|
||
function activateSingBox() {
|
||
const resources = readResources();
|
||
if (!resources.length) {
|
||
// 无资源时必须确保引擎停下（不能只在配置文件存在时停，否则全新安装会陷入重启循环）
|
||
try { serviceCommand('stop', 'nexusgate-sing-box'); } catch { /* 已经停了 */ }
|
||
if (fs.existsSync(SINGBOX_CONFIG)) fs.rmSync(SINGBOX_CONFIG, { force: true });
|
||
return;
|
||
}
|
||
if (!fs.existsSync(SINGBOX_BIN) || !fs.existsSync(SINGBOX_STATS_BIN)) throw new Error('sing-box 及统计组件未安装；请在入口机器运行 ng-agent engine install');
|
||
const candidate = `${SINGBOX_CONFIG}.candidate.json`;
|
||
const previous = `${SINGBOX_CONFIG}.previous`;
|
||
const next = `${JSON.stringify(combinedSingBoxConfig(resources), null, 2)}\n`;
|
||
if (fs.existsSync(SINGBOX_CONFIG) && fs.readFileSync(SINGBOX_CONFIG, 'utf8') === next && serviceHealth('nexusgate-sing-box').status === 'ready') return;
|
||
fs.mkdirSync(path.dirname(SINGBOX_CONFIG), { recursive: true, mode: 0o700 });
|
||
fs.writeFileSync(candidate, next, { mode: 0o600 });
|
||
try { run(SINGBOX_BIN, ['check', '-c', candidate]); }
|
||
catch (error) { fs.rmSync(candidate, { force: true }); throw error; }
|
||
if (fs.existsSync(SINGBOX_CONFIG)) fs.copyFileSync(SINGBOX_CONFIG, previous);
|
||
fs.renameSync(candidate, SINGBOX_CONFIG);
|
||
try { serviceCommand('restart', 'nexusgate-sing-box'); }
|
||
catch (error) {
|
||
if (fs.existsSync(previous)) fs.copyFileSync(previous, SINGBOX_CONFIG);
|
||
else fs.rmSync(SINGBOX_CONFIG, { force: true });
|
||
if (fs.existsSync(SINGBOX_CONFIG)) { try { serviceCommand('restart', 'nexusgate-sing-box'); } catch { /* preserve original error */ } }
|
||
throw error;
|
||
}
|
||
}
|
||
|
||
function safeResourcePath(resourceId) {
|
||
if (!/^[a-zA-Z0-9_-]{8,100}$/.test(String(resourceId))) throw new Error('Invalid resource id');
|
||
return path.join(RESOURCE_DIR, `${resourceId}.json`);
|
||
}
|
||
|
||
async function flushUsageBeforeChange() {
|
||
const live = readResources().some((resource) =>
|
||
resource.meta?.metricsTag && resource.meta.kind === 'node');
|
||
if (!live || serviceHealth('nexusgate-sing-box').status !== 'ready') return;
|
||
const report = queryUsage();
|
||
if (report.engineErrors.includes('sing-box')) throw new Error('sing-box 重载前无法读取入口计数；请先运行 ng-agent doctor');
|
||
// Await the controller's durable write before a restart destroys the
|
||
// in-memory counters. The periodic loop can report concurrently; the
|
||
// controller ignores repeated and out-of-order samples for one process.
|
||
await request('/api/agent/usage', { method:'POST', body:JSON.stringify(report) });
|
||
}
|
||
|
||
async function flushAllUsageBeforeRestart() {
|
||
await flushUsageBeforeChange();
|
||
}
|
||
|
||
async function applyResource(payload) {
|
||
if (!payload || !payload.resource || !payload.resource.id) throw new Error('Missing resource payload');
|
||
const artifacts = {};
|
||
const resource = materialize(payload.resource, artifacts);
|
||
if (resource.engine && resource.engine !== 'sing-box')
|
||
throw new Error(`Unsupported engine: ${resource.engine}（Xray 已弃用，只支持 sing-box）`);
|
||
for (const inbound of resource.inbounds || []) {
|
||
// Hysteria 2 无认证账户时拒绝启动未受保护的入口。
|
||
if (inbound.type === 'hysteria2') {
|
||
const users = inbound.users || [];
|
||
if (!users.length || users.some((user) => !user.password))
|
||
throw new Error('Hysteria 2 入口没有认证账户；拒绝启动未受保护的入口');
|
||
}
|
||
const tls = inbound.tls;
|
||
if (!tls) continue;
|
||
if (tls.reality && tls.reality.enabled) continue; // Reality 无证书，不走证书检查
|
||
const domain = tls.server_name;
|
||
if (!/^(?:[a-z0-9-]+\.)+[a-z]{2,63}$/.test(domain || '')) throw new Error('TLS 域名无效');
|
||
const expected = `/etc/nexusgate/tls/${domain}/`;
|
||
const cert = { certificateFile: tls.certificate_path, keyFile: tls.key_path };
|
||
if (!cert.certificateFile || !cert.keyFile)
|
||
throw new Error(`TLS certificate for ${domain} is missing; run ng-agent cert issue ${domain} EMAIL or ng-agent cert import`);
|
||
if (!cert.certificateFile.startsWith(expected) || !cert.keyFile.startsWith(expected) ||
|
||
!fs.existsSync(cert.certificateFile) || !fs.existsSync(cert.keyFile)) {
|
||
throw new Error(`TLS certificate for ${domain} is missing; run ng-agent cert issue ${domain} EMAIL or ng-agent cert import`);
|
||
}
|
||
run('openssl', ['x509', '-in', cert.certificateFile, '-noout', '-checkend', '86400']);
|
||
const nameCheck = run('openssl', ['x509', '-in', cert.certificateFile, '-noout', '-checkhost', domain]);
|
||
if (!nameCheck.includes('does match')) throw new Error(`TLS certificate does not match ${domain}`);
|
||
}
|
||
const target = safeResourcePath(resource.id);
|
||
const backup = fs.existsSync(target) ? fs.readFileSync(target) : null;
|
||
await flushUsageBeforeChange();
|
||
fs.writeFileSync(target, `${JSON.stringify(resource, null, 2)}\n`, { mode: 0o600 });
|
||
try {
|
||
activateSingBox();
|
||
lastEngineError = '';
|
||
}
|
||
catch (error) {
|
||
if (backup) fs.writeFileSync(target, backup, { mode: 0o600 }); else fs.rmSync(target, { force: true });
|
||
try { activateSingBox(); } catch { /* preserve original error */ }
|
||
throw error;
|
||
}
|
||
return { artifacts };
|
||
}
|
||
|
||
async function deleteResource(payload) {
|
||
const target = safeResourcePath(payload.resourceId);
|
||
if (!fs.existsSync(target)) return { alreadyAbsent: true };
|
||
const backup = fs.readFileSync(target);
|
||
await flushUsageBeforeChange();
|
||
fs.rmSync(target);
|
||
try { activateSingBox(); }
|
||
catch (error) { fs.writeFileSync(target, backup, { mode: 0o600 }); throw error; }
|
||
return { removed: true };
|
||
}
|
||
|
||
async function execute(job) {
|
||
log('Executing', job.id, job.action);
|
||
try {
|
||
let result;
|
||
if (job.action === 'apply_resource') result = await applyResource(job.payload);
|
||
else if (job.action === 'delete_resource') result = await deleteResource(job.payload);
|
||
else throw new Error(`Unsupported job action: ${job.action}`);
|
||
await request(`/api/agent/jobs/${encodeURIComponent(job.id)}/complete`, { method: 'POST', body: JSON.stringify({ success: true, result }) });
|
||
log('Completed', job.id);
|
||
} catch (error) {
|
||
log('Failed', job.id, safeError(error.message));
|
||
try { await request(`/api/agent/jobs/${encodeURIComponent(job.id)}/complete`, { method: 'POST', body: JSON.stringify({ success: false, error: safeError(error.message) }) }); }
|
||
catch (reportError) { log('Unable to report failure', reportError.message); }
|
||
}
|
||
}
|
||
|
||
function systemInfo() {
|
||
const cpus = os.cpus();
|
||
let memoryAvailable = os.freemem();
|
||
try {
|
||
const match = fs.readFileSync('/proc/meminfo', 'utf8').match(/^MemAvailable:\s+(\d+) kB/m);
|
||
if (match) memoryAvailable = Number(match[1]) * 1024;
|
||
} catch { /* Other supported systems still report free memory. */ }
|
||
let diskTotal = null, diskAvailable = null;
|
||
try {
|
||
const disk = fs.statfsSync('/');
|
||
diskTotal = disk.blocks * disk.bsize;
|
||
diskAvailable = disk.bavail * disk.bsize;
|
||
} catch { /* Older Node builds cannot report filesystem capacity. */ }
|
||
return {
|
||
hostname: os.hostname(), platform: os.platform(), release: os.release(), arch: os.arch(),
|
||
uptimeSeconds: Math.round(os.uptime()), load1: Number(os.loadavg()[0].toFixed(2)),
|
||
cpuCount: cpus.length, memoryTotal: os.totalmem(), memoryFree: os.freemem(),
|
||
memoryAvailable, diskTotal, diskAvailable
|
||
};
|
||
}
|
||
|
||
function engineHealth() {
|
||
const resources = readResources();
|
||
const installed = fs.existsSync(SINGBOX_BIN) && fs.existsSync(SINGBOX_STATS_BIN);
|
||
const service = serviceHealth('nexusgate-sing-box');
|
||
const result = {
|
||
status: installed ? service.status : 'error',
|
||
detail: installed ? service.detail : 'sing-box 及统计组件未安装；请在入口机器运行 ng-agent engine install',
|
||
singBoxInstalled: installed,
|
||
certificates: installedCertificates(),
|
||
singBoxStatus: service.status,
|
||
};
|
||
if (result.status === 'ready' && resources.length) {
|
||
try {
|
||
const missing = missingHysteriaListeners(resources);
|
||
if (missing.length) {
|
||
result.status = 'error';
|
||
result.detail = `Hysteria 2 UDP 未监听：${missing.join(', ')}`;
|
||
}
|
||
} catch (error) {
|
||
result.status = 'error';
|
||
result.detail = safeError(error.message);
|
||
}
|
||
}
|
||
return result;
|
||
}
|
||
|
||
function installedCertificates() {
|
||
const root = process.env.NG_TLS_DIR || '/etc/nexusgate/tls';
|
||
let domains;
|
||
try { domains = fs.readdirSync(root).slice(0, 50); } catch { return []; }
|
||
const fingerprint = JSON.stringify(domains.map((domain) => {
|
||
try {
|
||
const cert = fs.statSync(path.join(root, domain, 'fullchain.pem'));
|
||
const key = fs.statSync(path.join(root, domain, 'privkey.pem'));
|
||
return [domain, cert.mtimeMs, cert.size, key.mtimeMs, key.size];
|
||
} catch { return [domain]; }
|
||
}));
|
||
if (certificateCache.fingerprint === fingerprint && Date.now() - certificateCache.checkedAt < 300000)
|
||
return [...certificateCache.domains];
|
||
const valid = domains.filter((domain) => {
|
||
if (!/^(?:[a-z0-9-]+\.)+[a-z]{2,63}$/.test(domain)) return false;
|
||
const cert = path.join(root, domain, 'fullchain.pem');
|
||
const key = path.join(root, domain, 'privkey.pem');
|
||
if (!fs.existsSync(cert) || !fs.existsSync(key)) return false;
|
||
const expiry = spawnSync('openssl', ['x509', '-in', cert, '-noout', '-checkend', '86400'],
|
||
{ encoding:'utf8', timeout:3000 });
|
||
const host = spawnSync('openssl', ['x509', '-in', cert, '-noout', '-checkhost', domain],
|
||
{ encoding:'utf8', timeout:3000 });
|
||
if (expiry.status !== 0 || host.status !== 0 || !/does match/.test(host.stdout)) return false;
|
||
const verify = spawnSync('openssl', ['verify', '-verify_hostname', domain, '-untrusted', cert, cert],
|
||
{ encoding:'utf8', timeout:3000 });
|
||
if (verify.status !== 0) return false;
|
||
const certKey = spawnSync('openssl', ['x509', '-in', cert, '-pubkey', '-noout'], { encoding:'utf8', timeout:3000 });
|
||
const privateKey = spawnSync('openssl', ['pkey', '-in', key, '-pubout'], { encoding:'utf8', timeout:3000 });
|
||
return certKey.status === 0 && privateKey.status === 0 && certKey.stdout.trim() === privateKey.stdout.trim();
|
||
});
|
||
certificateCache = { fingerprint, checkedAt:Date.now(), domains:valid };
|
||
return [...valid];
|
||
}
|
||
|
||
function serviceHealth(service) {
|
||
const result = fs.existsSync('/run/systemd/system')
|
||
? spawnSync('systemctl', ['is-active', `${service}.service`], { encoding:'utf8', timeout:3000 })
|
||
: spawnSync('rc-service', [service, 'status'], { encoding:'utf8', timeout:3000 });
|
||
return { status: result.status === 0 ? 'ready' : 'error', detail: (result.status === 0 ? result.stdout : lastEngineError || result.stderr || result.stdout || '').trim().slice(0, 400) };
|
||
}
|
||
|
||
async function heartbeat() {
|
||
await request('/api/agent/heartbeat', { method: 'POST', body: JSON.stringify({ version: VERSION, system: systemInfo(), engine: engineHealth() }) });
|
||
try { fs.writeFileSync('/etc/nexusgate/last-heartbeat.json', `${JSON.stringify({ at:new Date().toISOString() })}\n`, { mode:0o600 }); }
|
||
catch (error) { log('Heartbeat reached controller but readiness file failed', error.message); }
|
||
}
|
||
|
||
function parseUsageStats(output) {
|
||
const parsed = JSON.parse(output);
|
||
// 零流量时 Go 端输出 {"stat":null}，视为无数据而非格式错误。
|
||
const statList = parsed && typeof parsed === 'object' ? (parsed.stat == null ? [] : parsed.stat) : null;
|
||
if (!Array.isArray(statList))
|
||
throw new Error('统计接口返回的 JSON stat 格式无效');
|
||
const inbounds = new Map(), users = new Map();
|
||
for (const stat of statList) {
|
||
const inboundMatch = /^inbound>>>(.+)>>>traffic>>>(uplink|downlink)$/.exec(stat.name || '');
|
||
const userMatch = inboundMatch ? null : /^user>>>(.+)>>>traffic>>>(uplink|downlink)$/.exec(stat.name || '');
|
||
if (!inboundMatch && !userMatch) continue;
|
||
// Protobuf JSON omits a scalar with its default value (zero). The engine can
|
||
// therefore return a named counter without a `value` field until traffic.
|
||
const value = Object.hasOwn(stat, 'value') ? Number(stat.value) : 0;
|
||
if (!Number.isSafeInteger(value) || value < 0 || stat.value === null || stat.value === '') {
|
||
const reason = typeof stat.value === 'string' && /^-?\d+$/.test(stat.value)
|
||
? (stat.value.startsWith('-') ? '负数' : '超出安全整数范围') : `格式 ${typeof stat.value}`;
|
||
throw new Error(`入口计数异常（${reason}）：${stat.name}；请运行 ng-agent doctor 查看原始统计`);
|
||
}
|
||
const bucket = inboundMatch ? inbounds : users;
|
||
const key = (inboundMatch || userMatch)[1];
|
||
const direction = (inboundMatch || userMatch)[2];
|
||
let entry = bucket.get(key);
|
||
if (!entry) { entry = { uplink: 0, downlink: 0 }; bucket.set(key, entry); }
|
||
entry[direction] = value;
|
||
}
|
||
return {
|
||
inbounds: [...inbounds].map(([tag, counters]) => ({ tag, ...counters })),
|
||
users: [...users].map(([name, counters]) => ({ name, ...counters })),
|
||
};
|
||
}
|
||
|
||
function serviceEpoch(service = 'nexusgate-sing-box') {
|
||
let pid;
|
||
if (fs.existsSync('/run/systemd/system')) {
|
||
const result = spawnSync('systemctl', ['show', '-p', 'MainPID', '--value', `${service}.service`], { encoding:'utf8', timeout:3000 });
|
||
if (result.status === 0) pid = Number(result.stdout.trim());
|
||
} else {
|
||
try { pid = Number(fs.readFileSync(`/run/${service}.pid`, 'utf8').trim()); } catch { /* service is not running */ }
|
||
}
|
||
if (!Number.isSafeInteger(pid) || pid <= 0) return null;
|
||
try {
|
||
// PID alone can be reused; the Linux process start tick distinguishes a new sing-box instance.
|
||
const stat = fs.readFileSync(`/proc/${pid}/stat`, 'utf8');
|
||
const startTick = stat.slice(stat.lastIndexOf(') ') + 2).trim().split(/\s+/)[19];
|
||
return startTick ? `${pid}:${startTick}` : null;
|
||
} catch { return null; }
|
||
}
|
||
|
||
function udpPortsForPid(pid) {
|
||
const sockets = new Set();
|
||
for (const fd of fs.readdirSync(`/proc/${pid}/fd`)) {
|
||
try {
|
||
const match = fs.readlinkSync(`/proc/${pid}/fd/${fd}`).match(/^socket:\[(\d+)\]$/);
|
||
if (match) sockets.add(match[1]);
|
||
} catch { /* a descriptor closed during inspection */ }
|
||
}
|
||
const ports = new Set();
|
||
for (const table of ['/proc/net/udp', '/proc/net/udp6']) {
|
||
const rows = fs.readFileSync(table, 'utf8').trim().split('\n').slice(1);
|
||
for (const row of rows) {
|
||
const fields = row.trim().split(/\s+/);
|
||
if (fields[3] === '07' && sockets.has(fields[9]))
|
||
ports.add(parseInt(fields[1].split(':').at(-1), 16));
|
||
}
|
||
}
|
||
return ports;
|
||
}
|
||
|
||
function missingHysteriaListeners(resources) {
|
||
const expected = [...new Set(resources.flatMap((item) => item.inbounds || [])
|
||
.filter((item) => item.type === 'hysteria2').map((item) => item.listen_port))];
|
||
if (!expected.length) return [];
|
||
const epoch = serviceEpoch('nexusgate-sing-box');
|
||
if (!epoch) throw new Error('无法确认 sing-box 进程，无法验证 Hysteria 2 UDP 监听');
|
||
const actual = udpPortsForPid(Number(epoch.split(':')[0]));
|
||
return expected.filter((port) => !actual.has(port));
|
||
}
|
||
|
||
function assertHysteriaListeners(resources) {
|
||
if (!resources.some((item) => (item.inbounds || []).some((inbound) => inbound.protocol === 'hysteria'))) return;
|
||
for (let attempt = 0; attempt < 10; attempt++) {
|
||
try {
|
||
const missing = missingHysteriaListeners(resources);
|
||
if (!missing.length) return;
|
||
} catch { /* systemd may not have published its MainPID yet */ }
|
||
if (attempt < 9) spawnSync('sleep', ['0.2']);
|
||
}
|
||
const missing = missingHysteriaListeners(resources);
|
||
throw new Error(`Hysteria 2 未监听 UDP ${missing.join(', ')}；检查 sing-box 配置与本机端口冲突`);
|
||
}
|
||
|
||
function queryUsage() {
|
||
// Count client ingress once; counting the exit as well doubles forwarded traffic.
|
||
const resources = readResources().filter((resource) => resource.meta && resource.meta.metricsTag &&
|
||
resource.meta.kind === 'node');
|
||
const samples = [], errors = [], engineErrors = [];
|
||
if (!resources.length) return { samples, error: null, engineErrors };
|
||
const service = 'nexusgate-sing-box';
|
||
const before = serviceEpoch(service);
|
||
const port = process.env.NG_SINGBOX_API_PORT || 10086;
|
||
const result = spawnSync(SINGBOX_STATS_BIN, [`--server=127.0.0.1:${port}`], { encoding: 'utf8', timeout: 15000 });
|
||
const after = serviceEpoch(service);
|
||
let failed = false;
|
||
if (before && after && before !== after) { errors.push('sing-box restarted during statistics query'); engineErrors.push('sing-box'); failed = true; }
|
||
else if (result.status !== 0) { errors.push(`sing-box statsquery failed: ${safeError(result.stderr || result.error?.message || result.stdout).slice(0, 160)}`); engineErrors.push('sing-box'); failed = true; }
|
||
const parsed = { inbounds: [], users: [] };
|
||
if (!failed) {
|
||
try {
|
||
const part = parseUsageStats(result.stdout);
|
||
parsed.inbounds.push(...part.inbounds);
|
||
parsed.users.push(...part.users);
|
||
} catch (error) { errors.push(`sing-box: ${safeError(error.message)}`); engineErrors.push('sing-box'); failed = true; }
|
||
}
|
||
if (!failed) {
|
||
const epoch = `sing-box:${before || serviceEpoch(service) || 'unknown'}`;
|
||
const tagNodes = new Map(resources.map((item) => [item.meta.metricsTag, item.meta.nodeId]));
|
||
for (const inbound of parsed.inbounds) {
|
||
const nodeId = tagNodes.get(inbound.tag);
|
||
if (!nodeId) continue;
|
||
samples.push({ kind: 'inbound', id: nodeId, uplink: inbound.uplink, downlink: inbound.downlink, epoch });
|
||
}
|
||
for (const user of parsed.users) {
|
||
const match = /^ng:(.+)$/.exec(user.name);
|
||
if (!match) continue;
|
||
samples.push({ kind: 'user', id: match[1], uplink: user.uplink, downlink: user.downlink, epoch });
|
||
}
|
||
if (!parsed.inbounds.length && !parsed.users.length) errors.push('sing-box 入口尚无统计记录');
|
||
}
|
||
return { samples, error: errors.join('; ') || null, engineErrors };
|
||
}
|
||
|
||
function readSingBoxObservations() {
|
||
if (!fs.existsSync(SINGBOX_LOG)) return { observations: [], offset: null };
|
||
let cursor = { offset: 0 };
|
||
try { cursor = JSON.parse(fs.readFileSync(SINGBOX_CURSOR, 'utf8')); } catch { /* first read */ }
|
||
const size = fs.statSync(SINGBOX_LOG).size;
|
||
if (size < cursor.offset) cursor.offset = 0;
|
||
const start = Math.max(cursor.offset, size - 2 * 1024 * 1024);
|
||
if (start >= size) return { observations: [], offset: null };
|
||
const buffer = Buffer.alloc(size - start);
|
||
const fd = fs.openSync(SINGBOX_LOG, 'r');
|
||
try { fs.readSync(fd, buffer, 0, buffer.length, start); } finally { fs.closeSync(fd); }
|
||
const end = buffer.lastIndexOf(10);
|
||
if (end < 0) return { observations: [], offset: null };
|
||
// AnyTLS 用户名本身就是 ng:<customerId>，身份直接写在日志行里，无需 tag 映射。
|
||
const entryTags = new Set(readResources().filter((item) =>
|
||
item.meta?.kind === 'node').map((item) => item.meta.metricsTag));
|
||
const sourceById = new Map(), unique = new Map();
|
||
for (const line of buffer.subarray(0, end).toString('utf8').split('\n')) {
|
||
const context = line.match(/\[(\d+)\s+\d+ms\]\s+inbound\/anytls\[([^\]]+)\]:\s+(.*)$/);
|
||
if (!context || !entryTags.has(context[2])) continue;
|
||
const key = `${context[2]}:${context[1]}`;
|
||
const source = context[3].match(/inbound connection from\s+(\[[^\]]+\]|[\d.]+):\d+/);
|
||
if (source) sourceById.set(key, source[1].replace(/^\[|\]$/g, ''));
|
||
// Only count authenticated AnyTLS sessions, never unauthenticated port probes.
|
||
const authed = context[3].match(/\[ng:([^\]]+)\] inbound connection to /);
|
||
if (authed && sourceById.has(key)) {
|
||
const customerId = authed[1];
|
||
const ip = sourceById.get(key);
|
||
unique.set(`${customerId}|${ip}`, { customerId, ip });
|
||
}
|
||
}
|
||
return { observations: [...unique.values()], offset: start + end + 1 };
|
||
}
|
||
|
||
async function usageLoop() {
|
||
while (true) {
|
||
await sleep(60000);
|
||
try {
|
||
let report;
|
||
try { report = queryUsage(); }
|
||
catch (error) { report = { samples: [], error: safeError(error.message) }; }
|
||
await request('/api/agent/usage', { method: 'POST', body: JSON.stringify(report) });
|
||
} catch (error) { log('Usage report failed', safeError(error.message)); }
|
||
try {
|
||
const sing = readSingBoxObservations();
|
||
if (sing.observations.length) await request('/api/agent/observations', { method: 'POST', body: JSON.stringify({ observations: sing.observations }) });
|
||
// A failed upload must not advance the cursor; retry the same log lines next cycle.
|
||
if (sing.offset !== null) fs.writeFileSync(SINGBOX_CURSOR, `${JSON.stringify({ offset: sing.offset })}\n`, { mode: 0o600 });
|
||
} catch (error) { log('IP observation report failed', safeError(error.message)); }
|
||
}
|
||
}
|
||
|
||
async function pollLoop() {
|
||
while (true) {
|
||
try {
|
||
const result = await request('/api/agent/poll', { method: 'POST', body: '{}' });
|
||
if (result.job) await execute(result.job);
|
||
else await sleep(POLL_MS);
|
||
} catch (error) { log('Poll failed', error.message); await sleep(Math.max(POLL_MS, 10000)); }
|
||
}
|
||
}
|
||
|
||
async function main() {
|
||
if (!CONTROLLER || !AGENT_KEY) throw new Error('NG_CONTROLLER and NG_AGENT_KEY are required');
|
||
ensureDirectories();
|
||
log(`NexusGate Agent v${VERSION} starting`);
|
||
// Keep the control channel alive even when a stale node configuration cannot start.
|
||
// The administrator can then see the engine error and a repair job can be claimed.
|
||
try { activateSingBox(); } catch (error) { lastEngineError = safeError(error.message); log('Initial sing-box activation failed', lastEngineError); }
|
||
try { await heartbeat(); } catch (error) { log('Initial heartbeat failed', error.message); }
|
||
setInterval(() => heartbeat().catch((error) => log('Heartbeat failed', error.message)), 30000).unref();
|
||
usageLoop();
|
||
await pollLoop();
|
||
}
|
||
|
||
if (require.main === module) {
|
||
const action = process.argv[2] === 'flush-usage' ? flushAllUsageBeforeRestart() : main();
|
||
action.catch((error) => { console.error(safeError(error.message)); process.exitCode = 1; });
|
||
}
|
||
|
||
module.exports = { activateSingBox, combinedSingBoxConfig, applyResource, realityKey,
|
||
readSingBoxObservations, parseUsageStats, queryUsage, installedCertificates,
|
||
udpPortsForPid, missingHysteriaListeners, systemInfo };
|