feat(proxy): operator-level proxy subscriptions (Karing-style) — hardened, ready for review (#7299)

* feat(proxy-subscriptions): src/lib/proxySubscription/parse.ts

* feat(proxy-subscriptions): src/lib/proxySubscription/subscriptionService.ts

* feat(proxy-subscriptions): src/lib/proxySubscription/index.ts

* feat(proxy-subscriptions): src/lib/db/migrations/123_proxy_subscriptions.sql

* feat(proxy-subscriptions): src/app/api/v1/management/proxy-subscriptions/route.ts

* feat(proxy-subscriptions): src/app/api/v1/management/proxy-subscriptions/[id]/route.ts

* feat(proxy-subscriptions): src/app/api/v1/management/proxy-subscriptions/[id]/refresh/route.ts

* feat(proxy-subscriptions): src/app/api/v1/management/proxy-subscriptions/[id]/nodes/route.ts

* feat(proxy-subscriptions): src/app/(dashboard)/dashboard/settings/components/proxy/SubscriptionTab.tsx

* feat(proxy-subscriptions): tests/unit/proxySubscription.parse.test.ts

* feat(proxy-subscriptions): tests/unit/proxySubscription.service.test.ts

* feat(proxy-subscriptions): docs/proxy-subscriptions.md

* feat(proxy-subscriptions): src/lib/db/proxies/types.ts

* feat(proxy-subscriptions): src/lib/db/proxies/mappers.ts

* feat(proxy-subscriptions): src/lib/db/proxies.ts

* feat(proxy-subscriptions): src/app/(dashboard)/dashboard/settings/components/ProxyTab.tsx

* i18n(proxy-subscriptions): add proxySubscriptionsTab key + use in ProxyTab

* i18n(proxy-subscriptions): add proxySubscriptionsTab key + use in ProxyTab

* i18n(proxy-subscriptions): add proxySubscriptionsTab key + use in ProxyTab

* i18n(proxy-subscriptions): add proxySubscriptionsTab key + use in ProxyTab

* test(proxy-subscriptions): extract isSubscriptionDue + unit tests

* test(proxy-subscriptions): extract isSubscriptionDue + unit tests

* test(proxy-subscriptions): extract isSubscriptionDue + unit tests

* test(proxy-subscriptions): add global->rule switch re-bind integration test

* feat(proxy-subscriptions): inline needs-local-core guidance in SubscriptionTab

* i18n: add proxySubscriptionsTab to en (rebased on current main)

* i18n: add proxySubscriptionsTab to zh-CN (rebased on current main)

* i18n: add proxySubscriptionsTab to pt-BR (rebased on current main)

* test(proxy-subscriptions): extract needsCore detection into pure module + unit tests

* test(proxy-subscriptions): extract needsCore detection into pure module + unit tests

* test(proxy-subscriptions): extract needsCore detection into pure module + unit tests

* refactor(proxy-subscriptions): extract scopes.ts into pure module + unit tests (#65)

* refactor(proxy-subscriptions): extract coreEndpoint.ts into pure module + unit tests (#65)

* refactor(proxy-subscriptions): extract subscriptionService.ts into pure module + unit tests (#65)

* refactor(proxy-subscriptions): extract proxySubscription.scopes.test.ts into pure module + unit tests (#65)

* refactor(proxy-subscriptions): extract proxySubscription.coreEndpoint.test.ts into pure module + unit tests (#65)

* security(proxy-subscriptions): fetchGuard.ts — SSRF guard + core scheme (#P0)

* security(proxy-subscriptions): coreEndpoint.ts — SSRF guard + core scheme (#P0)

* security(proxy-subscriptions): subscriptionService.ts — SSRF guard + core scheme (#P0)

* security(proxy-subscriptions): proxySubscription.fetchGuard.test.ts — SSRF guard + core scheme (#P0)

* security(proxy-subscriptions): proxySubscription.coreEndpoint.test.ts — SSRF guard + core scheme (#P0)

* refactor(proxy-subscriptions): subscriptionService.ts — concurrency lock / resilience / url redaction (#P1)

* refactor(proxy-subscriptions): url.ts — concurrency lock / resilience / url redaction (#P1)

* refactor(proxy-subscriptions): index.ts — concurrency lock / resilience / url redaction (#P1)

* refactor(proxy-subscriptions): route.ts — concurrency lock / resilience / url redaction (#P1)

* refactor(proxy-subscriptions): route.ts — concurrency lock / resilience / url redaction (#P1)

* refactor(proxy-subscriptions): proxySubscription.url.test.ts — concurrency lock / resilience / url redaction (#P1)

* i18n(proxy-subscriptions): subscriptionService.ts — stable error codes + locale keys (#P2-6)

* i18n(proxy-subscriptions): SubscriptionTab.tsx — stable error codes + locale keys (#P2-6)

* i18n(proxy-subscriptions): en.json — stable error codes + locale keys (#P2-6)

* i18n(proxy-subscriptions): zh-CN.json — stable error codes + locale keys (#P2-6)

* i18n(proxy-subscriptions): pt-BR.json — stable error codes + locale keys (#P2-6)

* i18n(proxy-subscriptions): en.json — normalize to LF line endings (#P2-6)

* i18n(proxy-subscriptions): zh-CN.json — normalize to LF line endings (#P2-6)

* i18n(proxy-subscriptions): pt-BR.json — normalize to LF line endings (#P2-6)

* enhance(proxy-subscriptions): reject subscription fetch if ANY resolved DNS address is blocked (P3-1)

* enhance(proxy-subscriptions): add withRetry() exponential-backoff helper (P3-2)

* enhance(proxy-subscriptions): DNS multi-record guard + retry/backoff fetch + batch scope writes + observability (P3-1..P3-4)

* enhance(proxy-subscriptions): batch addProxiesToScopePool() to drop N+1 writes (P3-3)

* enhance(proxy-subscriptions): add last_error_at + consecutive_failures observability columns (P3-4)

* enhance(proxy-subscriptions): surface consecutive failures + last error time in subscription cards (P3-4)

* test(proxy-subscriptions): cover multi-record DNS SSRF (block if ANY address internal) (P3-1)

* test(proxy-subscriptions): cover withRetry() first-success / retries / backoff / non-retryable stop (P3-2)

* fix(proxy-subscriptions): resolve js-yaml import + missing backup/generation-bump imports

Two bugs made the feature non-functional and its own test suite false:

1. parse.ts used `import yaml from "js-yaml"` (default import), but
   js-yaml@^5 is ESM-only with no default export — this threw a
   SyntaxError at module load, crashing every caller (index.ts
   re-exports parse.ts, so every API route hit this too). Switch to
   `import * as yaml`, matching how the rest of the codebase already
   imports js-yaml (hermes-agent.ts, openapiParser.ts, openapi/spec
   route.ts, guide-settings route.ts).

2. subscriptionService.ts's unapplySubscription() called backupDbFile()
   and bumpProxyRegistryGeneration() without importing either —
   backupDbFile exists but wasn't imported; bumpProxyRegistryGeneration
   was a private, non-exported function in db/proxies.ts. This threw a
   ReferenceError whenever a subscription with bound proxies was
   disabled/deleted (the normal path). Import backupDbFile from
   ../db/backup and export+import bumpProxyRegistryGeneration from
   ../db/proxies, matching the identical backup+bump pattern already
   used by deleteProxyById for the same proxy_assignments/proxy_registry
   mutation shape.

Fixing both unmasked a third, previously-unreachable bug (both crashes
happened before any test assertion could run): recomputeProxyEnabled()
checked `proxy_subscriptions.enabled = 1` alone, which stays true across
an unapply/disable cycle since unapplySubscription() never touches that
column — the proxyEnabled flag would get stuck on `true` even after the
subscription's proxies were fully detached. Changed the check to require
an actually-bound proxy_assignments row for an enabled subscription,
matching hasNonSubscriptionGlobalProxy()'s existing bound-check pattern.
Traced all 3 production call sites (mode/rule switch, disable, delete)
to confirm this doesn't change their outcome — only the previously-wrong
"unapply in isolation" case.

Also fixed the test file's own pre-existing bug: its provider_connections
inserts omitted created_at/updated_at (NOT NULL, no default in the
schema since 001_initial_schema.sql), which 0 assertions had ever
reached before because the SyntaxError always crashed the file first.

All 9 proxySubscription test files now run clean: 49/49 pass (previously
2 files crashed outright at import time, 0 assertions ever ran).

Separately confirmed via testing against the pristine PR head: this PR
has 3 more pre-existing gate failures unrelated to the above (file-size
on db/proxies.ts, cognitive-complexity, complexity, changelog-integrity
vs the current release tip) plus 3 pre-existing TS2345 errors in
parse.ts (lines 228/233/263, unrelated to the yaml import). All are the
PR's own scope/base-drift, out of scope for this fix — the PR has never
had a real CI run (base=main), so no gate has ever surfaced them; they
need the base retarget + a full CI pass called out in the plan file's
own remaining mandatory items.

Co-authored-by: diegosouzapw <8016841+diegosouzapw@users.noreply.github.com>

* fix(db): renumber proxy-subscriptions migrations to avoid version collision

release/v3.8.49 already ships 123_quota_auto_ping.sql and
124_generic_session_affinity_ttl.sql; this PR's 123/124 files collided,
tripping migrationRunner's version-collision guard and failing all
proxySubscription.service tests. Renumber to 127/128 (next free slots
after 126_reasoning_routing_rules.sql).

Co-authored-by: Diego Rodrigues de Sa e Souza <diegosouzapw@users.noreply.github.com>

* refactor(db): extract proxySubscriptions + registryGeneration to keep proxies.ts under file-size cap

Moves addProxiesToScopePool to ./proxySubscriptions.ts and the registry-generation
helpers to ./proxies/registryGeneration.ts (re-exported from proxies.ts), keeping the
module under its frozen 1177-line cap after the operator-proxy-subscriptions feature.

Co-authored-by: diegosouzapw <8016841+diegosouzapw@users.noreply.github.com>

* fix(db): renumber proxy-subscriptions migrations past release collision

Rebasing onto release/v3.8.49 surfaced a migration version collision: this
branch's 127_proxy_subscriptions.sql and 128_proxy_subscriptions_meta.sql
now collide with 127_usage_history_account_identity.sql and
128_auto_candidate_overrides.sql that landed on release since this branch
last synced. Renumbered to 131/132 (next free prefixes after the current
130_remove_unregistered_qwen_data.sql) and updated the in-file header
comments to match. No schema/behavior change.

Co-authored-by: diegosouzapw <8016841+diegosouzapw@users.noreply.github.com>

---------

Co-authored-by: diegosouzapw <8016841+diegosouzapw@users.noreply.github.com>
Co-authored-by: Diego Rodrigues de Sa e Souza <diegosouza.pw@gmail.com>
Co-authored-by: Diego Rodrigues de Sa e Souza <diegosouzapw@users.noreply.github.com>
Co-authored-by: xier2012 <xier2012@users.noreply.github.com>
This commit is contained in:
Adam
2026-07-22 00:16:41 +08:00
committed by GitHub
parent 124557cc4b
commit f31f3c081e
36 changed files with 3348 additions and 16 deletions

371
docs/proxy-subscriptions.md Normal file
View File

@@ -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=<selected provider 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.

View File

@@ -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" && <ProxyPoolTab />}
{activeTab === "free-pool" && <FreePoolTab />}
{activeTab === "documentation" && <DocumentationTab />}
{activeTab === "subscriptions" && <SubscriptionTab />}
</div>
</div>
);

View File

@@ -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<SubscriptionRecord[]>([]);
const [providers, setProviders] = useState<ProviderOption[]>([]);
const [loading, setLoading] = useState(false);
const [error, setError] = useState<string | null>(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<string | null>(null);
const [showForm, setShowForm] = useState(false);
const [editingId, setEditingId] = useState<string | null>(null);
const [form, setForm] = useState<FormState>(EMPTY_FORM);
const [saving, setSaving] = useState(false);
const [formError, setFormError] = useState<string | null>(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<string, unknown>) => ({
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<SubscriptionRecord["status"], string> = {
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 (
<div className="space-y-4">
<div className="flex items-center justify-between">
<p className="text-sm text-text-muted">
Provider
</p>
{!showForm && (
<Button size="sm" variant="primary" icon="add" onClick={() => setShowForm(true)}>
</Button>
)}
</div>
{error && (
<div className="rounded-md border border-red-500/30 bg-red-500/10 px-3 py-2 text-sm text-red-600">
{error}
</div>
)}
{showForm && (
<div className="rounded-lg border border-border bg-surface p-4 space-y-3">
<div className="flex items-center justify-between">
<h3 className="text-sm font-semibold">
{editingId ? "编辑订阅" : "新增订阅"}
</h3>
<Button size="sm" variant="secondary" icon="close" onClick={resetForm}>
</Button>
</div>
<div className="grid grid-cols-1 gap-3 md:grid-cols-2">
<label className="flex flex-col gap-1 text-sm">
<span className="text-text-muted"></span>
<input
className="rounded border border-border bg-surface px-2 py-1.5 text-text outline-none focus:border-primary"
value={form.name}
onChange={(e) => setForm({ ...form, name: e.target.value })}
placeholder="例如我的订阅A"
/>
</label>
<label className="flex flex-col gap-1 text-sm">
<span className="text-text-muted"></span>
<input
className="rounded border border-border bg-surface px-2 py-1.5 text-text outline-none focus:border-primary"
value={form.url}
onChange={(e) => setForm({ ...form, url: e.target.value })}
placeholder="https://.../subscribe?token=..."
/>
</label>
</div>
<div className="grid grid-cols-1 gap-3 md:grid-cols-2">
<div className="flex flex-col gap-1 text-sm">
<span className="text-text-muted"></span>
<div className="flex gap-2">
{(["global", "rule"] as const).map((m) => (
<button
key={m}
type="button"
onClick={() => setForm({ ...form, mode: m })}
className={`px-3 py-1.5 rounded text-sm border transition-colors ${
form.mode === m
? "border-primary bg-primary/20 text-primary"
: "border-border text-text-muted hover:text-text"
}`}
>
{m === "global" ? "全局模式" : "规则模式"}
</button>
))}
</div>
<span className="text-xs text-text-muted">
{form.mode === "global"
? "所有 Provider 流量都走该订阅的代理池。"
: "仅所选 Provider 的流量走代理,其余直连。"}
</span>
</div>
<label className="flex flex-col gap-1 text-sm">
<span className="text-text-muted"> SOCKS5/HTTP </span>
<input
className="rounded border border-border bg-surface px-2 py-1.5 text-text outline-none focus:border-primary"
value={form.localCoreEndpoint}
onChange={(e) => setForm({ ...form, localCoreEndpoint: e.target.value })}
placeholder="socks5://127.0.0.1:1080"
/>
<span className="text-xs text-text-muted">
127.0.0.1 / localhostSS/VMess/Trojan/VLESS sing-box/clash
</span>
</label>
</div>
{form.mode === "rule" && (
<div className="flex flex-col gap-1 text-sm">
<span className="text-text-muted"> Provider </span>
{providers.length === 0 ? (
<span className="text-xs text-text-muted"> Provider </span>
) : (
<div className="flex flex-wrap gap-2">
{providers.map((p) => {
const checked = form.ruleProviders.includes(p.id);
return (
<button
key={p.id}
type="button"
onClick={() =>
setForm({
...form,
ruleProviders: checked
? form.ruleProviders.filter((x) => x !== p.id)
: [...form.ruleProviders, p.id],
})
}
className={`px-3 py-1 rounded text-xs border transition-colors ${
checked
? "border-primary bg-primary/20 text-primary"
: "border-border text-text-muted hover:text-text"
}`}
>
{p.name}
</button>
);
})}
</div>
)}
</div>
)}
<div className="grid grid-cols-1 gap-3 md:grid-cols-2">
<label className="flex flex-col gap-1 text-sm">
<span className="text-text-muted"></span>
<input
type="number"
min={5}
className="rounded border border-border bg-surface px-2 py-1.5 text-text outline-none focus:border-primary"
value={form.updateIntervalMinutes}
onChange={(e) =>
setForm({ ...form, updateIntervalMinutes: Number(e.target.value) || 60 })
}
/>
</label>
<label className="flex items-center gap-2 text-sm pt-6">
<input
type="checkbox"
checked={form.enabled}
onChange={(e) => setForm({ ...form, enabled: e.target.checked })}
/>
<span></span>
</label>
</div>
{formError && <div className="text-sm text-red-600">{formError}</div>}
<div className="flex justify-end gap-2">
<Button size="sm" variant="secondary" onClick={resetForm}>
</Button>
<Button size="sm" variant="primary" icon="save" onClick={save} disabled={saving}>
{saving ? "保存中…" : editingId ? "保存修改" : "创建订阅"}
</Button>
</div>
</div>
)}
<div className="space-y-2">
{loading && <p className="text-sm text-text-muted"></p>}
{!loading && subs.length === 0 && (
<p className="text-sm text-text-muted"></p>
)}
{subs.map((sub) => {
const needsCoreNodes = (sub.lastNodes ?? []).filter(isNeedsCoreNode);
const showCoreHint = needsCoreNodes.length > 0 && !sub.localCoreEndpoint;
return (
<div
key={sub.id}
className="rounded-lg border border-border bg-surface p-3 flex flex-col gap-2"
>
<div className="flex items-center justify-between gap-2">
<div className="min-w-0">
<div className="flex items-center gap-2">
<span className="font-medium truncate">{sub.name}</span>
<span
className={`text-xs px-2 py-0.5 rounded border ${
statusBadge[sub.status] || statusBadge.empty
}`}
>
{sub.status === "ok" ? "正常" : sub.status === "error" ? "错误" : "空"}
</span>
<span className="text-xs px-2 py-0.5 rounded border border-border text-text-muted">
{sub.mode === "global" ? "全局" : "规则"}
</span>
<span className="text-xs px-2 py-0.5 rounded border border-border text-text-muted">
{sub.enabled ? "已启用" : "已停用"}
</span>
</div>
<p className="text-xs text-text-muted truncate mt-1" title={sub.url}>
{sub.url}
</p>
{resolveSubError(sub.error) && (
<p className="text-xs text-amber-600 mt-1 break-words">{resolveSubError(sub.error)}</p>
)}
{showCoreHint && (
<div className="mt-2 rounded-md border border-amber-500/30 bg-amber-500/10 px-3 py-2 text-xs text-amber-700 space-y-1.5">
<p className="font-medium">
{needsCoreNodes.length} SS / VMess / Trojan / VLESS
</p>
<p>
OmniRoute <code>sing-box</code> {" "}
<code>clashClash.Meta</code> SOCKS5/HTTP 127.0.0.1 / localhost
</p>
<div className="flex flex-wrap items-center gap-2">
<code className="rounded bg-surface px-2 py-1 border border-border">socks5://127.0.0.1:2080</code>
<button
type="button"
onClick={() => navigator.clipboard?.writeText("socks5://127.0.0.1:2080")}
className="px-2 py-1 rounded border border-border hover:border-primary/50"
>
</button>
<button
type="button"
onClick={() => startEdit(sub)}
className="px-2 py-1 rounded border border-border hover:border-primary/50"
>
</button>
</div>
</div>
)}
<p className="text-xs text-text-muted mt-1">
{sub.lastNodes?.length ?? 0}
{needsCoreNodes.length > 0 ? `${needsCoreNodes.length} 个需本地内核)` : ""}
{sub.lastFetchedAt ? ` · 上次同步:${sub.lastFetchedAt}` : ""}
{sub.consecutiveFailures > 0 ? ` · 连续失败 ${sub.consecutiveFailures}` : ""}
{sub.lastErrorAt ? ` · 上次错误:${sub.lastErrorAt}` : ""}
</p>
</div>
<div className="flex items-center gap-1.5 shrink-0">
<button
type="button"
disabled={busyId === sub.id}
onClick={() => toggleEnabled(sub)}
className="px-2 py-1 text-xs rounded border border-border hover:border-primary/50"
title={sub.enabled ? "停用" : "启用"}
>
{sub.enabled ? "停用" : "启用"}
</button>
<button
type="button"
disabled={busyId === sub.id}
onClick={() => refresh(sub)}
className="px-2 py-1 text-xs rounded border border-border hover:border-primary/50"
title="刷新节点"
>
</button>
<button
type="button"
disabled={busyId === sub.id}
onClick={() => startEdit(sub)}
className="px-2 py-1 text-xs rounded border border-border hover:border-primary/50"
title="编辑"
>
</button>
<button
type="button"
disabled={busyId === sub.id}
onClick={() => remove(sub)}
className="px-2 py-1 text-xs rounded border border-red-500/30 text-red-600 hover:bg-red-500/10"
title="删除"
>
</button>
</div>
</div>
</div>
); })}
</div>
</div>
);
}

View File

@@ -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");
}
}

View File

@@ -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");
}
}

View File

@@ -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<string, unknown>;
const payload: Partial<ProxySubscriptionPayload> = {};
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");
}
}

View File

@@ -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<string, unknown>;
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");
}
}

View File

@@ -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...",

View File

@@ -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...",

View File

@@ -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": "代理池",

View File

@@ -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);

View File

@@ -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);

View File

@@ -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.

View File

@@ -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 : "",
};

View File

@@ -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;
}

View File

@@ -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 {

View File

@@ -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<number> {
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;
}

View File

@@ -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<string>(["127.0.0.1", "::1", "localhost"]);
/** Only these URL schemes denote a usable local proxy-core endpoint. */
export const ALLOWED_CORE_SCHEMES = new Set<string>(["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;
}
}

View File

@@ -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;
}

View File

@@ -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<string>(["http:", "https:"]);
// Blocked IPv4 ranges (base, mask) as 32-bit ints.
const BLOCKED_IPV4: ReadonlyArray<readonly [number, number]> = [
[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
}

View File

@@ -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<void>;
}
const DEFAULTS = { maxAttempts: 3, baseDelayMs: 500, maxDelayMs: 5000 } as const;
export async function withRetry<T>(
fn: () => Promise<T>,
options: RetryOptions = {}
): Promise<T> {
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<void>((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;
}

View File

@@ -0,0 +1,3 @@
export * from "./parse";
export * from "./subscriptionService";
export * from "./url";

View File

@@ -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<string>([
"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<string, unknown>;
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;
}

View File

@@ -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<string> = new Set<DirectProxyType>(["http", "https", "socks5"]);
const NEEDS_CORE_PROTOCOLS: ReadonlySet<RawProxyProtocol> = new Set<RawProxyProtocol>([
"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<string, unknown>): 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<string, unknown>);
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<string, unknown> | null;
if (doc && Array.isArray(doc.proxies)) {
return collectFromArray(doc.proxies, "clash-yaml");
}
if (doc && Array.isArray((doc as Record<string, unknown>).outbounds)) {
return collectFromArray((doc as Record<string, unknown>).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<string, unknown>
> {
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];
}

View File

@@ -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 }];
}

View File

@@ -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<string, unknown>;
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<ProxySubscriptionRecord[]> {
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<ProxySubscriptionRecord | null> {
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<ProxySubscriptionRecord> {
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<ProxySubscriptionPayload>
): Promise<ProxySubscriptionRecord | null> {
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<ProxySubscriptionRecord | null> {
return updateSubscription(id, { enabled });
}
export async function deleteSubscription(id: string): Promise<boolean> {
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<void> {
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<string, string>): Promise<string> {
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<string> {
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<SyncResult> {
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<string, Promise<SyncResult>>();
/** Public entry point: de-dupes concurrent syncs, then runs the unsafe body. */
export async function syncSubscription(id: string): Promise<SyncResult> {
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<void> {
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<void> {
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<void> {
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<boolean> {
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<void> {
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<void> {
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<typeof setInterval> | 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;
}

View File

@@ -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;
}
}

View File

@@ -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()
);
});

View File

@@ -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);
});

View File

@@ -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);
});

View File

@@ -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]);
});

View File

@@ -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()
);
});

View File

@@ -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);
});

View File

@@ -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 }]
);
});

View File

@@ -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<typeof core.getDbInstance>,
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");
});

View File

@@ -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(""), "");
});