285 lines
11 KiB
JavaScript
285 lines
11 KiB
JavaScript
'use strict';
|
|
|
|
// External subscription aggregation (panel-go/extsub.go mirrors this
|
|
// file; design: SUB-AGGREGATION-DESIGN.md). The panel fetches a
|
|
// customer's external subscription sources, parses link lists, caches
|
|
// the result in data.subSourceCache and merges manual + external items
|
|
// into the subscription output behind the customer's own nodes.
|
|
//
|
|
// Safety: http/https only, SSRF guard rejects non-global IPs unless
|
|
// settings.allowPrivateSubSources is on, every redirect hop is
|
|
// re-validated, the connection is pinned to the validated IP (no DNS
|
|
// rebinding window), 10s timeout, 1MiB body cap, 200 items per source.
|
|
|
|
const http = require('node:http');
|
|
const https = require('node:https');
|
|
const dns = require('node:dns/promises');
|
|
const net = require('node:net');
|
|
const crypto = require('node:crypto');
|
|
const { isGlobalIPv4, isGlobalIPv6 } = require('./sysstat');
|
|
|
|
const LINK_SCHEMES = ['vless:', 'ss:', 'hysteria2:', 'anytls:', 'socks5:'];
|
|
const MAX_BODY_BYTES = 1024 * 1024;
|
|
const MAX_ITEMS = 200;
|
|
const FETCH_TIMEOUT_MS = 10000;
|
|
const MAX_REDIRECTS = 3;
|
|
const DEFAULT_UA = 'v2rayN/6.55';
|
|
|
|
function isGlobalIP(ip) {
|
|
const family = net.isIP(ip);
|
|
if (family === 4) return isGlobalIPv4(ip);
|
|
if (family === 6) return isGlobalIPv6(ip.replace(/^\[|\]$/g, ''));
|
|
return false;
|
|
}
|
|
|
|
function validateLinkUri(uri) {
|
|
if (typeof uri !== 'string' || !uri || uri.length > 2048) return false;
|
|
let url;
|
|
try { url = new URL(uri); } catch { return false; }
|
|
return LINK_SCHEMES.includes(url.protocol) && Boolean(url.hostname);
|
|
}
|
|
|
|
// Parse a fetched body into a normalized link list. Accepts a base64
|
|
// blob or plaintext lines; unsupported schemes and unparseable lines are
|
|
// dropped and counted, never fatal.
|
|
function parseLinkList(body) {
|
|
let text = String(body == null ? '' : body);
|
|
const compact = text.replace(/\s+/g, '');
|
|
if (compact.length >= 16 && /^[A-Za-z0-9+/=_-]+$/.test(compact)) {
|
|
try {
|
|
const normalized = compact.replace(/-/g, '+').replace(/_/g, '/');
|
|
const decoded = Buffer.from(normalized, 'base64').toString('utf8');
|
|
if (decoded.includes('://')) text = decoded;
|
|
} catch { /* keep plaintext */ }
|
|
}
|
|
const items = [];
|
|
let dropped = 0;
|
|
for (const rawLine of text.split(/\r?\n/)) {
|
|
const line = rawLine.trim();
|
|
if (!line) continue;
|
|
if (items.length >= MAX_ITEMS) { dropped += 1; continue; }
|
|
if (!validateLinkUri(line)) { dropped += 1; continue; }
|
|
let name = '';
|
|
const hash = line.lastIndexOf('#');
|
|
if (hash >= 0) {
|
|
try { name = decodeURIComponent(line.slice(hash + 1)); } catch { name = line.slice(hash + 1); }
|
|
}
|
|
items.push({ uri: line, name: name.slice(0, 120) });
|
|
}
|
|
return { items, dropped };
|
|
}
|
|
|
|
async function resolveChecked(hostname, allowPrivate) {
|
|
let addresses;
|
|
if (net.isIP(hostname)) {
|
|
addresses = [hostname];
|
|
} else {
|
|
const records = await dns.lookup(hostname, { all: true });
|
|
addresses = records.map((record) => record.address);
|
|
}
|
|
if (!addresses.length) throw new Error('域名解析无结果');
|
|
if (!allowPrivate && addresses.some((ip) => !isGlobalIP(ip))) {
|
|
throw new Error('订阅源指向非公网地址，已拒绝');
|
|
}
|
|
return addresses[0];
|
|
}
|
|
|
|
function fetchOnce(urlString, { userAgent, allowPrivate, deadline }) {
|
|
return new Promise((resolve, reject) => {
|
|
(async () => {
|
|
const url = new URL(urlString);
|
|
if (!['http:', 'https:'].includes(url.protocol)) throw new Error('仅支持 http/https 订阅源');
|
|
const hostname = url.hostname.replace(/^\[|\]$/g, '');
|
|
const ip = await resolveChecked(hostname, allowPrivate);
|
|
const isTls = url.protocol === 'https:';
|
|
const port = Number(url.port || (isTls ? 443 : 80));
|
|
const hostHeader = url.port ? `${hostname}:${url.port}` : hostname;
|
|
const options = {
|
|
hostname: ip, port, path: `${url.pathname}${url.search}`, method: 'GET',
|
|
headers: { host: hostHeader, 'user-agent': userAgent || DEFAULT_UA, accept: '*/*' },
|
|
timeout: Math.max(1000, deadline - Date.now()),
|
|
};
|
|
if (isTls) options.servername = hostname;
|
|
const transport = isTls ? https : http;
|
|
const req = transport.request(options, (res) => {
|
|
if (res.statusCode >= 300 && res.statusCode < 400 && res.headers.location) {
|
|
res.resume();
|
|
resolve({ redirect: new URL(res.headers.location, urlString).toString() });
|
|
return;
|
|
}
|
|
if (res.statusCode !== 200) {
|
|
res.resume();
|
|
reject(new Error(`订阅源返回 HTTP ${res.statusCode}`));
|
|
return;
|
|
}
|
|
const chunks = [];
|
|
let size = 0;
|
|
res.on('data', (chunk) => {
|
|
size += chunk.length;
|
|
if (size > MAX_BODY_BYTES) {
|
|
req.destroy(new Error('订阅源响应超过 1MiB'));
|
|
return;
|
|
}
|
|
chunks.push(chunk);
|
|
});
|
|
res.on('end', () => resolve({ body: Buffer.concat(chunks).toString('utf8') }));
|
|
res.on('error', reject);
|
|
});
|
|
req.on('timeout', () => req.destroy(new Error('订阅源请求超时')));
|
|
req.on('error', reject);
|
|
req.end();
|
|
})().catch(reject);
|
|
});
|
|
}
|
|
|
|
async function fetchSourceBody(source, settings) {
|
|
const allowPrivate = Boolean(settings && settings.allowPrivateSubSources);
|
|
const deadline = Date.now() + FETCH_TIMEOUT_MS;
|
|
let current = source.url;
|
|
for (let hop = 0; hop <= MAX_REDIRECTS; hop += 1) {
|
|
if (Date.now() > deadline) throw new Error('订阅源请求超时');
|
|
const result = await fetchOnce(current, { userAgent: source.userAgent, allowPrivate, deadline });
|
|
if (!result.redirect) return result.body;
|
|
if (hop === MAX_REDIRECTS) throw new Error('订阅源重定向次数过多');
|
|
current = result.redirect;
|
|
}
|
|
throw new Error('订阅源请求失败');
|
|
}
|
|
|
|
// Fetch + parse one source. Never throws: returns { ok, items, dropped, error }.
|
|
async function fetchExternalSource(source, settings) {
|
|
try {
|
|
const body = await fetchSourceBody(source, settings || {});
|
|
const { items, dropped } = parseLinkList(body);
|
|
return { ok: true, items, dropped, error: null };
|
|
} catch (error) {
|
|
return { ok: false, items: [], dropped: 0, error: String(error && error.message || error).slice(0, 240) };
|
|
}
|
|
}
|
|
|
|
function refreshMinutes(settings) {
|
|
const value = Number(settings && settings.externalSubRefreshMinutes);
|
|
if (!Number.isFinite(value) || value <= 0) return 60;
|
|
return Math.min(720, Math.max(15, value));
|
|
}
|
|
|
|
// Deterministic ±10% jitter from the source id so both implementations
|
|
// and restarts agree, and sources do not all fire in the same pass.
|
|
function jitterFactor(sourceId) {
|
|
const hash = crypto.createHash('sha256').update(String(sourceId)).digest();
|
|
return 1 + ((hash[0] % 21) - 10) / 100;
|
|
}
|
|
|
|
// Collect sources due for a refresh pass: [{ customerId, source }].
|
|
function dueSources(data, nowMs, limit = 3) {
|
|
const out = [];
|
|
const cache = data.subSourceCache || {};
|
|
for (const customer of data.customers) {
|
|
for (const source of customer.externalSources || []) {
|
|
if (!source.enabled) continue;
|
|
const entry = cache[source.id];
|
|
if (!entry || !entry.fetchedAt) { out.push({ customerId: customer.id, source }); continue; }
|
|
const base = refreshMinutes(data.settings) * 60000;
|
|
const backoff = Math.min(Math.pow(2, Math.min(entry.failCount || 0, 10)) * base, 24 * 3600 * 1000);
|
|
const due = Date.parse(entry.fetchedAt) + backoff * jitterFactor(source.id);
|
|
if (nowMs >= due) out.push({ customerId: customer.id, source });
|
|
if (out.length >= limit) return out;
|
|
}
|
|
if (out.length >= limit) break;
|
|
}
|
|
return out.slice(0, limit);
|
|
}
|
|
|
|
// Apply one fetch result to the cache inside a store transaction draft.
|
|
function applyFetchResult(data, sourceId, result, nowIso) {
|
|
data.subSourceCache ||= {};
|
|
const previous = data.subSourceCache[sourceId] || { fetchedAt: null, okAt: null, failCount: 0, lastError: null, droppedCount: 0, items: [] };
|
|
const entry = { ...previous, fetchedAt: nowIso };
|
|
if (result.ok) {
|
|
entry.okAt = nowIso;
|
|
entry.failCount = 0;
|
|
entry.lastError = null;
|
|
entry.droppedCount = result.dropped;
|
|
entry.items = result.items;
|
|
} else {
|
|
entry.failCount = (previous.failCount || 0) + 1;
|
|
entry.lastError = result.error;
|
|
}
|
|
data.subSourceCache[sourceId] = entry;
|
|
return entry;
|
|
}
|
|
|
|
// ---------- save-time normalization (shared by create/PATCH) ----------
|
|
|
|
function cleanTextLocal(value, max) {
|
|
return String(value == null ? '' : value).trim().slice(0, max);
|
|
}
|
|
|
|
function normalizeExternalSources(value, idFn) {
|
|
if (value == null) return undefined;
|
|
if (!Array.isArray(value)) { const e = new Error('外部订阅格式无效'); e.statusCode = 400; throw e; }
|
|
if (value.length > 5) { const e = new Error('每个客户最多 5 个外部订阅'); e.statusCode = 400; throw e; }
|
|
return value.map((raw) => {
|
|
const item = raw && typeof raw === 'object' ? raw : {};
|
|
const url = cleanTextLocal(item.url, 2048);
|
|
let parsed;
|
|
try { parsed = new URL(url); } catch { parsed = null; }
|
|
if (!parsed || !['http:', 'https:'].includes(parsed.protocol) || !parsed.hostname) {
|
|
const e = new Error('外部订阅地址无效'); e.statusCode = 400; throw e;
|
|
}
|
|
const next = {
|
|
id: typeof item.id === 'string' && item.id ? item.id.slice(0, 80) : idFn('src'),
|
|
url, remark: cleanTextLocal(item.remark, 80), enabled: item.enabled !== false,
|
|
};
|
|
const ua = cleanTextLocal(item.userAgent, 120);
|
|
if (ua) next.userAgent = ua;
|
|
return next;
|
|
});
|
|
}
|
|
|
|
function normalizeManualLinks(value, idFn) {
|
|
if (value == null) return undefined;
|
|
if (!Array.isArray(value)) { const e = new Error('外部链接格式无效'); e.statusCode = 400; throw e; }
|
|
if (value.length > 50) { const e = new Error('每个客户最多 50 条外部链接'); e.statusCode = 400; throw e; }
|
|
return value.map((raw) => {
|
|
const item = raw && typeof raw === 'object' ? raw : {};
|
|
const uri = cleanTextLocal(item.uri, 2048);
|
|
if (!validateLinkUri(uri)) { const e = new Error('外部链接无效或协议不受支持'); e.statusCode = 400; throw e; }
|
|
return {
|
|
id: typeof item.id === 'string' && item.id ? item.id.slice(0, 80) : idFn('lnk'),
|
|
uri, remark: cleanTextLocal(item.remark, 80), enabled: item.enabled !== false,
|
|
};
|
|
});
|
|
}
|
|
|
|
// ---------- merge helpers (used by lib/subscriptions.js) ----------
|
|
|
|
function dedupKey(uri) {
|
|
try {
|
|
const url = new URL(uri);
|
|
let userinfo = '';
|
|
try { userinfo = `${decodeURIComponent(url.username)}:${decodeURIComponent(url.password)}`; }
|
|
catch { userinfo = `${url.username}:${url.password}`; }
|
|
const hash = crypto.createHash('sha256').update(userinfo).digest('hex').slice(0, 16);
|
|
return `${url.protocol}//${url.hostname.toLowerCase()}:${url.port}#${hash}`;
|
|
} catch { return null; }
|
|
}
|
|
|
|
function withName(uri, name) {
|
|
const separator = uri.lastIndexOf('#');
|
|
const base = separator >= 0 ? uri.slice(0, separator) : uri;
|
|
return `${base}#${encodeURIComponent(name)}`;
|
|
}
|
|
|
|
function nameFromUri(uri) {
|
|
const hash = uri.lastIndexOf('#');
|
|
if (hash < 0) return '';
|
|
try { return decodeURIComponent(uri.slice(hash + 1)); } catch { return uri.slice(hash + 1); }
|
|
}
|
|
|
|
module.exports = {
|
|
LINK_SCHEMES, DEFAULT_UA, validateLinkUri, parseLinkList, fetchExternalSource,
|
|
refreshMinutes, dueSources, applyFetchResult, normalizeExternalSources, normalizeManualLinks,
|
|
dedupKey, withName, nameFromUri,
|
|
};
|