diff --git a/CLAUDE.md b/CLAUDE.md index 0cc8d5b..76d115b 100644 --- a/CLAUDE.md +++ b/CLAUDE.md @@ -74,6 +74,11 @@ so the UDS default stays v1. ### ArcSwap for RouteTable `ArcSwap` gives us a pointer-swap on writes (single atomic store) and a single `load()` on reads — no lock contention on the hot path. With ≤100 routes, rebuilding the full table on `updateConfig` costs ~microseconds (Arc pointer clones only). A partial-update scheme would be more complex without meaningful benefit. +### ArcSwap for ProtectionConfig (hot-swappable) +`ProtectionState.config: ArcSwap` holds *all* protection settings as one swappable snapshot — CIDR allowlist/blocklist, JA3 blocklist, rate limit parameters, concurrency cap, handshake timeout, and requireSni are all inside `ProtectionConfig`. This means one atomic `store()` in `updateConfig` reaches all checks for a listener with zero hot-path cost. `check()` calls `config.load()` once per connection, which is a single lock-free pointer load. Callers identify listeners by port (`JsHotConfig.protection: [{ port, protection }]`). + +**Invariant:** protection *presence* (None vs Some) is fixed at listener construction — the `Option>` in `ListenerState` never changes after `start()`. `update_config` returns an error for ports that match no listener or that were started without protection. The `symphony-server` reconcile path enforces this by including `hasProtection` in the per-listener signature: none→some and some→none transitions change the signature and trigger a seamless proxy recreate (SO_REUSEPORT, no bind gap); only contents-only changes reach the hot-swap path. + ### DashMap throughout `DashMap` provides lock-free concurrent access via internal sharding. Used for: - `protection.rs`: per-IP state (`ip_table: DashMap>`) @@ -83,6 +88,23 @@ so the UDS default stays v1. ### Token bucket via AtomicU32 CAS (×1000 fixed-point) The rate limit uses a fixed-point token count (×1000) in an `AtomicU32` with CAS retry loops. `Relaxed` ordering is correct here because the token bucket is inherently approximate — a small window of double-allowing at refill time is acceptable and expected. No mutex needed on the hot path. +Each IP now has **two independent token buckets** on `IpState`: a per-second bucket (`tokens`, `last_refill_ns`) and a sustained per-minute bucket (`sustained_tokens`, `sustained_last_refill_ns`). Both use the same ×1000 fixed-point and CAS idiom. Both are checked on admission; exhausting either blocks the connection. Max sustained burst: 4,294,967 connections (u32::MAX / 1000) — far above any realistic value. + +### Monotonic clock for all timing +`now_ns()` returns a monotonic offset from a process-wide `Instant` anchor (`static START: OnceLock`), NOT wall-clock nanoseconds. All penalty deadlines and bucket timestamps use this value. Trade-off: values cannot be interpreted as unix timestamps; they are only ever compared internally. Benefit: forward NTP steps cannot release a penalty-boxed IP early and backward steps cannot freeze bucket refills. + +### Penalty box via AtomicU64 deadline +When `penaltyBox.durationMs > 0`, exhausting any rate limit sets `IpState.penalty_deadline_ns = now_ns() + duration_ns`. While `now < deadline`, connections are blocked as `PenaltyBoxed`. Each blocked attempt while boxed also debits the token buckets (lazy refill + consume); if a bucket exhausts, the deadline is reset to `now + duration_ns` (extension). The IP is readmitted once the deadline passes without further extension. Penalty state lives on `IpState` and survives config hot-swaps. A `durationMs` hot-swap affects new stamps only; existing deadlines run out on the old duration. + +### IP state eviction (spawned in start()) +`ProtectionState::evict()` is spawned as a periodic background task (60 s interval) per listener with protection, in `start()` via the `shutdown_tx` broadcast pattern. Eviction uses **lazy bucket projection** — it computes what the token level *would* be if refilled by the current time (`now - last_refill_ns`), rather than relying on the stored token value (which is only updated on access). An entry is retained if: +- It is in the penalty box (`penalty_deadline_ns > now`). +- It has active connections (`active > 0`). +- Its per-second bucket would not yet be fully refilled (projected tokens < burst_fp). +- Its sustained bucket would not yet be fully refilled (projected sustained_tokens < sustained_burst_fp). + +This prevents an attacker from resetting their sustained window by pausing long enough for the eviction interval to pass. + ### SO_REUSEPORT per worker Each tokio worker thread gets its own listening socket on the same address via `SO_REUSEPORT`. The kernel distributes incoming connections across them using a hash of the 4-tuple. This eliminates the accept lock contention that would occur with a single accepting socket + channel dispatch, and scales linearly with CPU count. diff --git a/README.md b/README.md index e52ce37..945275c 100644 --- a/README.md +++ b/README.md @@ -17,7 +17,7 @@ symphony sits in front of your services and: - **Limits** routes with per-route token-bucket rate caps to prevent any one route from starving others - **Protects** connections with per-IP token-bucket rate limiting, concurrency limits, CIDR allowlist/blocklist, JA3 fingerprint blocking, TLS handshake timeout, and SNI-required enforcement - **Suspends** routes — hold incoming connections and fire an event; your code decides whether to proxy or reject each one -- **Hot-swaps** routes and protection config without restarting or dropping existing connections +- **Hot-swaps** routes and per-listener protection config (CIDR lists, JA3 blocklist, rate limits, concurrency caps, handshake timeout, requireSni) without restarting or dropping existing connections - Scales to **~1 million concurrent connections** via `SO_REUSEPORT`, tokio's multi-thread runtime, and lock-free data structures --- @@ -137,7 +137,9 @@ Both fields accept PEM-encoded strings or `Buffer`. The cert chain may include i | Field | Type | Default | Description | |---|---|---|---| -| `rateLimit` | `{ connectionsPerSecond, burst? }` | — | Token bucket per source IP | +| `rateLimit` | `{ connectionsPerSecond, burst? }` | — | Per-second token bucket per source IP | +| `sustained` | `{ connectionsPerMinute, burst? }` | — | Per-minute token bucket per source IP (independent of `rateLimit`) | +| `penaltyBox` | `{ durationMs? }` | — | Block an IP for `durationMs` after any rate limit exhaustion | | `maxConcurrentPerIp` | `number` | `0` (unlimited) | Max simultaneous connections per source IP | | `allowlist` | `string[]` | `[]` | CIDRs that bypass all checks | | `blocklist` | `string[]` | `[]` | CIDRs that are always blocked | @@ -440,6 +442,8 @@ The **carrier depends on `sourceAddressHeader`**: ```typescript protection: { rateLimit: { connectionsPerSecond: 50, burst: 100 }, + sustained: { connectionsPerMinute: 300, burst: 300 }, + penaltyBox: { durationMs: 600_000 }, // 10 minutes maxConcurrentPerIp: 200, allowlist: ['10.0.0.0/8', '172.16.0.0/12', '192.168.0.0/16'], requireSni: true, @@ -447,6 +451,41 @@ protection: { } ``` +### Sustained rate limits + +Use `sustained` to enforce a per-minute cap that is independent of the per-second `rateLimit` bucket. Both buckets are checked on every connection; exhausting either one blocks the connection. This lets you allow short bursts while still capping total volume over longer windows: + +```typescript +protection: { + rateLimit: { connectionsPerSecond: 50, burst: 200 }, // allow short bursts + sustained: { connectionsPerMinute: 600, burst: 600 }, // but cap at 10/s long-term +} +``` + +### Penalty box + +When `penaltyBox` is configured, exhausting any rate limit places the source IP in a penalty box for `durationMs` (default 10 minutes). While penalized, all connections from that IP are rejected outright without touching the token buckets — protecting the proxy from re-assembling per-connection state under a sustained attack. + +**Extension semantics:** while an IP is boxed, symphony still debits its token buckets on each connection attempt (to measure whether the attack is continuing). If a bucket is exhausted (the IP is still sending at the excess rate), the penalty deadline is reset to `now + durationMs` — effectively extending the penalty by a full `durationMs` from the moment of continued excess. If the IP stops attacking, the buckets refill and debits succeed; the deadline is not extended, so the IP is readmitted once the original deadline expires. + +```typescript +protection: { + rateLimit: { connectionsPerSecond: 50, burst: 100 }, + penaltyBox: { durationMs: 600_000 }, // 10 minutes (default) +} +``` + +Blocked events from penalized IPs have `reason: 'penalty_boxed'`. They appear under `penaltyBoxed` in `blockedIps()`: + +```typescript +const info = proxy.blockedIps(); +// info.penaltyBoxed — IPs currently in the penalty box +``` + +Penalty state is stored on per-IP runtime state and survives a configuration hot-swap. `penaltyBox` can be added or removed via `updateConfig` without restarting. + +**Hot-swap note on active deadlines:** changing `durationMs` via `updateConfig` affects new penalty stamps immediately, but IPs already in the box retain their current deadline until it expires or is re-stamped by a continued rate-limit hit. There is no retroactive recalculation of existing deadlines. + ### JA3 blocking JA3 fingerprints the TLS ClientHello by hashing a canonical string of the version, cipher suites, extensions, elliptic curves, and EC point formats using MD5. That MD5 is by specification, not a security claim — it allows lists of known-bad fingerprints to be compared cheaply. Collect JA3 fingerprints from your logs (available in the `blocked` event `ja3` field) and add known-bad clients: @@ -479,7 +518,27 @@ Matching is case-insensitive. JA4 fingerprints are always emitted as lowercase. ### Hot-swapping protection config -Protection config is per-listener and not currently hot-swappable via `updateConfig` (listeners would need to restart). To update protection, restart with a new config. Route changes do not require listener restarts. +Protection config is per-listener and fully hot-swappable via `updateConfig`. Push a new config atomically to each listener by port — no restart needed, in-flight connections are unaffected: + +```typescript +// Block a new CIDR range without restarting +proxy.updateConfig({ + protection: [ + { + port: 443, + protection: { + blocklist: ['198.51.100.0/24'], // added under attack + rateLimit: { connectionsPerSecond: 50, burst: 100 }, + requireSni: true, + }, + }, + ], +}); +``` + +The entire `ProtectionConfig` is replaced atomically (one pointer swap). Any field not included in the new config reverts to its default. Existing per-IP rate-limit token buckets are preserved across a swap; if burst decreases, tokens are capped at the new ceiling on the next refill — no underflow. + +**Transitions (none→some, some→none):** when using `symphony-server`, adding or removing a `protection` block in the config file triggers a seamless proxy recreate via SO_REUSEPORT — no bind gap, existing connections unaffected. Only contents-only changes (same presence, different CIDR/rate/etc.) stay on the pure hot-swap path. When calling `updateConfig()` directly, a listener that was started without protection cannot gain it and returns an error; restart the listener to change protection presence. --- @@ -503,9 +562,10 @@ const m = proxy.metrics(); // m.pendingSuspended — connections currently held waiting for resolveConnection() const blocked = proxy.blockedIps(); -// blocked.rateLimited — IPs with depleted token buckets +// blocked.rateLimited — IPs with a depleted per-second or sustained token bucket // blocked.concurrencyLimited — IPs at their maxConcurrentPerIp limit // blocked.cidrBlocklist — the configured static CIDR blocklist +// blocked.penaltyBoxed — IPs currently in the penalty box setInterval(() => { console.log('active:', proxy.metrics().activeConnections); @@ -523,9 +583,9 @@ proxy.updateConfig({ }); ``` -**What can be hot-swapped:** routes (destinations, TLS certs, suspension state). +**What can be hot-swapped:** routes (destinations, TLS certs, suspension state) and per-listener protection (CIDR allowlist/blocklist, JA3 blocklist, rate limits, sustained rate limits, penalty box, concurrency caps, handshake timeout, requireSni). -**What requires a restart:** listeners (bind address, port, protection config, idle timeout). +**What requires a restart:** bind address, port, idle timeout, worker threads. When calling `updateConfig()` directly, protection presence (None↔Some) cannot change at runtime — the listener must be restarted. The `symphony-server` bin handles this automatically via seamless recreate. --- diff --git a/__test__/protection.spec.ts b/__test__/protection.spec.ts index b1b4991..49526d9 100644 --- a/__test__/protection.spec.ts +++ b/__test__/protection.spec.ts @@ -1,5 +1,5 @@ /** - * Tests for the protection layer: rate limiting, blockedIps(), JA4 blocking. + * Tests for the protection layer: rate limiting, blockedIps(), JA4 blocking, and hot-swap via updateConfig(). */ import assert from 'node:assert/strict'; @@ -240,3 +240,368 @@ describe('Protection – JA4 blocklist', () => { assert.equal(blockedReason, 'ja4_blocked', `Expected ja4_blocked, got: ${blockedReason}`); }); }); + +describe('Protection – sustained rate limit', () => { + let proxyPort: number; + let echo: Awaited>; + let proxy: SymphonyProxy; + + before(async () => { + echo = await startEchoServer(); + proxyPort = await getFreePort(); + + // 100 CPM burst 3 → only 3 connections allowed in a short window, + // even though per-second limit is generous (1000 cps). + proxy = new SymphonyProxy({ + listeners: [ + { + host: '127.0.0.1', + port: proxyPort, + protection: { + rateLimit: { connectionsPerSecond: 1000, burst: 1000 }, + sustained: { connectionsPerMinute: 100, burst: 3 }, + }, + }, + ], + routes: [ + { + sni: 'localhost', + upstreams: [{ kind: 'tcp', host: '127.0.0.1', port: echo.port }], + terminateTls: false, + }, + ], + }); + await proxy.start(); + await sleep(50); + }); + + after(async () => { + await proxy.stop(); + await echo.close(); + }); + + it('blocks after sustained burst exhausted despite low per-second rate', async () => { + // Fire 6 rapid connections — the first 3 should be allowed (burst=3), + // then the sustained bucket is empty and further connections are blocked. + for (let i = 0; i < 6; i++) { + await openAndClose(proxyPort); + } + await sleep(50); + + const info = proxy.blockedIps(); + // 127.0.0.1 should be in rateLimited after sustained burst exhaustion + assert.ok( + info.rateLimited.includes('127.0.0.1') || info.rateLimited.length > 0, + `Expected 127.0.0.1 in rateLimited after sustained exhaustion, got: ${JSON.stringify(info)}`, + ); + }); + + it('sustained limit is independent of per-second limit', async () => { + // Even though per-second allows 1000 cps, sustained burst is 3 total. + // After exhaust: 4th connection is blocked → IP appears in rateLimited. + const info = proxy.blockedIps(); + assert.ok(Array.isArray(info.rateLimited), 'rateLimited must be an array'); + // penaltyBox not configured → penaltyBoxed must be empty + assert.deepEqual(info.penaltyBoxed, [], 'penaltyBoxed must be empty (no penaltyBox config)'); + }); +}); + +describe('Protection – penalty box', () => { + let proxyPort: number; + let echo: Awaited>; + let proxy: SymphonyProxy; + + before(async () => { + echo = await startEchoServer(); + proxyPort = await getFreePort(); + + // Tiny burst (1) + short penalty (300 ms) so the test stays fast. + proxy = new SymphonyProxy({ + listeners: [ + { + host: '127.0.0.1', + port: proxyPort, + protection: { + rateLimit: { connectionsPerSecond: 2, burst: 1 }, + penaltyBox: { durationMs: 300 }, + }, + }, + ], + routes: [ + { + sni: 'localhost', + upstreams: [{ kind: 'tcp', host: '127.0.0.1', port: echo.port }], + terminateTls: false, + }, + ], + }); + await proxy.start(); + await sleep(50); + }); + + after(async () => { + await proxy.stop(); + await echo.close(); + }); + + it('blocks immediately after rate limit exhaustion and emits penalty_boxed reason', async () => { + // Exhaust burst (1 connection), then one more triggers RateLimited → enters penalty box. + // Subsequent connections emit 'blocked' with reason 'penalty_boxed'. + const penaltyReasons: string[] = []; + proxy.on('blocked', (ev) => penaltyReasons.push(ev.reason)); + + // First connection consumes the only token + await openAndClose(proxyPort); + // Second: rate limited → enters penalty box + await openAndClose(proxyPort); + // Third: should be penalty_boxed + await openAndClose(proxyPort); + await sleep(50); + + assert.ok( + penaltyReasons.includes('penalty_boxed') || penaltyReasons.includes('rate_limited'), + `Expected a blocked event, got: ${JSON.stringify(penaltyReasons)}`, + ); + }); + + it('blockedIps() lists IP in penaltyBoxed while penalty is active', async () => { + // Drain the bucket again (it may have refilled slightly after prior test) + for (let i = 0; i < 5; i++) { + await openAndClose(proxyPort); + } + await sleep(30); + + const info = proxy.blockedIps(); + assert.ok( + info.penaltyBoxed.includes('127.0.0.1') || info.rateLimited.includes('127.0.0.1'), + `Expected 127.0.0.1 in penaltyBoxed or rateLimited, got: ${JSON.stringify(info)}`, + ); + }); + + it('readmits IP after penalty expires', async () => { + // Wait for the 300 ms penalty + some margin to expire. + await sleep(500); + + // Connection should now be allowed (penalty expired, bucket refilled at 2 cps). + const connected = await new Promise((resolve) => { + const s = net.createConnection({ port: proxyPort, host: '127.0.0.1' }, () => { + s.destroy(); + resolve(true); + }); + s.on('error', () => resolve(false)); + setTimeout(() => resolve(false), 1000); + }); + // Note: "connected" reflects TCP level — the proxy may immediately close but the TCP + // connect still succeeded. We just confirm no ECONNREFUSED. + assert.ok(connected, 'Expected connection to succeed after penalty expiry'); + }); +}); + +describe('Protection – penaltyBox hot-swap via updateConfig', () => { + let proxyPort: number; + let echo: Awaited>; + let proxy: SymphonyProxy; + + before(async () => { + echo = await startEchoServer(); + proxyPort = await getFreePort(); + + // Start without penaltyBox + proxy = new SymphonyProxy({ + listeners: [ + { + host: '127.0.0.1', + port: proxyPort, + protection: { + rateLimit: { connectionsPerSecond: 2, burst: 1 }, + }, + }, + ], + routes: [ + { + sni: 'localhost', + upstreams: [{ kind: 'tcp', host: '127.0.0.1', port: echo.port }], + terminateTls: false, + }, + ], + }); + await proxy.start(); + await sleep(50); + }); + + after(async () => { + await proxy.stop(); + await echo.close(); + }); + + it('penaltyBoxed is empty before penaltyBox config is added', async () => { + // Exhaust rate limit — no penalty box yet + for (let i = 0; i < 5; i++) { + await openAndClose(proxyPort); + } + await sleep(30); + + const info = proxy.blockedIps(); + assert.deepEqual(info.penaltyBoxed, [], 'penaltyBoxed must be empty without penaltyBox config'); + }); + + it('penaltyBoxed populated after hot-swapping in penaltyBox config', async () => { + // Hot-swap in a penaltyBox config (10 s penalty — long enough to observe) + proxy.updateConfig({ + protection: [ + { + port: proxyPort, + protection: { + rateLimit: { connectionsPerSecond: 2, burst: 1 }, + penaltyBox: { durationMs: 10_000 }, + }, + }, + ], + }); + await sleep(20); + + // Exhaust rate limit to enter the penalty box + const penaltyBoxedReason = new Promise((resolve) => { + const handler = (ev: { reason: string }) => { + if (ev.reason === 'penalty_boxed') { + proxy.off('blocked', handler); + resolve(true); + } + }; + proxy.on('blocked', handler); + setTimeout(() => { + proxy.off('blocked', handler); + resolve(false); + }, 2000); + }); + + // Multiple attempts: first exhausts rate, subsequent get penalty_boxed + for (let i = 0; i < 5; i++) { + await openAndClose(proxyPort); + } + + const saw = await penaltyBoxedReason; + assert.ok(saw, 'Expected "blocked" event with reason "penalty_boxed" after hot-swap'); + + const info = proxy.blockedIps(); + assert.ok( + info.penaltyBoxed.includes('127.0.0.1'), + `Expected 127.0.0.1 in penaltyBoxed, got: ${JSON.stringify(info.penaltyBoxed)}`, + ); + }); +}); + +describe('Protection – updateConfig hot-swap', () => { + let proxyPort: number; + let echo: Awaited>; + let proxy: SymphonyProxy; + + before(async () => { + echo = await startEchoServer(); + proxyPort = await getFreePort(); + + // Start with no CIDR blocklist — connections from 127.0.0.1 are allowed + proxy = new SymphonyProxy({ + listeners: [ + { + host: '127.0.0.1', + port: proxyPort, + protection: { + blocklist: [], // initially empty + }, + }, + ], + routes: [ + { + sni: 'localhost', + upstreams: [{ kind: 'tcp', host: '127.0.0.1', port: echo.port }], + terminateTls: false, + }, + ], + }); + await proxy.start(); + await sleep(50); + }); + + after(async () => { + await proxy.stop(); + await echo.close(); + }); + + it('connection succeeds before blocklist is pushed', async () => { + // Raw TCP connect should succeed (no blocklist) + const connected = await new Promise((resolve) => { + const s = net.createConnection({ port: proxyPort, host: '127.0.0.1' }, () => { + s.destroy(); + resolve(true); + }); + s.on('error', () => resolve(false)); + setTimeout(() => resolve(false), 2000); + }); + assert.ok(connected, 'Expected connection to succeed before blocklist update'); + }); + + it('blockedIps() reflects new blocklist after updateConfig — no restart', async () => { + // Push a blocklist covering 127.0.0.1 via updateConfig (no restart) + proxy.updateConfig({ + protection: [ + { + port: proxyPort, + protection: { blocklist: ['127.0.0.1/32'] }, + }, + ], + }); + await sleep(20); + + const info = proxy.blockedIps(); + assert.ok( + info.cidrBlocklist.includes('127.0.0.1/32'), + `Expected 127.0.0.1/32 in cidrBlocklist after hot-swap, got: ${JSON.stringify(info.cidrBlocklist)}`, + ); + }); + + it('new connections from blocklisted IP are blocked after updateConfig', async () => { + // Listen for the proxy's 'blocked' event — emitted when protection blocks a connection. + // sni::peek() blocks until the client sends data or closes; s.end() (client FIN) unblocks it. + const blockedEventFired = new Promise((resolve) => { + const handler = (event: { ip: string }) => { + if (event.ip === '127.0.0.1') { + proxy.off('blocked', handler); + resolve(true); + } + }; + proxy.on('blocked', handler); + setTimeout(() => { + proxy.off('blocked', handler); + resolve(false); + }, 1000); + }); + + // Connect and immediately half-close so sni::peek() returns EOF and the protection check runs. + await new Promise((resolve) => { + const s = net.createConnection({ port: proxyPort, host: '127.0.0.1' }, () => s.end()); + s.on('close', () => resolve()); + s.on('error', () => resolve()); + }); + + assert.ok( + await blockedEventFired, + 'Expected "blocked" event for 127.0.0.1 after blocklist hot-swap', + ); + }); + + it('removes blocklist via another updateConfig — connections allowed again', async () => { + proxy.updateConfig({ + protection: [ + { + port: proxyPort, + protection: { blocklist: [] }, + }, + ], + }); + await sleep(20); + + const info = proxy.blockedIps(); + assert.equal(info.cidrBlocklist.length, 0, 'Expected cidrBlocklist to be empty after removal'); + }); +}); diff --git a/__test__/proxy.spec.ts b/__test__/proxy.spec.ts index de26af7..aa28a83 100644 --- a/__test__/proxy.spec.ts +++ b/__test__/proxy.spec.ts @@ -261,3 +261,92 @@ describe('SymphonyProxy – hot config updateConfig', () => { assert.deepEqual(response, payload); }); }); + +describe('SymphonyProxy – updateConfig atomicity (routes + protection)', () => { + // Regression for: combined update with valid routes + invalid protection must leave + // BOTH in the old state. Previously routes were swapped before protection validation ran, + // causing partial application on error. + const certA = generateSelfSignedCert('atomicity-a.test'); + let proxyPort: number; + let echoA: Awaited>; + let proxy: SymphonyProxy; + + before(async () => { + echoA = await startEchoServer(); + proxyPort = await getFreePort(); + + proxy = new SymphonyProxy({ + listeners: [ + { + host: '127.0.0.1', + port: proxyPort, + protection: { blocklist: [] }, + }, + ], + routes: [ + { + sni: 'atomicity-a.test', + upstreams: [{ kind: 'tcp', host: '127.0.0.1', port: echoA.port }], + terminateTls: true, + cert: { certChain: certA.cert, privateKey: certA.key }, + }, + ], + }); + await proxy.start(); + await sleep(50); + }); + + after(async () => { + await proxy.stop(); + await echoA.close(); + }); + + it('failed combined update (valid routes + invalid protection) leaves old routes serving', async () => { + // Confirm the initial route serves. + const before = await tlsRoundTrip({ + port: proxyPort, + servername: 'atomicity-a.test', + caCert: certA.cert, + data: Buffer.from('before-atomic-test'), + rejectUnauthorized: false, + }); + assert.deepEqual(before, Buffer.from('before-atomic-test'), 'initial route must serve'); + + // Combined update: routes would remove atomicity-a.test, protection references a non-existent port. + // The call must throw and leave both routes AND protection unchanged. + assert.throws( + () => + proxy.updateConfig({ + routes: [ + { + sni: 'atomicity-b.test', // replaces atomicity-a.test + upstreams: [{ kind: 'tcp', host: '127.0.0.1', port: echoA.port }], + terminateTls: false, + }, + ], + protection: [ + { + port: 9999, // valid u16, but no such listener → error + protection: {}, + }, + ], + }), + /port 9999 matches no listener/, + 'updateConfig must throw when protection references a non-existent port', + ); + + // Old route must still resolve — not replaced by atomicity-b.test. + const after = await tlsRoundTrip({ + port: proxyPort, + servername: 'atomicity-a.test', + caCert: certA.cert, + data: Buffer.from('after-atomic-fail'), + rejectUnauthorized: false, + }); + assert.deepEqual( + after, + Buffer.from('after-atomic-fail'), + 'old route must still serve after failed combined update', + ); + }); +}); diff --git a/__test__/server.spec.ts b/__test__/server.spec.ts index 11a794e..165e15e 100644 --- a/__test__/server.spec.ts +++ b/__test__/server.spec.ts @@ -531,6 +531,114 @@ describe('symphony-server (route cert-file read failure is isolated)', () => { }); }); +describe('symphony-server (protection hot-swap via config file)', () => { + // Verifies that adding a protection block to a previously-unprotected listener forces + // a seamless recreate (fix 1: hasProtection in listenerSig), and that removing it + // recreates again without protection. Tests both transitions end-to-end. + const cert = generateSelfSignedCert('localhost'); + let dir: string; + let configPath: string; + let statusPath: string; + let proxyPort: number; + let echo: Awaited>; + let server: RunningServer; + + const baseRoute = () => ({ + sni: 'localhost', + upstreams: [{ kind: 'tcp', host: '127.0.0.1', port: echo.port }], + terminateTls: true, + cert: { certChain: cert.cert, privateKey: cert.key }, + }); + + before(async () => { + dir = fs.mkdtempSync(path.join(os.tmpdir(), 'symphony-prot-hot-')); + configPath = path.join(dir, 'config.json'); + statusPath = path.join(dir, 'status.json'); + + echo = await startEchoServer(); + proxyPort = await getFreePort(); + + // Start WITHOUT protection — 127.0.0.1 is allowed. + writeConfigAtomic(configPath, { + version: 1, + proxies: [{ + listeners: [{ host: '127.0.0.1', port: proxyPort }], + routes: [baseRoute()], + }], + }); + + server = spawnServer(configPath, statusPath); + await waitFor(() => fs.existsSync(statusPath)); + }); + + after(async () => { + await killServer(server); + await echo.close().catch(() => {}); + fs.rmSync(dir, { recursive: true, force: true }); + }); + + it('allows TLS connections before protection is added', async () => { + const data = Buffer.from('before-protection'); + const res = await tlsRoundTrip({ + port: proxyPort, + servername: 'localhost', + caCert: cert.cert, + data, + rejectUnauthorized: true, + }); + assert.deepEqual(res, data); + }); + + it('blocks connections after adding a protection blocklist — none→some forces recreate', async () => { + // Adding a protection block to a listener that had none changes hasProtection in the + // signature → server recreates the proxy with protection enabled, covering 127.0.0.1/32. + writeConfigAtomic(configPath, { + version: 1, + proxies: [{ + listeners: [{ host: '127.0.0.1', port: proxyPort, protection: { blocklist: ['127.0.0.1/32'] } }], + routes: [baseRoute()], + }], + }); + + // Poll until TLS from 127.0.0.1 is rejected (blocked pre-handshake). + await waitFor( + async () => { + try { + await tlsRoundTrip({ port: proxyPort, servername: 'localhost', caCert: cert.cert, data: Buffer.from('p'), rejectUnauthorized: true }); + return false; // still allowed — recreate not picked up yet + } catch { + return true; // blocked + } + }, + 8000, 150 + ); + }); + + it('allows connections again after removing protection — some→none forces another recreate', async () => { + writeConfigAtomic(configPath, { + version: 1, + proxies: [{ + listeners: [{ host: '127.0.0.1', port: proxyPort }], // no protection + routes: [baseRoute()], + }], + }); + + const data = Buffer.from('after-protection-removed'); + let res: Buffer | null = null; + await waitFor( + async () => { + try { + const r = await tlsRoundTrip({ port: proxyPort, servername: 'localhost', caCert: cert.cert, data, rejectUnauthorized: true }); + if (r.length === data.length && Buffer.compare(r, data) === 0) { res = r; return true; } + return false; + } catch { return false; } + }, + 8000, 150 + ); + assert.deepEqual(res, data, `traffic not unblocked. stderr:\n${server.getStderr()}`); + }); +}); + describe('symphony-server (status.json ownership guard)', () => { // During a version upgrade the replacement starts first (SO_REUSEPORT overlap) and // rewrites status.json with its own pid before the incumbent retires. stop() must only diff --git a/src/protection.rs b/src/protection.rs index 8a4861d..beefbcb 100644 --- a/src/protection.rs +++ b/src/protection.rs @@ -5,78 +5,170 @@ use ipnetwork::IpNetwork; use std::collections::HashSet; use std::net::IpAddr; use std::sync::atomic::{AtomicU32, AtomicU64, Ordering}; -use std::sync::Arc; -use std::time::{Duration, SystemTime, UNIX_EPOCH}; +use std::sync::{Arc, OnceLock}; +use std::time::{Duration, Instant}; + +// Process-wide monotonic anchor. All timing uses monotonic offsets from process start. +// Trade-off vs wall-clock: deadlines and bucket timestamps cannot be compared to unix time, +// but forward NTP steps no longer release penalty-boxed IPs early and backward steps no +// longer freeze legitimate bucket refills. Values are only compared internally. +static START: OnceLock = OnceLock::new(); fn now_ns() -> u64 { - SystemTime::now() - .duration_since(UNIX_EPOCH) - .unwrap_or(Duration::ZERO) + Instant::now() + .duration_since(*START.get_or_init(Instant::now)) .as_nanos() as u64 } // ── Configuration ───────────────────────────────────────────────────────────── +/// Snapshot of all protection settings for a listener. +/// Stored inside ArcSwap so every field is hot-swappable in one atomic pointer store. #[derive(Clone, Debug, Default)] pub struct ProtectionConfig { - /// Token bucket: max new connections/second per IP + /// Per-second token bucket: max new connections/second per IP. pub rate_limit_cps: Option, - /// Token bucket burst (defaults to rate_limit_cps if not set) + /// Per-second burst ceiling (defaults to rate_limit_cps if not set). pub rate_limit_burst: Option, - /// Max simultaneous connections per IP (0 = unlimited) + /// Sustained token bucket: max new connections/minute per IP. + /// Independent of the per-second bucket; either exhausting → block. + pub sustained_cpm: Option, + /// Sustained burst ceiling (defaults to sustained_cpm if not set). + /// Max useful value: 4,294,967 connections (u32::MAX / 1000 fixed-point units). + pub sustained_burst: Option, + /// Penalty box duration in ms. 0 = penalty box disabled (default). + /// When > 0: exhausting any rate limit places the IP in the penalty box for this duration. + /// While boxed, all connections are blocked. If the IP continues to hit rate limits while + /// boxed, the penalty is extended (reset to full duration from now). + pub penalty_box_duration_ms: u64, + /// Max simultaneous connections per IP (0 = unlimited). pub max_concurrent_per_ip: u32, - /// JA3 fingerprints to block (raw 16-byte MD5) + /// JA3 fingerprints to block (raw 16-byte MD5). pub ja3_blocklist: HashSet<[u8; 16]>, /// JA4 fingerprints to block. Values are stored lowercase; comparison is case-insensitive. /// JA4 strings are 36-char lowercase ASCII: t_<12hex>_<12hex>. pub ja4_blocklist: HashSet>, - /// Max ms for TLS handshake (0 = use default 10000) + /// Max ms for TLS handshake (0 = use default 10000). pub tls_handshake_timeout_ms: u64, - /// Reject connections without SNI + /// Reject connections without SNI. pub require_sni: bool, + /// CIDRs that bypass all protection checks. + pub allowlist: Vec, + /// CIDRs that are always blocked. + pub blocklist: Vec, + + // Precomputed derived constants — populated by ProtectionConfig::precompute(). + // These replace the method-based lookups on the hot check() path. + /// Per-second token refill rate (tokens per nanosecond). 0.0 = no limit. + pub tokens_per_ns: f64, + /// Per-second burst ceiling in fixed-point (×1000). 0 = no limit. + pub burst_fp: u32, + /// Sustained token refill rate (tokens per nanosecond). 0.0 = no limit. + pub sustained_tokens_per_ns: f64, + /// Sustained burst ceiling in fixed-point (×1000). 0 = no limit. + pub sustained_burst_fp: u32, } impl ProtectionConfig { - pub fn tls_handshake_timeout(&self) -> Duration { - let ms = if self.tls_handshake_timeout_ms == 0 { - 10_000 - } else { - self.tls_handshake_timeout_ms - }; - Duration::from_millis(ms) - } - - /// Token refill rate (tokens per nanosecond), or None if no rate limit. - pub fn tokens_per_ns(&self) -> Option { - self.rate_limit_cps.map(|cps| cps / 1_000_000_000.0) + /// Compute and cache float-derived constants from the source fields. + /// Must be called after setting all rate-limit fields, before ArcSwap storage. + pub fn precompute(mut self) -> Self { + self.burst_fp = (self.rate_limit_burst.or(self.rate_limit_cps).unwrap_or(0.0) * 1000.0) as u32; + self.tokens_per_ns = self.rate_limit_cps.map_or(0.0, |cps| cps / 1_000_000_000.0); + self.sustained_burst_fp = + (self.sustained_burst.or(self.sustained_cpm).unwrap_or(0.0) * 1000.0) as u32; + // 1 minute = 60_000_000_000 ns + self.sustained_tokens_per_ns = self.sustained_cpm.map_or(0.0, |cpm| cpm / 60_000_000_000.0); + self } - /// Burst ceiling in fixed-point (×1000). - pub fn burst_fp(&self) -> u32 { - let burst = self - .rate_limit_burst - .or(self.rate_limit_cps) - .unwrap_or(0.0); - (burst * 1000.0) as u32 + pub fn tls_handshake_timeout(&self) -> Duration { + let ms = if self.tls_handshake_timeout_ms == 0 { 10_000 } else { self.tls_handshake_timeout_ms }; + Duration::from_millis(ms) } } // ── Per-IP state ────────────────────────────────────────────────────────────── -struct IpState { - /// Token count in fixed-point ×1000. Max = burst_fp. +#[derive(Debug)] +pub(crate) struct IpState { + /// Per-second token count in fixed-point ×1000. Max = burst_fp. tokens: AtomicU32, last_refill_ns: AtomicU64, + /// Sustained (per-minute) token count in fixed-point ×1000. Max = sustained_burst_fp. + sustained_tokens: AtomicU32, + sustained_last_refill_ns: AtomicU64, /// Current active connections from this IP. - active: AtomicU32, + pub(crate) active: AtomicU32, + /// Penalty box deadline in monotonic ns (relative to START). 0 = not penalized. + penalty_deadline_ns: AtomicU64, } impl IpState { - fn new(burst_fp: u32) -> Self { + fn new(burst_fp: u32, sustained_burst_fp: u32) -> Self { + let now = now_ns(); Self { tokens: AtomicU32::new(burst_fp), - last_refill_ns: AtomicU64::new(now_ns()), + last_refill_ns: AtomicU64::new(now), + sustained_tokens: AtomicU32::new(sustained_burst_fp), + sustained_last_refill_ns: AtomicU64::new(now), active: AtomicU32::new(0), + penalty_deadline_ns: AtomicU64::new(0), + } + } + + /// Decrement the active counter. Called via the held Arc on connection close. + pub(crate) fn release(&self) { + self.active + .fetch_update(Ordering::Relaxed, Ordering::Relaxed, |v| Some(v.saturating_sub(1))) + .ok(); + } +} + +// ── Token bucket helpers ─────────────────────────────────────────────────────── + +/// Refills a token bucket from elapsed time, then tries to consume one token (1000 fp units). +/// Returns true if a token was consumed (connection allowed), false if bucket was empty. +/// Lock-free: uses Relaxed CAS retry loops. +fn refill_and_consume( + tokens: &AtomicU32, + last_refill_ns: &AtomicU64, + now: u64, + rate_per_ns: f64, + burst_fp: u32, +) -> bool { + // Refill phase — caps at burst_fp to handle burst decreases without underflow. + loop { + let last = last_refill_ns.load(Ordering::Relaxed); + let elapsed = now.saturating_sub(last); + let refill = ((elapsed as f64) * rate_per_ns * 1000.0) as u32; + if refill == 0 { + break; + } + let old_tokens = tokens.load(Ordering::Relaxed); + let new_tokens = old_tokens.saturating_add(refill).min(burst_fp); + if tokens + .compare_exchange(old_tokens, new_tokens, Ordering::Relaxed, Ordering::Relaxed) + .is_ok() + { + // Only the first CAS winner advances the refill timestamp; losers retry from above. + let _ = last_refill_ns.compare_exchange(last, now, Ordering::Relaxed, Ordering::Relaxed); + break; + } + } + + // Consume phase — 1 token = 1000 fixed-point units + const ONE_TOKEN: u32 = 1000; + loop { + let t = tokens.load(Ordering::Relaxed); + if t < ONE_TOKEN { + return false; + } + if tokens + .compare_exchange(t, t - ONE_TOKEN, Ordering::Relaxed, Ordering::Relaxed) + .is_ok() + { + return true; } } } @@ -91,6 +183,8 @@ pub enum BlockReason { NoSni, RateLimited, TooManyConnections, + /// IP is in the penalty box — placed here by prior rate limit exhaustion. + PenaltyBoxed, } impl BlockReason { @@ -102,15 +196,17 @@ impl BlockReason { Self::NoSni => "no_sni", Self::RateLimited => "rate_limited", Self::TooManyConnections => "too_many_connections", + Self::PenaltyBoxed => "penalty_boxed", } } } #[derive(Debug)] pub enum Decision { - /// Connection allowed and active counter incremented — release() must be called on close. - Allow, - /// Connection allowed via allowlist — active counter was NOT incremented; release() is a no-op. + /// Connection allowed and active counter incremented. + /// The caller MUST hold this Arc until connection close and call ip_state.release() on drop. + Allow(Arc), + /// Connection allowed via allowlist — active counter was NOT incremented. AllowBypassed, Block(BlockReason), } @@ -118,40 +214,39 @@ pub enum Decision { // ── ProtectionState ─────────────────────────────────────────────────────────── pub struct ProtectionState { + /// All protection settings in one ArcSwap snapshot — a single store() reaches all checks. pub config: ArcSwap, ip_table: DashMap>, - allowlist: Vec, - blocklist: Vec, } impl ProtectionState { - pub fn new( - config: ProtectionConfig, - allowlist: Vec, - blocklist: Vec, - ) -> Arc { + pub fn new(config: ProtectionConfig) -> Arc { Arc::new(Self { - config: ArcSwap::new(Arc::new(config)), + config: ArcSwap::new(Arc::new(config.precompute())), ip_table: DashMap::new(), - allowlist, - blocklist, }) } /// Check whether to allow or block a new connection. - /// If allowed, increments the IP's active counter. + /// On Allow, increments the IP's active counter and returns the IpState Arc the caller + /// must hold until connection close (call ip_state.release() on drop). pub fn check(&self, peer_ip: IpAddr, peek_info: &PeekInfo) -> Decision { + self.check_at(peer_ip, peek_info, now_ns()) + } + + /// Internal: same as check() but accepts an explicit `now_ns` timestamp for testing. + pub(crate) fn check_at(&self, peer_ip: IpAddr, peek_info: &PeekInfo, now: u64) -> Decision { let cfg = self.config.load(); - // 1. Allowlist — skip all other checks; active counter is NOT incremented - for network in &self.allowlist { + // 1. Allowlist — skip all other checks; active counter is NOT incremented. + for network in &cfg.allowlist { if network.contains(peer_ip) { return Decision::AllowBypassed; } } // 2. Blocklist - for network in &self.blocklist { + for network in &cfg.blocklist { if network.contains(peer_ip) { return Decision::Block(BlockReason::CidrBlocked); } @@ -179,58 +274,83 @@ impl ProtectionState { return Decision::Block(BlockReason::NoSni); } - // 5 & 6: Rate limit + concurrency — access IpState once - let state = self.get_or_create_state(peer_ip, cfg.burst_fp()); - - // 5. Token bucket rate limit - if let Some(rate) = cfg.tokens_per_ns() { - let now = now_ns(); - let burst_fp = cfg.burst_fp(); + // 5–8: IP-state checks — access IpState once + let state = self.get_or_create_state(peer_ip, cfg.burst_fp, cfg.sustained_burst_fp); - // Refill tokens - loop { - let last = state.last_refill_ns.load(Ordering::Relaxed); - let elapsed = now.saturating_sub(last); - let refill = ((elapsed as f64) * rate * 1000.0) as u32; - if refill == 0 { - break; + // 5. Penalty box — if the IP is currently penalized, debit buckets to detect continued + // excess and extend the deadline if found, then block outright. + let penalty_ms = cfg.penalty_box_duration_ms; + if penalty_ms > 0 { + let deadline = state.penalty_deadline_ns.load(Ordering::Relaxed); + if deadline > 0 && now < deadline { + // Debit both rate buckets to measure "continues to exceed". + // If either bucket is exhausted, the attacker is still hitting hard → extend. + let mut exceeded = false; + if cfg.tokens_per_ns > 0.0 + && !refill_and_consume( + &state.tokens, + &state.last_refill_ns, + now, + cfg.tokens_per_ns, + cfg.burst_fp, + ) { + exceeded = true; + } + if cfg.sustained_tokens_per_ns > 0.0 + && !refill_and_consume( + &state.sustained_tokens, + &state.sustained_last_refill_ns, + now, + cfg.sustained_tokens_per_ns, + cfg.sustained_burst_fp, + ) { + exceeded = true; } - let old_tokens = state.tokens.load(Ordering::Relaxed); - let new_tokens = old_tokens.saturating_add(refill).min(burst_fp); - // CAS: update tokens and last_refill atomically - if state - .tokens - .compare_exchange(old_tokens, new_tokens, Ordering::Relaxed, Ordering::Relaxed) - .is_ok() - { - // CAS on the timestamp so only the first winner advances the refill window. - // A losing compare_exchange here is harmless — another thread already wrote it. - let _ = state.last_refill_ns.compare_exchange( - last, now, Ordering::Relaxed, Ordering::Relaxed, + if exceeded { + // Extend deadline monotonically — fetch_max prevents inter-thread now_ns + // skew from regressing a deadline set by a concurrent writer. + state.penalty_deadline_ns.fetch_max( + now.saturating_add(penalty_ms.saturating_mul(1_000_000)), + Ordering::Relaxed, ); - break; } - // CAS failed — another thread beat us; retry + return Decision::Block(BlockReason::PenaltyBoxed); } + } - // Try to consume 1 token (= 1000 fixed-point units) - const ONE_TOKEN: u32 = 1000; - loop { - let tokens = state.tokens.load(Ordering::Relaxed); - if tokens < ONE_TOKEN { - return Decision::Block(BlockReason::RateLimited); - } - if state - .tokens - .compare_exchange(tokens, tokens - ONE_TOKEN, Ordering::Relaxed, Ordering::Relaxed) - .is_ok() - { - break; - } + // 6. Per-second rate limit + if cfg.tokens_per_ns > 0.0 + && !refill_and_consume(&state.tokens, &state.last_refill_ns, now, cfg.tokens_per_ns, cfg.burst_fp) + { + if penalty_ms > 0 { + state.penalty_deadline_ns.fetch_max( + now.saturating_add(penalty_ms.saturating_mul(1_000_000)), + Ordering::Relaxed, + ); } + return Decision::Block(BlockReason::RateLimited); } - // 6. Concurrency limit — atomic test-and-increment to avoid TOCTOU + // 7. Sustained rate limit (per-minute) + if cfg.sustained_tokens_per_ns > 0.0 + && !refill_and_consume( + &state.sustained_tokens, + &state.sustained_last_refill_ns, + now, + cfg.sustained_tokens_per_ns, + cfg.sustained_burst_fp, + ) + { + if penalty_ms > 0 { + state.penalty_deadline_ns.fetch_max( + now.saturating_add(penalty_ms.saturating_mul(1_000_000)), + Ordering::Relaxed, + ); + } + return Decision::Block(BlockReason::RateLimited); + } + + // 8. Concurrency limit — atomic test-and-increment to avoid TOCTOU if cfg.max_concurrent_per_ip > 0 { let max = cfg.max_concurrent_per_ip; let result = state.active.fetch_update(Ordering::Relaxed, Ordering::Relaxed, |v| { @@ -243,62 +363,197 @@ impl ProtectionState { } else { state.active.fetch_add(1, Ordering::Relaxed); } - Decision::Allow + Decision::Allow(state) } - /// Decrement the active counter for a peer IP. Call on connection close. + /// Decrement the active counter for a peer IP. Used by tests; production code uses + /// the Arc returned from check() so the decrement hits the same entry even + /// if an eviction ran between admission and close. + #[cfg(test)] pub fn release(&self, peer_ip: IpAddr) { if let Some(state) = self.ip_table.get(&peer_ip) { - state - .active - .fetch_update(Ordering::Relaxed, Ordering::Relaxed, |v| Some(v.saturating_sub(1))) - .ok(); + state.release(); } } - /// Returns (rate_limited_ips, concurrency_limited_ips) for the `blockedIps()` API. - pub fn blocked_ips(&self, max_concurrent: u32) -> (Vec, Vec) { + /// Returns (rate_limited_ips, concurrency_limited_ips, penalty_boxed_ips) for the `blockedIps()` API. + pub fn blocked_ips(&self) -> (Vec, Vec, Vec) { let cfg = self.config.load(); - let burst_fp = cfg.burst_fp(); + let max_concurrent = cfg.max_concurrent_per_ip; + let now = now_ns(); let mut rate_limited = Vec::new(); let mut concurrency_limited = Vec::new(); + let mut penalty_boxed = Vec::new(); for entry in self.ip_table.iter() { let ip = *entry.key(); let state = entry.value(); - if cfg.tokens_per_ns().is_some() && state.tokens.load(Ordering::Relaxed) < 1000 { + // Penalty box: only report when the feature is currently enabled in config. + // After a hot-swap that disables penaltyBox, check() stops enforcing it but + // stale deadlines remain on IpState until expiry — gate on config to avoid + // reporting IPs that are no longer actually blocked. + if cfg.penalty_box_duration_ms > 0 { + let deadline = state.penalty_deadline_ns.load(Ordering::Relaxed); + if deadline > 0 && now < deadline { + penalty_boxed.push(ip); + continue; + } + } + + // Rate-limited: apply lazy projection (same logic as evict_at) so a recovered + // idle IP isn't falsely reported as still limited. + let mut is_rate_limited = false; + if cfg.tokens_per_ns > 0.0 { + let last = state.last_refill_ns.load(Ordering::Relaxed); + let current = state.tokens.load(Ordering::Relaxed); + let refill = ((now.saturating_sub(last) as f64) * cfg.tokens_per_ns * 1000.0) as u32; + let projected = current.saturating_add(refill).min(cfg.burst_fp); + if projected < 1000 { + is_rate_limited = true; + } + } + if !is_rate_limited && cfg.sustained_tokens_per_ns > 0.0 { + let last = state.sustained_last_refill_ns.load(Ordering::Relaxed); + let current = state.sustained_tokens.load(Ordering::Relaxed); + let refill = + ((now.saturating_sub(last) as f64) * cfg.sustained_tokens_per_ns * 1000.0) as u32; + let projected = current.saturating_add(refill).min(cfg.sustained_burst_fp); + if projected < 1000 { + is_rate_limited = true; + } + } + if is_rate_limited { rate_limited.push(ip); } if max_concurrent > 0 { let active = state.active.load(Ordering::Relaxed); - // Track IPs that are AT the limit (>= max) — they'd be blocked - if active >= max_concurrent && burst_fp > 0 { + // Track IPs that are AT the limit (>= max) — they'd be blocked on next connect. + if active >= max_concurrent { concurrency_limited.push(ip); } } } - (rate_limited, concurrency_limited) + (rate_limited, concurrency_limited, penalty_boxed) + } + + /// Returns (rate_limited_ips, concurrency_limited_ips, penalty_boxed_ips) at a given time. + /// Test seam for fix 5 verification. + #[cfg(test)] + pub(crate) fn blocked_ips_at(&self, now: u64) -> (Vec, Vec, Vec) { + let cfg = self.config.load(); + let max_concurrent = cfg.max_concurrent_per_ip; + let mut rate_limited = Vec::new(); + let mut concurrency_limited = Vec::new(); + let mut penalty_boxed = Vec::new(); + + for entry in self.ip_table.iter() { + let ip = *entry.key(); + let state = entry.value(); + + if cfg.penalty_box_duration_ms > 0 { + let deadline = state.penalty_deadline_ns.load(Ordering::Relaxed); + if deadline > 0 && now < deadline { + penalty_boxed.push(ip); + continue; + } + } + + let mut is_rate_limited = false; + if cfg.tokens_per_ns > 0.0 { + let last = state.last_refill_ns.load(Ordering::Relaxed); + let current = state.tokens.load(Ordering::Relaxed); + let refill = ((now.saturating_sub(last) as f64) * cfg.tokens_per_ns * 1000.0) as u32; + if current.saturating_add(refill).min(cfg.burst_fp) < 1000 { + is_rate_limited = true; + } + } + if !is_rate_limited && cfg.sustained_tokens_per_ns > 0.0 { + let last = state.sustained_last_refill_ns.load(Ordering::Relaxed); + let current = state.sustained_tokens.load(Ordering::Relaxed); + let refill = + ((now.saturating_sub(last) as f64) * cfg.sustained_tokens_per_ns * 1000.0) as u32; + if current.saturating_add(refill).min(cfg.sustained_burst_fp) < 1000 { + is_rate_limited = true; + } + } + if is_rate_limited { + rate_limited.push(ip); + } + if max_concurrent > 0 && state.active.load(Ordering::Relaxed) >= max_concurrent { + concurrency_limited.push(ip); + } + } + + (rate_limited, concurrency_limited, penalty_boxed) } - /// Evict idle, fully-refilled IpState entries to bound memory. + /// Evict idle IpState entries to bound ip_table memory growth. /// Called by the background eviction task every 60 seconds. pub fn evict(&self) { + self.evict_at(now_ns()); + } + + /// Internal: same as evict() but accepts an explicit `now_ns` for testing. + /// + /// Retains an entry if ANY of: + /// - It is in the penalty box (penalty_deadline_ns > now). + /// - It has active connections. + /// - Its per-second bucket would not yet be fully refilled (lazy projection from last_refill_ns). + /// - Its sustained bucket would not yet be fully refilled (lazy projection). + /// + /// Lazy projection means an attacker cannot reset their sustained window by pausing + /// until eviction: the entry stays until the full burst_fp would be recovered. + pub(crate) fn evict_at(&self, now: u64) { let cfg = self.config.load(); - let burst_fp = cfg.burst_fp(); + self.ip_table.retain(|_, state| { - // Keep entry if: there are active connections, OR the bucket is not full - state.active.load(Ordering::Relaxed) > 0 - || (burst_fp > 0 && state.tokens.load(Ordering::Relaxed) < burst_fp) + // Keep: in the penalty box — but only when penaltyBox is currently enabled. + // A stale deadline left from a prior enabled window must not pin the entry + // after penaltyBox is disabled; otherwise re-enabling resurrects old deadlines. + let deadline = state.penalty_deadline_ns.load(Ordering::Relaxed); + if cfg.penalty_box_duration_ms > 0 && deadline > 0 && now < deadline { + return true; + } + + // Keep: active connections + if state.active.load(Ordering::Relaxed) > 0 { + return true; + } + + // Keep: per-second bucket would not yet be fully refilled + if cfg.tokens_per_ns > 0.0 { + let burst_fp = cfg.burst_fp; + let last = state.last_refill_ns.load(Ordering::Relaxed); + let current = state.tokens.load(Ordering::Relaxed); + let refill = ((now.saturating_sub(last) as f64) * cfg.tokens_per_ns * 1000.0) as u32; + if current.saturating_add(refill).min(burst_fp) < burst_fp { + return true; + } + } + + // Keep: sustained bucket would not yet be fully refilled + if cfg.sustained_tokens_per_ns > 0.0 { + let burst_fp = cfg.sustained_burst_fp; + let last = state.sustained_last_refill_ns.load(Ordering::Relaxed); + let current = state.sustained_tokens.load(Ordering::Relaxed); + let refill = + ((now.saturating_sub(last) as f64) * cfg.sustained_tokens_per_ns * 1000.0) as u32; + if current.saturating_add(refill).min(burst_fp) < burst_fp { + return true; + } + } + + false // evict: idle, not penalized, both buckets would be fully refilled }); } - fn get_or_create_state(&self, ip: IpAddr, burst_fp: u32) -> Arc { + fn get_or_create_state(&self, ip: IpAddr, burst_fp: u32, sustained_burst_fp: u32) -> Arc { if let Some(s) = self.ip_table.get(&ip) { return s.clone(); } - let state = Arc::new(IpState::new(burst_fp)); + let state = Arc::new(IpState::new(burst_fp, sustained_burst_fp)); self.ip_table.entry(ip).or_insert_with(|| state).clone() } } @@ -324,3 +579,782 @@ fn from_hex_digit(b: u8) -> Option { _ => None, } } + +#[cfg(test)] +mod tests { + use super::*; + use crate::sni::PeekInfo; + + fn ip(s: &str) -> IpAddr { + s.parse().unwrap() + } + + fn no_peek() -> PeekInfo { + PeekInfo::default() + } + + fn peek_with_ja3(hex: &str) -> PeekInfo { + PeekInfo { ja3: hex.to_string(), ..Default::default() } + } + + #[test] + fn cidr_block_hot_swap_adds_block() { + let state = ProtectionState::new(ProtectionConfig::default()); + let peer = ip("10.0.0.1"); + + // Initially allowed + assert!(matches!(state.check(peer, &no_peek()), Decision::Allow(_))); + state.release(peer); + + // Swap in a blocklist containing the peer + state.config.store(Arc::new( + ProtectionConfig { + blocklist: vec!["10.0.0.0/24".parse().unwrap()], + ..Default::default() + } + .precompute(), + )); + assert!(matches!(state.check(peer, &no_peek()), Decision::Block(BlockReason::CidrBlocked))); + } + + #[test] + fn cidr_block_hot_swap_removes_block() { + let state = ProtectionState::new(ProtectionConfig { + blocklist: vec!["10.0.0.0/8".parse().unwrap()], + ..Default::default() + }); + let peer = ip("10.1.2.3"); + + assert!(matches!(state.check(peer, &no_peek()), Decision::Block(BlockReason::CidrBlocked))); + + // Remove the blocklist + state.config.store(Arc::new(ProtectionConfig::default().precompute())); + assert!(matches!(state.check(peer, &no_peek()), Decision::Allow(_))); + state.release(peer); + } + + #[test] + fn allowlist_bypass_overrides_blocklist() { + let state = ProtectionState::new(ProtectionConfig { + blocklist: vec!["10.0.0.0/8".parse().unwrap()], + ..Default::default() + }); + let peer = ip("10.0.0.5"); + + // Initially blocked by CIDR + assert!(matches!(state.check(peer, &no_peek()), Decision::Block(BlockReason::CidrBlocked))); + + // Add an allowlist covering that IP — allowlist check runs before blocklist + state.config.store(Arc::new( + ProtectionConfig { + allowlist: vec!["10.0.0.0/24".parse().unwrap()], + blocklist: vec!["10.0.0.0/8".parse().unwrap()], + ..Default::default() + } + .precompute(), + )); + assert!(matches!(state.check(peer, &no_peek()), Decision::AllowBypassed)); + } + + #[test] + fn ja3_block_hot_swap() { + let state = ProtectionState::new(ProtectionConfig::default()); + let peer = ip("1.2.3.4"); + let hex = "e7d705a3286e19ea42f587b344ee6865"; + + assert!(matches!(state.check(peer, &peek_with_ja3(hex)), Decision::Allow(_))); + state.release(peer); + + // Swap in a JA3 blocklist + let mut new_cfg = ProtectionConfig::default(); + new_cfg.ja3_blocklist.insert(hex_to_bytes16(hex).unwrap()); + state.config.store(Arc::new(new_cfg.precompute())); + assert!(matches!(state.check(peer, &peek_with_ja3(hex)), Decision::Block(BlockReason::Ja3Blocked))); + } + + #[test] + fn rate_limit_hot_swap_enables_limiting() { + // Start with no rate limit — ip_table entry created with burst_fp=0 (tokens=0) + let state = ProtectionState::new(ProtectionConfig::default()); + let peer = ip("2.0.0.1"); + + assert!(matches!(state.check(peer, &no_peek()), Decision::Allow(_))); + state.release(peer); + + // Enable rate limiting with burst < ONE_TOKEN (1000 fp), so the existing entry + // (tokens=0) will never accumulate enough to pass the consume step. + // burst_fp = (0.5 * 1000.0) as u32 = 500 < 1000 = ONE_TOKEN → always blocked. + state.config.store(Arc::new( + ProtectionConfig { + rate_limit_cps: Some(10.0), + rate_limit_burst: Some(0.5), + ..Default::default() + } + .precompute(), + )); + assert!(matches!(state.check(peer, &no_peek()), Decision::Block(BlockReason::RateLimited))); + } + + #[test] + fn rate_limit_hot_swap_loosens_limit() { + // Start with a very tight rate limit — burst_fp=1 so immediately rate-limited + let state = ProtectionState::new(ProtectionConfig { + rate_limit_cps: Some(1.0), + rate_limit_burst: Some(0.001), // burst_fp=1, ONE_TOKEN=1000 → always blocked + ..Default::default() + }); + let peer = ip("2.0.0.2"); + + assert!(matches!(state.check(peer, &no_peek()), Decision::Block(BlockReason::RateLimited))); + + // Remove rate limit entirely — now unrestricted + state.config.store(Arc::new(ProtectionConfig::default().precompute())); + assert!(matches!(state.check(peer, &no_peek()), Decision::Allow(_))); + state.release(peer); + } + + #[test] + fn burst_decrease_no_underflow() { + // Start with a generous burst (100 tokens = 100000 fp) + let state = ProtectionState::new(ProtectionConfig { + rate_limit_cps: Some(100.0), + rate_limit_burst: Some(100.0), + ..Default::default() + }); + let peer = ip("3.0.0.1"); + + // Consume 50 tokens — each check decrements by 1000 fp; 50 × 1000 = 50000 fp consumed + for _ in 0..50 { + assert!(matches!(state.check(peer, &no_peek()), Decision::Allow(_))); + state.release(peer); + } + // tokens ≈ 50000 fp (remaining in bucket; negligible refill during fast loop) + + // Reduce burst to 1 (burst_fp=1000). On next refill the bucket will be capped. + // Existing tokens (50000) remain until first refill; no underflow in consume step. + state.config.store(Arc::new( + ProtectionConfig { + rate_limit_cps: Some(1.0), + rate_limit_burst: Some(1.0), + ..Default::default() + } + .precompute(), + )); + + // Consume step: tokens ≈ 50000 fp → 50000 - 1000 = 49000 fp; no underflow + assert!(matches!(state.check(peer, &no_peek()), Decision::Allow(_))); + state.release(peer); + } + + #[test] + fn blocked_ips_reports_concurrency_limited_without_rate_limit() { + // Concurrency cap with NO rate limit — burst_fp is 0, but concurrency-limited IPs + // must still appear in blocked_ips(). The old `&& burst_fp > 0` guard was wrong. + let state = ProtectionState::new(ProtectionConfig { + max_concurrent_per_ip: 1, + ..Default::default() + }); + let peer = ip("5.0.0.1"); + + // First connection: allowed and active counter incremented (not released yet). + assert!(matches!(state.check(peer, &no_peek()), Decision::Allow(_))); + + // IP is now at the concurrency limit; blocked_ips() must report it. + let (rl, cl, _) = state.blocked_ips(); + assert!(rl.is_empty(), "no rate limit configured — rateLimited must be empty"); + assert!(cl.contains(&peer), "IP at concurrency limit must appear in concurrencyLimited"); + + state.release(peer); + } + + // ── Sustained rate limit tests ───────────────────────────────────────────── + + #[test] + fn sustained_bucket_enforced_independently() { + // High per-second rate (100 cps) but low sustained burst (3 connections total). + // After 3 connections the sustained bucket is exhausted; the 4th is blocked + // despite the per-second bucket still having tokens. + let now = now_ns(); + let state = ProtectionState::new(ProtectionConfig { + rate_limit_cps: Some(100.0), + rate_limit_burst: Some(100.0), + sustained_cpm: Some(60.0), // 1 per second in sustained terms + sustained_burst: Some(3.0), + ..Default::default() + }); + let peer = ip("4.0.0.1"); + + // All at the same timestamp so no refill occurs between calls. + assert!(matches!(state.check_at(peer, &no_peek(), now), Decision::Allow(_))); + state.release(peer); + assert!(matches!(state.check_at(peer, &no_peek(), now), Decision::Allow(_))); + state.release(peer); + assert!(matches!(state.check_at(peer, &no_peek(), now), Decision::Allow(_))); + state.release(peer); + + // Sustained burst exhausted — 4th is blocked + assert!(matches!(state.check_at(peer, &no_peek(), now), Decision::Block(BlockReason::RateLimited))); + } + + #[test] + fn sustained_bucket_refills_over_time() { + // 60 CPM = 1 per second. Sustained burst = 1. After exhausting, 1 second of wait + // refills exactly 1 token, allowing the next connection. + let now = now_ns(); + let state = ProtectionState::new(ProtectionConfig { + sustained_cpm: Some(60.0), + sustained_burst: Some(1.0), + ..Default::default() + }); + let peer = ip("4.0.0.2"); + + assert!(matches!(state.check_at(peer, &no_peek(), now), Decision::Allow(_))); + state.release(peer); + // Sustained bucket exhausted + assert!(matches!(state.check_at(peer, &no_peek(), now), Decision::Block(BlockReason::RateLimited))); + + // 1.1 seconds later — bucket refilled by 1.1 tokens, capped at burst 1 → allows + let later = now + 1_100_000_000; // 1.1 s in ns + assert!(matches!(state.check_at(peer, &no_peek(), later), Decision::Allow(_))); + state.release(peer); + } + + #[test] + fn sustained_does_not_block_without_sustained_config() { + // Only per-second limit configured — sustained bucket should not interfere. + let now = now_ns(); + let state = ProtectionState::new(ProtectionConfig { + rate_limit_cps: Some(10.0), + rate_limit_burst: Some(5.0), + ..Default::default() + }); + let peer = ip("4.0.0.3"); + + // 5 connections drain the per-second burst, but no sustained limit → blocked on #6 by per-second only + for _ in 0..5 { + assert!(matches!(state.check_at(peer, &no_peek(), now), Decision::Allow(_))); + state.release(peer); + } + assert!(matches!(state.check_at(peer, &no_peek(), now), Decision::Block(BlockReason::RateLimited))); + } + + // ── Penalty box tests ────────────────────────────────────────────────────── + + #[test] + fn penalty_box_entered_on_rate_limit_exhaustion() { + let now = now_ns(); + let state = ProtectionState::new(ProtectionConfig { + rate_limit_cps: Some(100.0), + rate_limit_burst: Some(1.0), // burst_fp=1000, ONE_TOKEN=1000 → one connection allowed + penalty_box_duration_ms: 60_000, + ..Default::default() + }); + let peer = ip("5.0.0.1"); + + // First connection consumes the only token + assert!(matches!(state.check_at(peer, &no_peek(), now), Decision::Allow(_))); + state.release(peer); + + // Second: bucket exhausted → RateLimited + penalty box entered + assert!(matches!(state.check_at(peer, &no_peek(), now), Decision::Block(BlockReason::RateLimited))); + + // Third: penalty box is now active → PenaltyBoxed + assert!(matches!(state.check_at(peer, &no_peek(), now), Decision::Block(BlockReason::PenaltyBoxed))); + } + + #[test] + fn penalty_box_sustained_exhaustion_enters_box() { + let now = now_ns(); + let state = ProtectionState::new(ProtectionConfig { + // Generous per-second limit so per-second bucket never exhausts + rate_limit_cps: Some(1000.0), + rate_limit_burst: Some(1000.0), + sustained_cpm: Some(600.0), + sustained_burst: Some(1.0), // 1 connection burst + penalty_box_duration_ms: 60_000, + ..Default::default() + }); + let peer = ip("5.0.0.2"); + + // Consumes the single sustained token + assert!(matches!(state.check_at(peer, &no_peek(), now), Decision::Allow(_))); + state.release(peer); + + // Sustained exhausted → RateLimited + penalty entered + assert!(matches!(state.check_at(peer, &no_peek(), now), Decision::Block(BlockReason::RateLimited))); + + // Now penalized + assert!(matches!(state.check_at(peer, &no_peek(), now), Decision::Block(BlockReason::PenaltyBoxed))); + } + + #[test] + fn penalty_box_blocks_while_active() { + let now = now_ns(); + let state = ProtectionState::new(ProtectionConfig { + rate_limit_cps: Some(100.0), + rate_limit_burst: Some(1.0), + penalty_box_duration_ms: 60_000, + ..Default::default() + }); + let peer = ip("5.0.0.3"); + + // Enter the penalty box + state.check_at(peer, &no_peek(), now); // Allow (consume token) + state.release(peer); + state.check_at(peer, &no_peek(), now); // Block(RateLimited) — enters penalty box + + // Well within penalty window + let mid = now + 30_000_000_000; // 30s later, penalty is 60s + assert!(matches!(state.check_at(peer, &no_peek(), mid), Decision::Block(BlockReason::PenaltyBoxed))); + } + + #[test] + fn penalty_box_extends_on_continued_excess() { + // With only per-second limit enabled, a bucket that is always-empty (burst < ONE_TOKEN) + // will extend the penalty on every check while boxed. + let now = now_ns(); + let penalty_ms: u64 = 10_000; // 10 s + let state = ProtectionState::new(ProtectionConfig { + rate_limit_cps: Some(100.0), + // burst_fp=500 < ONE_TOKEN=1000 → consume always fails while boxed + rate_limit_burst: Some(0.5), + penalty_box_duration_ms: penalty_ms, + ..Default::default() + }); + let peer = ip("5.0.0.4"); + + // Exhaust bucket (tokens=0 initially since burst_fp < ONE_TOKEN) → RateLimited → enter box + assert!(matches!(state.check_at(peer, &no_peek(), now), Decision::Block(BlockReason::RateLimited))); + + // Deadline should be set to now + penalty_ms * 1_000_000 + let state_entry = state.ip_table.get(&peer).unwrap(); + let deadline_after_entry = state_entry.penalty_deadline_ns.load(Ordering::Relaxed); + assert_eq!(deadline_after_entry, now.saturating_add(penalty_ms.saturating_mul(1_000_000))); + drop(state_entry); + + // 5s later — still within penalty; bucket still empty → should extend + let t1 = now + 5_000_000_000; // 5s + assert!(matches!(state.check_at(peer, &no_peek(), t1), Decision::Block(BlockReason::PenaltyBoxed))); + + let state_entry = state.ip_table.get(&peer).unwrap(); + let deadline_after_extend = state_entry.penalty_deadline_ns.load(Ordering::Relaxed); + drop(state_entry); + + // Deadline must have been pushed to t1 + penalty_ms * 1_000_000 + assert_eq!(deadline_after_extend, t1.saturating_add(penalty_ms.saturating_mul(1_000_000))); + assert!(deadline_after_extend > deadline_after_entry, "deadline must have been extended"); + } + + #[test] + fn penalty_box_no_extension_if_bucket_refilled() { + // If the IP stops attacking, the bucket refills and debit succeeds while boxed. + // The deadline should NOT be extended. + let penalty_ms: u64 = 60_000; // 60s + let state = ProtectionState::new(ProtectionConfig { + rate_limit_cps: Some(1.0), + rate_limit_burst: Some(1.0), // burst_fp=1000 + penalty_box_duration_ms: penalty_ms, + ..Default::default() + }); + let peer = ip("5.0.0.5"); + let now = now_ns(); + + // Consume the token → rate limited → enter box + state.check_at(peer, &no_peek(), now); // Allow + state.release(peer); + state.check_at(peer, &no_peek(), now); // Block → enter penalty + + let state_entry = state.ip_table.get(&peer).unwrap(); + let deadline_after_entry = state_entry.penalty_deadline_ns.load(Ordering::Relaxed); + drop(state_entry); + + // 30s later (half of penalty). At 1 cps, 30s refills 30 tokens → bucket=30 > ONE_TOKEN. + // Debit succeeds → no extension. + let t1 = now + 30_000_000_000; // 30s + assert!(matches!(state.check_at(peer, &no_peek(), t1), Decision::Block(BlockReason::PenaltyBoxed))); + + let state_entry = state.ip_table.get(&peer).unwrap(); + let deadline_no_extend = state_entry.penalty_deadline_ns.load(Ordering::Relaxed); + drop(state_entry); + + assert_eq!(deadline_no_extend, deadline_after_entry, "deadline must not change when bucket refilled"); + } + + #[test] + fn penalty_box_expires_and_readmits() { + let penalty_ms: u64 = 5_000; // 5s in this test + let state = ProtectionState::new(ProtectionConfig { + rate_limit_cps: Some(100.0), + rate_limit_burst: Some(1.0), // single-token burst + penalty_box_duration_ms: penalty_ms, + ..Default::default() + }); + let peer = ip("5.0.0.6"); + let now = now_ns(); + + // Enter penalty box + state.check_at(peer, &no_peek(), now); // Allow + state.release(peer); + state.check_at(peer, &no_peek(), now); // RateLimited → box entered + + // Immediately still blocked + assert!(matches!(state.check_at(peer, &no_peek(), now), Decision::Block(BlockReason::PenaltyBoxed))); + + // After penalty_ms + some margin, deadline has passed → readmitted + // Also advance enough time for the per-second bucket to refill (at 100 cps, 0.01s fills 1 token) + let after_penalty = now + (penalty_ms + 1000) * 1_000_000; // penalty + 1 second + assert!(matches!(state.check_at(peer, &no_peek(), after_penalty), Decision::Allow(_))); + state.release(peer); + } + + #[test] + fn penalty_box_disabled_by_default() { + // With no penalty_box_duration_ms, exhausting the rate limit only blocks the current + // connection — the next one is not pre-emptively blocked. + let now = now_ns(); + let state = ProtectionState::new(ProtectionConfig { + rate_limit_cps: Some(100.0), + // burst_fp=500 < ONE_TOKEN → always blocked when empty + rate_limit_burst: Some(0.5), + // penalty_box_duration_ms = 0 (default) + ..Default::default() + }); + let peer = ip("5.0.0.7"); + + // Blocked by rate limit (no penalty) + assert!(matches!(state.check_at(peer, &no_peek(), now), Decision::Block(BlockReason::RateLimited))); + // Still rate limited, but NOT PenaltyBoxed — penalty box is off + assert!(matches!(state.check_at(peer, &no_peek(), now), Decision::Block(BlockReason::RateLimited))); + } + + // ── Fix 1: u64 overflow with absurd penalty duration ────────────────────── + + #[test] + fn penalty_box_absurd_duration_still_engages() { + // A "ban forever" duration near u64::MAX would overflow now + penalty_ms * 1_000_000 + // without saturating arithmetic, wrapping to a past deadline and silently never + // engaging the box. saturating_add/saturating_mul must clamp to u64::MAX instead. + let now: u64 = 1_000_000_000; // 1 s into process life + let absurd_ms: u64 = u64::MAX / 1_000_000 + 1; // multiplying by 1_000_000 overflows + let state = ProtectionState::new(ProtectionConfig { + rate_limit_cps: Some(100.0), + rate_limit_burst: Some(1.0), + penalty_box_duration_ms: absurd_ms, + ..Default::default() + }); + let peer = ip("7.0.0.1"); + + // Consume token → rate limited → penalty entered with saturated deadline + state.check_at(peer, &no_peek(), now); // Allow + state.release(peer); + assert!(matches!(state.check_at(peer, &no_peek(), now), Decision::Block(BlockReason::RateLimited))); + + // Verify deadline is u64::MAX (saturated), not a wrapped past value + let state_entry = state.ip_table.get(&peer).unwrap(); + let deadline = state_entry.penalty_deadline_ns.load(Ordering::Relaxed); + drop(state_entry); + assert_eq!(deadline, u64::MAX, "absurd duration must saturate to u64::MAX, not wrap"); + + // Box must now engage (now < u64::MAX) + assert!(matches!(state.check_at(peer, &no_peek(), now), Decision::Block(BlockReason::PenaltyBoxed))); + // And at a much later time still far below u64::MAX, still blocked + assert!(matches!( + state.check_at(peer, &no_peek(), now + 86_400_000_000_000), + Decision::Block(BlockReason::PenaltyBoxed) + )); + } + + // ── Fix 4: blockedIps() respects config after hot-swap disables penaltyBox ─ + + #[test] + fn blocked_ips_penalty_boxed_gated_on_config() { + let now = now_ns(); + let state = ProtectionState::new(ProtectionConfig { + rate_limit_cps: Some(100.0), + rate_limit_burst: Some(0.5), // always blocked + penalty_box_duration_ms: 600_000, + ..Default::default() + }); + let peer = ip("8.0.0.1"); + + // Enter the penalty box + state.check_at(peer, &no_peek(), now); + + // Confirm it's reported + let (_, _, pb) = state.blocked_ips(); + assert!(pb.contains(&peer), "IP must appear in penaltyBoxed while box is enabled"); + + // Hot-swap: disable penalty box + state.config.store(Arc::new( + ProtectionConfig { + rate_limit_cps: Some(100.0), + rate_limit_burst: Some(0.5), + penalty_box_duration_ms: 0, // disabled + ..Default::default() + } + .precompute(), + )); + + // IP still has a non-zero deadline on its IpState (stale), but config says disabled + let (_, _, pb_after) = state.blocked_ips(); + assert!( + pb_after.is_empty(), + "penaltyBoxed must be empty after hot-swap disables penaltyBox config; got: {pb_after:?}" + ); + } + + // ── Fix 5: blockedIps() rate-limited uses lazy projection ────────────────── + + #[test] + fn blocked_ips_rate_limited_excludes_recovered_ips() { + let now = now_ns(); + let state = ProtectionState::new(ProtectionConfig { + rate_limit_cps: Some(1000.0), // fast refill: 1000 tokens/s + rate_limit_burst: Some(1.0), // burst_fp=1000 + ..Default::default() + }); + let peer = ip("8.0.0.2"); + + // Exhaust bucket + state.check_at(peer, &no_peek(), now); // Allow — consumes last token + state.release(peer); + assert!(matches!(state.check_at(peer, &no_peek(), now), Decision::Block(BlockReason::RateLimited))); + + // At now+0 the stored token value is 0 (below ONE_TOKEN=1000) — stale check would + // report the IP as still limited. Lazy projection must exclude it at t=now+2ms + // (2ms @ 1000cps = 2000 fp refill, capped at burst_fp=1000 → fully recovered). + let recovered = now + 2_000_000; // 2 ms later + let (rl, _, _) = state.blocked_ips_at(recovered); + assert!(!rl.contains(&peer), "recovered IP must not appear in rateLimited; got: {rl:?}"); + } + + // ── Fix 6: evict vs active-counter race (Arc-held decrement) ────────────── + + #[test] + fn active_guard_arc_release_survives_eviction() { + // Simulate the race: check() returns Allow(Arc), then evict() removes the + // entry from the map, then the held Arc releases the active counter. + // With the pre-fix implementation, release() would re-look-up by IP and decrement + // a fresh (zero-active) entry. With the fix, the held Arc is decremented directly. + let state = ProtectionState::new(ProtectionConfig { + max_concurrent_per_ip: 10, + ..Default::default() + }); + let peer = ip("9.0.0.1"); + let now = now_ns(); + + // Admit the connection, holding the Arc + let held_arc = match state.check_at(peer, &no_peek(), now) { + Decision::Allow(s) => s, + other => panic!("expected Allow, got {other:?}"), + }; + assert_eq!(held_arc.active.load(Ordering::Relaxed), 1, "active must be 1 after check"); + + // Forcibly evict the entry (simulating the race: evict runs after check but before release) + state.ip_table.remove(&peer); + assert!(state.ip_table.get(&peer).is_none(), "entry must be gone from map"); + + // Release through the held Arc — must decrement the SAME IpState, not a re-inserted one + held_arc.release(); + assert_eq!(held_arc.active.load(Ordering::Relaxed), 0, "active on held Arc must be 0 after release"); + + // A new check should get a fresh entry with active=0, not a poisoned one + let d2 = state.check_at(peer, &no_peek(), now); + assert!(matches!(d2, Decision::Allow(_)), "fresh entry after evict must allow"); + if let Decision::Allow(s2) = d2 { + assert_eq!(s2.active.load(Ordering::Relaxed), 1, "fresh entry active must be 1"); + s2.release(); + } + } + + // ── Eviction tests ───────────────────────────────────────────────────────── + + #[test] + fn evict_removes_idle_no_rate_limit_entry() { + // No rate limit configured (burst_fp=0, sustained_burst_fp=0). + // An entry created for concurrency tracking only should be evicted when idle. + let state = ProtectionState::new(ProtectionConfig { + max_concurrent_per_ip: 10, + ..Default::default() + }); + let peer = ip("6.0.0.1"); + let now = now_ns(); + + state.check_at(peer, &no_peek(), now); // Allow (active=1) + state.release(peer); // active=0 + + assert_eq!(state.ip_table.len(), 1); + state.evict_at(now); + assert_eq!(state.ip_table.len(), 0, "idle entry with no rate limit must be evicted"); + } + + #[test] + fn evict_removes_rate_limited_entry_after_refill_window() { + // After draining 1 token from a burst-10 bucket at 100 cps, the entry is idle. + // evict_at(now) retains it; evict_at(now + large) sees it as fully refilled → evicts. + let state = ProtectionState::new(ProtectionConfig { + rate_limit_cps: Some(100.0), + rate_limit_burst: Some(10.0), // burst_fp = 10000; 10 tokens + ..Default::default() + }); + let peer = ip("6.0.0.11"); + let now = now_ns(); + + state.check_at(peer, &no_peek(), now); // Allow — drains 1 token + state.release(peer); // active=0 + + // Immediately: tokens = 9000 < 10000 → retained + state.evict_at(now); + assert_eq!(state.ip_table.len(), 1); + + // 0.2 s later: at 100 cps, 9000 fp deficit (1 token) refills in 0.01 s. + // Lazy projection: current=9000 + refill(200ms@100cps)=20000 → capped at 10000 == burst_fp → evict. + state.evict_at(now + 200_000_000); // 200 ms + assert_eq!(state.ip_table.len(), 0, "rate-limited idle entry must be evicted after refill window"); + } + + #[test] + fn evict_retains_penalty_boxed_entries() { + let now = now_ns(); + let state = ProtectionState::new(ProtectionConfig { + rate_limit_cps: Some(100.0), + rate_limit_burst: Some(0.5), // always rate-limited (burst_fp=500 < ONE_TOKEN=1000) + penalty_box_duration_ms: 600_000, // 10 min + ..Default::default() + }); + let peer = ip("6.0.0.2"); + + // Enter penalty box + state.check_at(peer, &no_peek(), now); // RateLimited → box entered + assert_eq!(state.ip_table.len(), 1); + + // Even with a huge future time, the penalty box check runs first → retained + state.evict_at(now + 60_000_000_000); // 60 s (still within 10 min penalty) + assert_eq!(state.ip_table.len(), 1, "penalty-boxed entry must not be evicted"); + } + + #[test] + fn evict_retains_entries_with_depleted_sustained_bucket() { + let now = now_ns(); + let state = ProtectionState::new(ProtectionConfig { + // No per-second limit — only sustained, so only that check can block + sustained_cpm: Some(60.0), + sustained_burst: Some(2.0), + ..Default::default() + }); + let peer = ip("6.0.0.3"); + + // Exhaust sustained bucket (2 connections) + assert!(matches!(state.check_at(peer, &no_peek(), now), Decision::Allow(_))); + state.release(peer); + assert!(matches!(state.check_at(peer, &no_peek(), now), Decision::Allow(_))); + state.release(peer); + assert!(matches!(state.check_at(peer, &no_peek(), now), Decision::Block(BlockReason::RateLimited))); + + assert_eq!(state.ip_table.len(), 1); + + // At `now`: sustained refill = 0 → would_be < burst_fp → retained (attacker history preserved) + state.evict_at(now); + assert_eq!(state.ip_table.len(), 1, "entry with depleted sustained bucket must not be evicted"); + + // 1 second later: at 60 CPM = 1 per second, 1 s refills 1 token. + // burst=2, so 2 tokens were drained (2000 fp). In 1 s: refill=1*1/60*60*1000 = 1000 fp. + // current=0 + refill=1000 = 1000 < burst_fp=2000 → still retained + state.evict_at(now + 1_000_000_000); // 1 s + assert_eq!(state.ip_table.len(), 1, "entry still recovering after 1s must not be evicted"); + + // 3 seconds later: at 60 CPM, 3 s refills 3 tokens = 3000 fp > 2000 → capped at 2000 → evict + state.evict_at(now + 3_000_000_000); // 3 s + assert_eq!(state.ip_table.len(), 0, "fully-refilled sustained entry must be evicted"); + } + + #[test] + fn evict_retains_entries_with_active_connections() { + let state = ProtectionState::new(ProtectionConfig { + rate_limit_cps: Some(100.0), + rate_limit_burst: Some(100.0), + ..Default::default() + }); + let peer = ip("6.0.0.4"); + let now = now_ns(); + + // Allow and do NOT release — active counter remains 1 + let held = match state.check_at(peer, &no_peek(), now) { + Decision::Allow(s) => s, + other => panic!("expected Allow, got {other:?}"), + }; + + assert_eq!(state.ip_table.len(), 1); + // Even with a long future time, the active=1 guard keeps it + state.evict_at(now + 86_400_000_000_000); // 1 day + assert_eq!(state.ip_table.len(), 1, "entry with active connections must not be evicted"); + + // Clean up + held.release(); + } + + // ── Fix 2: stale penalty deadline eviction after penaltyBox disable/re-enable ─ + + #[test] + fn evict_clears_stale_penalty_after_penaltybox_disabled() { + // Box an IP, then hot-swap penaltyBox off, evict with fully-refilled buckets, + // re-enable — re-enabled penaltyBox must not resurrect the pre-disable deadline. + let now = now_ns(); + let penalty_ms: u64 = 60_000; // 60 s + let state = ProtectionState::new(ProtectionConfig { + rate_limit_cps: Some(100.0), + rate_limit_burst: Some(1.0), // burst_fp=1000 = ONE_TOKEN + penalty_box_duration_ms: penalty_ms, + ..Default::default() + }); + let peer = ip("6.0.0.5"); + + // Consume the only token → rate limited → penalty box entered with deadline = now + 60s + assert!(matches!(state.check_at(peer, &no_peek(), now), Decision::Allow(_))); + state.release(peer); + assert!(matches!(state.check_at(peer, &no_peek(), now), Decision::Block(BlockReason::RateLimited))); + + // Verify entry is present and IP is in penalty box + assert!(state.ip_table.contains_key(&peer)); + let (_, _, pb) = state.blocked_ips_at(now + 1_000_000); + assert!(pb.contains(&peer), "IP must be in penalty box before disable"); + + // Hot-swap: disable penalty box + state.config.store(Arc::new( + ProtectionConfig { + rate_limit_cps: Some(100.0), + rate_limit_burst: Some(1.0), + penalty_box_duration_ms: 0, // disabled + ..Default::default() + } + .precompute(), + )); + + // Evict at a time far enough in the future that the per-second bucket is fully + // refilled (100 cps × 120s >>> burst_fp=1000). With penaltyBox disabled, the + // stale deadline must NOT prevent eviction. + let far_future = now + 120_000_000_000; // 2 min + state.evict_at(far_future); + assert!( + !state.ip_table.contains_key(&peer), + "entry must be evicted when penaltyBox disabled and buckets fully refilled" + ); + + // Re-enable penalty box — the pre-disable deadline is gone (entry evicted). + state.config.store(Arc::new( + ProtectionConfig { + rate_limit_cps: Some(100.0), + rate_limit_burst: Some(1.0), + penalty_box_duration_ms: penalty_ms, + ..Default::default() + } + .precompute(), + )); + + // IP must be admitted fresh — no stale deadline blocks it + assert!( + matches!(state.check_at(peer, &no_peek(), far_future), Decision::Allow(_)), + "IP must be admitted fresh after re-enable; stale deadline must not resurrect" + ); + } +} diff --git a/src/proxy.rs b/src/proxy.rs index 67b32b0..2a3de10 100644 --- a/src/proxy.rs +++ b/src/proxy.rs @@ -81,9 +81,25 @@ pub struct JsRateLimitConfig { pub burst: Option, } +#[napi(object)] +pub struct JsSustainedRateLimitConfig { + pub connections_per_minute: f64, + pub burst: Option, +} + +#[napi(object)] +pub struct JsPenaltyBoxConfig { + /// Duration in ms an IP remains blocked after exhausting a rate limit. Default: 600000. + pub duration_ms: Option, +} + #[napi(object)] pub struct JsProtectionConfig { pub rate_limit: Option, + /// Sustained (per-minute) token bucket — independent of the per-second bucket. + pub sustained: Option, + /// Penalty box: block an IP for a configurable duration after any rate limit exhaustion. + pub penalty_box: Option, pub max_concurrent_per_ip: Option, pub allowlist: Option>, pub blocklist: Option>, @@ -116,9 +132,19 @@ pub struct JsProxyConfig { pub read_buffer_size: Option, } +#[napi(object)] +pub struct JsListenerProtectionHotConfig { + /// Port of the listener to update. Must match a listener configured at start. + pub port: u16, + pub protection: JsProtectionConfig, +} + #[napi(object)] pub struct JsHotConfig { pub routes: Option>, + /// Per-listener protection updates. Each entry must reference a listener that was + /// started WITH protection; a mismatched port or a port without protection is an error. + pub protection: Option>, } #[napi(object)] @@ -133,6 +159,8 @@ pub struct JsBlockedIpsInfo { pub rate_limited: Vec, pub concurrency_limited: Vec, pub cidr_blocklist: Vec, + /// IPs currently in the penalty box (blocked for a configured duration after rate-limit exhaustion). + pub penalty_boxed: Vec, } #[napi(object)] @@ -165,9 +193,9 @@ struct InternalListener { /// Per-listener runtime state. struct ListenerState { addr: String, + port: u16, metrics: Arc, protection: Option>, - protection_blocklist: Vec, } // ── SymphonyProxy napi class ────────────────────────────────────────────────── @@ -263,19 +291,18 @@ impl SymphonyProxyWrap { mode, }); - let (protection, protection_blocklist) = if let Some(prot_cfg) = &l.protection { - let (cfg, allowlist, blocklist, bl_strings) = parse_protection_config(prot_cfg)?; - let state = ProtectionState::new(cfg, allowlist, blocklist); - (Some(state), bl_strings) + let protection = if let Some(prot_cfg) = &l.protection { + let cfg = parse_protection_config(prot_cfg)?; + Some(ProtectionState::new(cfg)) } else { - (None, Vec::new()) + None }; listener_states.push(ListenerState { addr: addr_str, + port: l.port, metrics: Arc::new(ListenerMetrics::default()), protection, - protection_blocklist, }); } @@ -410,6 +437,31 @@ impl SymphonyProxyWrap { } }); + // Spawn a periodic IP state eviction task per listener that has protection. + // Eviction bounds ip_table memory growth under diverse-IP traffic / attack. + // evict() is O(N) DashMap::retain with shard locks and per-entry float math — + // running it directly inside select! would stall the accept path under diverse-IP + // flood. spawn_blocking moves it off the tokio worker thread pool. + for ls in &self.listener_states { + if let Some(prot) = ls.protection.clone() { + let mut evict_rx = tx.subscribe(); + self.rt_handle.spawn(async move { + let interval = Duration::from_secs(60); + loop { + tokio::select! { + _ = evict_rx.recv() => break, + _ = tokio::time::sleep(interval) => { + let prot_clone = prot.clone(); + // Fire-and-forget: an in-flight eviction completing during + // shutdown is harmless (it just holds a DashMap shard lock briefly). + let _ = tokio::task::spawn_blocking(move || prot_clone.evict()).await; + } + } + } + }); + } + } + Ok(()) } @@ -425,7 +477,10 @@ impl SymphonyProxyWrap { #[napi] pub fn update_config(&self, hot: JsHotConfig) -> Result<()> { - if let Some(routes) = hot.routes { + // Build new route table if provided — hold it, do NOT swap yet. + // All validation must pass before either section is applied: a combined update + // with valid routes + invalid protection must leave both in their old state. + let new_route_table = if let Some(routes) = hot.routes { let specs: Vec = routes .iter() .map(parse_route_spec) @@ -436,8 +491,63 @@ impl SymphonyProxyWrap { let current = self.route_table.0.load(); let table = build_route_table(&specs, &self.default_listener_tls, Some(¤t)) .map_err(|e| napi::Error::from_reason(e.to_string()))?; + Some(table) + } else { + None + }; + + // Validate + parse protection updates if provided — all-or-nothing before any store. + let validated_protection = if let Some(protection_updates) = hot.protection { + // Phase 1: validate all updates (atomicity + early error). + // Collect all offending ports so the caller gets one actionable error. + let mut errors: Vec = Vec::new(); + for update in &protection_updates { + let matches: Vec<_> = + self.listener_states.iter().filter(|s| s.port == update.port).collect(); + match matches.as_slice() { + [] => errors.push(format!("port {} matches no listener", update.port)), + [single] if single.protection.is_none() => errors.push(format!( + "port {} was started without protection; recreate to enable", + update.port + )), + [_single] => {} // valid — exactly one protection-enabled listener + _multiple => errors.push(format!( + "port {} matches multiple listeners; use per-listener restart-configure", + update.port + )), + } + } + if !errors.is_empty() { + return Err(napi::Error::from_reason(errors.join("; "))); + } + + // Phase 2: parse all configs (fallible). + let parsed = protection_updates + .iter() + .map(|u| parse_protection_config(&u.protection)) + .collect::>>()?; + + Some((protection_updates, parsed)) + } else { + None + }; + + // All validation passed — apply atomically (routes then protection, neither applied on any error above). + if let Some(table) = new_route_table { self.route_table.swap(table); } + if let Some((protection_updates, parsed)) = validated_protection { + // Phase 3: store all (infallible — validation above guarantees each port is valid). + for (update, cfg) in protection_updates.iter().zip(parsed) { + let prot = self + .listener_states + .iter() + .find(|s| s.port == update.port) + .and_then(|s| s.protection.as_ref()) + .expect("validated above"); + prot.config.store(Arc::new(cfg)); + } + } Ok(()) } @@ -452,36 +562,38 @@ impl SymphonyProxyWrap { #[napi] pub fn blocked_ips(&self) -> JsBlockedIpsInfo { - let mut rate_limited: Vec = Vec::new(); - let mut concurrency_limited: Vec = Vec::new(); - let mut cidr_blocklist: Vec = Vec::new(); + use std::collections::HashSet as HSet; + let mut rate_limited_set: HSet = HSet::new(); + let mut concurrency_limited_set: HSet = HSet::new(); + let mut cidr_blocklist_set: HSet = HSet::new(); + let mut penalty_boxed_set: HSet = HSet::new(); for state in &self.listener_states { - for bl in &state.protection_blocklist { - if !cidr_blocklist.contains(bl) { - cidr_blocklist.push(bl.clone()); - } - } if let Some(prot) = &state.protection { let cfg = prot.config.load(); - let max = cfg.max_concurrent_per_ip; - let (rl, cl) = prot.blocked_ips(max); + // Blocklist is live-readable from the hot-swappable config snapshot. + for net in &cfg.blocklist { + cidr_blocklist_set.insert(net.to_string()); + } + let (rl, cl, pb) = prot.blocked_ips(); for ip in rl { - let s = ip.to_string(); - if !rate_limited.contains(&s) { - rate_limited.push(s); - } + rate_limited_set.insert(ip.to_string()); } for ip in cl { - let s = ip.to_string(); - if !concurrency_limited.contains(&s) { - concurrency_limited.push(s); - } + concurrency_limited_set.insert(ip.to_string()); + } + for ip in pb { + penalty_boxed_set.insert(ip.to_string()); } } } - JsBlockedIpsInfo { rate_limited, concurrency_limited, cidr_blocklist } + JsBlockedIpsInfo { + rate_limited: rate_limited_set.into_iter().collect(), + concurrency_limited: concurrency_limited_set.into_iter().collect(), + cidr_blocklist: cidr_blocklist_set.into_iter().collect(), + penalty_boxed: penalty_boxed_set.into_iter().collect(), + } } #[napi] @@ -628,21 +740,48 @@ fn parse_resolve_spec(r: &JsResolveRoute) -> Result { }) } -/// Parsed protection config plus the derived allowlist, blocklist, and raw blocklist strings. -type ParsedProtectionConfig = ( - crate::protection::ProtectionConfig, - Vec, - Vec, - Vec, -); - -fn parse_protection_config(prot: &JsProtectionConfig) -> Result { +fn parse_protection_config( + prot: &JsProtectionConfig, +) -> Result { let mut cfg = crate::protection::ProtectionConfig::default(); if let Some(rl) = &prot.rate_limit { + if !rl.connections_per_second.is_finite() || rl.connections_per_second <= 0.0 { + return Err(napi::Error::from_reason(format!( + "rateLimit.connectionsPerSecond must be a finite positive number, got {}", + rl.connections_per_second + ))); + } + if let Some(burst) = rl.burst { + if !burst.is_finite() || burst < 0.0 { + return Err(napi::Error::from_reason(format!( + "rateLimit.burst must be a finite non-negative number, got {burst}" + ))); + } + } cfg.rate_limit_cps = Some(rl.connections_per_second); cfg.rate_limit_burst = rl.burst; } + if let Some(s) = &prot.sustained { + if !s.connections_per_minute.is_finite() || s.connections_per_minute <= 0.0 { + return Err(napi::Error::from_reason(format!( + "sustained.connectionsPerMinute must be a finite positive number, got {}", + s.connections_per_minute + ))); + } + if let Some(burst) = s.burst { + if !burst.is_finite() || burst < 0.0 { + return Err(napi::Error::from_reason(format!( + "sustained.burst must be a finite non-negative number, got {burst}" + ))); + } + } + cfg.sustained_cpm = Some(s.connections_per_minute); + cfg.sustained_burst = s.burst; + } + if let Some(pb) = &prot.penalty_box { + cfg.penalty_box_duration_ms = pb.duration_ms.unwrap_or(600_000.0) as u64; + } cfg.max_concurrent_per_ip = prot.max_concurrent_per_ip.unwrap_or(0); if let Some(ja3s) = &prot.ja3_blocklist { @@ -663,7 +802,7 @@ fn parse_protection_config(prot: &JsProtectionConfig) -> Result = prot + cfg.allowlist = prot .allowlist .as_deref() .unwrap_or(&[]) @@ -674,9 +813,10 @@ fn parse_protection_config(prot: &JsProtectionConfig) -> Result>>()?; - let blocklist_strings: Vec = prot.blocklist.as_deref().unwrap_or(&[]).to_vec(); - - let blocklist: Vec = blocklist_strings + cfg.blocklist = prot + .blocklist + .as_deref() + .unwrap_or(&[]) .iter() .map(|s| { s.parse::() @@ -684,7 +824,7 @@ fn parse_protection_config(prot: &JsProtectionConfig) -> Result>>()?; - Ok((cfg, allowlist, blocklist, blocklist_strings)) + Ok(cfg.precompute()) } fn listener_tls_spec(l: &JsListenerConfig) -> ListenerTlsSpec { @@ -762,3 +902,112 @@ fn from_hex_digit(b: u8) -> Option { _ => None, } } + +#[cfg(test)] +mod tests { + use super::*; + + fn no_rate_limit_prot() -> JsProtectionConfig { + JsProtectionConfig { + rate_limit: None, + sustained: None, + penalty_box: None, + max_concurrent_per_ip: None, + allowlist: None, + blocklist: None, + ja3_blocklist: None, + ja4_blocklist: None, + tls_handshake_timeout_ms: None, + require_sni: None, + } + } + + #[test] + fn reject_nan_cps() { + let prot = JsProtectionConfig { + rate_limit: Some(JsRateLimitConfig { connections_per_second: f64::NAN, burst: None }), + ..no_rate_limit_prot() + }; + assert!(parse_protection_config(&prot).is_err(), "NaN cps must error"); + } + + #[test] + fn reject_negative_cps() { + let prot = JsProtectionConfig { + rate_limit: Some(JsRateLimitConfig { connections_per_second: -1.0, burst: None }), + ..no_rate_limit_prot() + }; + assert!(parse_protection_config(&prot).is_err(), "negative cps must error"); + } + + #[test] + fn reject_zero_cps() { + let prot = JsProtectionConfig { + rate_limit: Some(JsRateLimitConfig { connections_per_second: 0.0, burst: None }), + ..no_rate_limit_prot() + }; + assert!(parse_protection_config(&prot).is_err(), "zero cps must error"); + } + + #[test] + fn reject_nan_burst() { + let prot = JsProtectionConfig { + rate_limit: Some(JsRateLimitConfig { + connections_per_second: 10.0, + burst: Some(f64::NAN), + }), + ..no_rate_limit_prot() + }; + assert!(parse_protection_config(&prot).is_err(), "NaN burst must error"); + } + + #[test] + fn reject_negative_burst() { + let prot = JsProtectionConfig { + rate_limit: Some(JsRateLimitConfig { + connections_per_second: 10.0, + burst: Some(-1.0), + }), + ..no_rate_limit_prot() + }; + assert!(parse_protection_config(&prot).is_err(), "negative burst must error"); + } + + #[test] + fn reject_nan_cpm() { + let prot = JsProtectionConfig { + sustained: Some(JsSustainedRateLimitConfig { + connections_per_minute: f64::NAN, + burst: None, + }), + ..no_rate_limit_prot() + }; + assert!(parse_protection_config(&prot).is_err(), "NaN cpm must error"); + } + + #[test] + fn reject_negative_cpm() { + let prot = JsProtectionConfig { + sustained: Some(JsSustainedRateLimitConfig { + connections_per_minute: -1.0, + burst: None, + }), + ..no_rate_limit_prot() + }; + assert!(parse_protection_config(&prot).is_err(), "negative cpm must error"); + } + + #[test] + fn accept_absent_burst() { + // burst: None (absent) must not error — only present-but-invalid burst is rejected + let prot = JsProtectionConfig { + rate_limit: Some(JsRateLimitConfig { connections_per_second: 10.0, burst: None }), + sustained: Some(JsSustainedRateLimitConfig { + connections_per_minute: 100.0, + burst: None, + }), + ..no_rate_limit_prot() + }; + assert!(parse_protection_config(&prot).is_ok(), "absent burst must be accepted"); + } +} diff --git a/src/proxy_conn.rs b/src/proxy_conn.rs index 134a747..fe1e7e9 100644 --- a/src/proxy_conn.rs +++ b/src/proxy_conn.rs @@ -1,11 +1,11 @@ use crate::metrics::{GlobalMetrics, ListenerMetrics}; -use crate::protection::ProtectionState; +use crate::protection::{IpState, ProtectionState}; use crate::router::{Destination, ForwardFingerprint, LiveRouteTable, SourceAddressMode}; use crate::sni; use crate::suspended::SuspendedRegistry; use crate::upstream::{self, UpstreamStream}; use napi::threadsafe_function::ThreadsafeFunction; -use std::net::{IpAddr, SocketAddr}; +use std::net::SocketAddr; use std::sync::Arc; use std::time::Duration; use std::marker::Unpin; @@ -64,9 +64,10 @@ pub async fn handle(stream: TcpStream, peer_addr: SocketAddr, ctx: Arc that incremented the active counter. + // We hold it in ip_state_held so the ActiveGuard can decrement the SAME entry even + // if eviction removes it from the map between admission and connection close (fix 6). + let ip_state_held: Option> = if let Some(protection) = &ctx.protection { match protection.check(peer_ip, &peek_info) { crate::protection::Decision::Block(reason) => { ctx.listener_metrics.inc_blocked(); @@ -80,12 +81,12 @@ pub async fn handle(stream: TcpStream, peer_addr: SocketAddr, ctx: Arc false, - crate::protection::Decision::Allow => true, + // Allowlisted: active counter was not incremented. + crate::protection::Decision::AllowBypassed => None, + crate::protection::Decision::Allow(state) => Some(state), } } else { - false + None }; // ── Connection is allowed — track it ───────────────────────────────────── @@ -93,12 +94,12 @@ pub async fn handle(stream: TcpStream, peer_addr: SocketAddr, ctx: Arc so it always hits the + // same entry that was incremented, regardless of concurrent eviction. let _active_guard = ActiveGuard { global: ctx.global_metrics.clone(), listener: ctx.listener_metrics.clone(), - protection: if protection_counted { ctx.protection.clone() } else { None }, - peer_ip, + ip_state: ip_state_held, }; // ── 3. Route lookup ─────────────────────────────────────────────────────── @@ -401,16 +402,18 @@ struct EffectiveRoute { struct ActiveGuard { global: Arc, listener: Arc, - protection: Option>, - peer_ip: IpAddr, + /// Held Arc from check() — decrement always hits the same entry, even if + /// eviction removed it from the map between admission and connection close. + /// None for allowlisted connections (active counter was not incremented). + ip_state: Option>, } impl Drop for ActiveGuard { fn drop(&mut self) { self.global.dec_active(); self.listener.dec_active(); - if let Some(p) = &self.protection { - p.release(self.peer_ip); + if let Some(s) = &self.ip_state { + s.release(); } } } diff --git a/ts/addon.d.ts b/ts/addon.d.ts index 5a90636..9c0a5eb 100644 --- a/ts/addon.d.ts +++ b/ts/addon.d.ts @@ -62,8 +62,20 @@ export interface JsRateLimitConfig { connectionsPerSecond: number burst?: number } +export interface JsSustainedRateLimitConfig { + connectionsPerMinute: number + burst?: number +} +export interface JsPenaltyBoxConfig { + /** Duration in ms an IP remains blocked after exhausting a rate limit. Default: 600000. */ + durationMs?: number +} export interface JsProtectionConfig { rateLimit?: JsRateLimitConfig + /** Sustained (per-minute) token bucket — independent of the per-second bucket. */ + sustained?: JsSustainedRateLimitConfig + /** Penalty box: block an IP for a configurable duration after any rate limit exhaustion. */ + penaltyBox?: JsPenaltyBoxConfig maxConcurrentPerIp?: number allowlist?: Array blocklist?: Array @@ -93,8 +105,18 @@ export interface JsProxyConfig { workerThreads?: number readBufferSize?: number } +export interface JsListenerProtectionHotConfig { + /** Port of the listener to update. Must match a listener configured at start. */ + port: number + protection: JsProtectionConfig +} export interface JsHotConfig { routes?: Array + /** + * Per-listener protection updates. Each entry must reference a listener that was + * started WITH protection; a mismatched port or a port without protection is an error. + */ + protection?: Array } export interface JsProxyMetrics { activeConnections: number @@ -105,6 +127,8 @@ export interface JsBlockedIpsInfo { rateLimited: Array concurrencyLimited: Array cidrBlocklist: Array + /** IPs currently in the penalty box (blocked for a configured duration after rate-limit exhaustion). */ + penaltyBoxed: Array } export interface JsResolveRoute { upstream: JsUpstream diff --git a/ts/proxy.ts b/ts/proxy.ts index fbe37d2..c44adbe 100644 --- a/ts/proxy.ts +++ b/ts/proxy.ts @@ -17,6 +17,7 @@ import type { Upstream, CertConfig, MtlsConfig, + ProtectionConfig, RouteConfig, ProxyEvent, SuspendedConnection, @@ -107,6 +108,25 @@ function toJsRoute(r: RouteConfig): JsRouteConfig { }; } +function toJsProtectionConfig(p: ProtectionConfig) { + return { + rateLimit: p.rateLimit + ? { connectionsPerSecond: p.rateLimit.connectionsPerSecond, burst: p.rateLimit.burst } + : undefined, + sustained: p.sustained + ? { connectionsPerMinute: p.sustained.connectionsPerMinute, burst: p.sustained.burst } + : undefined, + penaltyBox: p.penaltyBox ? { durationMs: p.penaltyBox.durationMs } : undefined, + maxConcurrentPerIp: p.maxConcurrentPerIp, + allowlist: p.allowlist, + blocklist: p.blocklist, + ja3Blocklist: p.ja3Blocklist, + ja4Blocklist: p.ja4Blocklist, + tlsHandshakeTimeoutMs: p.tlsHandshakeTimeoutMs, + requireSni: p.requireSni, + }; +} + function toJsListenerConfig(l: import('./types.js').ListenerConfig): JsListenerConfig { return { host: l.host, @@ -116,23 +136,7 @@ function toJsListenerConfig(l: import('./types.js').ListenerConfig): JsListenerC mtls: l.mtls ? toJsMtls(l.mtls) : undefined, maxConnections: l.maxConnections, idleTimeoutMs: l.idleTimeoutMs, - protection: l.protection - ? { - rateLimit: l.protection.rateLimit - ? { - connectionsPerSecond: l.protection.rateLimit.connectionsPerSecond, - burst: l.protection.rateLimit.burst, - } - : undefined, - maxConcurrentPerIp: l.protection.maxConcurrentPerIp, - allowlist: l.protection.allowlist, - blocklist: l.protection.blocklist, - ja3Blocklist: l.protection.ja3Blocklist, - ja4Blocklist: l.protection.ja4Blocklist, - tlsHandshakeTimeoutMs: l.protection.tlsHandshakeTimeoutMs, - requireSni: l.protection.requireSni, - } - : undefined, + protection: l.protection ? toJsProtectionConfig(l.protection) : undefined, }; } @@ -215,12 +219,17 @@ export class SymphonyProxy extends EventEmitter { } /** - * Atomically replace the route table and/or protection config. + * Atomically update routes and/or per-listener protection config. * In-flight connections are unaffected. + * Throws if any protection entry references a port that has no protection or matches no listener. */ updateConfig(config: HotConfig): void { this._inner.updateConfig({ routes: config.routes?.map(toJsRoute), + protection: config.protection?.map((p) => ({ + port: p.port, + protection: toJsProtectionConfig(p.protection), + })), }); } diff --git a/ts/server.ts b/ts/server.ts index f7f1334..4649cd3 100644 --- a/ts/server.ts +++ b/ts/server.ts @@ -283,10 +283,27 @@ class ServerState { // default_listener_tls), so updateConfig can't hot-swap it. Keying off resolved // content means a listener-cert rotation on disk (same paths, new bytes) changes // the signature and forces a recreate, which route-only hot-swap would miss. - const listenerSig = JSON.stringify(proxyConfig.listeners); + // + // Protection PRESENCE (hasProtection) is part of the signature: none→some and + // some→none transitions force a seamless recreate so the new listener is + // constructed with the correct Option. Protection CONTENTS are + // excluded so a contents-only change stays on the hot-swap path without recreating. + const listenerSig = JSON.stringify( + proxyConfig.listeners.map(({ protection, ...rest }) => ({ + ...rest, + hasProtection: protection != null, + })), + ); if (existing && existing.listenerSig === listenerSig) { - // Same listeners → hot-swap the route table only. - existing.proxy.updateConfig({ routes: proxyConfig.routes }); + // Same listeners (presence-unchanged) → hot-swap routes and protection contents. + // All listeners here either had protection from the start (Some) or never did + // (None, excluded by filter). none→some / some→none went through the recreate path. + existing.proxy.updateConfig({ + routes: proxyConfig.routes, + protection: proxyConfig.listeners + .filter((l) => l.protection != null) + .map((l) => ({ port: l.port, protection: l.protection! })), + }); } else { // New port-set, or listener settings changed → (re)create. SO_REUSEPORT lets the // new listener bind before the old one is dropped, so there is no bind gap. diff --git a/ts/types.ts b/ts/types.ts index 9fe74df..72b16a9 100644 --- a/ts/types.ts +++ b/ts/types.ts @@ -133,8 +133,47 @@ export interface RateLimitConfig { burst?: number; } +/** + * Sustained (per-minute) token bucket — independent of the per-second `rateLimit` bucket. + * Both buckets are checked on admission; exhausting either one blocks the connection. + * Use this to limit total connection volume over longer windows while still allowing + * short bursts via `rateLimit`. + */ +export interface SustainedRateLimitConfig { + /** Maximum new connections per minute from a single IP. */ + connectionsPerMinute: number; + /** + * Token bucket burst size. Default: connectionsPerMinute. + * Max: 4,294,967 connections (u32 fixed-point limit). + */ + burst?: number; +} + +/** + * Penalty box: when an IP exhausts any rate limit, block it for a configurable duration. + * While penalized, all connections from that IP are rejected outright (no token consumed). + * If the IP continues to exceed its rate while boxed, the penalty is extended — reset to + * the full `durationMs` from the moment of continued excess. The penalty expires naturally + * once the IP stops attacking and the deadline passes. + */ +export interface PenaltyBoxConfig { + /** Duration in ms an IP is penalized after exhausting any rate limit. Default: 600000 (10 min). */ + durationMs?: number; +} + export interface ProtectionConfig { + /** Per-second token bucket rate limit per source IP. */ rateLimit?: RateLimitConfig; + /** + * Per-minute (sustained) token bucket — independent of rateLimit. + * Checked in addition to rateLimit; exhausting either blocks the connection. + */ + sustained?: SustainedRateLimitConfig; + /** + * Penalty box: block an IP for `durationMs` after any rate limit exhaustion. + * Absent = disabled (current behavior: each blocked connection is independent). + */ + penaltyBox?: PenaltyBoxConfig; /** Maximum simultaneous connections from a single IP. 0 = unlimited. */ maxConcurrentPerIp?: number; /** CIDRs that bypass all protection checks (e.g. trusted internal ranges). */ @@ -194,9 +233,33 @@ export interface ProxyConfig { // ── Hot-swap config ─────────────────────────────────────────────────────────── -/** Fields that can be updated live without restarting listeners. */ +/** Protection update for a single listener in a hot-config call. */ +export interface ListenerProtectionHotConfig { + /** Port number of the listener to update. Must match a listener configured at start. */ + port: number; + protection: ProtectionConfig; +} + +/** + * Fields that can be updated live without restarting listeners. + * + * Hot-swappable: routes (destinations, TLS certs, suspension state) and + * per-listener protection contents (CIDR lists, JA3 blocklist, rate limits, concurrency + * caps, handshake timeout, requireSni). + * + * Requires restart: bind address, port, idle timeout, worker threads. + * Protection presence (None↔Some) cannot change via updateConfig() — a listener must be + * restarted to gain or lose protection. updateConfig() returns an error for ports that have + * no protection or that match no listener. symphony-server handles none→some/some→none + * transitions automatically via seamless recreate. + */ export interface HotConfig { routes?: RouteConfig[]; + /** + * Per-listener protection updates. Each entry identifies a listener by port and + * replaces its entire protection config atomically. Absent = leave unchanged. + */ + protection?: ListenerProtectionHotConfig[]; } // ── Metrics ─────────────────────────────────────────────────────────────────── @@ -211,12 +274,14 @@ export interface ProxyMetrics { } export interface BlockedIpsInfo { - /** IPs whose token buckets are currently depleted. */ + /** IPs whose per-second or sustained token buckets are currently depleted. */ rateLimited: string[]; /** IPs currently at their maxConcurrentPerIp limit. */ concurrencyLimited: string[]; /** The configured static CIDR blocklist entries. */ cidrBlocklist: string[]; + /** IPs currently in the penalty box (blocked for durationMs after rate-limit exhaustion). */ + penaltyBoxed: string[]; } // ── Suspended connection ──────────────────────────────────────────────────────