diff --git a/docs/proxy-subscriptions.md b/docs/proxy-subscriptions.md new file mode 100644 index 0000000000..8bc7c08853 --- /dev/null +++ b/docs/proxy-subscriptions.md @@ -0,0 +1,371 @@ +# Operator Proxy Subscriptions (Karing-style) + +> Feature design + implementation notes for OmniRoute's operator-level proxy +> subscription flow. This is the v1 cut: a single operator pastes subscription +> links, picks a mode (global or rule), and OmniRoute binds the resulting proxy +> pool into the existing scope resolution. Multi-tenant per-API-key, advanced +> traffic rules, latency-driven per-rule weights, and so on are explicitly +> out-of-scope and listed in §7. + +--- + +## 1. Motivation + +Today, OmniRoute's proxy pool is hand-curated: every node lives in +`proxy_registry` with hand-written host/port/credentials, and every binding to +the upstream dispatchers (account → provider → combo → global → direct) is a +manual `proxy_assignments` row. Operators who already maintain a Clash/V2Ray/ +sing-box subscription (e.g. from an airport service) have to retype every node +into OmniRoute and re-bind them whenever the upstream list changes. + +The goal of v1 is to make OmniRoute first-class for **operator-supplied** +subscriptions, similar to how Karing / Clash / sing-box let users paste a +`https://...` URL and have the client manage the lifecycle. + +## 2. User stories + +| # | As a(n) | I want to | So that | +|---|---------|-----------|---------| +| U1 | Operator | paste a subscription URL once | I don't retype nodes every time the airport refreshes | +| U2 | Operator | toggle the subscription on/off | I can fall back to direct without deleting the URL | +| U3 | Operator | pick **global** mode | every provider's traffic exits via the subscription | +| U4 | Operator | pick **rule** mode and select specific providers | only selected providers route through the proxy; others stay direct | +| U5 | Operator | supply a local sing-box/clash SOCKS5 endpoint | SS/VMess/Trojan/VLESS nodes (which OmniRoute's dispatcher can't speak natively) become usable through a local kernel bridge | +| U6 | Operator | see fetch status and a recent redacted node summary | I can debug "why is this empty / erroring" without leaking credentials | + +## 3. Non-goals (v1) + +- Per-API-key subscription overrides (multi-tenant). v1 is operator-only. +- Per-provider traffic rules beyond `global` / `rule-on-selected-providers`. +- Latency-based smart routing between subscription nodes and other pools + (existing `resolveProxyForConnectionFromRegistry` already does this for the + global pool; v1 just feeds subscription nodes into it). +- Auto-importing URL/password from headers or query params. +- SSRF mitigation beyond loopback-only local-core endpoints (the subscription + URL itself is operator-controlled, so we trust it the same way we trust + upstream provider URLs today). + +## 4. Architecture + +``` + ┌─────────────────────────────────────────┐ + │ dashboard / settings / 代理 / 订阅代理 │ + │ (client component, SubscriptionTab) │ + └──────────────────┬──────────────────────┘ + │ fetch + ▼ + ┌────────────────────────────────────────────────────────┐ + │ /api/v1/management/proxy-subscriptions │ + │ ├ GET list │ + │ ├ POST create │ + │ ├ GET /:id │ + │ ├ PATCH /:id │ + │ ├ DELETE /:id │ + │ ├ POST /:id/refresh │ + │ └ GET /:id/nodes │ + └────────────────────────┬───────────────────────────────┘ + │ uses + ▼ + ┌────────────────────────────────────────────────────────┐ + │ src/lib/proxySubscription/ │ + │ ├ parse.ts (Clash YAML / V2Ray JSON / URIs) │ + │ ├ subscriptionService.ts │ + │ │ CRUD, sync, apply, unapply, scheduler │ + │ └ index.ts (barrel) │ + └──────────┬─────────────────────────────┬───────────────┘ + │ upsert/scope-bind │ DB + ▼ ▼ + ┌─────────────────────────┐ ┌──────────────────────────┐ + │ proxy_registry │ │ proxy_subscriptions │ + │ (existing) + │ │ (NEW — subscription │ + │ subscription_id column │ │ metadata + scheduler │ + │ + status/health checks │ │ state) │ + └─────────────────────────┘ └──────────────────────────┘ + │ + ▼ (existing) + resolveProxyForConnectionFromRegistry + hasBlockingProxyAssignment (fail-closed) + proxyDispatcher (open-sse/utils/proxyDispatcher) +``` + +Key design decision: **we do not invent a new scope or routing pipeline**. We +upsert subscription-derived nodes into `proxy_registry` with `source = +'subscription'` + `subscription_id`, and then `applySubscription()` walks the +existing `addProxyToScopePool(scope, scopeId, proxyId)` API. This means: + +- Existing rotation, health checks, and fail-closed guards apply for free. +- Existing dashboards (ProxyPoolTab, SourceToggleBar, GlobalConfigTab) work + unchanged — subscription nodes just appear in the pool with a `source` + badge. +- Deleting/disabling a subscription cleanly removes its bindings without + touching manual proxies. + +## 5. Data model + +### 5.1 New table `proxy_subscriptions` + +| Column | Type | Notes | +|---|---|---| +| `id` | TEXT PK | UUID | +| `name` | TEXT NOT NULL | display name | +| `url` | TEXT NOT NULL | subscription URL | +| `enabled` | INTEGER NOT NULL DEFAULT 0 | 1 = active | +| `mode` | TEXT NOT NULL DEFAULT `'global'` | `'global'` or `'rule'` | +| `rule_providers` | TEXT NULL | JSON array of provider IDs (mode='rule' only) | +| `local_core_endpoint` | TEXT NULL | loopback SOCKS5/HTTP for SS/VMess/etc. (e.g. `socks5://127.0.0.1:2080`) | +| `update_interval_minutes` | INTEGER NOT NULL DEFAULT 60 | background refresh cadence | +| `last_fetched_at` | TEXT NULL | ISO timestamp of last successful fetch | +| `status` | TEXT NOT NULL DEFAULT `'empty'` | `'ok'` / `'error'` / `'empty'` | +| `error` | TEXT NULL | last error / warning text (redacted) | +| `last_nodes` | TEXT NULL | JSON array, redacted node summaries | +| `created_at` | TEXT NOT NULL | ISO | +| `updated_at` | TEXT NOT NULL | ISO | + +Index: `idx_proxy_subscriptions_enabled (enabled)` for the scheduler tick. + +### 5.2 Extended `proxy_registry` + +Added one column: + +| Column | Type | Notes | +|---|---|---| +| `subscription_id` | TEXT NULL | FK by convention (no enforced FK; subscription row lives in `proxy_subscriptions`) | + +Existing rows on upgrade: `subscription_id = NULL`, behavior unchanged. +Migration: `ALTER TABLE proxy_registry ADD COLUMN subscription_id TEXT;` +(applied as `123_proxy_subscriptions.sql`, idempotent via the migration +runner's `ALTER` semantics). + +### 5.3 Extended `proxy_subscriptions` test isolation + +The migration runner applies new migrations automatically; the only places +that need to know about the new column are `types.ts` and `mappers.ts` (one +extra field each) and `proxies.ts` (3 SQL statements: INSERT/UPDATE/SELECT). + +## 6. Modes + +### 6.1 Global mode + +- Pool bound to `scope='global', scope_id=NULL`. +- `proxyEnabled` setting forced to `true` whenever any subscription (or any + non-subscription global proxy) is active. +- All provider traffic exits via the subscription pool, with rotation/health + applied by the existing `resolveProxyForConnectionFromRegistry`. + +### 6.2 Rule mode + +- Pool bound to `scope='provider', scope_id=` for each + selected provider. +- Providers NOT in the list fall through to direct (their own provider-level + proxy or no proxy). +- Toggling a subscription from global → rule first calls `unapplySubscription` + to detach the previous global bindings, then re-syncs. + +## 7. Protocol support + +The existing `proxyDispatcher` only speaks **http / https / socks5 / vercel / +deno / cloudflare**. v1 follows that: + +| Parser-detected type | Goes into pool directly? | Needs `localCoreEndpoint`? | +|---|---|---| +| `http` / `https` | yes | no | +| `socks5` | yes | no | +| `ss` / `ssr` | no | yes (sing-box/clash → loopback SOCKS5) | +| `vmess` / `vless` | no | yes | +| `trojan` | no | yes | +| `hysteria` / `tuic` / `wireguard` | no | yes | +| `relay` (vercel/deno/cloudflare) | yes | no | + +Without `localCoreEndpoint`, SS-class nodes are surfaced in the status as a +warning but **not routed**. This matches the "fail-closed, but don't lie about +capability" policy: we never silently drop traffic; we report unrouteable +nodes and let the operator decide. + +## 8. Parser (`src/lib/proxySubscription/parse.ts`) + +Hand-rolled, no external dependency. Inputs accepted: + +1. **Clash / Clash.Meta YAML** — `proxies:` array, with `type` dispatch. +2. **Base64-wrapped URI list** — `parseSubscription` detects base64 by length + and charset, decodes, then URI-parses. +3. **V2RayN-style JSON-array-of-URI** — uses `vmess://` / `vless://` URIs. +4. **Plain URI list** — `ss://`, `vmess://`, `vless://`, `trojan://`, + `hysteria://`, `tuic://`, `wireguard://`, `socks5://`, `http(s)://`. + +Output: + +```ts +type ParsedSubscription = { + nodes: DirectlyUsableNode[]; // http/https/socks5/relay + needsCore: NeedsCoreNode[]; // ss/vmess/... — redacted summary + rawProtocols: string[]; // for diagnostics + parserWarnings: string[]; // per-line parse errors, redacted +}; + +type DirectlyUsableNode = { + name: string; + type: "http" | "https" | "socks5" | "vercel" | "deno" | "cloudflare"; + host: string; + port: number; + username?: string; + password?: string; +}; +``` + +`redactedNodeSummary` returns a JSON-serializable array of `{name, type, +host, port, hasCredentials}` with credentials omitted. This is what gets +persisted in `last_nodes` for the operator UI. + +## 9. Security + +- **SSRF on `localCoreEndpoint`**: the only SSRF surface here is the local + core endpoint (the subscription URL itself is operator-supplied). Allowed + hosts: `127.0.0.1`, `::1`, `localhost`. Any other host is rejected at parse + time with a `subscription_needs_core_endpoint_invalid` status. +- **No outbound to operator-internal hosts** from a subscription URL. The URL + fetch goes through Node's `fetch` (same trust model as the existing + `proxyLatency` health checks and the provider ping tasks). The operator + already trusts the URL by pasting it. +- **Fail-closed**: if a subscription's proxy is dead but still bound to a + scope, `hasBlockingProxyAssignment` returns true and traffic fails closed — + matches existing policy for any pool proxy. The operator can always disable + the subscription or remove the binding. +- **No secret echo**: `last_nodes` is redacted; the UI never sends secrets + back. `password` / `username` are stored encrypted at rest by the existing + `proxy_registry` encryption path. +- **No cross-tenant write**: the API routes are gated by `requireManagementAuth` + (dashboard session OR a manage-scope API key). Per-API-key overrides are + explicitly out-of-scope. + +## 10. UI + +A new sub-tab **"订阅代理"** in `dashboard / settings / 代理`, placed after +"documentation". List view shows: + +- Name + URL (truncated, with full URL in `title` attribute) +- Status badge: `ok` / `error` / `empty` +- Enabled switch (optimistic toggle) +- Action buttons: edit / refresh / delete + +The edit form has: + +- Name (text, required) +- URL (text, required, validated as URL) +- Mode toggle (global / rule) +- Provider multi-select (visible only in rule mode; populated from + `/api/providers`) +- Local core endpoint (text, optional; placeholder `socks5://127.0.0.1:2080`) +- Update interval (number, default 60 minutes) +- Enabled toggle + +When `status === 'error'`, an inline warning banner shows `subscription.error`. +When `status === 'ok'` and there are nodes that needed a local core, a soft +warning banner shows which protocols were skipped. + +## 11. Migration & rollout + +1. New migration `123_proxy_subscriptions.sql` runs on first DB open after + upgrade (auto-discovered by the existing migration runner). +2. The migration is **idempotent**: `ALTER TABLE … ADD COLUMN …` against an + already-migrated DB is a no-op in SQLite when wrapped in the runner's + "ignore duplicate column" path. See the existing + `040_oneproxy_proxy_fields.sql` and `093_proxy_enable_toggles.sql` + precedents. +3. No backfill: existing rows get `subscription_id = NULL`, which the + service treats as "manual, not subscription-managed". +4. UI hides the tab when there are zero subscriptions, but the API is always + available — that's intentional, so headless operators can manage + subscriptions via API only. + +## 12. Auto-refresh + +`startSubscriptionScheduler()` is idempotent and: + +- Skips in the browser (`typeof window !== "undefined"`). +- Skips under `NODE_ENV=test`. +- Otherwise starts a 60s `setInterval` that: + - Lists enabled subscriptions. + - For each, computes `due = now - lastFetchedAt >= updateIntervalMinutes * 60_000`. + - Calls `syncSubscription` for due ones, swallowing errors (logged). +- The interval timer is `.unref()`'d so it never blocks process exit. + +The scheduler is started on: +- First `GET /api/v1/management/proxy-subscriptions` (dashboard open). +- Any `syncSubscription` call (defensive — for CLI / automation paths that + bypass the GET). + +## 13. Testing strategy + +`tests/unit/proxySubscription.parse.test.ts` — 7 pure-parser cases, no DB, +runnable in <1s: + +1. Clash YAML with `direct` (http) and `needsCore` (ss) nodes. +2. Base64-wrapped URI list (decoded correctly). +3. V2Ray JSON-array-of-URI (vmess / vless). +4. Plain URI list (mixed protocols). +5. Clash.Meta outbounds (socks5). +6. Empty / unknown input → `nodes=[]`, `needsCore=[]`, parserWarnings filled. +7. `redactedNodeSummary` strips credentials. + +`tests/unit/proxySubscription.service.test.ts` — 4 integration tests using +`process.env.DATA_DIR` + `core.resetDbInstance()`: + +1. **Global**: create enabled global subscription → `syncSubscription` → + verify pool rows in `proxy_registry` with `subscription_id` set → + `resolveProxyForConnectionFromRegistry` returns one of those rows → + `proxyEnabled` is true. +2. **Rule**: create enabled rule subscription on provider P1 → verify only + P1's scope is bound, P2's scope is untouched. +3. **Fail-closed**: subscription fetch URL is unreachable → `status='error'`, + pool is empty, but if pool ever had rows they are cleaned up; + `hasBlockingProxyAssignment` returns false (no dead proxies in any scope). +4. **Delete**: delete subscription → registry rows for that subscription are + removed with `force: true` (manual deletions can't cascade-block it) → + `proxyEnabled` recomputed. + +Test runner command: + +```bash +node --import tsx/esm \ + --import ./open-sse/utils/setupPolyfill.ts \ + --import ./tests/_setup/isolateDataDir.ts \ + --test \ + tests/unit/proxySubscription.parse.test.ts \ + tests/unit/proxySubscription.service.test.ts +``` + +## 14. Future work (NOT in v1) + +- Per-API-key subscription overrides (multi-tenant; needs a `key_subscription_overrides` table). +- Per-provider traffic rules with domain matchers (would slot into the existing `interceptionRules` table). +- Latency-weighted rotation across subscription pools (we already have `ProxyRotationStrategy = "latency"`; just expose it in the UI). +- Proxying the subscription fetch itself through a separate egress (so operators can fetch behind a corporate firewall). +- Browser-side preview of a parsed subscription before saving (currently must save → wait → see nodes). + +## 15. Files touched / added + +**Added (new):** + +- `src/lib/proxySubscription/parse.ts` +- `src/lib/proxySubscription/subscriptionService.ts` +- `src/lib/proxySubscription/index.ts` +- `src/lib/db/migrations/123_proxy_subscriptions.sql` +- `src/app/api/v1/management/proxy-subscriptions/route.ts` +- `src/app/api/v1/management/proxy-subscriptions/[id]/route.ts` +- `src/app/api/v1/management/proxy-subscriptions/[id]/refresh/route.ts` +- `src/app/api/v1/management/proxy-subscriptions/[id]/nodes/route.ts` +- `src/app/(dashboard)/dashboard/settings/components/proxy/SubscriptionTab.tsx` +- `tests/unit/proxySubscription.parse.test.ts` +- `tests/unit/proxySubscription.service.test.ts` +- `docs/proxy-subscriptions.md` (this file) + +**Modified (minimal):** + +- `src/lib/db/proxies/types.ts` — `+ subscriptionId: string | null` on + `ProxyRegistryRecord`; `+ subscriptionId?: string | null` on `ProxyPayload`. +- `src/lib/db/proxies/mappers.ts` — `mapProxyRow` reads + `subscription_id` from the row. +- `src/lib/db/proxies.ts` — INSERT / UPDATE / SELECT add `subscription_id`. +- `src/app/(dashboard)/dashboard/settings/components/ProxyTab.tsx` — adds + one new sub-tab ("订阅代理") + the `literal` fallback for labels that + aren't in the i18n catalog yet. \ No newline at end of file diff --git a/src/app/(dashboard)/dashboard/settings/components/ProxyTab.tsx b/src/app/(dashboard)/dashboard/settings/components/ProxyTab.tsx index 3cfc2902b9..01b308edd1 100644 --- a/src/app/(dashboard)/dashboard/settings/components/ProxyTab.tsx +++ b/src/app/(dashboard)/dashboard/settings/components/ProxyTab.tsx @@ -6,14 +6,16 @@ import GlobalConfigTab from "./proxy/GlobalConfigTab"; import ProxyPoolTab from "./proxy/ProxyPoolTab"; import FreePoolTab from "./proxy/FreePoolTab"; import DocumentationTab from "./proxy/DocumentationTab"; +import SubscriptionTab from "./proxy/SubscriptionTab"; -type TabId = "global-config" | "proxy-pool" | "free-pool" | "documentation"; +type TabId = "global-config" | "proxy-pool" | "free-pool" | "documentation" | "subscriptions"; const TABS: Array<{ id: TabId; labelKey: string; fallback: string }> = [ { id: "global-config", labelKey: "proxyGlobalConfigTab", fallback: "Global config" }, { id: "proxy-pool", labelKey: "proxyPoolTab", fallback: "Proxy pool" }, { id: "free-pool", labelKey: "freePoolTab", fallback: "Free pool" }, { id: "documentation", labelKey: "proxyDocumentationTab", fallback: "Documentation" }, + { id: "subscriptions", labelKey: "proxySubscriptionsTab", fallback: "Subscriptions" }, ]; export default function ProxyTab() { @@ -64,6 +66,7 @@ export default function ProxyTab() { {activeTab === "proxy-pool" && } {activeTab === "free-pool" && } {activeTab === "documentation" && } + {activeTab === "subscriptions" && } ); diff --git a/src/app/(dashboard)/dashboard/settings/components/proxy/SubscriptionTab.tsx b/src/app/(dashboard)/dashboard/settings/components/proxy/SubscriptionTab.tsx new file mode 100644 index 0000000000..7ed9dc1770 --- /dev/null +++ b/src/app/(dashboard)/dashboard/settings/components/proxy/SubscriptionTab.tsx @@ -0,0 +1,520 @@ +"use client"; + +import { useCallback, useEffect, useState } from "react"; +import { useTranslations } from "next-intl"; +import { Button } from "@/shared/components"; +import { isNeedsCoreNode } from "@/lib/proxySubscription/needsCore"; + +interface SubscriptionRecord { + id: string; + name: string; + url: string; + enabled: boolean; + mode: "global" | "rule"; + ruleProviders: string[] | null; + localCoreEndpoint: string | null; + updateIntervalMinutes: number; + lastFetchedAt: string | null; + status: "ok" | "error" | "empty"; + error: string | null; + lastNodes: unknown[] | null; + lastErrorAt: string | null; + consecutiveFailures: number; +} + +interface ProviderOption { + id: string; + name: string; + provider: string; +} + +type FormState = { + name: string; + url: string; + mode: "global" | "rule"; + ruleProviders: string[]; + localCoreEndpoint: string; + updateIntervalMinutes: number; + enabled: boolean; +}; + +const EMPTY_FORM: FormState = { + name: "", + url: "", + mode: "global", + ruleProviders: [], + localCoreEndpoint: "", + updateIntervalMinutes: 60, + enabled: true, +}; + +export default function SubscriptionTab() { + const [subs, setSubs] = useState([]); + const [providers, setProviders] = useState([]); + const [loading, setLoading] = useState(false); + const [error, setError] = useState(null); + + const t = useTranslations("settings"); + + // Resolve a subscription `error` value into a localized message. Values are + // either a `{ code, detail? }` JSON (user-facing, i18n'd) or a plain + // diagnostic string (technical fetch/sync errors) shown verbatim. + const resolveSubError = useCallback((raw: string | null): string | null => { + if (!raw) return null; + try { + const parsed = JSON.parse(raw) as { code?: string; detail?: string }; + if (parsed?.code) { + const base = t(`proxySubscription.error.${parsed.code}`); + return parsed.detail ? `${base}（${parsed.detail}）` : base; + } + } catch { + // plain diagnostic string — show as-is + } + return raw; + }, [t]); + const [busyId, setBusyId] = useState(null); + + const [showForm, setShowForm] = useState(false); + const [editingId, setEditingId] = useState(null); + const [form, setForm] = useState(EMPTY_FORM); + const [saving, setSaving] = useState(false); + const [formError, setFormError] = useState(null); + + const load = useCallback(async () => { + setLoading(true); + setError(null); + try { + const res = await fetch("/api/v1/management/proxy-subscriptions"); + if (!res.ok) throw new Error("加载订阅列表失败"); + const data = await res.json(); + setSubs(Array.isArray(data.items) ? data.items : []); + } catch (e) { + setError(e instanceof Error ? e.message : String(e)); + } finally { + setLoading(false); + } + }, []); + + const loadProviders = useCallback(async () => { + try { + const res = await fetch("/api/providers"); + if (!res.ok) return; + const data = await res.json(); + const connections = Array.isArray(data.connections) ? data.connections : []; + setProviders( + connections.map((c: Record) => ({ + id: String(c.id), + name: typeof c.name === "string" && c.name ? c.name : String(c.provider), + provider: String(c.provider), + })) + ); + } catch { + /* non-fatal: rule mode just won't offer a provider picker */ + } + }, []); + + useEffect(() => { + load(); + loadProviders(); + }, [load, loadProviders]); + + const resetForm = () => { + setForm(EMPTY_FORM); + setEditingId(null); + setFormError(null); + setShowForm(false); + }; + + const startEdit = (sub: SubscriptionRecord) => { + setEditingId(sub.id); + setForm({ + name: sub.name, + url: sub.url, + mode: sub.mode, + ruleProviders: sub.ruleProviders ?? [], + localCoreEndpoint: sub.localCoreEndpoint ?? "", + updateIntervalMinutes: sub.updateIntervalMinutes, + enabled: sub.enabled, + }); + setShowForm(true); + }; + + const save = async () => { + setSaving(true); + setFormError(null); + try { + if (!form.name.trim()) throw new Error("请填写名称"); + if (!form.url.trim()) throw new Error("请填写订阅链接"); + if (form.mode === "rule" && form.ruleProviders.length === 0) { + throw new Error("规则模式下请至少选择一个 Provider"); + } + const payload = { + name: form.name.trim(), + url: form.url.trim(), + mode: form.mode, + ruleProviders: form.mode === "rule" ? form.ruleProviders : null, + localCoreEndpoint: form.localCoreEndpoint.trim() || null, + updateIntervalMinutes: Number(form.updateIntervalMinutes) || 60, + enabled: form.enabled, + }; + const res = editingId + ? await fetch(`/api/v1/management/proxy-subscriptions/${editingId}`, { + method: "PATCH", + headers: { "Content-Type": "application/json" }, + body: JSON.stringify(payload), + }) + : await fetch("/api/v1/management/proxy-subscriptions", { + method: "POST", + headers: { "Content-Type": "application/json" }, + body: JSON.stringify(payload), + }); + if (!res.ok) { + const data = await res.json().catch(() => ({})); + throw new Error(data.error || "保存失败"); + } + resetForm(); + await load(); + } catch (e) { + setFormError(e instanceof Error ? e.message : String(e)); + } finally { + setSaving(false); + } + }; + + const toggleEnabled = async (sub: SubscriptionRecord) => { + setBusyId(sub.id); + try { + const res = await fetch(`/api/v1/management/proxy-subscriptions/${sub.id}`, { + method: "PATCH", + headers: { "Content-Type": "application/json" }, + body: JSON.stringify({ enabled: !sub.enabled }), + }); + if (!res.ok) throw new Error("切换开关失败"); + await load(); + } catch (e) { + setError(e instanceof Error ? e.message : String(e)); + } finally { + setBusyId(null); + } + }; + + const refresh = async (sub: SubscriptionRecord) => { + setBusyId(sub.id); + try { + const res = await fetch(`/api/v1/management/proxy-subscriptions/${sub.id}/refresh`, { + method: "POST", + }); + if (!res.ok) throw new Error("刷新失败"); + await load(); + } catch (e) { + setError(e instanceof Error ? e.message : String(e)); + } finally { + setBusyId(null); + } + }; + + const remove = async (sub: SubscriptionRecord) => { + if (!window.confirm(`确定删除订阅「${sub.name}」？相关代理节点也会一并移除。`)) return; + setBusyId(sub.id); + try { + const res = await fetch(`/api/v1/management/proxy-subscriptions/${sub.id}`, { + method: "DELETE", + }); + if (!res.ok) throw new Error("删除失败"); + await load(); + } catch (e) { + setError(e instanceof Error ? e.message : String(e)); + } finally { + setBusyId(null); + } + }; + + const statusBadge: Record = { + ok: "bg-green-500/15 text-green-600 border-green-500/30", + error: "bg-red-500/15 text-red-600 border-red-500/30", + empty: "bg-yellow-500/15 text-yellow-600 border-yellow-500/30", + }; + + return ( +
+
+

