'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, };