+ 粘贴你的代理订阅链接，开启后即可按全局或规则（指定 Provider）模式走代理。 + 订阅节点会自动同步进代理池，并复用既有的轮询、健康检查与防泄漏机制。 +

+ {!showForm && ( + + )} +
+ + {error && ( +
+ {error} +
+ )} + + {showForm && ( +
+
+

+ {editingId ? "编辑订阅" : "新增订阅"} +

+ +
+ +
+ + + +
+ +
+
+ 模式 +
+ {(["global", "rule"] as const).map((m) => ( + + ))} +
+ + {form.mode === "global" + ? "所有 Provider 流量都走该订阅的代理池。" + : "仅所选 Provider 的流量走代理，其余直连。"} + +
+ + +
+ + {form.mode === "rule" && ( +
+ 按 Provider 路由（多选） + {providers.length === 0 ? ( + 正在加载 Provider 列表… + ) : ( +
+ {providers.map((p) => { + const checked = form.ruleProviders.includes(p.id); + return ( + + ); + })} +
+ )} +
+ )} + +
+ + +
+ + {formError &&
{formError}
} + +
+ + +
+
+ )} + +
+ {loading &&

加载中…

} + {!loading && subs.length === 0 && ( +

还没有任何订阅。点击「新增订阅」开始吧。

+ )} + {subs.map((sub) => { + const needsCoreNodes = (sub.lastNodes ?? []).filter(isNeedsCoreNode); + const showCoreHint = needsCoreNodes.length > 0 && !sub.localCoreEndpoint; + return ( +
+
+
+
+ {sub.name} + + {sub.status === "ok" ? "正常" : sub.status === "error" ? "错误" : "空"} + + + {sub.mode === "global" ? "全局" : "规则"} + + + {sub.enabled ? "已启用" : "已停用"} + +
+

+ {sub.url} +

+ {resolveSubError(sub.error) && ( +

{resolveSubError(sub.error)}

+ )} + {showCoreHint && ( +
+

+ 该订阅有 {needsCoreNodes.length} 个节点需要本地代理内核（SS / VMess / Trojan / VLESS 等），当前未被路由。 +

+

+ 这些协议无法被 OmniRoute 直接转发。请在本机启动一个 sing-box 或{" "} + clash（Clash.Meta） 内核，并把它暴露为一个 SOCKS5/HTTP 端点，然后在「编辑」中填入该端点（仅接受 127.0.0.1 / localhost）。 +

+
+ socks5://127.0.0.1:2080 + + +
+
+ )} +

+ 节点数：{sub.lastNodes?.length ?? 0} + {needsCoreNodes.length > 0 ? `（${needsCoreNodes.length} 个需本地内核）` : ""} + {sub.lastFetchedAt ? ` · 上次同步：${sub.lastFetchedAt}` : ""} + {sub.consecutiveFailures > 0 ? ` · 连续失败 ${sub.consecutiveFailures} 次` : ""} + {sub.lastErrorAt ? ` · 上次错误：${sub.lastErrorAt}` : ""} +

+
+
+ + + + +
+
+
+ ); })} +
+
+ ); +} diff --git a/src/app/api/v1/management/proxy-subscriptions/[id]/nodes/route.ts b/src/app/api/v1/management/proxy-subscriptions/[id]/nodes/route.ts new file mode 100644 index 0000000000..d6c1322a18 --- /dev/null +++ b/src/app/api/v1/management/proxy-subscriptions/[id]/nodes/route.ts @@ -0,0 +1,31 @@ +import { requireManagementAuth } from "@/lib/api/requireManagementAuth"; +import { createErrorResponseFromUnknown } from "@/lib/api/errorResponse"; +import { getSubscriptionById } from "@/lib/proxySubscription"; + +type RouteCtx = { params: Promise<{ id: string }> }; + +/** + * GET /api/v1/management/proxy-subscriptions/:id/nodes — return the last-parsed + * node summary for display without re-fetching the (possibly slow) subscription. + */ +export async function GET(request: Request, ctx: RouteCtx) { + const authError = await requireManagementAuth(request); + if (authError) return authError; + try { + const { id } = await ctx.params; + const sub = await getSubscriptionById(id); + if (!sub) return Response.json({ error: "Subscription not found" }, { status: 404 }); + return Response.json({ + id: sub.id, + name: sub.name, + mode: sub.mode, + enabled: sub.enabled, + status: sub.status, + error: sub.error, + lastFetchedAt: sub.lastFetchedAt, + nodes: sub.lastNodes ?? [], + }); + } catch (error) { + return createErrorResponseFromUnknown(error, "Failed to load proxy subscription nodes"); + } +} diff --git a/src/app/api/v1/management/proxy-subscriptions/[id]/refresh/route.ts b/src/app/api/v1/management/proxy-subscriptions/[id]/refresh/route.ts new file mode 100644 index 0000000000..f9a1511326 --- /dev/null +++ b/src/app/api/v1/management/proxy-subscriptions/[id]/refresh/route.ts @@ -0,0 +1,22 @@ +import { requireManagementAuth } from "@/lib/api/requireManagementAuth"; +import { createErrorResponseFromUnknown } from "@/lib/api/errorResponse"; +import { syncSubscription } from "@/lib/proxySubscription"; + +type RouteCtx = { params: Promise<{ id: string }> }; + +/** + * POST /api/v1/management/proxy-subscriptions/:id/refresh — re-fetch + re-parse + * the subscription, sync its nodes into proxy_registry, and (re)bind the pool. + * Returns the SyncResult (node counts, bound count, status, warning). + */ +export async function POST(request: Request, ctx: RouteCtx) { + const authError = await requireManagementAuth(request); + if (authError) return authError; + try { + const { id } = await ctx.params; + const result = await syncSubscription(id); + return Response.json(result); + } catch (error) { + return createErrorResponseFromUnknown(error, "Failed to refresh proxy subscription"); + } +} diff --git a/src/app/api/v1/management/proxy-subscriptions/[id]/route.ts b/src/app/api/v1/management/proxy-subscriptions/[id]/route.ts new file mode 100644 index 0000000000..c19b5d6b9c --- /dev/null +++ b/src/app/api/v1/management/proxy-subscriptions/[id]/route.ts @@ -0,0 +1,77 @@ +import { requireManagementAuth } from "@/lib/api/requireManagementAuth"; +import { createErrorResponseFromUnknown } from "@/lib/api/errorResponse"; +import { + getSubscriptionById, + updateSubscription, + deleteSubscription, + redactSubscriptionUrl, + type ProxySubscriptionPayload, +} from "@/lib/proxySubscription"; + +type RouteCtx = { params: Promise<{ id: string }> }; + +/** + * GET /api/v1/management/proxy-subscriptions/:id — fetch one subscription. + * PATCH /api/v1/management/proxy-subscriptions/:id — update (name/url/mode/ + * ruleProviders/localCoreEndpoint/updateIntervalMinutes/enabled). + * DELETE /api/v1/management/proxy-subscriptions/:id — remove (unbinds + drops + * the synced proxy rows). + */ +export async function GET(_request: Request, ctx: RouteCtx) { + const authError = await requireManagementAuth(_request); + if (authError) return authError; + try { + const { id } = await ctx.params; + const sub = await getSubscriptionById(id); + if (!sub) return Response.json({ error: "Subscription not found" }, { status: 404 }); + return Response.json({ ...sub, url: redactSubscriptionUrl(sub.url) }); + } catch (error) { + return createErrorResponseFromUnknown(error, "Failed to load proxy subscription"); + } +} + +export async function PATCH(request: Request, ctx: RouteCtx) { + const authError = await requireManagementAuth(request); + if (authError) return authError; + try { + const { id } = await ctx.params; + const body = await request.json().catch(() => null); + if (!body || typeof body !== "object") { + return Response.json({ error: "Invalid JSON body" }, { status: 400 }); + } + const b = body as Record; + const payload: Partial = {}; + if (typeof b.name === "string") payload.name = b.name.trim(); + if (typeof b.url === "string") payload.url = b.url.trim(); + if (typeof b.mode === "string") payload.mode = b.mode === "rule" ? "rule" : "global"; + if (typeof b.enabled === "boolean") payload.enabled = b.enabled; + if (typeof b.localCoreEndpoint === "string") { + payload.localCoreEndpoint = b.localCoreEndpoint.trim() || null; + } + if (typeof b.updateIntervalMinutes === "number") { + payload.updateIntervalMinutes = b.updateIntervalMinutes; + } + if (Array.isArray(b.ruleProviders)) { + payload.ruleProviders = b.ruleProviders.filter((x) => typeof x === "string"); + } + + const updated = await updateSubscription(id, payload); + if (!updated) return Response.json({ error: "Subscription not found" }, { status: 404 }); + return Response.json({ ...updated, url: redactSubscriptionUrl(updated.url) }); + } catch (error) { + return createErrorResponseFromUnknown(error, "Failed to update proxy subscription"); + } +} + +export async function DELETE(request: Request, ctx: RouteCtx) { + const authError = await requireManagementAuth(request); + if (authError) return authError; + try { + const { id } = await ctx.params; + const ok = await deleteSubscription(id); + if (!ok) return Response.json({ error: "Subscription not found" }, { status: 404 }); + return Response.json({ deleted: true }); + } catch (error) { + return createErrorResponseFromUnknown(error, "Failed to delete proxy subscription"); + } +} diff --git a/src/app/api/v1/management/proxy-subscriptions/route.ts b/src/app/api/v1/management/proxy-subscriptions/route.ts new file mode 100644 index 0000000000..4ba9c68d8f --- /dev/null +++ b/src/app/api/v1/management/proxy-subscriptions/route.ts @@ -0,0 +1,84 @@ +import { requireManagementAuth } from "@/lib/api/requireManagementAuth"; +import { createErrorResponseFromUnknown } from "@/lib/api/errorResponse"; +import { + listSubscriptions, + createSubscription, + startSubscriptionScheduler, + redactSubscriptionUrl, + type ProxySubscriptionPayload, +} from "@/lib/proxySubscription"; + +/** + * GET /api/v1/management/proxy-subscriptions — list all operator subscriptions. + * POST /api/v1/management/proxy-subscriptions — create a subscription. + * + * A subscription is an operator-supplied proxy link (Karing-style). On create + * (and whenever enabled), its nodes are fetched + synced into proxy_registry + * and bound through the existing account/provider/global scope resolution. + */ +export async function GET(request: Request) { + const authError = await requireManagementAuth(request); + if (authError) return authError; + try { + // Best-effort: once the operator opens the UI, ensure the auto-refresh + // ticker is running (idempotent; no-op in test env). + startSubscriptionScheduler(); + const items = await listSubscriptions(); + // Redact credentials in the subscription URL before sending to the client. + const safe = items.map((it) => ({ ...it, url: redactSubscriptionUrl(it.url) })); + return Response.json({ items: safe }); + } catch (error) { + return createErrorResponseFromUnknown(error, "Failed to list proxy subscriptions"); + } +} + +function parsePayload(body: unknown): ProxySubscriptionPayload | { error: string } { + if (!body || typeof body !== "object") return { error: "Invalid JSON body" }; + const b = body as Record; + const name = typeof b.name === "string" ? b.name.trim() : ""; + const url = typeof b.url === "string" ? b.url.trim() : ""; + if (!name) return { error: "name is required" }; + if (!url) return { error: "url is required" }; + + const mode = b.mode === "rule" ? "rule" : "global"; + let ruleProviders: string[] | null = null; + if (Array.isArray(b.ruleProviders)) { + ruleProviders = b.ruleProviders.filter((x) => typeof x === "string"); + } + if (mode === "rule" && (!ruleProviders || ruleProviders.length === 0)) { + return { error: "ruleProviders is required when mode is 'rule'" }; + } + + const localCoreEndpoint = + typeof b.localCoreEndpoint === "string" && b.localCoreEndpoint.trim() + ? b.localCoreEndpoint.trim() + : null; + const updateIntervalMinutes = Number(b.updateIntervalMinutes) || 60; + const enabled = b.enabled === true; + + return { + name, + url, + mode, + ruleProviders, + localCoreEndpoint, + updateIntervalMinutes, + enabled, + }; +} + +export async function POST(request: Request) { + const authError = await requireManagementAuth(request); + if (authError) return authError; + try { + const body = await request.json().catch(() => null); + const parsed = parsePayload(body); + if ("error" in parsed) { + return Response.json({ error: parsed.error }, { status: 400 }); + } + const created = await createSubscription(parsed); + return Response.json({ ...created, url: redactSubscriptionUrl(created.url) }, { status: 201 }); + } catch (error) { + return createErrorResponseFromUnknown(error, "Failed to create proxy subscription"); + } +} diff --git a/src/i18n/messages/en.json b/src/i18n/messages/en.json index 1e3c40d0d8..f92b07449d 100644 --- a/src/i18n/messages/en.json +++ b/src/i18n/messages/en.json @@ -5240,6 +5240,14 @@ "proxyPoolTab": "Proxy Pool", "freePoolTab": "Free Pool", "proxyDocumentationTab": "Documentation", + "proxySubscriptionsTab": "Subscriptions", + "proxySubscription": { + "error": { + "LOCAL_CORE_ENDPOINT_INVALID": "Local proxy-core endpoint is invalid; SS/VMess/Trojan/VLESS nodes were ignored.", + "NEEDS_CORE_NOT_CONFIGURED": "This subscription has nodes that need a local proxy core (SS/VMess/Trojan/VLESS); they are not routed until you configure the local-core SOCKS5 endpoint.", + "NO_USABLE_NODES": "Subscription yielded no usable nodes (http/https/socks5 or nodes with a local core endpoint)." + } + }, "bulkHealthcheck": "Bulk Healthcheck", "bulkHealthcheckDesc": "Test all configured proxies against a target URL to find which ones work.", "healthcheckTesting": "Testing...", diff --git a/src/i18n/messages/pt-BR.json b/src/i18n/messages/pt-BR.json index 4f3885fe17..d0aaa95585 100644 --- a/src/i18n/messages/pt-BR.json +++ b/src/i18n/messages/pt-BR.json @@ -5240,6 +5240,14 @@ "proxyPoolTab": "Pool de Proxy", "freePoolTab": "Pool Gratuito", "proxyDocumentationTab": "Documentação", + "proxySubscriptionsTab": "Assinaturas", + "proxySubscription": { + "error": { + "LOCAL_CORE_ENDPOINT_INVALID": "O endpoint do core local é inválido; nós SS/VMess/Trojan/VLESS foram ignorados.", + "NEEDS_CORE_NOT_CONFIGURED": "Esta assinatura tem nós que precisam de um core local (SS/VMess/Trojan/VLESS); eles não são roteados até você configurar o endpoint SOCKS5 do core local.", + "NO_USABLE_NODES": "A assinatura não retornou nós utilizáveis (http/https/socks5 ou nós com endpoint de core local)." + } + }, "bulkHealthcheck": "Verificação de integridade em massa", "bulkHealthcheckDesc": "Teste todos os proxies configurados contra uma URL de destino para encontrar quais funcionam.", "healthcheckTesting": "Testando...", diff --git a/src/i18n/messages/zh-CN.json b/src/i18n/messages/zh-CN.json index 55aea159f4..4bf440bd84 100644 --- a/src/i18n/messages/zh-CN.json +++ b/src/i18n/messages/zh-CN.json @@ -5870,6 +5870,14 @@ "proxyPoolTab": "代理池", "freePoolTab": "免费泳池", "proxyDocumentationTab": "文档", + "proxySubscriptionsTab": "订阅代理", + "proxySubscription": { + "error": { + "LOCAL_CORE_ENDPOINT_INVALID": "本地内核端点格式无效，已忽略 SS/VMess/Trojan/VLESS 节点。", + "NEEDS_CORE_NOT_CONFIGURED": "订阅包含需本地内核的节点（SS/VMess/Trojan/VLESS 等），配置“本地内核 SOCKS5 端点”后才会被路由。", + "NO_USABLE_NODES": "订阅未解析出任何可用节点（http/https/socks5 或带本地内核端点的节点）。" + } + }, "perKeyProxyEnabled": "启用按密钥代理分配", "perKeyProxyEnabledDesc": "启用后，每个提供商连接可以使用自己的代理分配", "proxyPool": "代理池", diff --git a/src/lib/db/migrations/131_proxy_subscriptions.sql b/src/lib/db/migrations/131_proxy_subscriptions.sql new file mode 100644 index 0000000000..5461e857ee --- /dev/null +++ b/src/lib/db/migrations/131_proxy_subscriptions.sql @@ -0,0 +1,44 @@ +-- 131_proxy_subscriptions.sql +-- User-supplied proxy subscriptions (Karing-style): the operator pastes a +-- subscription URL that resolves to a pool of proxy nodes. Nodes are synced +-- into proxy_registry (source='subscription', subscription_id set) and bound +-- through the existing account/provider/global scope resolution. +-- +-- Columns: +-- name — human label +-- url — subscription link +-- enabled — 0/1 master on/off switch +-- mode — 'global' (bind pool to global scope) or +-- 'rule' (bind pool to selected provider scopes only) +-- rule_providers — JSON array of provider ids used in 'rule' mode +-- local_core_endpoint — optional local SOCKS5/HTTP endpoint (e.g. a running +-- sing-box/clash) that fronts SS/VMess/Trojan/VLESS nodes +-- update_interval_minutes — auto-refresh period +-- last_fetched_at — last successful/attempted fetch timestamp +-- status — 'ok' | 'error' | 'empty' +-- error — last error / warning message +-- last_nodes — redacted node summary (JSON) for display without re-fetch + +CREATE TABLE IF NOT EXISTS proxy_subscriptions ( + id TEXT PRIMARY KEY, + name TEXT NOT NULL, + url TEXT NOT NULL, + enabled INTEGER NOT NULL DEFAULT 0, + mode TEXT NOT NULL DEFAULT 'global', + rule_providers TEXT, + local_core_endpoint TEXT, + update_interval_minutes INTEGER NOT NULL DEFAULT 60, + last_fetched_at TEXT, + status TEXT NOT NULL DEFAULT 'empty', + error TEXT, + last_nodes TEXT, + created_at TEXT NOT NULL DEFAULT (datetime('now')), + updated_at TEXT NOT NULL DEFAULT (datetime('now')) +); + +CREATE INDEX IF NOT EXISTS idx_proxy_subscriptions_enabled ON proxy_subscriptions(enabled); + +-- Tag subscription-sourced rows in the registry so refresh/cleanup can scope to them. +ALTER TABLE proxy_registry ADD COLUMN subscription_id TEXT; + +CREATE INDEX IF NOT EXISTS idx_proxy_registry_subscription ON proxy_registry(subscription_id); diff --git a/src/lib/db/migrations/132_proxy_subscriptions_meta.sql b/src/lib/db/migrations/132_proxy_subscriptions_meta.sql new file mode 100644 index 0000000000..cb2a1f700a --- /dev/null +++ b/src/lib/db/migrations/132_proxy_subscriptions_meta.sql @@ -0,0 +1,10 @@ +-- 132_proxy_subscriptions_meta.sql +-- Observability columns for proxy subscriptions. Track the last error time and +-- a consecutive-failure counter so an operator can tell a transient blip +-- (one failed scheduled refresh) apart from a persistently broken subscription +-- (cloudflare outage vs. a typo'd URL / dead endpoint). + +ALTER TABLE proxy_subscriptions ADD COLUMN last_error_at TEXT; +ALTER TABLE proxy_subscriptions ADD COLUMN consecutive_failures INTEGER NOT NULL DEFAULT 0; + +CREATE INDEX IF NOT EXISTS idx_proxy_subscriptions_last_error ON proxy_subscriptions(last_error_at); diff --git a/src/lib/db/proxies.ts b/src/lib/db/proxies.ts index c574f23d9c..37c3db5928 100755 --- a/src/lib/db/proxies.ts +++ b/src/lib/db/proxies.ts @@ -28,11 +28,14 @@ import { redactProxySecrets, } from "./proxies/mappers"; import { isGlobalProxyEnabled, PROXY_ALIVE_PREDICATE } from "./proxies/guards"; +import { bumpProxyRegistryGeneration } from "./proxies/registryGeneration"; export { hasBlockingProxyAssignment, hasBlockingProxyAssignmentForProvider, } from "./proxies/guards"; export { extractRelayAuth, redactProxySecrets } from "./proxies/mappers"; +export { addProxiesToScopePool } from "./proxySubscriptions"; +export { bumpProxyRegistryGeneration, getProxyRegistryGeneration } from "./proxies/registryGeneration"; import { normalizeRotationScopeId, clearRotationState, @@ -50,16 +53,6 @@ export { resolveProxyForScopeFromRegistry, }; -let proxyRegistryGeneration = 0; - -function bumpProxyRegistryGeneration() { - proxyRegistryGeneration++; -} - -export function getProxyRegistryGeneration() { - return proxyRegistryGeneration; -} - // Mutate legacy proxyConfig rows directly so these writes stay inside the same // SQLite transaction as the proxy registry row and assignment upsert. function clearLegacyProxyForAssignment( @@ -132,8 +125,8 @@ function insertProxyRow( ) { db.prepare( `INSERT INTO proxy_registry - (id, name, type, host, port, username, password, region, notes, status, source, family, created_at, updated_at) - VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)` + (id, name, type, host, port, username, password, region, notes, status, source, family, subscription_id, created_at, updated_at) + VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)` ).run( id, payload.name, @@ -147,6 +140,7 @@ function insertProxyRow( payload.status || "active", payload.source || "manual", payload.family || "auto", + payload.subscriptionId ?? null, now, now ); @@ -170,12 +164,16 @@ function updateProxyRow( // Omitted credentials mean preserve; explicitly provided blanks clear stored auth. username: incomingUsername === undefined ? existing.username : incomingUsername, password: incomingPassword === undefined ? existing.password : incomingPassword, + // subscription_id: only override when the caller explicitly passes it (string|null); + // otherwise preserve whatever the existing row already carries. + subscriptionId: + payload.subscriptionId === undefined ? existing.subscriptionId : payload.subscriptionId, updatedAt: now, }; db.prepare( `UPDATE proxy_registry - SET name = ?, type = ?, host = ?, port = ?, username = ?, password = ?, region = ?, notes = ?, status = ?, source = ?, family = ?, updated_at = ? + SET name = ?, type = ?, host = ?, port = ?, username = ?, password = ?, region = ?, notes = ?, status = ?, source = ?, family = ?, subscription_id = ?, updated_at = ? WHERE id = ?` ).run( merged.name, @@ -189,6 +187,7 @@ function updateProxyRow( merged.status || "active", merged.source || "manual", merged.family || "auto", + merged.subscriptionId ?? null, merged.updatedAt, id ); @@ -263,7 +262,7 @@ export async function listProxies(options?: { const offset = options?.offset ?? 0; const db = getDbInstance(); let sql = - "SELECT id, name, type, host, port, username, password, region, notes, status, source, family, created_at, updated_at FROM proxy_registry ORDER BY datetime(updated_at) DESC, name ASC"; + "SELECT id, name, type, host, port, username, password, region, notes, status, source, family, subscription_id, created_at, updated_at FROM proxy_registry ORDER BY datetime(updated_at) DESC, name ASC"; const params: unknown[] = []; if (limit !== undefined) { sql += " LIMIT ? OFFSET ?"; @@ -290,7 +289,7 @@ function getProxyRowById( const includeSecrets = options?.includeSecrets === true; const row = db .prepare( - "SELECT id, name, type, host, port, username, password, region, notes, status, source, family, created_at, updated_at FROM proxy_registry WHERE id = ?" + "SELECT id, name, type, host, port, username, password, region, notes, status, source, family, subscription_id, created_at, updated_at FROM proxy_registry WHERE id = ?" ) .get(id); if (!row) return null; @@ -592,6 +591,9 @@ export async function addProxyToScopePool( return row ? mapAssignmentRow(row) : null; } +// addProxiesToScopePool moved to ./proxySubscriptions.ts to keep this file under +// the frozen size cap; re-exported below for existing callers. + /** * Remove one proxy from a scope's pool (#6365). Returns true if a row was deleted. * Leaves other pool members (and the rotation cursor) intact. diff --git a/src/lib/db/proxies/mappers.ts b/src/lib/db/proxies/mappers.ts index c475ad1b64..06248bc880 100644 --- a/src/lib/db/proxies/mappers.ts +++ b/src/lib/db/proxies/mappers.ts @@ -26,6 +26,7 @@ export function mapProxyRow(row: unknown): ProxyRegistryRecord { status: typeof r.status === "string" ? r.status : "active", source: typeof r.source === "string" ? r.source : "manual", family: typeof r.family === "string" ? r.family : "auto", + subscriptionId: typeof r.subscription_id === "string" ? r.subscription_id : null, createdAt: typeof r.created_at === "string" ? r.created_at : "", updatedAt: typeof r.updated_at === "string" ? r.updated_at : "", }; diff --git a/src/lib/db/proxies/registryGeneration.ts b/src/lib/db/proxies/registryGeneration.ts new file mode 100644 index 0000000000..4fab9f78a4 --- /dev/null +++ b/src/lib/db/proxies/registryGeneration.ts @@ -0,0 +1,12 @@ +// Shared registry-generation counter for the proxy registry. Split out so both +// `proxies.ts` and `proxySubscriptions.ts` can bump/read it without a cyclic +// import between the two sibling modules. +let proxyRegistryGeneration = 0; + +export function bumpProxyRegistryGeneration() { + proxyRegistryGeneration++; +} + +export function getProxyRegistryGeneration() { + return proxyRegistryGeneration; +} diff --git a/src/lib/db/proxies/types.ts b/src/lib/db/proxies/types.ts index c734f20a9d..11af7a5bc1 100644 --- a/src/lib/db/proxies/types.ts +++ b/src/lib/db/proxies/types.ts @@ -27,6 +27,8 @@ export interface ProxyRegistryRecord { status: string; source: string; family: string; + /** Set when this registry row was synced from a proxy subscription (#subscription-feature). */ + subscriptionId: string | null; createdAt: string; updatedAt: string; } @@ -53,6 +55,8 @@ export interface ProxyPayload { status?: string; source?: string; family?: string; + /** Optional link to a proxy subscription that created this row (#subscription-feature). */ + subscriptionId?: string | null; } export interface ProxyAssignmentPayload { diff --git a/src/lib/db/proxySubscriptions.ts b/src/lib/db/proxySubscriptions.ts new file mode 100644 index 0000000000..e4dddd6348 --- /dev/null +++ b/src/lib/db/proxySubscriptions.ts @@ -0,0 +1,58 @@ +// Proxy-subscription-specific pool operations, split out of `proxies.ts` to keep +// that module under its frozen size cap. Re-exported from `proxies.ts` so callers +// (subscriptionService.ts et al.) can keep importing from the original module. +import { getDbInstance } from "./core"; +import { backupDbFile } from "./backup"; +import { normalizeScope, normalizeAssignmentScopeId } from "./proxies/mappers"; +import { bumpProxyRegistryGeneration } from "./proxies/registryGeneration"; + +/** + * Add MULTIPLE proxies to a scope's rotation POOL in a single batched write + * (#6365). Idempotent per (scope, scope_id, proxy_id): existing members are + * skipped. New members are appended after the current highest `position` so + * round-robin order is stable. Returns the number of proxies actually added. + * Prefer this over N calls to `addProxyToScopePool` when binding a whole pool + * (e.g. a synced subscription's node list). + */ +export async function addProxiesToScopePool( + scope: string, + scopeId: string | null, + proxyIds: string[] +): Promise { + const normalizedScope = normalizeScope(scope); + const normalizedScopeId = normalizeAssignmentScopeId(normalizedScope, scopeId); + if (normalizedScope !== "global" && !normalizedScopeId) { + throw new Error("scopeId is required for non-global proxy assignments"); + } + const unique = [...new Set((proxyIds || []).filter(Boolean))]; + if (unique.length === 0) return 0; + + const db = getDbInstance(); + const maxRow = db + .prepare("SELECT MAX(position) AS maxPos FROM proxy_assignments WHERE scope = ? AND scope_id IS ?") + .get(normalizedScope, normalizedScopeId) as { maxPos?: number | null } | undefined; + const base = maxRow && typeof maxRow.maxPos === "number" ? maxRow.maxPos + 1 : 0; + const now = new Date().toISOString(); + + const exists = db.prepare( + "SELECT 1 FROM proxy_assignments WHERE scope = ? AND scope_id IS ? AND proxy_id = ? LIMIT 1" + ); + const insert = db.prepare( + `INSERT INTO proxy_assignments (proxy_id, scope, scope_id, position, created_at, updated_at) + VALUES (?, ?, ?, ?, ?, ?)` + ); + + let added = 0; + unique.forEach((pid, i) => { + if (!exists.get(normalizedScope, normalizedScopeId, pid)) { + insert.run(pid, normalizedScope, normalizedScopeId, base + i, now, now); + added++; + } + }); + + if (added > 0) { + backupDbFile("pre-write"); + bumpProxyRegistryGeneration(); + } + return added; +} diff --git a/src/lib/proxySubscription/coreEndpoint.ts b/src/lib/proxySubscription/coreEndpoint.ts new file mode 100644 index 0000000000..b421dbb46a --- /dev/null +++ b/src/lib/proxySubscription/coreEndpoint.ts @@ -0,0 +1,34 @@ +/** + * Pure, dependency-free validation of a subscription's local proxy-core + * endpoint. + * + * Extracted from `subscriptionService.isLocalCoreEndpointAllowed` so the + * security gate is unit-testable without the full DB / Next.js stack. + */ + +export const ALLOWED_LOCAL_CORE_HOSTS = new Set(["127.0.0.1", "::1", "localhost"]); + +/** Only these URL schemes denote a usable local proxy-core endpoint. */ +export const ALLOWED_CORE_SCHEMES = new Set(["http:", "https:", "socks5:"]); + +/** + * Whether `endpoint` is an acceptable local proxy-core address. + * + * Only loopback hosts over a proxy scheme (http/https/socks5) are permitted. + * A subscription's `localCoreEndpoint` becomes the single SOCKS5/HTTP address + * OmniRoute routes SS/VMess/Trojan/VLESS/etc. traffic through, so it must + * never point at a remote host — and a non-proxy scheme (file:/ftp:/…) is + * meaningless and rejected. + */ +export function isLocalCoreEndpointAllowed(endpoint: string | null): boolean { + if (!endpoint) return false; + try { + const u = new URL(endpoint); + const host = u.hostname.toLowerCase(); + if (!ALLOWED_LOCAL_CORE_HOSTS.has(host)) return false; + if (!ALLOWED_CORE_SCHEMES.has(u.protocol)) return false; + return true; + } catch { + return false; + } +} diff --git a/src/lib/proxySubscription/due.ts b/src/lib/proxySubscription/due.ts new file mode 100644 index 0000000000..a426d39b10 --- /dev/null +++ b/src/lib/proxySubscription/due.ts @@ -0,0 +1,31 @@ +/** + * Pure, dependency-free helper for the auto-refresh scheduler. + * + * Extracted from `subscriptionService.startSubscriptionScheduler` so the + * "is this subscription due for a refresh?" rule is unit-testable without the + * full Next.js / better-sqlite3 stack. + */ + +export interface DueCheckInput { + enabled: boolean; + lastFetchedAt: string | null; + updateIntervalMinutes: number; +} + +/** + * Whether `sub` should be auto-refreshed at time `now` (epoch milliseconds). + * + * Rules: + * - Disabled subscriptions are never due. + * - A subscription that has never been fetched (or has an unparseable + * `lastFetchedAt`) is immediately due. + * - Otherwise it is due once the elapsed time since the last fetch is at + * least `updateIntervalMinutes` (clamped to >= 0). + */ +export function isSubscriptionDue(sub: DueCheckInput, now: number = Date.now()): boolean { + if (!sub.enabled) return false; + const last = sub.lastFetchedAt ? Date.parse(sub.lastFetchedAt) : NaN; + if (!Number.isFinite(last)) return true; + const intervalMs = Math.max(0, sub.updateIntervalMinutes) * 60_000; + return now - last >= intervalMs; +} diff --git a/src/lib/proxySubscription/fetchGuard.ts b/src/lib/proxySubscription/fetchGuard.ts new file mode 100644 index 0000000000..5391c2f3a2 --- /dev/null +++ b/src/lib/proxySubscription/fetchGuard.ts @@ -0,0 +1,111 @@ +/** + * Pure, dependency-free guard for the *source* of a proxy subscription. + * + * The subscription URL is fetched server-side (see `subscriptionService + * .fetchSubscriptionContent`). Without a guard, an operator — or a compromised + * subscription link — could point OmniRoute at internal services or cloud + * metadata (SSRF). Only http/https to non-internal hosts are allowed: + * loopback / private / link-local (incl. 169.254.0.0/16 cloud metadata) / + * unspecified addresses are blocked. + * + * Hostname resolution is re-checked at fetch time (also using the IP-range + * helpers here) so a hostname that resolves to an internal address is still + * refused. Splitting the logic into pure functions keeps it unit-testable + * without DNS / the full stack. + */ + +/** Only these URL schemes may be used to *fetch* a subscription. */ +export const ALLOWED_FETCH_SCHEMES = new Set(["http:", "https:"]); + +// Blocked IPv4 ranges (base, mask) as 32-bit ints. +const BLOCKED_IPV4: ReadonlyArray = [ + [0x00000000, 0xff000000], // 0.0.0.0/8 unspecified + [0x7f000000, 0xff000000], // 127.0.0.0/8 loopback + [0x0a000000, 0xff000000], // 10.0.0.0/8 private + [0xac100000, 0xfff00000], // 172.16.0.0/12 private + [0xc0a80000, 0xffff0000], // 192.168.0.0/16 private + [0xa9fe0000, 0xffff0000], // 169.254.0.0/16 link-local (cloud metadata) +]; + +const IPV4_RE = /^(\d{1,3})\.(\d{1,3})\.(\d{1,3})\.(\d{1,3})$/; + +export function isIpv4Literal(host: string): boolean { + return IPV4_RE.test(host); +} + +/** Parse an IPv4 literal to an unsigned 32-bit int, or null if invalid. */ +export function ipv4ToLong(host: string): number | null { + const m = IPV4_RE.exec(host); + if (!m) return null; + const parts = m.slice(1).map(Number); + if (parts.some((p) => p > 255 || Number.isNaN(p))) return null; + // Weighted sum by powers of 256 — no bitwise ops, so octets >= 128 can't + // overflow into negative numbers the way `<< 24` would under 32-bit signed. + return (parts[0] * 16777216 + parts[1] * 65536 + parts[2] * 256 + parts[3]) >>> 0; +} + +export function isIpv4Blocked(ip: string): boolean { + const n = ipv4ToLong(ip); + if (n === null) return false; + // `&` yields a signed 32-bit int; coerce both sides to unsigned before + // comparing so masked results with the high bit set aren't negative. + return BLOCKED_IPV4.some(([base, mask]) => ((n & mask) >>> 0) === (base >>> 0)); +} + +/** Blocked IPv6 addresses: loopback, unspecified, link-local, ULA. */ +export function isIpv6Blocked(ip: string): boolean { + const h = ip.toLowerCase(); + if (h === "::1") return true; // loopback + if (h === "::") return true; // unspecified + if (h.startsWith("fe80")) return true; // link-local + if (h.startsWith("fc") || h.startsWith("fd")) return true; // unique local + return false; +} + +/** Whether `host` is an IP literal (v4 or v6). Hostnames return false. */ +export function isIpLiteral(host: string): boolean { + if (isIpv4Literal(host)) return true; + // IPv6 literals contain ":" and consist only of hex digits + ":". + return host.includes(":") && /^([0-9a-fA-F:]+)$/.test(host); +} + +/** + * True if ANY resolved address is in a blocked (internal) range. Used after + * DNS resolution: a hostname may resolve to MULTIPLE records (e.g. both a + * public and a private IP). Checking only the first record would let an + * attacker reach an internal service via a hostname that also has a public + * record — so we refuse if any resolved address is internal. `family` follows + * the `dns` module convention (4 = IPv4, 6 = IPv6; missing ⇒ treat as v4). + */ +export function isAnyResolvedAddressBlocked( + addrs: ReadonlyArray<{ address: string; family?: number }> +): boolean { + return addrs.some(({ address, family }) => { + const fam = family === 6 ? 6 : 4; + return fam === 6 ? isIpv6Blocked(address) : isIpv4Blocked(address); + }); +} + +/** + * Structural check (no DNS). True only if the scheme is allowed AND, when the + * host is an IP literal, it is not in a blocked range. Hostnames pass the + * structural check — they are resolved and re-checked at fetch time. + */ +export function isSubscriptionFetchUrlAllowed(url: string): boolean { + let u: URL; + try { + u = new URL(url); + } catch { + return false; + } + if (!ALLOWED_FETCH_SCHEMES.has(u.protocol)) return false; + // Strip enclosing brackets from an IPv6 literal host ("[::1]" → "::1"). + const rawHost = u.hostname.toLowerCase(); + const host = rawHost.startsWith("[") && rawHost.endsWith("]") ? rawHost.slice(1, -1) : rawHost; + if (host === "") return false; + if (isIpLiteral(host)) { + if (isIpv4Literal(host)) return !isIpv4Blocked(host); + return !isIpv6Blocked(host); + } + return true; // hostname: resolved + checked at fetch time +} diff --git a/src/lib/proxySubscription/fetchRetry.ts b/src/lib/proxySubscription/fetchRetry.ts new file mode 100644 index 0000000000..e7b4192f42 --- /dev/null +++ b/src/lib/proxySubscription/fetchRetry.ts @@ -0,0 +1,48 @@ +/** + * Pure, dependency-free retry helper for transient failures (used by the + * subscription fetch path). No DB / network imports, so it is unit-testable + * with a fake async function. + * + * Retries on transient errors with bounded exponential backoff: + * delay(attempt) = min(maxDelayMs, baseDelayMs * 2 ** attempt) + * The caller decides what is retryable via `isRetryable` (e.g. network/5xx/429 + * are retryable; a 4xx or an SSRF-block is not). + */ +export interface RetryOptions { + maxAttempts?: number; + baseDelayMs?: number; + maxDelayMs?: number; + /** Return false to stop retrying immediately (re-throw the error). */ + isRetryable?: (err: unknown) => boolean; + /** Injectable for tests; defaults to a real setTimeout sleep. */ + sleep?: (ms: number) => Promise; +} + +const DEFAULTS = { maxAttempts: 3, baseDelayMs: 500, maxDelayMs: 5000 } as const; + +export async function withRetry( + fn: () => Promise, + options: RetryOptions = {} +): Promise { + const maxAttempts = options.maxAttempts ?? DEFAULTS.maxAttempts; + const baseDelayMs = options.baseDelayMs ?? DEFAULTS.baseDelayMs; + const maxDelayMs = options.maxDelayMs ?? DEFAULTS.maxDelayMs; + const isRetryable = options.isRetryable ?? (() => true); + const sleep = options.sleep ?? ((ms: number) => new Promise((r) => setTimeout(r, ms))); + + let lastErr: unknown; + for (let attempt = 0; attempt < maxAttempts; attempt++) { + try { + return await fn(); + } catch (e) { + lastErr = e; + // Last attempt: no point delaying, just surface the error. + if (attempt === maxAttempts - 1) break; + // Permanent error (caller says so): stop immediately. + if (!isRetryable(e)) throw e; + const delay = Math.min(maxDelayMs, baseDelayMs * 2 ** attempt); + await sleep(delay); + } + } + throw lastErr; +} diff --git a/src/lib/proxySubscription/index.ts b/src/lib/proxySubscription/index.ts new file mode 100644 index 0000000000..bd76b11b65 --- /dev/null +++ b/src/lib/proxySubscription/index.ts @@ -0,0 +1,3 @@ +export * from "./parse"; +export * from "./subscriptionService"; +export * from "./url"; diff --git a/src/lib/proxySubscription/needsCore.ts b/src/lib/proxySubscription/needsCore.ts new file mode 100644 index 0000000000..6dc6b6c320 --- /dev/null +++ b/src/lib/proxySubscription/needsCore.ts @@ -0,0 +1,41 @@ +/** + * Pure, dependency-free helpers for detecting proxy nodes that require a + * local proxy core (sing-box / clash). + * + * Extracted from the SubscriptionTab UI so the rule is unit-testable without + * React / DOM. The redacted node summary (from `parse.redactedNodeSummary`) + * carries `type` for directly-usable nodes (http/https/socks5) and only + * `rawProtocol` for nodes that need a local core. + */ + +export const NEEDS_CORE_PROTOCOLS = new Set([ + "ss", + "vmess", + "trojan", + "vless", + "tuic", + "hysteria", + "wireguard", +]); + +/** + * Whether a redacted node summary represents a node that OmniRoute cannot + * forward directly and therefore needs a local sing-box / clash core. + * + * A node needs a core when it has no usable `type` (i.e. it is not a direct + * http/https/socks5 node) and its `rawProtocol` is one of the core protocols. + */ +export function isNeedsCoreNode(n: unknown): boolean { + if (!n || typeof n !== "object") return false; + const r = n as Record; + if (typeof r.type === "string") return false; // directly-usable node + return typeof r.rawProtocol === "string" && NEEDS_CORE_PROTOCOLS.has(r.rawProtocol); +} + +/** Count how many of the given redacted node summaries need a local core. */ +export function countNeedsCoreNodes(nodes: unknown[] | null | undefined): number { + if (!Array.isArray(nodes)) return 0; + let count = 0; + for (const n of nodes) if (isNeedsCoreNode(n)) count += 1; + return count; +} diff --git a/src/lib/proxySubscription/parse.ts b/src/lib/proxySubscription/parse.ts new file mode 100644 index 0000000000..13d9b545bc --- /dev/null +++ b/src/lib/proxySubscription/parse.ts @@ -0,0 +1,326 @@ +/** + * Proxy subscription parser. + * + * Turns a subscription body (base64-wrapped list, Clash/Clash.Meta YAML, + * V2Ray/Clash JSON, or a plain list of URIs) into a normalized node model. + * + * OmniRoute's request dispatcher (open-sse/utils/proxyDispatcher) only speaks + * http / https / socks5 (+ edge relay). Subscriptions whose nodes are + * Shadowsocks / VMess / VLESS / Trojan / TUIC / Hysteria / WireGuard cannot be + * used directly — they require a local proxy core (sing-box / clash) that + * exposes a SOCKS5/HTTP endpoint. Those nodes are reported separately as + * `needsCore` so the caller can either bind the operator-supplied + * `local_core_endpoint` or surface a clear "needs a local core" error. + * + * Source: operator-supplied subscription feature (Karing-style proxy). + */ +import * as yaml from "js-yaml"; + +export type DirectProxyType = "http" | "https" | "socks5"; +export type RawProxyProtocol = + | DirectProxyType + | "ss" + | "ssr" + | "vmess" + | "vless" + | "trojan" + | "tuic" + | "hysteria" + | "hysteria2" + | "wireguard" + | "snell" + | "unknown"; + +const DIRECT_TYPES: ReadonlySet = new Set(["http", "https", "socks5"]); +const NEEDS_CORE_PROTOCOLS: ReadonlySet = new Set([ + "ss", + "ssr", + "vmess", + "vless", + "trojan", + "tuic", + "hysteria", + "hysteria2", + "wireguard", + "snell", +]); + +export interface SubscriptionNode { + name: string; + type: DirectProxyType; + host: string; + port: number; + username?: string; + password?: string; + rawProtocol: RawProxyProtocol; +} + +export interface NeedsCoreNode { + name: string; + rawProtocol: RawProxyProtocol; + host?: string; + port?: number; + detail: string; +} + +export interface ParsedSubscription { + nodes: SubscriptionNode[]; + needsCore: NeedsCoreNode[]; + format: + | "clash-yaml" + | "clash-json" + | "v2ray-json" + | "lines" + | "base64-lines" + | "empty" + | "unknown"; +} + +function looksLikeBase64(s: string): boolean { + // Require length >= 16 and only base64 alphabet (allow trailing =), no spaces. + if (s.length < 16) return false; + if (/\s/.test(s)) return false; + return /^[A-Za-z0-9+/]+={0,2}$/.test(s); +} + +function tryDecodeBase64(s: string): string | null { + if (!looksLikeBase64(s)) return null; + try { + const buf = Buffer.from(s, "base64"); + // Reject if it doesn't round-trip (i.e. not actually base64 text). + // Strip padding from BOTH sides: original may have `=` stripped by the + // caller, while Node's base64 encoding always emits canonical padding. + const reencoded = buf.toString("base64").replace(/=+$/, ""); + if (reencoded !== s.replace(/=+$/, "")) return null; + const text = buf.toString("utf-8"); + if (!text || /[\x00-\x08\x0e-\x1f]/.test(text)) return null; + return text; + } catch { + return null; + } +} + +function asProtocol(raw: unknown): RawProxyProtocol { + if (typeof raw !== "string") return "unknown"; + const t = raw.toLowerCase().trim(); + if (DIRECT_TYPES.has(t)) return t as DirectProxyType; + if (NEEDS_CORE_PROTOCOLS.has(t as RawProxyProtocol)) return t as RawProxyProtocol; + return "unknown"; +} + +function nodeFromClashObject(obj: Record): SubscriptionNode | NeedsCoreNode | null { + if (!obj || typeof obj !== "object") return null; + const name = typeof obj.name === "string" ? obj.name : ""; + const type = asProtocol(obj.type); + const host = typeof obj.server === "string" ? obj.server : ""; + const port = Number(obj.port) || 0; + if (!name || !host || !port) return null; + if (DIRECT_TYPES.has(type)) { + return { + name, + type: type as DirectProxyType, + host, + port, + username: typeof obj.username === "string" && obj.username ? obj.username : undefined, + password: typeof obj.password === "string" && obj.password ? obj.password : undefined, + rawProtocol: type as RawProxyProtocol, + }; + } + if (NEEDS_CORE_PROTOCOLS.has(type)) { + return { + name, + rawProtocol: type as RawProxyProtocol, + host, + port, + detail: `${type}://${host}:${port}`, + }; + } + return null; +} + +function nodeFromUri(uri: string): SubscriptionNode | NeedsCoreNode | null { + let parsed: URL; + try { + parsed = new URL(uri); + } catch { + return null; + } + const scheme = parsed.protocol.replace(":", "").toLowerCase(); + + // vmess:// is special: the URL carries a base64(JSON) blob in place of + // host:port (`vmess://eyJhZGQiOi...`). Handle it BEFORE the host/port check. + if (scheme === "vmess") { + const b64 = (parsed.pathname || "").replace(/^\//, ""); + const tag = decodeURIComponent((parsed.hash || "").replace(/^#/, "")); + const name = tag || `vmess-${(parsed.hostname || "").slice(0, 8) || "node"}`; + let host = parsed.hostname || undefined; + let port = parsed.port ? Number(parsed.port) : undefined; + try { + if (b64) { + const json = JSON.parse(Buffer.from(b64, "base64").toString("utf-8")); + if (json && typeof json.add === "string") { + host = json.add; + port = Number(json.port) || port; + } + } + } catch { + // keep generic + } + const detail = host && port ? `vmess://${host}:${port}` : `vmess://${(b64 || "").slice(0, 16)}`; + return { + name, + rawProtocol: "vmess", + host, + port, + detail, + }; + } + + const name = decodeURIComponent(parsed.hash.replace(/^#/, "")) || parsed.hostname; + const host = parsed.hostname; + const port = Number(parsed.port); + if (!host || !port) return null; + + if (DIRECT_TYPES.has(scheme)) { + return { + name, + type: scheme as DirectProxyType, + host, + port, + username: parsed.username ? decodeURIComponent(parsed.username) : undefined, + password: parsed.password ? decodeURIComponent(parsed.password) : undefined, + rawProtocol: scheme as RawProxyProtocol, + }; + } + + if (scheme === "ss") { + // SIP002: ss://base64(user:pass)@host:port OR ss://method:pass@host:port + // We only need host/port for direct usability; SS itself needs a core, so + // report as needsCore but carry the host/port for the operator's reference. + return { + name, + rawProtocol: "ss", + host, + port, + detail: `ss://${host}:${port}`, + }; + } + + if (NEEDS_CORE_PROTOCOLS.has(scheme as RawProxyProtocol)) { + return { + name, + rawProtocol: scheme as RawProxyProtocol, + host, + port, + detail: `${scheme}://${host}:${port}`, + }; + } + + return null; +} + +function collectFromArray(items: unknown[], format: ParsedSubscription["format"]): ParsedSubscription { + const nodes: SubscriptionNode[] = []; + const needsCore: NeedsCoreNode[] = []; + for (const item of items) { + if (typeof item === "string") { + const n = nodeFromUri(item.trim()); + if (n) (isDirect(n) ? nodes : needsCore).push(n); + continue; + } + if (item && typeof item === "object") { + const n = nodeFromClashObject(item as Record); + if (n) (isDirect(n) ? nodes : needsCore).push(n); + } + } + return { nodes, needsCore, format }; +} + +function isDirect(n: SubscriptionNode | NeedsCoreNode): n is SubscriptionNode { + return (n as SubscriptionNode).type !== undefined; +} + +function parseClashYaml(content: string): ParsedSubscription { + try { + const doc = yaml.load(content) as Record | null; + if (doc && Array.isArray(doc.proxies)) { + return collectFromArray(doc.proxies, "clash-yaml"); + } + if (doc && Array.isArray((doc as Record).outbounds)) { + return collectFromArray((doc as Record).outbounds as unknown[], "clash-yaml"); + } + } catch { + // fall through to unknown + } + return { nodes: [], needsCore: [], format: "unknown" }; +} + +function parseLineList(lines: string[]): ParsedSubscription { + const nodes: SubscriptionNode[] = []; + const needsCore: NeedsCoreNode[] = []; + for (const line of lines) { + const n = nodeFromUri(line); + if (n) (isDirect(n) ? nodes : needsCore).push(n); + } + return { nodes, needsCore, format: "lines" }; +} + +/** Parse a subscription body into a normalized node model. */ +export function parseSubscription(body: string): ParsedSubscription { + const text = (body || "").trim(); + if (!text) return { nodes: [], needsCore: [], format: "empty" }; + + const decoded = tryDecodeBase64(text); + const base64Used = decoded !== null && decoded !== text; + const content = decoded ?? text; + + if ( + /^\s*proxies\s*:/m.test(content) || + /^\s*proxy-providers\s*:/m.test(content) || + /^\s*outbounds\s*:/m.test(content) + ) { + const res = parseClashYaml(content); + return base64Used ? { ...res, format: "base64-lines" } : res; + } + + if (content.startsWith("{") || content.startsWith("[")) { + try { + const json = JSON.parse(content); + if (Array.isArray(json)) return collectFromArray(json, "v2ray-json"); + if (json && Array.isArray(json.proxies)) return collectFromArray(json.proxies, "clash-json"); + if (json && Array.isArray(json.outbounds)) return collectFromArray(json.outbounds, "v2ray-json"); + } catch { + // fall through + } + } + + const lines = content.split(/\r?\n/).map((s) => s.trim()).filter(Boolean); + if (lines.length > 0 && lines.some((l) => /^[a-zA-Z][a-zA-Z0-9+.\-]*:\/\//.test(l))) { + const res = parseLineList(lines); + return base64Used ? { ...res, format: "base64-lines" } : res; + } + + return { nodes: [], needsCore: [], format: "unknown" }; +} + +/** Redacted node summary for storage/display (no secrets). */ +export function redactedNodeSummary(parsed: ParsedSubscription): Array< + Record +> { + const direct = parsed.nodes.map((n) => ({ + name: n.name, + type: n.type, + host: n.host, + port: n.port, + rawProtocol: n.rawProtocol, + hasAuth: Boolean(n.username || n.password), + })); + const core = parsed.needsCore.map((n) => ({ + name: n.name, + rawProtocol: n.rawProtocol, + host: n.host, + port: n.port, + detail: n.detail, + })); + return [...direct, ...core]; +} diff --git a/src/lib/proxySubscription/scopes.ts b/src/lib/proxySubscription/scopes.ts new file mode 100644 index 0000000000..2f3746f35b --- /dev/null +++ b/src/lib/proxySubscription/scopes.ts @@ -0,0 +1,31 @@ +/** + * Pure, dependency-free resolution of which scope(s) a subscription's synced + * proxy pool should be bound to. + * + * Extracted from `subscriptionService.resolveTargetScopes` so the routing + * rule is unit-testable without the full DB / Next.js stack. + */ + +export type TargetScope = { scope: "global" | "provider"; scopeId: string | null }; + +export interface ScopeInput { + mode: "global" | "rule"; + ruleProviders?: string[] | null; +} + +/** + * Resolve the target scopes for a subscription. + * + * - `global` mode (or `rule` mode with no providers selected) binds the + * global scope, so every provider's traffic is proxied. + * - `rule` mode with providers binds one provider scope per selected + * provider, so only those providers' traffic is proxied and the rest go + * direct. + */ +export function resolveTargetScopes(sub: ScopeInput): TargetScope[] { + if (sub.mode === "rule" && sub.ruleProviders && sub.ruleProviders.length > 0) { + return sub.ruleProviders.map((p) => ({ scope: "provider" as const, scopeId: p })); + } + // global mode, or rule mode with no providers selected → bind global. + return [{ scope: "global" as const, scopeId: null }]; +} diff --git a/src/lib/proxySubscription/subscriptionService.ts b/src/lib/proxySubscription/subscriptionService.ts new file mode 100644 index 0000000000..94b74d0d3d --- /dev/null +++ b/src/lib/proxySubscription/subscriptionService.ts @@ -0,0 +1,700 @@ +/** + * Proxy subscription service (Karing-style, operator-supplied). + * + * A subscription is a URL the operator pastes. We fetch + parse it into a pool + * of proxy nodes, sync those nodes into `proxy_registry` (source='subscription', + * subscription_id set), and bind the pool through the EXISTING scope resolution + * (account/provider/global) — so subscriptions inherit rotation, health checks, + * and the fail-closed guard for free. + * + * Modes: + * - 'global': pool bound to the global scope (all provider traffic proxied). + * - 'rule': pool bound only to the selected provider scopes (others direct). + * + * Protocol support: + * - http/https/socks5 nodes are used directly. + * - ss/vmess/vless/trojan/tuic/hysteria/wireguard nodes need a local proxy + * core (sing-box/clash) exposing a SOCKS5/HTTP endpoint; supply it via + * `localCoreEndpoint` and we bind that single endpoint (the core does the + * protocol translation + node selection). Without it, those nodes are + * reported but not routed. + */ +import { randomUUID } from "crypto"; +import { getDbInstance } from "../db/core"; +import { backupDbFile } from "../db/backup"; +import { + addProxiesToScopePool, + bumpProxyRegistryGeneration, + deleteProxyById, + upsertProxy, +} from "../db/proxies"; +import { bumpProxyConfigGeneration } from "../db/settings"; +import { isSubscriptionDue } from "./due"; +import { isLocalCoreEndpointAllowed } from "./coreEndpoint"; +import { resolveTargetScopes } from "./scopes"; +import { + isSubscriptionFetchUrlAllowed, + isIpLiteral, + isAnyResolvedAddressBlocked, +} from "./fetchGuard"; +import { withRetry } from "./fetchRetry"; +import { parseSubscription, redactedNodeSummary, type ParsedSubscription } from "./parse"; + +export type ProxySubscriptionMode = "global" | "rule"; +export type ProxySubscriptionStatus = "ok" | "error" | "empty"; + +/** Stable, language-neutral error codes stored in the subscription `error` + * column (as JSON) so the dashboard can localize them via i18n instead of + * showing server-side strings. */ +export type ProxySubscriptionErrorCode = + | "LOCAL_CORE_ENDPOINT_INVALID" + | "NEEDS_CORE_NOT_CONFIGURED" + | "NO_USABLE_NODES"; + +/** Encode a user-facing error as `{ code, detail? }` for i18n on the client. */ +export function subscriptionErrorCode(code: ProxySubscriptionErrorCode, detail?: string): string { + return JSON.stringify(detail ? { code, detail } : { code }); +} + +export interface ProxySubscriptionRecord { + id: string; + name: string; + url: string; + enabled: boolean; + mode: ProxySubscriptionMode; + ruleProviders: string[] | null; + localCoreEndpoint: string | null; + updateIntervalMinutes: number; + lastFetchedAt: string | null; + status: ProxySubscriptionStatus; + error: string | null; + lastNodes: unknown[] | null; + lastErrorAt: string | null; + consecutiveFailures: number; + createdAt: string; + updatedAt: string; +} + +export interface ProxySubscriptionPayload { + name: string; + url: string; + enabled?: boolean; + mode?: ProxySubscriptionMode; + ruleProviders?: string[] | null; + localCoreEndpoint?: string | null; + updateIntervalMinutes?: number; +} + +export interface SyncResult { + subscriptionId: string; + nodes: number; + needsCore: number; + boundProxies: number; + status: ProxySubscriptionStatus; + error: string | null; + applied: boolean; +} + +const SUBSCRIPTION_FETCH_TIMEOUT_MS = 15_000; + +// ───────────────────────────── Row mapping ───────────────────────────── + +function mapSubscriptionRow(row: unknown): ProxySubscriptionRecord { + const r = (row && typeof row === "object" ? row : {}) as Record; + const parseList = (v: unknown): string[] | null => { + if (typeof v !== "string" || !v.trim()) return null; + try { + const arr = JSON.parse(v); + return Array.isArray(arr) ? arr.filter((x) => typeof x === "string") : null; + } catch { + return null; + } + }; + const parseNodes = (v: unknown): unknown[] | null => { + if (typeof v !== "string" || !v.trim()) return null; + try { + const arr = JSON.parse(v); + return Array.isArray(arr) ? arr : null; + } catch { + return null; + } + }; + return { + id: typeof r.id === "string" ? r.id : "", + name: typeof r.name === "string" ? r.name : "", + url: typeof r.url === "string" ? r.url : "", + enabled: Number(r.enabled) !== 0, + mode: r.mode === "rule" ? "rule" : "global", + ruleProviders: parseList(r.rule_providers), + localCoreEndpoint: typeof r.local_core_endpoint === "string" ? r.local_core_endpoint : null, + updateIntervalMinutes: Number(r.update_interval_minutes) || 60, + lastFetchedAt: typeof r.last_fetched_at === "string" ? r.last_fetched_at : null, + status: (r.status as ProxySubscriptionStatus) || "empty", + error: typeof r.error === "string" ? r.error : null, + lastNodes: parseNodes(r.last_nodes), + lastErrorAt: typeof r.last_error_at === "string" ? r.last_error_at : null, + consecutiveFailures: Number(r.consecutive_failures) || 0, + createdAt: typeof r.created_at === "string" ? r.created_at : "", + updatedAt: typeof r.updated_at === "string" ? r.updated_at : "", + }; +} + +// ───────────────────────────── CRUD ───────────────────────────── + +export async function listSubscriptions(): Promise { + const db = getDbInstance(); + const rows = db + .prepare( + "SELECT id, name, url, enabled, mode, rule_providers, local_core_endpoint, update_interval_minutes, last_fetched_at, status, error, last_nodes, last_error_at, consecutive_failures, created_at, updated_at FROM proxy_subscriptions ORDER BY datetime(updated_at) DESC, name ASC" + ) + .all(); + return rows.map(mapSubscriptionRow); +} + +export async function getSubscriptionById(id: string): Promise { + const db = getDbInstance(); + const row = db + .prepare( + "SELECT id, name, url, enabled, mode, rule_providers, local_core_endpoint, update_interval_minutes, last_fetched_at, status, error, last_nodes, last_error_at, consecutive_failures, created_at, updated_at FROM proxy_subscriptions WHERE id = ?" + ) + .get(id); + return row ? mapSubscriptionRow(row) : null; +} + +export async function createSubscription( + payload: ProxySubscriptionPayload +): Promise { + const id = randomUUID(); + const now = new Date().toISOString(); + const enabled = payload.enabled === true ? 1 : 0; + const db = getDbInstance(); + db.prepare( + `INSERT INTO proxy_subscriptions + (id, name, url, enabled, mode, rule_providers, local_core_endpoint, update_interval_minutes, status, created_at, updated_at) + VALUES (?, ?, ?, ?, ?, ?, ?, ?, 'empty', ?, ?)` + ).run( + id, + payload.name, + payload.url, + enabled, + payload.mode || "global", + payload.ruleProviders ? JSON.stringify(payload.ruleProviders) : null, + payload.localCoreEndpoint || null, + payload.updateIntervalMinutes || 60, + now, + now + ); + const created = (await getSubscriptionById(id))!; + if (created.enabled) { + await syncSubscription(id); + } + // Re-read so the returned record reflects the post-sync status/error/lastNodes. + return (await getSubscriptionById(id))!; +} + +export async function updateSubscription( + id: string, + payload: Partial +): Promise { + const existing = await getSubscriptionById(id); + if (!existing) return null; + const db = getDbInstance(); + const name = payload.name ?? existing.name; + const url = payload.url ?? existing.url; + const mode = payload.mode ?? existing.mode; + const ruleProviders = payload.ruleProviders !== undefined ? payload.ruleProviders : existing.ruleProviders; + const localCoreEndpoint = + payload.localCoreEndpoint !== undefined ? payload.localCoreEndpoint : existing.localCoreEndpoint; + const updateIntervalMinutes = payload.updateIntervalMinutes ?? existing.updateIntervalMinutes; + const now = new Date().toISOString(); + + const enabledChanged = payload.enabled !== undefined && payload.enabled !== existing.enabled; + const enabled = payload.enabled !== undefined ? (payload.enabled ? 1 : 0) : existing.enabled ? 1 : 0; + + db.prepare( + `UPDATE proxy_subscriptions + SET name = ?, url = ?, enabled = ?, mode = ?, rule_providers = ?, local_core_endpoint = ?, update_interval_minutes = ?, updated_at = ? + WHERE id = ?` + ).run( + name, + url, + enabled, + mode, + ruleProviders ? JSON.stringify(ruleProviders) : null, + localCoreEndpoint || null, + updateIntervalMinutes, + now, + id + ); + + const updated = (await getSubscriptionById(id))!; + + // Re-evaluate binding. + if (updated.enabled) { + if (payload.mode !== undefined || payload.ruleProviders !== undefined) { + // Routing targets changed: detach from the old scopes first, then + // re-fetch + bind to the new targets. applySubscription is idempotent per + // scope, but a global→rule switch must drop the previous global binding. + await unapplySubscription(id); + await syncSubscription(id); + } else if ( + enabledChanged || + payload.url !== undefined || + payload.localCoreEndpoint !== undefined + ) { + // Same scopes: just refresh nodes (re-applies idempotently to same scopes). + await syncSubscription(id); + } + } else { + // Disabled: detach everything. + await unapplySubscription(id); + } + return getSubscriptionById(id); +} + +export async function setSubscriptionEnabled(id: string, enabled: boolean): Promise { + return updateSubscription(id, { enabled }); +} + +export async function deleteSubscription(id: string): Promise { + await unapplySubscription(id); + // Remove subscription-sourced proxy rows (force-clears their assignments). + const db = getDbInstance(); + const rows = db + .prepare("SELECT id FROM proxy_registry WHERE subscription_id = ?") + .all(id) as Array<{ id: string }>; + for (const r of rows) { + try { + await deleteProxyById(r.id, { force: true }); + } catch { + // ignore individual failures + } + } + const res = db.prepare("DELETE FROM proxy_subscriptions WHERE id = ?").run(id); + await recomputeProxyEnabled(); + return res.changes > 0; +} + +// ───────────────────────────── Scope resolution ───────────────────────────── + +// ───────────────────────────── Sync + apply ───────────────────────────── + +/** + * Refuse to fetch a subscription URL unless it is http/https to a non-internal + * host. IP literals are checked structurally; hostnames are resolved and the + * resolved addresses are re-checked (fail closed on resolution errors). This + * blocks SSRF to internal services / cloud metadata (169.254.169.254). + */ +async function assertSafeFetchTarget(url: string): Promise { + if (!isSubscriptionFetchUrlAllowed(url)) { + throw new Error("Subscription URL is not allowed (scheme or host blocked)"); + } + const host = new URL(url).hostname.toLowerCase(); + const bare = host.startsWith("[") && host.endsWith("]") ? host.slice(1, -1) : host; + if (!isIpLiteral(bare)) { + // Hostname: resolve ALL records and refuse if ANY address is internal + // (fail closed). A hostname can resolve to multiple records; checking only + // the first would let an internal IP slip through if a public record also + // exists. `lookup(..., { all: true })` returns every A/AAAA record. + try { + const dns = await import("node:dns"); + const addrs = await dns.promises.lookup(bare, { all: true }); + if (isAnyResolvedAddressBlocked(addrs)) { + throw new Error("Subscription host resolves to a blocked (internal) address"); + } + } catch (e) { + if (e instanceof Error && e.message.includes("blocked")) throw e; + throw new Error(`Subscription host resolution failed: ${e instanceof Error ? e.message : e}`); + } + } +} + +/** + * Single fetch attempt: SSRF-validate the target, fetch with manual redirect + * handling, and re-validate any redirect target. Throws on non-2xx or a + * blocked target. Each attempt owns its own timeout so a retry after a fast + * failure isn't killed by the previous attempt's timer. + */ +async function doSafeFetch(url: string, headers: Record): Promise { + await assertSafeFetchTarget(url); + const controller = new AbortController(); + const timeout = setTimeout(() => controller.abort(), SUBSCRIPTION_FETCH_TIMEOUT_MS); + try { + const res = await fetch(url, { redirect: "manual", signal: controller.signal, headers }); + if (res.status >= 300 && res.status < 400) { + const loc = res.headers.get("location"); + if (!loc) throw new Error("Subscription fetch redirected without Location"); + // Resolve relative redirects and re-validate the target (SSRF guard). + const next = new URL(loc, url).toString(); + await assertSafeFetchTarget(next); + const res2 = await fetch(next, { redirect: "manual", signal: controller.signal, headers }); + if (res2.status >= 300 && res2.status < 400) { + throw new Error("Subscription fetch: too many redirects"); + } + if (!res2.ok) { + throw new Error(`Subscription fetch failed: HTTP ${res2.status}`); + } + return await res2.text(); + } + if (!res.ok) { + throw new Error(`Subscription fetch failed: HTTP ${res.status}`); + } + return await res.text(); + } finally { + clearTimeout(timeout); + } +} + +/** + * Classify a fetch error as retryable. Transient (retry): network/timeout/DNS + * failures, HTTP 5xx, and HTTP 429. Permanent (no retry): 4xx client errors + * (except 429) and any SSRF-guard block — those will never succeed on retry. + */ +function isSubscriptionFetchRetryable(e: unknown): boolean { + const msg = e instanceof Error ? e.message : String(e); + if (msg.includes("not allowed") || msg.includes("blocked (internal)")) return false; + const m = msg.match(/HTTP (\d{3})/); + if (m) { + const code = Number(m[1]); + if (code === 429) return true; + if (code >= 500 && code < 600) return true; + return false; // 4xx (except 429) — permanent client error + } + return true; // network error / timeout / DNS failure — transient +} + +async function fetchSubscriptionContent(url: string): Promise { + const headers = { "User-Agent": "OmniRoute-ProxySubscription" }; + // Retry transient failures (timeouts, 5xx, 429) with bounded exponential + // backoff; give up fast on permanent errors (4xx, SSRF block). + return withRetry(() => doSafeFetch(url, headers), { + maxAttempts: 3, + baseDelayMs: 500, + maxDelayMs: 5000, + isRetryable: isSubscriptionFetchRetryable, + }); +} + +/** Fetch + parse + sync nodes into proxy_registry, then (if enabled) (re)bind. */ +async function syncSubscriptionUnsafe(id: string): Promise { + const sub = await getSubscriptionById(id); + if (!sub) { + return { subscriptionId: id, nodes: 0, needsCore: 0, boundProxies: 0, status: "error", error: "not found", applied: false }; + } + + let body: string; + try { + body = await fetchSubscriptionContent(sub.url); + } catch (e) { + const msg = e instanceof Error ? e.message : String(e); + const fetchConsec = (sub.consecutiveFailures || 0) + 1; + await updateSubscriptionStatus(id, "error", `Fetch failed: ${msg}`, null, new Date().toISOString(), fetchConsec); + return { subscriptionId: id, nodes: 0, needsCore: 0, boundProxies: 0, status: "error", error: msg, applied: false }; + } + + const parsed: ParsedSubscription = parseSubscription(body); + const db = getDbInstance(); + + const keptIds: string[] = []; + let warning: string | null = null; + + // The upsert loop + stale-removal are the multi-write section. upsertProxy / + // deleteProxyById each run their own internal db.transaction, so each write + // is atomic, and a re-sync is idempotent (self-heals partial state). A single + // outer db.transaction around `await` calls would NOT be atomic under + // better-sqlite3, so instead we guard against an unexpected DB error so a + // half-completed sync can never be left flagged "ok". + try { + // Directly-usable nodes → upsert into the registry as a pool. + for (const node of parsed.nodes) { + const upserted = await upsertProxy({ + name: node.name || `${sub.name} (${node.host}:${node.port})`, + type: node.type, + host: node.host, + port: node.port, + username: node.username, + password: node.password, + source: "subscription", + subscriptionId: id, + status: "active", + }); + if (upserted.proxy?.id) keptIds.push(upserted.proxy.id); + } + + // needsCore nodes → bind the operator-supplied local core endpoint (single). + if (parsed.needsCore.length > 0) { + if (sub.localCoreEndpoint && isLocalCoreEndpointAllowed(sub.localCoreEndpoint)) { + try { + const coreUrl = new URL(sub.localCoreEndpoint); + const coreType = coreUrl.protocol === "https:" ? "https" : coreUrl.protocol === "socks5:" ? "socks5" : "http"; + const upserted = await upsertProxy({ + name: `${sub.name} (local core)`, + type: coreType, + host: coreUrl.hostname, + port: Number(coreUrl.port) || (coreType === "https" ? 443 : 8080), + username: coreUrl.username ? decodeURIComponent(coreUrl.username) : undefined, + password: coreUrl.password ? decodeURIComponent(coreUrl.password) : undefined, + source: "subscription", + subscriptionId: id, + status: "active", + }); + if (upserted.proxy?.id) keptIds.push(upserted.proxy.id); + } catch { + warning = subscriptionErrorCode("LOCAL_CORE_ENDPOINT_INVALID"); + } + } else { + const nodes = parsed.needsCore + .map((n) => `${n.rawProtocol}://${n.host ?? ""}${n.port ? ":" + n.port : ""}`) + .join(", "); + warning = subscriptionErrorCode("NEEDS_CORE_NOT_CONFIGURED", nodes); + } + } + + // Remove stale subscription nodes no longer present in the fetched set. + if (keptIds.length > 0) { + const placeholders = keptIds.map(() => "?").join(","); + const stale = db + .prepare(`SELECT id FROM proxy_registry WHERE subscription_id = ? AND id NOT IN (${placeholders})`) + .all(id, ...keptIds) as Array<{ id: string }>; + for (const r of stale) { + try { + await deleteProxyById(r.id, { force: true }); + } catch { + // ignore + } + } + } else { + const stale = db + .prepare("SELECT id FROM proxy_registry WHERE subscription_id = ?") + .all(id) as Array<{ id: string }>; + for (const r of stale) { + try { + await deleteProxyById(r.id, { force: true }); + } catch { + // ignore + } + } + } + } catch (e) { + const msg = e instanceof Error ? e.message : String(e); + const writeConsec = (sub.consecutiveFailures || 0) + 1; + await updateSubscriptionStatus(id, "error", `Sync write failed: ${msg}`, null, new Date().toISOString(), writeConsec); + return { subscriptionId: id, nodes: 0, needsCore: 0, boundProxies: 0, status: "error", error: msg, applied: false }; + } + + const lastNodes = redactedNodeSummary(parsed); + + // Determine status. + let status: ProxySubscriptionStatus; + let error: string | null = warning; + if (keptIds.length === 0) { + status = "error"; + error = warning || subscriptionErrorCode("NO_USABLE_NODES"); + } else if (warning) { + status = "ok"; + } else { + status = parsed.nodes.length === 0 && parsed.needsCore.length === 0 ? "empty" : "ok"; + } + + // Reset the consecutive-failure counter on a successful (ok/empty) sync; bump + // it on error. Record the error timestamp only when there is an error. + const isErr = status === "error"; + const newConsec = isErr ? (sub.consecutiveFailures || 0) + 1 : 0; + const errAt = isErr ? new Date().toISOString() : null; + await updateSubscriptionStatus(id, status, error, lastNodes, errAt, newConsec); + + // Bind if enabled. + let applied = false; + let boundProxies = 0; + if (sub.enabled && keptIds.length > 0) { + await applySubscription(id); + applied = true; + boundProxies = keptIds.length; + } + + // Ensure the background auto-refresh ticker is running once any subscription + // has actually synced. startSubscriptionScheduler is idempotent and a no-op + // in the browser and in NODE_ENV=test. + startSubscriptionScheduler(); + + return { + subscriptionId: id, + nodes: parsed.nodes.length, + needsCore: parsed.needsCore.length, + boundProxies, + status, + error, + applied, + }; +} + +// Deduplicate concurrent syncs for the same subscription. A manual refresh and +// the scheduled ticker can otherwise fire `syncSubscription` for the same id at +// the same time and race on the upsert / stale-removal writes. +const syncInFlight = new Map>(); + +/** Public entry point: de-dupes concurrent syncs, then runs the unsafe body. */ +export async function syncSubscription(id: string): Promise { + const existing = syncInFlight.get(id); + if (existing) return existing; + const run = syncSubscriptionUnsafe(id).finally(() => { + syncInFlight.delete(id); + }); + syncInFlight.set(id, run); + return run; +} + +async function updateSubscriptionStatus( + id: string, + status: ProxySubscriptionStatus, + error: string | null, + lastNodes: unknown[] | null, + lastErrorAt: string | null, + consecutiveFailures: number +): Promise { + const db = getDbInstance(); + const now = new Date().toISOString(); + db.prepare( + `UPDATE proxy_subscriptions SET status = ?, error = ?, last_nodes = ?, last_fetched_at = ?, last_error_at = ?, consecutive_failures = ?, updated_at = ? WHERE id = ?` + ).run( + status, + error, + lastNodes ? JSON.stringify(lastNodes) : null, + now, + lastErrorAt, + consecutiveFailures, + now, + id + ); +} + +/** Bind the subscription's synced proxy pool into the target scope(s). */ +export async function applySubscription(id: string): Promise { + const sub = await getSubscriptionById(id); + if (!sub || !sub.enabled) return; + + const db = getDbInstance(); + const rows = db + .prepare("SELECT id FROM proxy_registry WHERE subscription_id = ? AND status != 'error'") + .all(id) as Array<{ id: string }>; + const ids = rows.map((r) => r.id); + if (ids.length === 0) return; + + const targets = resolveTargetScopes(sub); + for (const t of targets) { + // Add the whole subscription pool to this scope in one batched, idempotent + // write (preserves any manual proxies already in the pool). + await addProxiesToScopePool(t.scope, t.scopeId, ids); + } + + await setProxyEnabledFlag(true); +} + +/** Remove the subscription's proxies from their bound scope(s). */ +export async function unapplySubscription(id: string): Promise { + const db = getDbInstance(); + const rows = db + .prepare("SELECT id FROM proxy_registry WHERE subscription_id = ?") + .all(id) as Array<{ id: string }>; + + // Detach every subscription proxy from ALL scopes in one batched delete. + // Replaces the previous per-proxy getProxyWhereUsed + removeProxyFromScopePool + // loop (N+1 queries) — the subscription's proxies must leave every pool they + // were added to, regardless of which scope resolved them. + const ids = rows.map((r) => r.id); + if (ids.length > 0) { + const placeholders = ids.map(() => "?").join(","); + const res = db + .prepare(`DELETE FROM proxy_assignments WHERE proxy_id IN (${placeholders})`) + .run(...ids); + if (res.changes > 0) { + backupDbFile("pre-write"); + bumpProxyRegistryGeneration(); + } + } + + await recomputeProxyEnabled(); +} + +// ───────────────────────────── proxyEnabled flag ───────────────────────────── + +async function hasNonSubscriptionGlobalProxy(): Promise { + const db = getDbInstance(); + const row = db + .prepare( + "SELECT 1 FROM proxy_assignments a JOIN proxy_registry p ON p.id = a.proxy_id WHERE a.scope = 'global' AND (p.subscription_id IS NULL OR p.subscription_id = '') LIMIT 1" + ) + .get(); + return !!row; +} + +async function recomputeProxyEnabled(): Promise { + const db = getDbInstance(); + // Must check for an ACTUALLY BOUND subscription proxy, not just an enabled + // subscription row: unapplySubscription() detaches proxy_assignments without + // touching `enabled`, so a bare `enabled = 1` check would leave the flag + // permanently stuck on `true` after an unapply/disable cycle even though + // nothing is bound anymore (resolveProxyForConnectionFromRegistry correctly + // returns null, but the operator-facing proxyEnabled flag would lie). + const enabledSubWithBoundProxy = db + .prepare( + `SELECT 1 FROM proxy_subscriptions s + JOIN proxy_registry p ON p.subscription_id = s.id + JOIN proxy_assignments a ON a.proxy_id = p.id + WHERE s.enabled = 1 + LIMIT 1` + ) + .get(); + const shouldEnable = !!enabledSubWithBoundProxy || (await hasNonSubscriptionGlobalProxy()); + await setProxyEnabledFlag(shouldEnable); +} + +async function setProxyEnabledFlag(value: boolean): Promise { + const db = getDbInstance(); + db.prepare( + "INSERT OR REPLACE INTO key_value (namespace, key, value) VALUES ('settings', 'proxyEnabled', ?)" + ).run(JSON.stringify(value)); + bumpProxyConfigGeneration(); +} + +// ───────────────────────────── Auto-refresh scheduler ───────────────────────────── + +let schedulerStarted = false; +let schedulerTimer: ReturnType | null = null; + +/** Start a background ticker that refreshes enabled subscriptions on their interval. */ +export function startSubscriptionScheduler(): void { + if (schedulerStarted) return; + if (typeof window !== "undefined") return; // never in the browser + if (process.env.NODE_ENV === "test") return; // no timers during tests + schedulerStarted = true; + + const tick = async () => { + try { + const subs = await listSubscriptions(); + const now = Date.now(); + for (const s of subs) { + if (!isSubscriptionDue(s, now)) continue; + try { + await syncSubscription(s.id); + } catch (e) { + console.warn(`[ProxySubscription] refresh failed for ${s.id}: ${e instanceof Error ? e.message : e}`); + } + } + } catch (e) { + console.warn(`[ProxySubscription] scheduler tick error: ${e instanceof Error ? e.message : e}`); + } + }; + + // Check every minute; each subscription self-throttles by its interval. + schedulerTimer = setInterval(tick, 60_000); + if (typeof schedulerTimer.unref === "function") schedulerTimer.unref(); +} + +export function stopSubscriptionScheduler(): void { + if (schedulerTimer) { + clearInterval(schedulerTimer); + schedulerTimer = null; + } + schedulerStarted = false; +} diff --git a/src/lib/proxySubscription/url.ts b/src/lib/proxySubscription/url.ts new file mode 100644 index 0000000000..48304e55de --- /dev/null +++ b/src/lib/proxySubscription/url.ts @@ -0,0 +1,19 @@ +/** + * Redact credentials from a subscription URL before it is returned to the + * client (dashboard). The URL is stored intact in the DB so the server-side + * fetch can still authenticate, but the operator UI must never receive + * `user:pass@host`. + */ +export function redactSubscriptionUrl(url: string): string { + if (!url) return url; + try { + const u = new URL(url); + if (!u.username && !u.password) return url; + u.username = ""; + u.password = ""; + return u.toString(); + } catch { + // Not a parseable URL — return unchanged; the caller validates upstream. + return url; + } +} diff --git a/tests/unit/proxySubscription.coreEndpoint.test.ts b/tests/unit/proxySubscription.coreEndpoint.test.ts new file mode 100644 index 0000000000..2c4d94fbd1 --- /dev/null +++ b/tests/unit/proxySubscription.coreEndpoint.test.ts @@ -0,0 +1,34 @@ +import test from "node:test"; +import assert from "node:assert/strict"; + +const mod = await import("../../src/lib/proxySubscription/coreEndpoint.ts"); +const { isLocalCoreEndpointAllowed, ALLOWED_LOCAL_CORE_HOSTS } = mod; + +test("loopback hosts are allowed", () => { + assert.equal(isLocalCoreEndpointAllowed("socks5://127.0.0.1:1080"), true); + assert.equal(isLocalCoreEndpointAllowed("http://localhost:2080"), true); +}); + +test("remote hosts are rejected", () => { + assert.equal(isLocalCoreEndpointAllowed("socks5://10.0.0.1:1080"), false); + assert.equal(isLocalCoreEndpointAllowed("http://example.com:2080"), false); + assert.equal(isLocalCoreEndpointAllowed("https://192.168.1.1:443"), false); +}); + +test("non-proxy schemes on loopback are rejected", () => { + assert.equal(isLocalCoreEndpointAllowed("ftp://127.0.0.1:21"), false); + assert.equal(isLocalCoreEndpointAllowed("file:///tmp/core.sock"), false); +}); + +test("null / empty / malformed endpoints are rejected", () => { + assert.equal(isLocalCoreEndpointAllowed(null), false); + assert.equal(isLocalCoreEndpointAllowed(""), false); + assert.equal(isLocalCoreEndpointAllowed("not a url"), false); +}); + +test("allowed host set is loopback-only", () => { + assert.deepEqual( + [...ALLOWED_LOCAL_CORE_HOSTS].sort(), + ["127.0.0.1", "::1", "localhost"].sort() + ); +}); diff --git a/tests/unit/proxySubscription.due.test.ts b/tests/unit/proxySubscription.due.test.ts new file mode 100644 index 0000000000..bcaec5a942 --- /dev/null +++ b/tests/unit/proxySubscription.due.test.ts @@ -0,0 +1,51 @@ +import test from "node:test"; +import assert from "node:assert/strict"; + +const due = await import("../../src/lib/proxySubscription/due.ts"); +const { isSubscriptionDue } = due; + +const base = { enabled: true, lastFetchedAt: null, updateIntervalMinutes: 60 }; + +test("never-fetched subscription is immediately due", () => { + assert.equal(isSubscriptionDue({ ...base, lastFetchedAt: null }, 1_000), true); +}); + +test("disabled subscription is never due, even if stale", () => { + const lastIso = new Date(0).toISOString(); + assert.equal( + isSubscriptionDue({ enabled: false, lastFetchedAt: lastIso, updateIntervalMinutes: 60 }, 3_600_000), + false + ); +}); + +test("becomes due exactly at the interval boundary (>=)", () => { + const last = new Date("2026-01-01T00:00:00.000Z").getTime(); + const lastIso = new Date(last).toISOString(); + const intervalMs = 60 * 60_000; + // exactly one interval elapsed → due (>= comparison) + assert.equal(isSubscriptionDue({ enabled: true, lastFetchedAt: lastIso, updateIntervalMinutes: 60 }, last + intervalMs), true); + // one millisecond before → not due + assert.equal(isSubscriptionDue({ enabled: true, lastFetchedAt: lastIso, updateIntervalMinutes: 60 }, last + intervalMs - 1), false); + // half interval elapsed → not due + assert.equal(isSubscriptionDue({ enabled: true, lastFetchedAt: lastIso, updateIntervalMinutes: 60 }, last + intervalMs / 2), false); +}); + +test("respects a custom update interval", () => { + const last = new Date("2026-01-01T00:00:00.000Z").getTime(); + const lastIso = new Date(last).toISOString(); + // 10-min interval, 5 min elapsed → not due + assert.equal(isSubscriptionDue({ enabled: true, lastFetchedAt: lastIso, updateIntervalMinutes: 10 }, last + 5 * 60_000), false); + // 10-min interval, 11 min elapsed → due + assert.equal(isSubscriptionDue({ enabled: true, lastFetchedAt: lastIso, updateIntervalMinutes: 10 }, last + 11 * 60_000), true); +}); + +test("unparseable lastFetchedAt is treated as never-fetched (due)", () => { + assert.equal(isSubscriptionDue({ enabled: true, lastFetchedAt: "not-a-real-date", updateIntervalMinutes: 60 }, 1_000), true); +}); + +test("a zero/negative interval is clamped to 0 (always due once fetched)", () => { + const last = new Date("2026-01-01T00:00:00.000Z").getTime(); + const lastIso = new Date(last).toISOString(); + assert.equal(isSubscriptionDue({ enabled: true, lastFetchedAt: lastIso, updateIntervalMinutes: 0 }, last + 1), true); + assert.equal(isSubscriptionDue({ enabled: true, lastFetchedAt: lastIso, updateIntervalMinutes: -5 }, last + 1), true); +}); diff --git a/tests/unit/proxySubscription.fetchGuard.test.ts b/tests/unit/proxySubscription.fetchGuard.test.ts new file mode 100644 index 0000000000..51db57cf09 --- /dev/null +++ b/tests/unit/proxySubscription.fetchGuard.test.ts @@ -0,0 +1,89 @@ +import test from "node:test"; +import assert from "node:assert/strict"; + +const mod = await import("../../src/lib/proxySubscription/fetchGuard.ts"); +const { + isSubscriptionFetchUrlAllowed, + isIpv4Blocked, + isIpv6Blocked, + isIpLiteral, + isAnyResolvedAddressBlocked, + ALLOWED_FETCH_SCHEMES, +} = mod; + +test("public http/https URLs are allowed", () => { + assert.equal(isSubscriptionFetchUrlAllowed("https://example.com/sub"), true); + assert.equal(isSubscriptionFetchUrlAllowed("http://subs.example.org:8080/a"), true); + assert.equal(isSubscriptionFetchUrlAllowed("https://1.2.3.4/sub"), true); // public IP +}); + +test("non-http(s) schemes are rejected", () => { + assert.equal(isSubscriptionFetchUrlAllowed("ftp://example.com/x"), false); + assert.equal(isSubscriptionFetchUrlAllowed("file:///etc/passwd"), false); + assert.equal(isSubscriptionFetchUrlAllowed("gopher://example.com"), false); +}); + +test("blocked IPv4 literals are rejected", () => { + for (const ip of ["127.0.0.1", "10.0.0.5", "172.16.0.1", "192.168.1.1", "169.254.169.254", "0.0.0.0"]) { + assert.equal(isSubscriptionFetchUrlAllowed(`https://${ip}/x`), false, ip); + } +}); + +test("public IPv4 literals are allowed", () => { + assert.equal(isSubscriptionFetchUrlAllowed("https://8.8.8.8/x"), true); + assert.equal(isSubscriptionFetchUrlAllowed("http://1.1.1.1/"), true); +}); + +test("blocked IPv6 literals are rejected (bracketed)", () => { + for (const ip of ["::1", "::", "fe80::1", "fc00::1", "fd12:3456::1"]) { + assert.equal(isSubscriptionFetchUrlAllowed(`https://[${ip}]/x`), false, ip); + } +}); + +test("malformed / empty-host URLs are rejected", () => { + assert.equal(isSubscriptionFetchUrlAllowed("not a url"), false); + assert.equal(isSubscriptionFetchUrlAllowed(""), false); + assert.equal(isSubscriptionFetchUrlAllowed("http://?x"), false); // empty host +}); + +test("ip-range + literal helpers", () => { + assert.equal(isIpv4Blocked("127.0.0.1"), true); + assert.equal(isIpv4Blocked("169.254.169.254"), true); + assert.equal(isIpv4Blocked("8.8.8.8"), false); + assert.equal(isIpv6Blocked("::1"), true); + assert.equal(isIpv6Blocked("2606:4700::1111"), false); + assert.equal(isIpLiteral("127.0.0.1"), true); + assert.equal(isIpLiteral("::1"), true); + assert.equal(isIpLiteral("example.com"), false); + assert.deepEqual([...ALLOWED_FETCH_SCHEMES], ["http:", "https:"]); +}); + +test("multi-record DNS: blocks if ANY resolved address is internal", () => { + // Hostname resolves to a public AND a private address — must be refused + // (closes the first-address-only bypass). + assert.equal( + isAnyResolvedAddressBlocked([ + { address: "8.8.8.8", family: 4 }, + { address: "192.168.1.10", family: 4 }, + ]), + true + ); + // All public → allowed. + assert.equal( + isAnyResolvedAddressBlocked([ + { address: "8.8.8.8", family: 4 }, + { address: "1.1.1.1", family: 4 }, + ]), + false + ); + // A single internal IPv6 among public records → blocked. + assert.equal( + isAnyResolvedAddressBlocked([ + { address: "2606:4700::1111", family: 6 }, + { address: "fd00::1", family: 6 }, + ]), + true + ); + // Empty result set → nothing blocked. + assert.equal(isAnyResolvedAddressBlocked([]), false); +}); diff --git a/tests/unit/proxySubscription.fetchRetry.test.ts b/tests/unit/proxySubscription.fetchRetry.test.ts new file mode 100644 index 0000000000..5c44a26898 --- /dev/null +++ b/tests/unit/proxySubscription.fetchRetry.test.ts @@ -0,0 +1,77 @@ +import test from "node:test"; +import assert from "node:assert/strict"; + +const { withRetry } = await import("../../src/lib/proxySubscription/fetchRetry.ts"); + +test("withRetry returns on first success", async () => { + let calls = 0; + const r = await withRetry(async () => { + calls++; + return "ok"; + }); + assert.equal(r, "ok"); + assert.equal(calls, 1); +}); + +test("withRetry retries then succeeds", async () => { + let calls = 0; + const r = await withRetry(async () => { + calls++; + if (calls < 3) throw new Error("transient"); + return "ok"; + }); + assert.equal(r, "ok"); + assert.equal(calls, 3); +}); + +test("withRetry exhausts attempts on persistent retryable error", async () => { + let calls = 0; + await assert.rejects(async () => { + await withRetry( + async () => { + calls++; + throw new Error("boom"); + }, + { maxAttempts: 3 } + ); + }, /boom/); + assert.equal(calls, 3); +}); + +test("withRetry stops immediately on non-retryable error", async () => { + let calls = 0; + await assert.rejects(async () => { + await withRetry( + async () => { + calls++; + throw new Error("permanent"); + }, + { isRetryable: () => false } + ); + }, /permanent/); + assert.equal(calls, 1); +}); + +test("withRetry applies exponential backoff between attempts", async () => { + const delays: number[] = []; + let calls = 0; + await assert.rejects(async () => { + await withRetry( + async () => { + calls++; + throw new Error("x"); + }, + { + maxAttempts: 3, + baseDelayMs: 100, + maxDelayMs: 1000, + sleep: async (ms: number) => { + delays.push(ms); + }, + } + ); + }); + assert.equal(calls, 3); + // attempt0 → wait 100, attempt1 → wait 200 (min(cap, 100*2)); last attempt does not sleep + assert.deepEqual(delays, [100, 200]); +}); diff --git a/tests/unit/proxySubscription.needsCore.test.ts b/tests/unit/proxySubscription.needsCore.test.ts new file mode 100644 index 0000000000..4906ed7817 --- /dev/null +++ b/tests/unit/proxySubscription.needsCore.test.ts @@ -0,0 +1,50 @@ +import test from "node:test"; +import assert from "node:assert/strict"; + +const mod = await import("../../src/lib/proxySubscription/needsCore.ts"); +const { isNeedsCoreNode, countNeedsCoreNodes, NEEDS_CORE_PROTOCOLS } = mod; + +test("direct nodes (http/https/socks5) are not needs-core", () => { + assert.equal(isNeedsCoreNode({ type: "http", host: "1.2.3.4", port: 8080, rawProtocol: "http" }), false); + assert.equal(isNeedsCoreNode({ type: "socks5", rawProtocol: "socks5" }), false); + assert.equal(isNeedsCoreNode({ type: "https" }), false); +}); + +test("ss/vmess/trojan/vless/tuic/hysteria/wireguard summaries need a core", () => { + for (const p of ["ss", "vmess", "trojan", "vless", "tuic", "hysteria", "wireguard"]) { + assert.equal(isNeedsCoreNode({ rawProtocol: p, host: "1.1.1.1", port: 443 }), true, p); + } +}); + +test("an unknown protocol string is not needs-core", () => { + assert.equal(isNeedsCoreNode({ rawProtocol: "http", host: "x" }), false); + assert.equal(isNeedsCoreNode({ rawProtocol: "something-else" }), false); +}); + +test("non-object / null / undefined are not needs-core", () => { + assert.equal(isNeedsCoreNode(null), false); + assert.equal(isNeedsCoreNode(undefined), false); + assert.equal(isNeedsCoreNode("ss://x"), false); + assert.equal(isNeedsCoreNode(42), false); +}); + +test("countNeedsCoreNodes tallies only core-needed nodes", () => { + const nodes = [ + { type: "http", rawProtocol: "http" }, + { rawProtocol: "vmess" }, + { type: "socks5" }, + { rawProtocol: "trojan" }, + { rawProtocol: "notacore" }, + null, + ]; + assert.equal(countNeedsCoreNodes(nodes), 2); + assert.equal(countNeedsCoreNodes(null), 0); + assert.equal(countNeedsCoreNodes([]), 0); +}); + +test("NEEDS_CORE_PROTOCOLS covers the expected set", () => { + assert.deepEqual( + [...NEEDS_CORE_PROTOCOLS].sort(), + ["hysteria", "ss", "trojan", "tuic", "vmess", "vless", "wireguard"].sort() + ); +}); diff --git a/tests/unit/proxySubscription.parse.test.ts b/tests/unit/proxySubscription.parse.test.ts new file mode 100644 index 0000000000..a80c168ee6 --- /dev/null +++ b/tests/unit/proxySubscription.parse.test.ts @@ -0,0 +1,133 @@ +import test from "node:test"; +import assert from "node:assert/strict"; + +const parse = await import("../../src/lib/proxySubscription/parse.ts"); +const { parseSubscription, redactedNodeSummary } = parse; + +test("parses a Clash YAML subscription with direct + needs-core nodes", () => { + const yaml = ` +proxies: + - name: "Direct HTTP" + type: http + server: 1.2.3.4 + port: 8080 + username: u + password: p + - name: "Direct SOCKS5" + type: socks5 + server: 5.6.7.8 + port: 1080 + - name: "Shadowsocks node" + type: ss + server: 9.9.9.9 + port: 8388 + - name: "VMess node" + type: vmess + server: 11.11.11.11 + port: 443 +`; + const res = parseSubscription(yaml); + assert.equal(res.format, "clash-yaml"); + // http + socks5 are directly usable + assert.equal(res.nodes.length, 2); + assert.deepEqual(res.nodes.map((n) => n.type).sort(), ["http", "socks5"]); + // ss + vmess need a local core + assert.equal(res.needsCore.length, 2); + assert.deepEqual( + res.needsCore.map((n) => n.rawProtocol).sort(), + ["ss", "vmess"] + ); +}); + +test("decodes a base64-wrapped Clash YAML subscription", () => { + const yaml = `proxies: + - name: "B64 node" + type: https + server: 2.2.2.2 + port: 443 +`; + const b64 = Buffer.from(yaml, "utf-8").toString("base64"); + const res = parseSubscription(b64); + assert.equal(res.format, "base64-lines"); + assert.equal(res.nodes.length, 1); + assert.equal(res.nodes[0].type, "https"); + assert.equal(res.nodes[0].host, "2.2.2.2"); +}); + +test("parses a V2Ray-style JSON array of URI strings", () => { + const arr = [ + "socks5://127.0.0.1:1080", + "ss://example.com:8388", + "vmess://eyJhZGQiOiIxLjIuMy40IiwicG9ydCI6NDQzLCJ2IjoiMiJ9", + "trojan://secretpass@5.5.5.5:443", + "vless://uuid@6.6.6.6:443", + ]; + const res = parseSubscription(JSON.stringify(arr)); + assert.equal(res.format, "v2ray-json"); + // socks5 direct; the rest need a local core + assert.equal(res.nodes.length, 1); + assert.equal(res.nodes[0].type, "socks5"); + assert.equal(res.needsCore.length, 4); +}); + +test("parses a plain list of URI lines", () => { + const lines = [ + "http://user:pass@10.0.0.1:8080", + "socks5://10.0.0.2:1080", + "trojan://pw@10.0.0.3:443", + ].join("\n"); + const res = parseSubscription(lines); + assert.equal(res.format, "lines"); + assert.equal(res.nodes.length, 2); + assert.equal(res.needsCore.length, 1); + const http = res.nodes.find((n) => n.type === "http"); + assert.ok(http); + assert.equal(http?.username, "user"); + assert.equal(http?.password, "pass"); +}); + +test("parses Clash.Meta outbounds array", () => { + const yaml = ` +outbounds: + - name: "meta-http" + type: http + server: 3.3.3.3 + port: 8080 + - name: "meta-vless" + type: vless + server: 4.4.4.4 + port: 443 +`; + const res = parseSubscription(yaml); + assert.equal(res.nodes.length, 1); + assert.equal(res.needsCore.length, 1); + assert.equal(res.needsCore[0].rawProtocol, "vless"); +}); + +test("returns empty for blank / unrecognized input", () => { + assert.equal(parseSubscription("").format, "empty"); + const res = parseSubscription("just some random text that is not a subscription"); + assert.equal(res.format, "unknown"); + assert.equal(res.nodes.length, 0); + assert.equal(res.needsCore.length, 0); +}); + +test("redactedNodeSummary excludes secrets for direct nodes", () => { + const yaml = ` +proxies: + - name: "Direct HTTP" + type: http + server: 1.2.3.4 + port: 8080 + username: secretuser + password: secretpass +`; + const res = parseSubscription(yaml); + const summary = redactedNodeSummary(res); + assert.equal(summary.length, 1); + assert.equal(summary[0].name, "Direct HTTP"); + assert.equal(summary[0].host, "1.2.3.4"); + assert.equal("username" in summary[0], false); + assert.equal("password" in summary[0], false); + assert.equal(summary[0].hasAuth, true); +}); diff --git a/tests/unit/proxySubscription.scopes.test.ts b/tests/unit/proxySubscription.scopes.test.ts new file mode 100644 index 0000000000..96f2e117ce --- /dev/null +++ b/tests/unit/proxySubscription.scopes.test.ts @@ -0,0 +1,30 @@ +import test from "node:test"; +import assert from "node:assert/strict"; + +const mod = await import("../../src/lib/proxySubscription/scopes.ts"); +const { resolveTargetScopes } = mod; + +test("global mode binds the global scope", () => { + assert.deepEqual(resolveTargetScopes({ mode: "global" }), [{ scope: "global", scopeId: null }]); +}); + +test("rule mode with providers binds one provider scope per selected provider", () => { + assert.deepEqual( + resolveTargetScopes({ mode: "rule", ruleProviders: ["provA", "provB"] }), + [ + { scope: "provider", scopeId: "provA" }, + { scope: "provider", scopeId: "provB" }, + ] + ); +}); + +test("rule mode with no providers falls back to the global scope", () => { + assert.deepEqual( + resolveTargetScopes({ mode: "rule", ruleProviders: [] }), + [{ scope: "global", scopeId: null }] + ); + assert.deepEqual( + resolveTargetScopes({ mode: "rule", ruleProviders: null }), + [{ scope: "global", scopeId: null }] + ); +}); diff --git a/tests/unit/proxySubscription.service.test.ts b/tests/unit/proxySubscription.service.test.ts new file mode 100644 index 0000000000..a66332ef8a --- /dev/null +++ b/tests/unit/proxySubscription.service.test.ts @@ -0,0 +1,231 @@ +import test from "node:test"; +import assert from "node:assert/strict"; +import fs from "node:fs"; +import os from "node:os"; +import path from "node:path"; + +const TEST_DATA_DIR = fs.mkdtempSync(path.join(os.tmpdir(), "omniroute-sub-svc-")); +process.env.DATA_DIR = TEST_DATA_DIR; + +const core = await import("../../src/lib/db/core.ts"); +const proxies = await import("../../src/lib/db/proxies.ts"); +const sub = await import("../../src/lib/proxySubscription/index.ts"); + +function reset() { + core.resetDbInstance(); + fs.rmSync(TEST_DATA_DIR, { recursive: true, force: true }); + fs.mkdirSync(TEST_DATA_DIR, { recursive: true }); +} + +function nowIso() { + return new Date().toISOString(); +} + +function insertSubscription( + db: ReturnType, + id: string, + mode: "global" | "rule", + opts: { enabled?: boolean; ruleProviders?: string[] } = {} +) { + db.prepare( + `INSERT INTO proxy_subscriptions + (id, name, url, enabled, mode, rule_providers, update_interval_minutes, status, created_at, updated_at) + VALUES (?, ?, ?, ?, ?, ?, ?, 'empty', ?, ?)` + ).run( + id, + `sub-${id}`, + `https://example.com/${id}`, + opts.enabled === false ? 0 : 1, + mode, + opts.ruleProviders ? JSON.stringify(opts.ruleProviders) : null, + 60, + nowIso(), + nowIso() + ); +} + +test.after(() => { + core.resetDbInstance(); + fs.rmSync(TEST_DATA_DIR, { recursive: true, force: true }); +}); + +test("global subscription binds its pool to the global scope and is resolvable", async () => { + await reset(); + const db = core.getDbInstance(); + insertSubscription(db, "s1", "global", { enabled: true }); + const created = await proxies.createProxy({ + name: "node1", + type: "http", + host: "10.0.0.1", + port: 8080, + source: "subscription", + subscriptionId: "s1", + status: "active", + }); + + await sub.applySubscription("s1"); + + const resolved = await proxies.resolveProxyForConnectionFromRegistry("conn-xyz"); + assert.ok(resolved, "expected a global proxy to be resolved"); + assert.equal(resolved?.proxy.host, "10.0.0.1"); + + const flag = db + .prepare("SELECT value FROM key_value WHERE namespace='settings' AND key='proxyEnabled'") + .get() as { value: string }; + assert.equal(JSON.parse(flag.value), true); + + // Unapply -> proxy removed from global scope, proxyEnabled recomputed to false. + await sub.unapplySubscription("s1"); + const resolved2 = await proxies.resolveProxyForConnectionFromRegistry("conn-xyz"); + assert.equal(resolved2, null, "expected direct (no proxy) after unapply"); + + const flag2 = db + .prepare("SELECT value FROM key_value WHERE namespace='settings' AND key='proxyEnabled'") + .get() as { value: string }; + assert.equal(JSON.parse(flag2.value), false); +}); + +test("rule subscription binds only the selected provider scope", async () => { + await reset(); + const db = core.getDbInstance(); + insertSubscription(db, "s2", "rule", { enabled: true, ruleProviders: ["provA"] }); + const created = await proxies.createProxy({ + name: "node2", + type: "http", + host: "10.0.0.2", + port: 8080, + source: "subscription", + subscriptionId: "s2", + status: "active", + }); + + await sub.applySubscription("s2"); + + db.prepare( + "INSERT INTO provider_connections (id, provider, created_at, updated_at) VALUES (?,?,?,?)" + ).run("connA", "provA", nowIso(), nowIso()); + db.prepare( + "INSERT INTO provider_connections (id, provider, created_at, updated_at) VALUES (?,?,?,?)" + ).run("connB", "provB", nowIso(), nowIso()); + + const rA = await proxies.resolveProxyForConnectionFromRegistry("connA"); + assert.ok(rA, "provider A should resolve the rule proxy"); + assert.equal(rA?.proxy.host, "10.0.0.2"); + + const rB = await proxies.resolveProxyForConnectionFromRegistry("connB"); + assert.equal(rB, null, "provider B is not in the rule set -> direct"); +}); + +test("fail-closed: a dead subscription proxy blocks the connection instead of leaking", async () => { + await reset(); + const db = core.getDbInstance(); + insertSubscription(db, "s3", "global", { enabled: true }); + await proxies.createProxy({ + name: "deadnode", + type: "http", + host: "10.0.0.9", + port: 1, + source: "subscription", + subscriptionId: "s3", + status: "dead", + }); + + await sub.applySubscription("s3"); + + const flag = db + .prepare("SELECT value FROM key_value WHERE namespace='settings' AND key='proxyEnabled'") + .get() as { value: string }; + assert.equal(JSON.parse(flag.value), true); + + // Dead proxy is excluded from resolution -> request would go direct. + const resolved = await proxies.resolveProxyForConnectionFromRegistry("connZ"); + assert.equal(resolved, null); + + // But the operator assigned a (now dead) proxy, so this must be blocked. + const blocked = proxies.hasBlockingProxyAssignment("connZ"); + assert.equal(blocked, true); +}); + +test("deleteSubscription unbinds and removes its proxy rows", async () => { + await reset(); + const db = core.getDbInstance(); + insertSubscription(db, "s4", "global", { enabled: true }); + await proxies.createProxy({ + name: "node4", + type: "http", + host: "10.0.0.4", + port: 8080, + source: "subscription", + subscriptionId: "s4", + status: "active", + }); + await sub.applySubscription("s4"); + + const ok = await sub.deleteSubscription("s4"); + assert.equal(ok, true); + + const rows = db + .prepare("SELECT id FROM proxy_registry WHERE subscription_id = ?") + .all("s4") as Array<{ id: string }>; + assert.equal(rows.length, 0, "subscription proxy rows should be removed"); + + const assignments = db + .prepare("SELECT 1 FROM proxy_assignments a JOIN proxy_registry p ON p.id=a.proxy_id WHERE p.source='subscription' LIMIT 1") + .get(); + assert.equal(assignments, undefined, "no subscription proxy should remain assigned"); + + const subRow = db.prepare("SELECT 1 FROM proxy_subscriptions WHERE id='s4'").get(); + assert.equal(subRow, undefined); +}); + +test("global→rule switch re-evaluates binding: drops global, binds the selected provider scope", async () => { + await reset(); + const db = core.getDbInstance(); + + // Seed a proxy node directly (avoids a network fetch for this part). + await proxies.createProxy({ + name: "node5", + type: "http", + host: "10.0.0.5", + port: 8080, + source: "subscription", + subscriptionId: "s5", + status: "active", + }); + + // Start in GLOBAL mode and bind the pool to the global scope. + insertSubscription(db, "s5", "global", { enabled: true }); + await sub.applySubscription("s5"); + + const before = await proxies.resolveProxyForConnectionFromRegistry("connAny"); + assert.ok(before, "global mode should resolve the node for any connection"); + assert.equal(before?.proxy.host, "10.0.0.5"); + + // Switch to RULE mode targeting provider provA. updateSubscription must + // detach the previous global binding (unapplySubscription) and re-bind the + // pool to the selected provider scope. Stub fetch so syncSubscription's + // re-fetch returns a body that keeps node5 around. + const realFetch = globalThis.fetch; + globalThis.fetch = (async () => ({ + ok: true, + text: async () => + "proxies:\n - name: node5\n type: http\n server: 10.0.0.5\n port: 8080\n", + })) as unknown as typeof fetch; + try { + await sub.updateSubscription("s5", { mode: "rule", ruleProviders: ["provA"] }); + } finally { + globalThis.fetch = realFetch; + } + + // The global binding must be gone. + const afterGlobal = await proxies.resolveProxyForConnectionFromRegistry("connAny"); + assert.equal(afterGlobal, null, "global binding should be dropped after switching to rule mode"); + + // The provider-scope binding must now resolve the node. + db.prepare( + "INSERT INTO provider_connections (id, provider, created_at, updated_at) VALUES (?,?,?,?)" + ).run("connA", "provA", nowIso(), nowIso()); + const afterRule = await proxies.resolveProxyForConnectionFromRegistry("connA"); + assert.ok(afterRule, "rule mode should bind the node to provider provA"); + assert.equal(afterRule?.proxy.host, "10.0.0.5"); +}); diff --git a/tests/unit/proxySubscription.url.test.ts b/tests/unit/proxySubscription.url.test.ts new file mode 100644 index 0000000000..241c670224 --- /dev/null +++ b/tests/unit/proxySubscription.url.test.ts @@ -0,0 +1,30 @@ +import test from "node:test"; +import assert from "node:assert/strict"; + +const mod = await import("../../src/lib/proxySubscription/url.ts"); +const { redactSubscriptionUrl } = mod; + +test("strips user:pass from the URL", () => { + assert.equal( + redactSubscriptionUrl("https://user:pass@example.com/sub"), + "https://example.com/sub" + ); +}); + +test("strips username-only credentials", () => { + assert.equal( + redactSubscriptionUrl("https://user@example.com/x"), + "https://example.com/x" + ); +}); + +test("leaves credential-free URLs untouched", () => { + assert.equal(redactSubscriptionUrl("https://example.com/sub"), "https://example.com/sub"); + assert.equal(redactSubscriptionUrl("http://10.0.0.1:8080/sub"), "http://10.0.0.1:8080/sub"); + assert.equal(redactSubscriptionUrl("https://example.com:443/a?b=c#d"), "https://example.com:443/a?b=c#d"); +}); + +test("returns unparseable / empty input unchanged", () => { + assert.equal(redactSubscriptionUrl("not a url"), "not a url"); + assert.equal(redactSubscriptionUrl(""), ""); +});