diff --git a/CLAUDE.md b/CLAUDE.md index 723c66d..0cc8d5b 100644 --- a/CLAUDE.md +++ b/CLAUDE.md @@ -39,7 +39,7 @@ TCP accept (SO_REUSEPORT per worker thread) | `src/http_proxy.rs` | HTTP/1.1 header framing and rewrite helpers shared by the HTTP listener | | `src/sni.rs` | MSG_PEEK ClientHello parser; SNI extraction; JA3 fingerprint | | `src/router.rs` | RouteTable (exact + wildcard HashMap); ArcSwap hot-swap | -| `src/upstream.rs` | UpstreamStream enum (Tcp/Uds); connect(); TCP_NODELAY | +| `src/upstream.rs` | UpstreamStream enum (Tcp/Uds); connect(); TCP_NODELAY; PROXY v1/v2 header encoders; HTTP header injection (XFF, X-JA3/X-JA4) | | `src/balancer.rs` | UdsBalancer: AtomicU32 least-connections; IP affinity DashMap | | `src/tls.rs` | rustls ServerConfig builder; SHA-256 deduplication cache | | `src/mtls.rs` | SymphonyClientVerifier wrapping WebPkiClientVerifier | @@ -56,6 +56,21 @@ TCP accept (SO_REUSEPORT per worker thread) ### MSG_PEEK for SNI/JA3 A single `stream.peek(&mut buf[..512])` reads the ClientHello without consuming any bytes. Cost: 1 syscall, 512-byte stack buffer, zero heap allocation. This gives us both the SNI (for routing) and the JA3 fingerprint (for protection) before the TLS handshake begins. The alternative — a custom TLS acceptor that extracts SNI internally — would require modifying rustls internals. +### Source-address & fingerprint forwarding (`upstream.rs` + `proxy_conn.rs`) +Per route, `sourceAddressHeader` picks how the client IP reaches the upstream: PROXY v1 (text), +PROXY **v2** (binary, TLV-capable), `X-Forwarded-For` injection, or none. `forwardFingerprint` +(`ja3`/`ja4`/`none`) additionally forwards the ClientHello fingerprint symphony already computes +in `sni.rs`, so backends can make their own bot/abuse decisions on it. Carrier follows the mode: +a PROXY v2 TLV (custom types `0xE0`=JA3 / `0xE1`=JA4, in HAProxy's `0xE0–0xEF` private range) under +`proxyProtocolV2` — which works even in passthrough since it prefixes the raw TLS bytes — otherwise +an injected `X-JA3`/`X-JA4` header. Header injection is gated on `l7_http1` (terminated TLS *and* a +non-h2 negotiated ALPN); it's a no-op in passthrough or on an h2 upstream, so text is never spliced +into TLS ciphertext or an h2 frame stream. `inject_request_headers` also strips any client-supplied +copy of an injected header so the value is authoritative (anti-spoof). `proxy_conn.rs` threads the +fingerprint + the connection's `local_addr` (the PROXY v2 destination) into `apply_source_header`. +Keep v2 **opt-in**: HAProxy/nginx speak it, but Harper core's own UDS reader currently parses v1 only, +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. diff --git a/README.md b/README.md index f548036..e52ce37 100644 --- a/README.md +++ b/README.md @@ -97,7 +97,8 @@ console.log('proxy listening on :443'); | `maxConnectionsPerSecond` | `number` | — | Route-wide new-connection rate cap (token bucket). Connections are silently dropped when exhausted. | | `burst` | `number` | `maxConnectionsPerSecond` | Token bucket burst ceiling for the route rate limit | | `http2` | `boolean` | `false` | Advertise `h2` in ALPN so clients negotiate HTTP/2. Raw H2 frames flow through to the upstream unchanged. Requires `terminateTls: true`. | -| `sourceAddressHeader` | `'proxyProtocol' \| 'xForwardedFor' \| 'none'` | `'proxyProtocol'` for UDS, `'none'` for TCP | How the real client IP is forwarded to the upstream. See [Source address forwarding](#source-address-forwarding). | +| `sourceAddressHeader` | `'proxyProtocol' \| 'proxyProtocolV2' \| 'xForwardedFor' \| 'none'` | `'proxyProtocol'` for UDS, `'none'` for TCP | How the real client IP is forwarded to the upstream. See [Source address forwarding](#source-address-forwarding). | +| `forwardFingerprint` | `'ja3' \| 'ja4' \| 'none'` | `'none'` | Forward the client TLS fingerprint downstream. See [Forwarding the fingerprint](#forwarding-the-fingerprint-downstream). | ### `Upstream` @@ -344,11 +345,12 @@ Connections that exceed the limit are silently dropped (TCP RST). This is a glob ## Source address forwarding -Use `sourceAddressHeader` on a route to control how the real client IP is communicated to the upstream. This only applies when `terminateTls: true` (TLS is terminated by the proxy). +Use `sourceAddressHeader` on a route to control how the real client IP is communicated to the upstream. The PROXY protocol carriers work whether or not TLS is terminated (they prefix the connection); `'xForwardedFor'` requires `terminateTls: true` and a plaintext HTTP/1 upstream, since it rewrites the HTTP request. | Value | Behaviour | |---|---| -| `'proxyProtocol'` | Sends a PROXY protocol v1 header (`PROXY TCP4 0\r\n`) before any application data. Default for UDS upstreams. | +| `'proxyProtocol'` | Sends a PROXY protocol v1 (text) header (`PROXY TCP4 0\r\n`) before any application data. Default for UDS upstreams. | +| `'proxyProtocolV2'` | Sends a PROXY protocol v2 (binary) header before any application data. v2 adds a TLV section — the carrier for `forwardFingerprint` below. Keep it opt-in: the consumer must speak v2 (nginx/HAProxy do; Harper core's own UDS reader currently parses v1 only). | | `'xForwardedFor'` | Reads the first chunk of the HTTP request, inserts an `X-Forwarded-For` header after the request line, then copies the rest verbatim. No per-request parsing overhead for keep-alive connections. Default for TCP upstreams (disabled). | | `'none'` | Does not forward source address information. Default for TCP upstreams. | @@ -392,6 +394,43 @@ Bun.serve({ }); ``` +### Forwarding the fingerprint downstream + +symphony computes the client's JA3/JA4 fingerprint from the ClientHello (the same value used for `ja3Blocklist`/`ja4Blocklist`). Set `forwardFingerprint` to also hand it to the upstream, so a backend behind symphony can make its own bot/abuse decisions on it. + +| `forwardFingerprint` | Value forwarded | +|---|---| +| `'ja3'` | The JA3 MD5 hex (32 chars) | +| `'ja4'` | The JA4 fingerprint | +| `'none'` (default) | Nothing forwarded | + +The **carrier depends on `sourceAddressHeader`**: + +- With `'proxyProtocolV2'`, the fingerprint rides a PROXY v2 **TLV** — type `0xE0` for JA3, `0xE1` for JA4 (in HAProxy's `0xE0–0xEF` private range). This works even in passthrough (`terminateTls: false`), since the header prefixes the raw TLS bytes. +- Otherwise, symphony injects an **`X-JA3` / `X-JA4` HTTP header**. This requires a plaintext HTTP/1 upstream (`terminateTls: true` and not `http2`); it is skipped for passthrough or HTTP/2 upstreams (use `'proxyProtocolV2'` there). Any client-supplied `X-JA3`/`X-JA4` is stripped so the injected value is authoritative and can't be spoofed. + +```typescript +// TLV carrier — works for any upstream that speaks PROXY v2, including passthrough +{ + sni: 'app.example.com', + upstreams: [{ kind: 'tcp', host: '10.0.0.5', port: 8080 }], + terminateTls: true, + cert: { certChain, privateKey }, + sourceAddressHeader: 'proxyProtocolV2', + forwardFingerprint: 'ja4', +} + +// HTTP-header carrier — for HTTP/1 backends that read request headers +{ + sni: 'app.example.com', + upstreams: [{ kind: 'uds', path: '/run/app/worker.sock' }], + terminateTls: true, + cert: { certChain, privateKey }, + sourceAddressHeader: 'xForwardedFor', + forwardFingerprint: 'ja3', // upstream reads X-JA3 alongside X-Forwarded-For +} +``` + --- ## Protection diff --git a/__test__/proxy-protocol-v2.spec.ts b/__test__/proxy-protocol-v2.spec.ts new file mode 100644 index 0000000..bd341d5 --- /dev/null +++ b/__test__/proxy-protocol-v2.spec.ts @@ -0,0 +1,370 @@ +/** + * Integration tests for PROXY protocol v2 emission and downstream JA3/JA4 fingerprint + * forwarding (issue #9). + * + * These tests require the native addon to be built: + * npm run build:debug + */ + +import assert from 'node:assert/strict'; +import * as tls from 'node:tls'; +import { after, before, describe, it } from 'node:test'; +import { SymphonyProxy } from '../ts/proxy.js'; +import { + generateSelfSignedCert, + getFreePort, + startCaptureServer, + startTlsEchoServer, + tlsRoundTrip, + sleep, +} from './util.js'; + +const PROXY_V2_SIGNATURE = Buffer.from([0x0d, 0x0a, 0x0d, 0x0a, 0x00, 0x0d, 0x0a, 0x51, 0x55, 0x49, 0x54, 0x0a]); +const PP2_TYPE_JA3 = 0xe0; + +interface ParsedV2 { + command: number; + famProto: number; + srcIp: string; + srcPort: number; + tlvs: Map; + rest: Buffer; +} + +/** Parse a PROXY protocol v2 header (IPv4 only, sufficient for these localhost tests). */ +function parseProxyV2(buf: Buffer): ParsedV2 { + assert.ok(buf.subarray(0, 12).equals(PROXY_V2_SIGNATURE), 'v2 signature'); + const command = buf[12]; + const famProto = buf[13]; + const len = buf.readUInt16BE(14); + const body = buf.subarray(16, 16 + len); + assert.equal(famProto, 0x11, 'AF_INET | STREAM'); + const srcIp = `${body[0]}.${body[1]}.${body[2]}.${body[3]}`; + const srcPort = body.readUInt16BE(8); + // TLVs begin after the 12-byte IPv4 address block. + const tlvs = new Map(); + let off = 12; + while (off + 3 <= body.length) { + const type = body[off]; + const tlvLen = body.readUInt16BE(off + 1); + tlvs.set(type, body.subarray(off + 3, off + 3 + tlvLen)); + off += 3 + tlvLen; + } + return { command, famProto, srcIp, srcPort, tlvs, rest: buf.subarray(16 + len) }; +} + +/** Open a TLS connection through the proxy, send `data`, and resolve once written. */ +function tlsSend(port: number, servername: string, caCert: string, data: string): Promise { + return new Promise((resolve, reject) => { + const socket = tls.connect({ port, host: '127.0.0.1', servername, ca: caCert, rejectUnauthorized: false }, () => { + socket.write(data, () => resolve(socket)); + }); + socket.on('error', reject); + }); +} + +const HTTP_REQUEST = 'GET / HTTP/1.1\r\nHost: localhost\r\n\r\n'; + +describe('PROXY protocol v2 + fingerprint forwarding', () => { + const cert = generateSelfSignedCert('localhost'); + + it('emits a v2 header carrying the JA3 fingerprint as a TLV', async () => { + const capture = await startCaptureServer(); + const proxyPort = await getFreePort(); + const proxy = new SymphonyProxy({ + listeners: [{ host: '127.0.0.1', port: proxyPort }], + routes: [ + { + sni: 'localhost', + upstreams: [{ kind: 'tcp', host: '127.0.0.1', port: capture.port }], + terminateTls: true, + cert: { certChain: cert.cert, privateKey: cert.key }, + sourceAddressHeader: 'proxyProtocolV2', + forwardFingerprint: 'ja3', + }, + ], + }); + await proxy.start(); + await sleep(50); + + const socket = await tlsSend(proxyPort, 'localhost', cert.cert, HTTP_REQUEST); + const received = await capture.received; + const parsed = parseProxyV2(received); + + assert.equal(parsed.command, 0x21, 'version 2 | PROXY command'); + assert.equal(parsed.srcIp, '127.0.0.1', 'source IP is the real client'); + const ja3 = parsed.tlvs.get(PP2_TYPE_JA3); + assert.ok(ja3, 'JA3 TLV present'); + assert.match(ja3!.toString('ascii'), /^[0-9a-f]{32}$/, 'JA3 is a 32-char md5 hex'); + assert.equal(parsed.rest.toString('ascii'), HTTP_REQUEST, 'application data follows the header'); + + socket.destroy(); + await proxy.stop(); + await capture.close(); + }); + + it('emits a v2 header with no TLV when forwardFingerprint is off', async () => { + const capture = await startCaptureServer(); + const proxyPort = await getFreePort(); + const proxy = new SymphonyProxy({ + listeners: [{ host: '127.0.0.1', port: proxyPort }], + routes: [ + { + sni: 'localhost', + upstreams: [{ kind: 'tcp', host: '127.0.0.1', port: capture.port }], + terminateTls: true, + cert: { certChain: cert.cert, privateKey: cert.key }, + sourceAddressHeader: 'proxyProtocolV2', + }, + ], + }); + await proxy.start(); + await sleep(50); + + const socket = await tlsSend(proxyPort, 'localhost', cert.cert, HTTP_REQUEST); + const parsed = parseProxyV2(await capture.received); + assert.equal(parsed.tlvs.size, 0, 'no TLVs without forwardFingerprint'); + assert.equal(parsed.rest.toString('ascii'), HTTP_REQUEST); + + socket.destroy(); + await proxy.stop(); + await capture.close(); + }); + + it('injects an X-JA3 header alongside X-Forwarded-For for L7 upstreams', async () => { + const capture = await startCaptureServer(); + const proxyPort = await getFreePort(); + const proxy = new SymphonyProxy({ + listeners: [{ host: '127.0.0.1', port: proxyPort }], + routes: [ + { + sni: 'localhost', + upstreams: [{ kind: 'tcp', host: '127.0.0.1', port: capture.port }], + terminateTls: true, + cert: { certChain: cert.cert, privateKey: cert.key }, + sourceAddressHeader: 'xForwardedFor', + forwardFingerprint: 'ja3', + }, + ], + }); + await proxy.start(); + await sleep(50); + + const socket = await tlsSend(proxyPort, 'localhost', cert.cert, HTTP_REQUEST); + const text = (await capture.received).toString('ascii'); + + assert.match(text, /\r\nX-Forwarded-For: 127\.0\.0\.1\r\n/, 'X-Forwarded-For injected'); + assert.match(text, /\r\nX-JA3: [0-9a-f]{32}\r\n/, 'X-JA3 injected after the request line'); + assert.ok(text.startsWith('GET / HTTP/1.1\r\n'), 'request line preserved'); + + socket.destroy(); + await proxy.stop(); + await capture.close(); + }); + + // Passthrough forwards raw TLS bytes to a TLS upstream. A header carrier must be a no-op here: + // splicing X-JA3 into the ClientHello ciphertext would break the upstream handshake. A working + // end-to-end round-trip proves nothing was injected. + it('does not inject a fingerprint header in passthrough mode', async () => { + const upstream = await startTlsEchoServer(cert.cert, cert.key); + const proxyPort = await getFreePort(); + const proxy = new SymphonyProxy({ + listeners: [{ host: '127.0.0.1', port: proxyPort }], + routes: [ + { + sni: 'localhost', + upstreams: [{ kind: 'tcp', host: '127.0.0.1', port: upstream.port }], + terminateTls: false, + forwardFingerprint: 'ja3', + }, + ], + }); + await proxy.start(); + await sleep(50); + + const payload = Buffer.from('passthrough-ok'); + const response = await tlsRoundTrip({ + port: proxyPort, + servername: 'localhost', + caCert: cert.cert, + data: payload, + }); + assert.deepEqual(response, payload, 'end-to-end TLS round-trip intact (no injected header)'); + + await proxy.stop(); + await upstream.close(); + }); + + // Finding 1 (Critical — Slowloris): a client that completes the TLS handshake and then stalls + // without sending any HTTP request must be dropped once the idle timeout elapses, not held open + // forever. The header read now lives inside the idle-timeout-bounded copy. + it('drops a client that stalls post-TLS without sending a request', async () => { + const capture = await startCaptureServer(); + const proxyPort = await getFreePort(); + const proxy = new SymphonyProxy({ + listeners: [{ host: '127.0.0.1', port: proxyPort, idleTimeoutMs: 300 }], + routes: [ + { + sni: 'localhost', + upstreams: [{ kind: 'tcp', host: '127.0.0.1', port: capture.port }], + terminateTls: true, + cert: { certChain: cert.cert, privateKey: cert.key }, + sourceAddressHeader: 'xForwardedFor', + forwardFingerprint: 'ja3', + }, + ], + }); + await proxy.start(); + await sleep(50); + + // Complete the TLS handshake but never send an HTTP request. + const socket = await new Promise((resolve, reject) => { + const s = tls.connect( + { port: proxyPort, host: '127.0.0.1', servername: 'localhost', ca: cert.cert, rejectUnauthorized: false }, + () => resolve(s) + ); + s.on('error', reject); + }); + + const closedInTime = await new Promise((resolve) => { + const guard = setTimeout(() => resolve(false), 2000); + socket.on('close', () => { + clearTimeout(guard); + resolve(true); + }); + }); + assert.ok(closedInTime, 'proxy dropped the stalled connection after the idle timeout'); + + socket.destroy(); + await proxy.stop(); + await capture.close(); + }); + + // Finding 2 (High): a client-supplied X-JA3 / X-Forwarded-For must be stripped end-to-end so the + // injected values are authoritative and cannot be spoofed. + it('strips client-supplied fingerprint and forwarded-for headers', async () => { + const capture = await startCaptureServer(); + const proxyPort = await getFreePort(); + const proxy = new SymphonyProxy({ + listeners: [{ host: '127.0.0.1', port: proxyPort }], + routes: [ + { + sni: 'localhost', + upstreams: [{ kind: 'tcp', host: '127.0.0.1', port: capture.port }], + terminateTls: true, + cert: { certChain: cert.cert, privateKey: cert.key }, + sourceAddressHeader: 'xForwardedFor', + forwardFingerprint: 'ja3', + }, + ], + }); + await proxy.start(); + await sleep(50); + + const spoofed = + 'GET / HTTP/1.1\r\nHost: localhost\r\nX-JA3: deadbeefdeadbeefdeadbeefdeadbeef\r\n' + + 'X-Forwarded-For: 9.9.9.9\r\n\r\n'; + const socket = await tlsSend(proxyPort, 'localhost', cert.cert, spoofed); + const text = (await capture.received).toString('ascii'); + + assert.match(text, /\r\nX-JA3: [0-9a-f]{32}\r\n/, 'authoritative X-JA3 present'); + assert.match(text, /\r\nX-Forwarded-For: 127\.0\.0\.1\r\n/, 'authoritative X-Forwarded-For present'); + assert.ok(!text.includes('deadbeef'), 'spoofed X-JA3 stripped'); + assert.ok(!text.includes('9.9.9.9'), 'spoofed X-Forwarded-For stripped'); + assert.equal((text.match(/X-JA3:/gi) ?? []).length, 1, 'exactly one X-JA3'); + + socket.destroy(); + await proxy.stop(); + await capture.close(); + }); + + // Finding 2 (keep-alive): every request on a keep-alive connection is rewritten, so a spoofed + // header on a *second* pipelined request is stripped too — not just the first read. + it('injects and strips on every pipelined keep-alive request', async () => { + const capture = await startCaptureServer(); + const proxyPort = await getFreePort(); + const proxy = new SymphonyProxy({ + listeners: [{ host: '127.0.0.1', port: proxyPort }], + routes: [ + { + sni: 'localhost', + upstreams: [{ kind: 'tcp', host: '127.0.0.1', port: capture.port }], + terminateTls: true, + cert: { certChain: cert.cert, privateKey: cert.key }, + sourceAddressHeader: 'xForwardedFor', + forwardFingerprint: 'ja3', + }, + ], + }); + await proxy.start(); + await sleep(50); + + const pipelined = + 'GET /one HTTP/1.1\r\nHost: localhost\r\n\r\n' + + 'GET /two HTTP/1.1\r\nHost: localhost\r\nX-JA3: spoofed0000000000000000000000000\r\n\r\n'; + const socket = await tlsSend(proxyPort, 'localhost', cert.cert, pipelined); + const text = (await capture.received).toString('ascii'); + + assert.ok(text.includes('GET /one HTTP/1.1\r\n'), 'first request forwarded'); + assert.ok(text.includes('GET /two HTTP/1.1\r\n'), 'second request forwarded'); + assert.equal((text.match(/X-JA3: [0-9a-f]{32}\r\n/g) ?? []).length, 2, 'both requests get an authoritative X-JA3'); + assert.ok(!text.includes('spoofed'), 'spoofed X-JA3 on the second request is stripped'); + + socket.destroy(); + await proxy.stop(); + await capture.close(); + }); + + // XFF + h2 rejection at config time is covered by h2-dispatch.spec.ts (route dropped). + // With http2 advertised the client negotiates h2, so the upstream receives binary h2 frames. + // The fingerprint header carrier must be a runtime no-op on the negotiated-h2 connection — + // the captured bytes must be exactly what the client sent. + it('negotiates h2 and does not inject the fingerprint header into the HTTP/2 stream', async () => { + const capture = await startCaptureServer(); + const proxyPort = await getFreePort(); + const proxy = new SymphonyProxy({ + listeners: [{ host: '127.0.0.1', port: proxyPort }], + routes: [ + { + sni: 'localhost', + upstreams: [{ kind: 'tcp', host: '127.0.0.1', port: capture.port }], + terminateTls: true, + cert: { certChain: cert.cert, privateKey: cert.key }, + http2: true, + sourceAddressHeader: 'none', + forwardFingerprint: 'ja3', + }, + ], + }); + await proxy.start(); + await sleep(50); + + const preface = 'PRI * HTTP/2.0\r\n\r\nSM\r\n\r\n'; + let socket!: tls.TLSSocket; + const alpn = await new Promise((resolve, reject) => { + socket = tls.connect( + { + port: proxyPort, + host: '127.0.0.1', + servername: 'localhost', + ALPNProtocols: ['h2', 'http/1.1'], + ca: cert.cert, + rejectUnauthorized: false, + }, + () => { + socket.write(preface); + resolve(socket.alpnProtocol); + } + ); + socket.on('error', reject); + }); + + assert.equal(alpn, 'h2', 'route.http2 must reach ALPN so h2 negotiates'); + const received = (await capture.received).toString('ascii'); + assert.equal(received, preface, 'the h2 stream is forwarded verbatim — no header spliced in'); + + socket.destroy(); + await proxy.stop(); + await capture.close(); + }); +}); diff --git a/__test__/util.ts b/__test__/util.ts index 849acd4..b9c4a09 100644 --- a/__test__/util.ts +++ b/__test__/util.ts @@ -172,6 +172,45 @@ export function startEchoServer(): Promise { }); } +export interface CaptureServer { + port: number; + /** Resolves with all bytes received on the first connection (debounced 150ms after the last chunk). */ + received: Promise; + close(): Promise; +} + +/** + * Start a plain TCP server that captures the raw bytes an upstream would receive — used to + * inspect the PROXY protocol header / injected HTTP headers symphony prepends. Does not echo. + */ +export function startCaptureServer(): Promise { + return new Promise((resolve, reject) => { + let resolveReceived!: (b: Buffer) => void; + const received = new Promise((res) => { + resolveReceived = res; + }); + const chunks: Buffer[] = []; + let timer: NodeJS.Timeout | undefined; + const server = net.createServer((socket) => { + socket.on('data', (c: Buffer) => { + chunks.push(c); + if (timer) clearTimeout(timer); + timer = setTimeout(() => resolveReceived(Buffer.concat(chunks)), 150); + }); + }); + server.listen(0, '127.0.0.1', () => { + const { port } = server.address() as net.AddressInfo; + resolve({ + port, + received, + close: () => + new Promise((res, rej) => server.close((e) => (e ? rej(e) : res()))), + }); + }); + server.on('error', reject); + }); +} + /** Start a TLS echo server (for terminate-TLS tests). */ export function startTlsEchoServer(certPem: string, keyPem: string): Promise { return new Promise((resolve, reject) => { diff --git a/package.json b/package.json index 669a12f..11f1c88 100644 --- a/package.json +++ b/package.json @@ -31,7 +31,7 @@ "artifacts": "napi artifacts", "prepublishOnly": "napi prepublish -t npm", "version": "napi version", - "test": "tsc -p tsconfig.test.json && node --test --test-force-exit dist-test/__test__/protection.spec.js dist-test/__test__/proxy.spec.js dist-test/__test__/suspended.spec.js dist-test/__test__/http-listener.spec.js dist-test/__test__/server.spec.js dist-test/__test__/h2-dispatch.spec.js", + "test": "tsc -p tsconfig.test.json && node --test --test-force-exit dist-test/__test__/protection.spec.js dist-test/__test__/proxy.spec.js dist-test/__test__/proxy-protocol-v2.spec.js dist-test/__test__/suspended.spec.js dist-test/__test__/http-listener.spec.js dist-test/__test__/server.spec.js dist-test/__test__/h2-dispatch.spec.js", "test:integration": "tsc -p tsconfig.test.json && node --test --test-force-exit dist-test/__test__/harper-integration.spec.js", "benchmark": "tsc -p tsconfig.test.json && node dist-test/__test__/benchmark.js", "benchmark:throughput": "tsc -p tsconfig.test.json && node dist-test/__test__/benchmark-throughput.js", diff --git a/src/http_listener.rs b/src/http_listener.rs index 4107351..af4c701 100644 --- a/src/http_listener.rs +++ b/src/http_listener.rs @@ -121,7 +121,7 @@ async fn handle_http(mut stream: TcpStream, peer_addr: SocketAddr, ctx: Arc pair, Ok(Err(e)) => { tracing::debug!("http :80 header read error from {}: {e}", peer_addr.ip()); diff --git a/src/http_proxy.rs b/src/http_proxy.rs index d2a6b30..780260a 100644 --- a/src/http_proxy.rs +++ b/src/http_proxy.rs @@ -12,109 +12,168 @@ use std::io; use tokio::io::{AsyncRead, AsyncReadExt, AsyncWrite, AsyncWriteExt}; const MAX_HEADER_SIZE: usize = 64 * 1024; +/// Cap on a chunked-framing control line (`[;ext]CRLF`, or a trailer line). Generous for +/// chunk extensions while keeping a line that never terminates from growing the buffer forever. +const MAX_CHUNK_LINE: usize = 1024; + +fn bad(msg: &str) -> io::Error { + io::Error::new(io::ErrorKind::InvalidData, msg) +} // ── Header framing ───────────────────────────────────────────────────────────── -/// Read from `reader` until the first `\r\n\r\n` (end of HTTP headers). +/// Read from `reader` into `carry` until it holds a complete header block, returning the offset +/// just past that block's terminating `\r\n\r\n`. /// -/// Returns `(header_block, excess)`: -/// - `header_block` includes the trailing `\r\n\r\n`. -/// - `excess` contains any bytes read beyond the header boundary. -/// - Both vecs are empty if EOF is reached before any data arrives. -pub async fn read_http_headers( +/// `carry` may already hold bytes left over from an earlier message, and on return everything +/// past the returned offset is the body / the next request — so this composes across the requests +/// of a keep-alive connection. `Ok(None)` means a clean EOF with nothing buffered; an EOF part-way +/// through a header block is `UnexpectedEof`. The block is capped at `MAX_HEADER_SIZE`. +pub async fn read_header_block( reader: &mut R, -) -> io::Result<(Vec, Vec)> { - let mut buf: Vec = Vec::with_capacity(2048); + carry: &mut Vec, +) -> io::Result> { let mut tmp = [0u8; 4096]; + // Bytes before this offset can't begin a match that a later read completes. + let mut scanned = 0usize; loop { - if buf.len() > MAX_HEADER_SIZE { - return Err(io::Error::new( - io::ErrorKind::InvalidData, - "HTTP headers exceed 64 KB limit", - )); + if let Some(rel) = carry[scanned..].windows(4).position(|w| w == b"\r\n\r\n") { + return Ok(Some(scanned + rel + 4)); } + scanned = carry.len().saturating_sub(3); - // Search only in the most-recently-appended region. - let search_start = buf.len().saturating_sub(3 + tmp.len()); // overlap for boundary splits - if let Some(rel) = buf[search_start..].windows(4).position(|w| w == b"\r\n\r\n") { - let end = search_start + rel + 4; - let excess = buf[end..].to_vec(); - buf.truncate(end); - return Ok((buf, excess)); + if carry.len() > MAX_HEADER_SIZE { + return Err(bad("HTTP headers exceed 64 KB limit")); } let n = reader.read(&mut tmp).await?; if n == 0 { - return Ok((Vec::new(), Vec::new())); // EOF before headers complete + return if carry.is_empty() { + Ok(None) + } else { + Err(io::Error::new(io::ErrorKind::UnexpectedEof, "closed mid-header")) + }; } - buf.extend_from_slice(&tmp[..n]); + carry.extend_from_slice(&tmp[..n]); + } +} + +/// Read from `reader` until the first `\r\n\r\n` (end of HTTP headers). +/// +/// Returns `(header_block, excess)`: +/// - `header_block` includes the trailing `\r\n\r\n`. +/// - `excess` contains any bytes read beyond the header boundary. +/// - Both vecs are empty if EOF is reached before any data arrives. +pub async fn read_http_headers( + reader: &mut R, +) -> io::Result<(Vec, Vec)> { + let mut carry: Vec = Vec::with_capacity(2048); + match read_header_block(reader, &mut carry).await { + Ok(Some(end)) => { + let excess = carry[end..].to_vec(); + carry.truncate(end); + Ok((carry, excess)) + } + Ok(None) => Ok((Vec::new(), Vec::new())), + // An EOF part-way through the headers stays a soft "nothing to serve" for this caller. + Err(e) if e.kind() == io::ErrorKind::UnexpectedEof => Ok((Vec::new(), Vec::new())), + Err(e) => Err(e), } } // ── Header field helpers ─────────────────────────────────────────────────────── -/// Return the trimmed value of the first header whose name matches `name_lower` -/// (case-insensitive comparison; caller must pass lowercase). Skips the -/// request/status line. -fn header_value<'a>(headers: &'a str, name_lower: &str) -> Option<&'a str> { - for line in headers.split("\r\n").skip(1) { - if let Some(colon) = line.find(':') { - if line[..colon].trim().to_ascii_lowercase() == name_lower { - return Some(line[colon + 1..].trim()); +/// Iterate the header fields of `block` as raw `(name, value)` byte slices, +/// skipping the request/status line and stopping at the blank line. Byte-oriented +/// so a header value carrying obs-text (0x80–0xFF, valid per RFC 7230 §3.2.6) +/// never poisons parsing of *other* fields the way whole-block UTF-8 validation did. +fn header_fields(block: &[u8]) -> impl Iterator { + let body_start = block.len(); // fields never span past the block the caller framed + let mut rest = match block.windows(2).position(|w| w == b"\r\n") { + Some(p) => &block[p + 2..body_start], + None => &[][..], + }; + std::iter::from_fn(move || { + loop { + let line_end = rest.windows(2).position(|w| w == b"\r\n")?; + let line = &rest[..line_end]; + rest = &rest[line_end + 2..]; + if line.is_empty() { + return None; // end of headers } + let Some(colon) = line.iter().position(|&b| b == b':') else { + continue; + }; + return Some((trim_ascii(&line[..colon]), trim_ascii(&line[colon + 1..]))); } - } - None + }) +} + +fn trim_ascii(bytes: &[u8]) -> &[u8] { + let start = bytes.iter().position(|b| !b.is_ascii_whitespace()).unwrap_or(bytes.len()); + let end = bytes.iter().rposition(|b| !b.is_ascii_whitespace()).map_or(start, |p| p + 1); + &bytes[start..end] +} + +/// The trimmed value of the first header named `name` (case-insensitive). +fn header_field<'a>(block: &'a [u8], name: &str) -> Option<&'a [u8]> { + header_fields(block) + .find(|(n, _)| n.eq_ignore_ascii_case(name.as_bytes())) + .map(|(_, v)| v) +} + +/// Return `true` if any `Connection` header (there may be several, each a +/// comma-separated token list) contains `token`. +fn connection_has_token(block: &[u8], token: &str) -> bool { + header_fields(block) + .filter(|(n, _)| n.eq_ignore_ascii_case(b"connection")) + .flat_map(|(_, v)| v.split(|&b| b == b',')) + .any(|t| trim_ascii(t).eq_ignore_ascii_case(token.as_bytes())) } /// Parse the `Content-Length` header value. pub fn content_length(headers: &[u8]) -> Option { - let text = std::str::from_utf8(headers).ok()?; - header_value(text, "content-length")?.parse().ok() + std::str::from_utf8(header_field(headers, "content-length")?) + .ok()? + .parse() + .ok() } /// Return `true` if `Transfer-Encoding: chunked` is present. pub fn is_transfer_encoding_chunked(headers: &[u8]) -> bool { - let text = match std::str::from_utf8(headers) { - Ok(t) => t, - Err(_) => return false, - }; - header_value(text, "transfer-encoding") - .map(|v| v.to_ascii_lowercase().contains("chunked")) + header_field(headers, "transfer-encoding") + .map(|v| { + v.split(|&b| b == b',') + .any(|t| trim_ascii(t).eq_ignore_ascii_case(b"chunked")) + }) .unwrap_or(false) } /// Return `true` if the `Connection: close` header is present. pub fn is_connection_close(headers: &[u8]) -> bool { - let text = match std::str::from_utf8(headers) { - Ok(t) => t, - Err(_) => return false, - }; - header_value(text, "connection") - .map(|v| v.trim().eq_ignore_ascii_case("close")) - .unwrap_or(false) + connection_has_token(headers, "close") } /// Return `true` if a protocol upgrade is requested (e.g. WebSocket). +/// `Connection` may be split across several fields (`Connection: keep-alive` + +/// `Connection: Upgrade`) — every field's token list counts. pub fn is_upgrade(headers: &[u8]) -> bool { - let text = match std::str::from_utf8(headers) { - Ok(t) => t, - Err(_) => return false, - }; - header_value(text, "upgrade").is_some() - && header_value(text, "connection") - .map(|v| v.to_ascii_lowercase().contains("upgrade")) - .unwrap_or(false) + header_fields(headers).any(|(n, _)| n.eq_ignore_ascii_case(b"upgrade")) + && connection_has_token(headers, "upgrade") } /// Parse the HTTP response status code (first line: `HTTP/1.x NNN ...`). pub fn status_code(headers: &[u8]) -> u16 { - std::str::from_utf8(headers) - .ok() - .and_then(|s| s.split_whitespace().nth(1)) - .and_then(|code| code.parse().ok()) - .unwrap_or(200) + parse_status(headers).unwrap_or(200) +} + +/// Status code of a response head, or `None` when the status line is malformed. +fn parse_status(headers: &[u8]) -> Option { + let line_end = headers.windows(2).position(|w| w == b"\r\n")?; + let mut fields = headers[..line_end].split(|&b| b == b' ').filter(|f| !f.is_empty()); + let code = fields.nth(1)?; + std::str::from_utf8(code).ok()?.parse().ok() } /// Byte slice of the HTTP request method (first token before a space). @@ -138,15 +197,14 @@ pub fn request_target(headers: &[u8]) -> &[u8] { } /// Return the trimmed value of the `Host` header (case-insensitive), or `None` -/// if absent or the headers aren't valid UTF-8. Strips any `:port` suffix. +/// if absent or the value isn't valid UTF-8. Strips any `:port` suffix. /// IPv6 literals such as `[::1]:80` are returned as `[::1]`. /// /// Returns `None` if the value contains a CR, LF, or NUL byte — guarding /// against response-splitting attacks when the result is interpolated into /// a redirect `Location` header. pub fn host_header(headers: &[u8]) -> Option<&str> { - let text = std::str::from_utf8(headers).ok()?; - let value = header_value(text, "host")?; + let value = std::str::from_utf8(header_field(headers, "host")?).ok()?; if value.bytes().any(|b| matches!(b, b'\r' | b'\n' | 0)) { return None; } @@ -297,7 +355,7 @@ where if read == 0 { return Err(io::Error::new( io::ErrorKind::UnexpectedEof, - "upstream closed mid-body", + "closed mid-body", )); } writer.write_all(&buf[..read]).await?; @@ -305,3 +363,808 @@ where } Ok(()) } + +// ── Request-stream rewriting (client → upstream) ─────────────────────────────── + +/// A header symphony owns end-to-end on the client→upstream path. +/// +/// Any client-supplied copy is dropped, and `value` — when symphony has an authoritative one to +/// state — is inserted after the request line. A `None` value still strips: a client must never be +/// able to smuggle a value through on a request where symphony has nothing of its own to say. +pub struct HeaderRewrite { + pub name: &'static str, + pub value: Option, +} + +/// How the body of the request described by `block` is framed — i.e. where the *next* request on +/// this connection starts. +enum RequestBody { + /// No body; the next request follows the header block immediately. + None, + Length(u64), + Chunked, +} + +/// Rewrite one request's header block: drop every client-supplied copy of a `rewrites` name, and +/// insert symphony's authoritative values right after the request line. `block` must be a complete +/// header block including the trailing `\r\n\r\n`. +/// +/// Malformed framing is refused rather than forwarded — a header block we can't parse exactly is +/// one whose stripping we can't guarantee. Obs-fold continuations (RFC 7230 §3.2.4 deprecates +/// them) are rejected in particular because a fold following a stripped header would silently +/// re-attach the attacker's value to the preceding header. +fn rewrite_request_head(block: &[u8], rewrites: &[HeaderRewrite]) -> io::Result> { + let Some(rl_end) = block.windows(2).position(|w| w == b"\r\n").map(|p| p + 2) else { + return Err(bad("malformed HTTP request line")); + }; + + let mut out = Vec::with_capacity(block.len() + 128); + out.extend_from_slice(&block[..rl_end]); + for r in rewrites { + if let Some(value) = &r.value { + out.extend_from_slice(format!("{}: {}\r\n", r.name, value).as_bytes()); + } + } + + let mut i = rl_end; + while i < block.len() { + let rel = block[i..] + .windows(2) + .position(|w| w == b"\r\n") + .ok_or_else(|| bad("unterminated header line"))?; + if rel == 0 { + out.extend_from_slice(b"\r\n"); // end-of-headers blank line + break; + } + let line = &block[i..i + rel]; + if matches!(line[0], b' ' | b'\t') { + return Err(bad("obs-fold header continuation is not allowed")); + } + let colon = line + .iter() + .position(|&b| b == b':') + .ok_or_else(|| bad("header line has no colon"))?; + let name = &line[..colon]; + // `Foo : v` — whitespace before the colon is a classic smuggling primitive: peers + // disagree on whether the name is `Foo` or `Foo `. + if name.is_empty() || name.iter().any(|b| b.is_ascii_whitespace()) { + return Err(bad("malformed header name")); + } + if !rewrites.iter().any(|r| r.name.as_bytes().eq_ignore_ascii_case(name)) { + out.extend_from_slice(&block[i..i + rel + 2]); + } + i += rel + 2; + } + + Ok(out) +} + +/// Determine the body framing of the request in `block`. Byte-oriented: a header +/// value carrying obs-text (0x80–0xFF, valid per RFC 7230 §3.2.6) must not get the +/// request rejected — only the framing fields themselves need to parse. +fn request_body(block: &[u8]) -> io::Result { + let mut transfer_encoding = false; + let mut chunked = false; + let mut length: Option = None; + + for (name, value) in header_fields(block) { + if name.eq_ignore_ascii_case(b"transfer-encoding") { + transfer_encoding = true; + // Only a *final* `chunked` leaves us frame boundaries we can follow. + chunked = value + .rsplit(|&b| b == b',') + .next() + .map(|last| trim_ascii(last).eq_ignore_ascii_case(b"chunked")) + .unwrap_or(false); + } else if name.eq_ignore_ascii_case(b"content-length") { + let n: u64 = std::str::from_utf8(value) + .ok() + .and_then(|v| v.parse().ok()) + .ok_or_else(|| bad("malformed Content-Length"))?; + if length.map(|prev| prev != n).unwrap_or(false) { + return Err(bad("conflicting Content-Length headers")); + } + length = Some(n); + } + } + + if transfer_encoding { + // TE+CL together is the classic request-smuggling primitive, and a non-chunked TE gives + // us no boundary at all. Refuse instead of guessing where the next request begins. + if length.is_some() { + return Err(bad("both Transfer-Encoding and Content-Length present")); + } + if !chunked { + return Err(bad("unsupported Transfer-Encoding")); + } + return Ok(RequestBody::Chunked); + } + + Ok(match length { + Some(0) | None => RequestBody::None, + Some(n) => RequestBody::Length(n), + }) +} + +/// Copy `n` body bytes through, draining anything already buffered in `carry` before reading more. +async fn copy_body(reader: &mut R, writer: &mut W, carry: &mut Vec, n: u64) -> io::Result<()> +where + R: AsyncRead + Unpin, + W: AsyncWrite + Unpin, +{ + let buffered = (carry.len() as u64).min(n) as usize; + writer.write_all(&carry[..buffered]).await?; + carry.drain(..buffered); + let remaining = n - buffered as u64; + if remaining > 0 { + copy_exact(reader, writer, remaining).await?; + } + Ok(()) +} + +/// Read one CRLF-terminated control line, drawing from `carry` first. The returned line includes +/// its CRLF and is removed from `carry`; any bytes past it stay buffered. +async fn read_control_line( + reader: &mut R, + carry: &mut Vec, +) -> io::Result> { + let mut tmp = [0u8; 256]; + let mut scanned = 0usize; + loop { + if let Some(rel) = carry[scanned..].windows(2).position(|w| w == b"\r\n") { + let end = scanned + rel + 2; + let line = carry[..end].to_vec(); + carry.drain(..end); + return Ok(line); + } + scanned = carry.len().saturating_sub(1); + + if carry.len() > MAX_CHUNK_LINE { + return Err(bad("chunked framing line exceeds limit")); + } + let n = reader.read(&mut tmp).await?; + if n == 0 { + return Err(io::Error::new(io::ErrorKind::UnexpectedEof, "closed mid-chunk")); + } + carry.extend_from_slice(&tmp[..n]); + } +} + +/// Stream a chunked body through verbatim, following its frame markers only so we know where the +/// next request begins. Chunk data and trailers are passed through untouched — backends surface +/// trailers separately from headers, so they aren't a header-spoofing surface. +async fn copy_chunked_body(reader: &mut R, writer: &mut W, carry: &mut Vec) -> io::Result<()> +where + R: AsyncRead + Unpin, + W: AsyncWrite + Unpin, +{ + loop { + let line = read_control_line(reader, carry).await?; + let size_field = line[..line.len() - 2].split(|&b| b == b';').next().unwrap_or(&[]); + let size_text = std::str::from_utf8(size_field) + .map_err(|_| bad("malformed chunk size"))? + .trim(); + let size = u64::from_str_radix(size_text, 16).map_err(|_| bad("malformed chunk size"))?; + writer.write_all(&line).await?; + + if size == 0 { + // Trailer section: lines until a blank one closes the body. + loop { + let trailer = read_control_line(reader, carry).await?; + writer.write_all(&trailer).await?; + if trailer == b"\r\n" { + return Ok(()); + } + } + } + + copy_body(reader, writer, carry, size).await?; + let crlf = read_control_line(reader, carry).await?; + if crlf != b"\r\n" { + return Err(bad("missing CRLF after chunk data")); + } + writer.write_all(&crlf).await?; + } +} + +/// Per-request metadata the request pump hands the response pump so responses can be framed and +/// matched back to what asked for them. +struct RequestMeta { + /// HEAD responses carry framing headers but no body. + head_only: bool, + /// CONNECT: a 2xx response establishes a tunnel. + connect: bool, + /// Upgrade request: a 101 response establishes a tunnel. + upgrade: bool, + /// Present on tunnel candidates: the response pump reports whether the upstream accepted. + verdict: Option>, +} + +/// Rewrite the header block of **every** HTTP/1 request on the client→upstream half of a +/// connection, applying `rewrites` to each. +/// +/// Each request's header block is framed in full before it is rewritten (bounded at +/// `MAX_HEADER_SIZE`), so a client can't slip a spoofed header past the strip by fragmenting it +/// across TCP segments, pushing it beyond a fixed read size, or sending it on a later keep-alive +/// or pipelined request. Bodies are framed — not buffered — purely to locate the next request. +/// +/// A CONNECT/Upgrade request does **not** switch this pump to raw passthrough by itself: the +/// switch is gated on the response pump reporting that the upstream actually accepted the tunnel +/// (101, or 2xx for CONNECT). Without that gate, a client could pipeline a spoofed-header request +/// behind an upgrade the upstream rejects-but-keeps-alive and have it forwarded verbatim. +async fn rewrite_request_stream( + reader: &mut R, + writer: &mut W, + rewrites: &[HeaderRewrite], + meta_tx: &tokio::sync::mpsc::UnboundedSender, +) -> io::Result<()> +where + R: AsyncRead + Unpin, + W: AsyncWrite + Unpin, +{ + let mut carry: Vec = Vec::with_capacity(2048); + loop { + let Some(end) = read_header_block(reader, &mut carry).await? else { + return Ok(()); // clean EOF between requests + }; + let head = rewrite_request_head(&carry[..end], rewrites)?; + let is_connect = request_method(&carry[..end]).eq_ignore_ascii_case(b"CONNECT"); + let is_tunnel_candidate = is_connect || is_upgrade(&carry[..end]); + // CONNECT has no body by definition; an Upgrade request still uses normal framing + // until the upstream switches protocols. + let body = if is_connect { RequestBody::None } else { request_body(&carry[..end])? }; + let head_only = request_method(&carry[..end]).eq_ignore_ascii_case(b"HEAD"); + + let (verdict_tx, verdict_rx) = if is_tunnel_candidate { + let (tx, rx) = tokio::sync::oneshot::channel(); + (Some(tx), Some(rx)) + } else { + (None, None) + }; + // Send meta before the head so the response pump's queue is in request order. + if meta_tx + .send(RequestMeta { head_only, connect: is_connect, upgrade: is_tunnel_candidate && !is_connect, verdict: verdict_tx }) + .is_err() + { + return Ok(()); // response pump is gone — the connection is closing + } + writer.write_all(&head).await?; + carry.drain(..end); + + match body { + RequestBody::None => {} + RequestBody::Length(n) => copy_body(reader, writer, &mut carry, n).await?, + RequestBody::Chunked => copy_chunked_body(reader, writer, &mut carry).await?, + } + + if let Some(verdict_rx) = verdict_rx { + match verdict_rx.await { + Ok(true) => { + // Tunnel established: the upgraded protocol isn't ours to parse. + writer.write_all(&carry).await?; + carry.clear(); + tokio::io::copy(reader, writer).await?; + return Ok(()); + } + // Upstream declined the tunnel and kept the connection: whatever the client + // pipelined behind the upgrade is HTTP and stays under the rewriter. + Ok(false) => {} + Err(_) => return Ok(()), // response pump ended (upstream closed) + } + } + } +} + +/// How the body of the response described by a head block is framed. +enum ResponseBody { + None, + Length(u64), + Chunked, + /// No framing headers on a body-bearing response: the body runs to connection close. + ReadToEnd, +} + +fn response_body(head: &[u8], status: u16, head_only: bool) -> io::Result { + if head_only || !response_has_body(status, if head_only { b"HEAD" } else { b"GET" }) { + return Ok(ResponseBody::None); + } + if is_transfer_encoding_chunked(head) { + return Ok(ResponseBody::Chunked); + } + match content_length(head) { + Some(0) => Ok(ResponseBody::None), + Some(n) => Ok(ResponseBody::Length(n)), + None => Ok(ResponseBody::ReadToEnd), + } +} + +/// Forward upstream responses to the client, framing each one so tunnel verdicts can be +/// reported back to the request pump (see `rewrite_request_stream`). +async fn forward_response_stream( + reader: &mut R, + writer: &mut W, + meta_rx: &mut tokio::sync::mpsc::UnboundedReceiver, +) -> io::Result<()> +where + R: AsyncRead + Unpin, + W: AsyncWrite + Unpin, +{ + let mut carry: Vec = Vec::with_capacity(2048); + loop { + let Some(end) = read_header_block(reader, &mut carry).await? else { + return Ok(()); // upstream closed between responses + }; + let status = parse_status(&carry[..end]).ok_or_else(|| bad("malformed response status line"))?; + + // Interim responses (100 Continue etc.) precede the real one; 101 is the upgrade accept. + if (100..200).contains(&status) && status != 101 { + writer.write_all(&carry[..end]).await?; + carry.drain(..end); + continue; + } + + let Some(mut meta) = meta_rx.recv().await else { + // Response without a matching forwarded request: refuse to guess at framing. + return Err(bad("upstream response with no corresponding request")); + }; + + let tunnel_established = + (meta.upgrade && status == 101) || (meta.connect && (200..300).contains(&status)); + if let Some(verdict) = meta.verdict.take() { + let _ = verdict.send(tunnel_established); + } else if status == 101 { + return Err(bad("unsolicited 101 response")); + } + + let framing = response_body(&carry[..end], status, meta.head_only)?; + writer.write_all(&carry[..end]).await?; + carry.drain(..end); + + if tunnel_established { + // Tunnel: the rest of the upstream stream is opaque; pass it through. + writer.write_all(&carry).await?; + carry.clear(); + tokio::io::copy(reader, writer).await?; + return Ok(()); + } + + match framing { + ResponseBody::None => {} + ResponseBody::Length(n) => copy_body(reader, writer, &mut carry, n).await?, + ResponseBody::Chunked => copy_chunked_body(reader, writer, &mut carry).await?, + ResponseBody::ReadToEnd => { + writer.write_all(&carry).await?; + carry.clear(); + tokio::io::copy(reader, writer).await?; + return Ok(()); + } + } + } +} + +/// Bidirectional HTTP/1 proxying where the client→upstream half rewrites every request head and +/// tunnel switches (CONNECT / Upgrade) are gated on the upstream actually accepting them. The two +/// pumps run concurrently; each writer is shut down when its source reaches EOF (keep-alive +/// half-close), mirroring `copy_bidirectional`. +pub async fn proxy_http1_rewriting( + client: &mut C, + upstream: &mut U, + rewrites: &[HeaderRewrite], +) -> io::Result<()> +where + C: AsyncRead + AsyncWrite + Unpin, + U: AsyncRead + AsyncWrite + Unpin, +{ + let (mut client_read, mut client_write) = tokio::io::split(client); + let (mut upstream_read, mut upstream_write) = tokio::io::split(upstream); + let (meta_tx, mut meta_rx) = tokio::sync::mpsc::unbounded_channel(); + + let client_to_upstream = async { + let result = rewrite_request_stream(&mut client_read, &mut upstream_write, rewrites, &meta_tx).await; + drop(meta_tx); // release the response pump's queue so it can finish + let _ = upstream_write.shutdown().await; + result + }; + let upstream_to_client = async { + let result = forward_response_stream(&mut upstream_read, &mut client_write, &mut meta_rx).await; + let _ = client_write.shutdown().await; + result + }; + let (a, b) = tokio::join!(client_to_upstream, upstream_to_client); + a.and(b) +} + +#[cfg(test)] +mod tests { + use super::*; + use std::collections::VecDeque; + use std::pin::Pin; + use std::task::{Context, Poll}; + + const JA3: &str = "0123456789abcdef0123456789abcdef"; + + /// A reader that yields its data in caller-defined chunks — one chunk per `poll_read` — so a + /// test can reproduce TCP fragmentation exactly (a header split across two segments, etc.). + struct ChunkReader { + chunks: VecDeque>, + } + + impl ChunkReader { + fn new(chunks: &[&[u8]]) -> Self { + Self { chunks: chunks.iter().map(|c| c.to_vec()).collect() } + } + } + + impl AsyncRead for ChunkReader { + fn poll_read( + mut self: Pin<&mut Self>, + _cx: &mut Context<'_>, + buf: &mut tokio::io::ReadBuf<'_>, + ) -> Poll> { + if let Some(front) = self.chunks.front_mut() { + let n = front.len().min(buf.remaining()); + buf.put_slice(&front[..n]); + front.drain(..n); + if front.is_empty() { + self.chunks.pop_front(); + } + } + Poll::Ready(Ok(())) + } + } + + fn ja3_rewrite(value: Option<&str>) -> Vec { + vec![HeaderRewrite { name: "X-JA3", value: value.map(str::to_string) }] + } + + async fn run(chunks: &[&[u8]], rewrites: &[HeaderRewrite]) -> io::Result { + let mut reader = ChunkReader::new(chunks); + let mut out: Vec = Vec::new(); + // Keep the receiver alive so meta sends succeed; none of these requests are + // tunnel candidates, so no verdict is awaited. + let (meta_tx, _meta_rx) = tokio::sync::mpsc::unbounded_channel(); + rewrite_request_stream(&mut reader, &mut out, rewrites, &meta_tx).await?; + Ok(String::from_utf8(out).unwrap()) + } + + /// Drive the full request/response pump pair over in-memory duplex pipes: the test + /// scripts the client's bytes and the upstream's responses, and gets back what each + /// side received. + async fn run_pair(client_sends: &[u8], upstream_sends: &[u8], rewrites: &[HeaderRewrite]) -> (String, String) { + let (mut client_side, mut proxy_client_side) = tokio::io::duplex(64 * 1024); + let (mut proxy_upstream_side, mut upstream_side) = tokio::io::duplex(64 * 1024); + + let proxy = proxy_http1_rewriting(&mut proxy_client_side, &mut proxy_upstream_side, rewrites); + + let client_data = client_sends.to_vec(); + let client = async move { + client_side.write_all(&client_data).await.unwrap(); + client_side.shutdown().await.unwrap(); + let mut received = Vec::new(); + let _ = client_side.read_to_end(&mut received).await; + received + }; + let upstream_data = upstream_sends.to_vec(); + let upstream = async move { + let mut received = Vec::new(); + // Read everything the proxy forwards, then answer with the scripted responses. + // (Write first for tunnel bytes to flow; ordering is fine over duplex buffers.) + upstream_side.write_all(&upstream_data).await.unwrap(); + upstream_side.shutdown().await.unwrap(); + let _ = upstream_side.read_to_end(&mut received).await; + received + }; + + let (_, client_received, upstream_received) = tokio::join!(proxy, client, upstream); + ( + String::from_utf8_lossy(&upstream_received).into_owned(), + String::from_utf8_lossy(&client_received).into_owned(), + ) + } + + #[tokio::test] + async fn injects_after_request_line() { + let got = run( + &[b"GET / HTTP/1.1\r\nHost: x\r\n\r\n"], + &ja3_rewrite(Some(JA3)), + ) + .await + .unwrap(); + assert_eq!(got, format!("GET / HTTP/1.1\r\nX-JA3: {JA3}\r\nHost: x\r\n\r\n")); + } + + #[tokio::test] + async fn strips_client_supplied_copy_case_insensitive() { + let got = run( + &[b"POST /p HTTP/1.1\r\nx-ja3: deadbeef\r\nHost: x\r\nContent-Length: 2\r\n\r\nhi"], + &ja3_rewrite(Some(JA3)), + ) + .await + .unwrap(); + assert_eq!( + got, + format!("POST /p HTTP/1.1\r\nX-JA3: {JA3}\r\nHost: x\r\nContent-Length: 2\r\n\r\nhi") + ); + assert!(!got.contains("deadbeef")); + } + + // Finding 2 (a): a spoofed header split across two TCP segments must still be stripped. + #[tokio::test] + async fn strips_header_fragmented_across_reads() { + let got = run( + &[b"GET / HTTP/1.1\r\nX-JA", b"3: deadbeef\r\nHost: x\r\n\r", b"\n"], + &ja3_rewrite(Some(JA3)), + ) + .await + .unwrap(); + assert_eq!(got, format!("GET / HTTP/1.1\r\nX-JA3: {JA3}\r\nHost: x\r\n\r\n")); + assert!(!got.contains("deadbeef")); + } + + // Finding 2 (c): a second keep-alive request must be rewritten too, not passed through raw. + #[tokio::test] + async fn rewrites_every_keep_alive_request() { + let got = run( + &[ + b"GET /1 HTTP/1.1\r\nHost: x\r\n\r\n", + b"GET /2 HTTP/1.1\r\nX-JA3: spoofed\r\nHost: x\r\n\r\n", + ], + &ja3_rewrite(Some(JA3)), + ) + .await + .unwrap(); + assert_eq!( + got, + format!( + "GET /1 HTTP/1.1\r\nX-JA3: {JA3}\r\nHost: x\r\n\r\n\ + GET /2 HTTP/1.1\r\nX-JA3: {JA3}\r\nHost: x\r\n\r\n" + ) + ); + assert!(!got.contains("spoofed")); + } + + // A body between two pipelined requests must pass through verbatim, and the request after it + // must still be rewritten (proves body framing locates the next request correctly). + #[tokio::test] + async fn frames_content_length_body_then_rewrites_next() { + let got = run( + &[ + b"POST /1 HTTP/1.1\r\nContent-Length: 5\r\n\r\nhello", + b"GET /2 HTTP/1.1\r\nX-JA3: spoofed\r\n\r\n", + ], + &ja3_rewrite(Some(JA3)), + ) + .await + .unwrap(); + assert_eq!( + got, + format!( + "POST /1 HTTP/1.1\r\nX-JA3: {JA3}\r\nContent-Length: 5\r\n\r\nhello\ + GET /2 HTTP/1.1\r\nX-JA3: {JA3}\r\n\r\n" + ) + ); + assert!(!got.contains("spoofed")); + } + + #[tokio::test] + async fn frames_chunked_body_then_rewrites_next() { + let got = run( + &[ + b"POST /1 HTTP/1.1\r\nTransfer-Encoding: chunked\r\n\r\n5\r\nhello\r\n0\r\n\r\n", + b"GET /2 HTTP/1.1\r\nX-JA3: spoofed\r\n\r\n", + ], + &ja3_rewrite(Some(JA3)), + ) + .await + .unwrap(); + assert_eq!( + got, + format!( + "POST /1 HTTP/1.1\r\nX-JA3: {JA3}\r\nTransfer-Encoding: chunked\r\n\r\n5\r\nhello\r\n0\r\n\r\n\ + GET /2 HTTP/1.1\r\nX-JA3: {JA3}\r\n\r\n" + ) + ); + assert!(!got.contains("spoofed")); + } + + // Finding 2 (Medium): with no authoritative value the client copy is still stripped, and no + // bogus replacement is injected. + #[tokio::test] + async fn strips_even_when_no_replacement_value() { + let got = run( + &[b"GET / HTTP/1.1\r\nX-JA3: spoofed\r\nHost: x\r\n\r\n"], + &ja3_rewrite(None), + ) + .await + .unwrap(); + assert_eq!(got, "GET / HTTP/1.1\r\nHost: x\r\n\r\n"); + assert!(!got.contains("spoofed")); + assert!(!got.contains("X-JA3")); + } + + // Finding 2 (Slowloris-adjacent bound): a header block past the 64 KB cap is refused, not + // buffered unbounded. + #[tokio::test] + async fn oversized_header_block_is_rejected() { + let mut req = b"GET / HTTP/1.1\r\n".to_vec(); + req.extend_from_slice(b"X-Pad: "); + req.resize(req.len() + 70 * 1024, b'A'); + req.extend_from_slice(b"\r\n\r\n"); + let mut reader = ChunkReader::new(&[&req]); + let mut out: Vec = Vec::new(); + let (meta_tx, _meta_rx) = tokio::sync::mpsc::unbounded_channel(); + let err = rewrite_request_stream(&mut reader, &mut out, &ja3_rewrite(Some(JA3)), &meta_tx) + .await + .unwrap_err(); + assert_eq!(err.kind(), io::ErrorKind::InvalidData); + } + + // Finding 2 (partial-first-read corruption): a request line arriving before its terminator + // must not have headers prepended before it — the whole block is framed first. + #[tokio::test] + async fn partial_request_line_is_not_corrupted() { + let got = run( + &[b"GET /longpath HT", b"TP/1.1\r\nHost: x\r\n\r\n"], + &ja3_rewrite(Some(JA3)), + ) + .await + .unwrap(); + assert!(got.starts_with("GET /longpath HTTP/1.1\r\n"), "request line intact: {got:?}"); + assert_eq!(got, format!("GET /longpath HTTP/1.1\r\nX-JA3: {JA3}\r\nHost: x\r\n\r\n")); + } + + // An empty stream (client connects, sends nothing, closes) is a clean no-op — no spurious head. + #[tokio::test] + async fn empty_stream_is_clean() { + let got = run(&[b""], &ja3_rewrite(Some(JA3))).await.unwrap(); + assert_eq!(got, ""); + } + + // Request-smuggling primitives are refused rather than forwarded ambiguously. + #[tokio::test] + async fn rejects_te_and_cl_together() { + let err = run( + &[b"POST / HTTP/1.1\r\nTransfer-Encoding: chunked\r\nContent-Length: 5\r\n\r\n0\r\n\r\n"], + &ja3_rewrite(Some(JA3)), + ) + .await + .unwrap_err(); + assert_eq!(err.kind(), io::ErrorKind::InvalidData); + } + + #[tokio::test] + async fn rejects_obs_fold_continuation() { + let err = run( + &[b"GET / HTTP/1.1\r\nX-JA3: a\r\n\tcontinued\r\nHost: x\r\n\r\n"], + &ja3_rewrite(Some(JA3)), + ) + .await + .unwrap_err(); + assert_eq!(err.kind(), io::ErrorKind::InvalidData); + } + + #[tokio::test] + async fn no_rewrites_leaves_stream_verbatim() { + let got = run(&[b"GET / HTTP/1.1\r\nX-JA3: keep\r\n\r\n"], &[]).await.unwrap(); + assert_eq!(got, "GET / HTTP/1.1\r\nX-JA3: keep\r\n\r\n"); + } + + // Obs-text (0x80–0xFF) in an unrelated header value is valid per RFC 7230 §3.2.6 and + // must not get the request rejected — only the framing fields need to parse. + #[tokio::test] + async fn forwards_obs_text_header_values() { + let req: &[u8] = b"GET / HTTP/1.1\r\nX-Custom: caf\xE9\r\nHost: x\r\n\r\n"; + let mut reader = ChunkReader::new(&[req]); + let mut out: Vec = Vec::new(); + let (meta_tx, _meta_rx) = tokio::sync::mpsc::unbounded_channel(); + rewrite_request_stream(&mut reader, &mut out, &ja3_rewrite(Some(JA3)), &meta_tx) + .await + .unwrap(); + assert!(out.windows(4).any(|w| w == b"caf\xE9"), "obs-text header forwarded intact"); + assert!(String::from_utf8_lossy(&out).contains(&format!("X-JA3: {JA3}"))); + } + + // Multi-field Connection (`Connection: keep-alive` + `Connection: Upgrade`) is a valid + // upgrade request and must be classified as a tunnel candidate. + #[test] + fn upgrade_across_multiple_connection_fields() { + let head = b"GET /ws HTTP/1.1\r\nConnection: keep-alive\r\nConnection: Upgrade\r\nUpgrade: websocket\r\n\r\n"; + assert!(is_upgrade(head)); + let head_tokens = b"GET /ws HTTP/1.1\r\nConnection: keep-alive, Upgrade\r\nUpgrade: websocket\r\n\r\n"; + assert!(is_upgrade(head_tokens)); + let no_upgrade = b"GET / HTTP/1.1\r\nConnection: keep-alive\r\n\r\n"; + assert!(!is_upgrade(no_upgrade)); + } + + // ── Tunnel verdict gating (pump pair) ───────────────────────────────────── + + const UPGRADE_REQ: &[u8] = + b"GET /ws HTTP/1.1\r\nHost: x\r\nConnection: Upgrade\r\nUpgrade: websocket\r\nX-JA3: spoofed\r\n\r\n"; + + // Upstream accepts the upgrade: everything after the 101 flows raw, both directions. + #[tokio::test] + async fn tunnel_after_accepted_upgrade() { + let client_sends = [UPGRADE_REQ, b"\x88\x00raw-client-frames"].concat(); + let upstream_sends: &[u8] = + b"HTTP/1.1 101 Switching Protocols\r\nUpgrade: websocket\r\nConnection: Upgrade\r\n\r\n\x82\x01raw-upstream-frames"; + let (upstream_got, client_got) = run_pair(&client_sends, upstream_sends, &ja3_rewrite(Some(JA3))).await; + assert!(upstream_got.contains(&format!("X-JA3: {JA3}")), "upgrade head rewritten"); + assert!(!upstream_got.contains("spoofed")); + assert!(upstream_got.contains("raw-client-frames"), "client tunnel bytes flow"); + assert!(client_got.contains("101 Switching Protocols")); + assert!(client_got.contains("raw-upstream-frames"), "upstream tunnel bytes flow"); + } + + // The heskew scenario: upstream REJECTS the upgrade but keeps the connection alive. A + // request the client pipelined behind the upgrade must still be rewritten — not passed + // through raw. + #[tokio::test] + async fn rejected_upgrade_keeps_rewriting_pipelined_requests() { + let client_sends = [ + UPGRADE_REQ, + b"GET /2 HTTP/1.1\r\nHost: x\r\nX-JA3: smuggled\r\n\r\n" as &[u8], + ] + .concat(); + let upstream_sends: &[u8] = b"HTTP/1.1 400 Bad Request\r\nContent-Length: 0\r\n\r\n\ + HTTP/1.1 200 OK\r\nContent-Length: 2\r\n\r\nok"; + let (upstream_got, client_got) = run_pair(&client_sends, upstream_sends, &ja3_rewrite(Some(JA3))).await; + assert!(!upstream_got.contains("spoofed"), "upgrade head still stripped"); + assert!(!upstream_got.contains("smuggled"), "pipelined request must not bypass the rewriter"); + let occurrences = upstream_got.matches(&format!("X-JA3: {JA3}")).count(); + assert_eq!(occurrences, 2, "both requests carry the authoritative header"); + assert!(client_got.contains("400 Bad Request")); + assert!(client_got.contains("ok")); + } + + // CONNECT accepted (2xx) tunnels; CONNECT rejected keeps framing. + #[tokio::test] + async fn connect_verdicts() { + let accepted = run_pair( + &[b"CONNECT db:5432 HTTP/1.1\r\nHost: db\r\n\r\n" as &[u8], b"opaque-bytes"].concat(), + b"HTTP/1.1 200 Connection Established\r\n\r\ntunnel-back", + &ja3_rewrite(Some(JA3)), + ) + .await; + assert!(accepted.0.contains("opaque-bytes"), "client bytes tunnel after 2xx"); + assert!(accepted.1.contains("tunnel-back")); + + let rejected = run_pair( + &[ + b"CONNECT db:5432 HTTP/1.1\r\nHost: db\r\n\r\n" as &[u8], + b"GET /next HTTP/1.1\r\nX-JA3: smuggled\r\n\r\n", + ] + .concat(), + b"HTTP/1.1 403 Forbidden\r\nContent-Length: 0\r\n\r\nHTTP/1.1 200 OK\r\nContent-Length: 0\r\n\r\n", + &ja3_rewrite(Some(JA3)), + ) + .await; + assert!(!rejected.0.contains("smuggled"), "post-CONNECT request stays under the rewriter"); + } + + // An interim 100 Continue is forwarded without consuming the request's response slot. + #[tokio::test] + async fn interim_response_then_final() { + let (upstream_got, client_got) = run_pair( + b"POST / HTTP/1.1\r\nHost: x\r\nContent-Length: 2\r\nExpect: 100-continue\r\n\r\nhi", + b"HTTP/1.1 100 Continue\r\n\r\nHTTP/1.1 200 OK\r\nContent-Length: 4\r\n\r\ndone", + &ja3_rewrite(Some(JA3)), + ) + .await; + assert!(upstream_got.contains("hi")); + assert!(client_got.contains("100 Continue")); + assert!(client_got.contains("done")); + } + + // HEAD responses carry framing headers but no body: the next response must still frame. + #[tokio::test] + async fn head_response_body_is_not_consumed() { + let (_, client_got) = run_pair( + b"HEAD / HTTP/1.1\r\nHost: x\r\n\r\nGET /2 HTTP/1.1\r\nHost: x\r\n\r\n", + b"HTTP/1.1 200 OK\r\nContent-Length: 100\r\n\r\nHTTP/1.1 200 OK\r\nContent-Length: 3\r\n\r\nyes", + &ja3_rewrite(Some(JA3)), + ) + .await; + assert!(client_got.contains("Content-Length: 100")); + assert!(client_got.contains("yes")); + } +} diff --git a/src/proxy.rs b/src/proxy.rs index 5d4d5af..64cf6c6 100644 --- a/src/proxy.rs +++ b/src/proxy.rs @@ -4,7 +4,8 @@ use crate::metrics::{GlobalMetrics, ListenerMetrics}; use crate::protection::ProtectionState; use crate::proxy_conn::{ConnContext, JsEvent}; use crate::router::{ - build_route_table, ListenerTlsSpec, LiveRouteTable, RouteSpec, SourceAddressMode, UpstreamSpec, + build_route_table, ForwardFingerprint, ListenerTlsSpec, LiveRouteTable, RouteSpec, + SourceAddressMode, UpstreamSpec, }; use crate::suspended::{build_resolved_route, ResolveSpec, ResolveUpstream, SuspendedRegistry}; use ipnetwork::IpNetwork; @@ -63,8 +64,13 @@ pub struct JsRouteConfig { /// Token bucket burst ceiling (defaults to `maxConnectionsPerSecond`). pub burst: Option, /// How the real client IP is forwarded to the upstream. - /// "proxyProtocol" (default for UDS), "xForwardedFor", or "none" (default for TCP). + /// "proxyProtocol" (v1, default for UDS), "proxyProtocolV2", "xForwardedFor", or + /// "none" (default for TCP). pub source_address_header: Option, + /// Which client TLS fingerprint to forward downstream: "ja3", "ja4", or "none" + /// (default). Carried as a PROXY v2 TLV under "proxyProtocolV2", otherwise as an + /// injected X-JA3/X-JA4 HTTP header. + pub forward_fingerprint: Option, /// Advertise h2 in ALPN so clients can negotiate HTTP/2. Default: false. pub http2: Option, } @@ -136,6 +142,7 @@ pub struct JsResolveRoute { pub cert: Option, pub mtls: Option, pub source_address_header: Option, + pub forward_fingerprint: Option, pub http2: Option, } @@ -510,8 +517,9 @@ fn parse_route_spec(r: &JsRouteConfig) -> Result { let has_uds = upstreams.iter().any(|u| matches!(u, UpstreamSpec::Uds { .. })); let source_address_mode = parse_source_address_mode(r.source_address_header.as_deref(), has_uds)?; + let forward_fingerprint = parse_forward_fingerprint(r.forward_fingerprint.as_deref())?; - let result = Ok(RouteSpec { + let spec = RouteSpec { sni: r.sni.clone(), upstreams, terminate_tls: r.terminate_tls, @@ -524,15 +532,26 @@ fn parse_route_spec(r: &JsRouteConfig) -> Result { max_cps: r.max_connections_per_second, burst: r.burst, source_address_mode, + forward_fingerprint, http2: r.http2.unwrap_or(false), - }); + }; - if let Ok(ref spec) = result { - if spec.http2 && !spec.terminate_tls { - eprintln!("symphony: route '{}': http2=true has no effect when terminateTls=false (passthrough mode)", spec.sni); - } + if spec.http2 && !spec.terminate_tls { + eprintln!("symphony: route '{}': http2=true has no effect when terminateTls=false (passthrough mode)", spec.sni); + } + // An injected X-JA3/X-JA4 header needs a plaintext HTTP/1 upstream (terminated, not h2); the + // runtime skips it otherwise. The PROXY v2 TLV carrier works everywhere (it prefixes the raw + // bytes), so steer non-HTTP/1 routes to it. + if forward_fingerprint != ForwardFingerprint::None + && source_address_mode != SourceAddressMode::ProxyProtocolV2 + && (!spec.terminate_tls || spec.http2) + { + eprintln!( + "symphony: route '{}': forwardFingerprint via HTTP header has no effect on a non-HTTP/1 upstream (terminateTls=false or http2=true); use sourceAddressHeader='proxyProtocolV2'", + spec.sni + ); } - result + Ok(spec) } fn parse_upstream_spec(u: &JsUpstream, sni: &str) -> Result { @@ -594,6 +613,7 @@ fn parse_resolve_spec(r: &JsResolveRoute) -> Result { let has_uds = matches!(&upstream, ResolveUpstream::Uds { .. }); let source_address_mode = parse_source_address_mode(r.source_address_header.as_deref(), has_uds)?; + let forward_fingerprint = parse_forward_fingerprint(r.forward_fingerprint.as_deref())?; Ok(ResolveSpec { upstream, @@ -603,6 +623,7 @@ fn parse_resolve_spec(r: &JsResolveRoute) -> Result { mtls_ca_pem: r.mtls.as_ref().map(|m| pem_bytes(&m.client_ca_cert)), require_client_cert: r.mtls.as_ref().and_then(|m| m.require_client_cert).unwrap_or(false), source_address_mode, + forward_fingerprint, http2: r.http2.unwrap_or(false), }) } @@ -693,10 +714,11 @@ fn parse_source_address_mode( ) -> Result { match value { Some("proxyProtocol") => Ok(SourceAddressMode::ProxyProtocol), + Some("proxyProtocolV2") => Ok(SourceAddressMode::ProxyProtocolV2), Some("xForwardedFor") => Ok(SourceAddressMode::XForwardedFor), Some("none") => Ok(SourceAddressMode::None), Some(other) => Err(napi::Error::from_reason(format!( - "unknown sourceAddressHeader value '{other}'; expected 'proxyProtocol', 'xForwardedFor', or 'none'" + "unknown sourceAddressHeader value '{other}'; expected 'proxyProtocol', 'proxyProtocolV2', 'xForwardedFor', or 'none'" ))), // Default: proxyProtocol for UDS upstreams, none for TCP None => Ok(if has_uds_upstreams { @@ -707,6 +729,17 @@ fn parse_source_address_mode( } } +fn parse_forward_fingerprint(value: Option<&str>) -> Result { + match value { + None | Some("none") => Ok(ForwardFingerprint::None), + Some("ja3") => Ok(ForwardFingerprint::Ja3), + Some("ja4") => Ok(ForwardFingerprint::Ja4), + Some(other) => Err(napi::Error::from_reason(format!( + "unknown forwardFingerprint value '{other}'; expected 'ja3', 'ja4', or 'none'" + ))), + } +} + fn num_cpus() -> u32 { std::thread::available_parallelism() .map(|n| n.get() as u32) diff --git a/src/proxy_conn.rs b/src/proxy_conn.rs index 6a8ee72..61f87dd 100644 --- a/src/proxy_conn.rs +++ b/src/proxy_conn.rs @@ -1,6 +1,6 @@ use crate::metrics::{GlobalMetrics, ListenerMetrics}; use crate::protection::ProtectionState; -use crate::router::{Destination, LiveRouteTable, SourceAddressMode}; +use crate::router::{Destination, ForwardFingerprint, LiveRouteTable, SourceAddressMode}; use crate::sni; use crate::suspended::SuspendedRegistry; use crate::upstream::{self, UpstreamStream}; @@ -56,6 +56,9 @@ pub struct ConnContext { pub async fn handle(stream: TcpStream, peer_addr: SocketAddr, ctx: Arc) { let peer_ip = peer_addr.ip(); + // The address the client connected to — the PROXY v2 destination. Captured before the + // stream is consumed by the TLS handshake. + let local_addr = stream.local_addr().ok(); // ── 1. Peek: extract SNI + JA3 ─────────────────────────────────────────── let peek_info = sni::peek(&stream).await; @@ -148,6 +151,7 @@ pub async fn handle(stream: TcpStream, peer_addr: SocketAddr, ctx: Arc dest, _ => &effective_route.destination, }; - // Static routes reject XFF+h2 at config time, but suspended-route - // resolutions supply their own source mode — never splice a header - // into an h2 preface; drop the injection instead. - let source_mode = if negotiated_h2 && source_mode == SourceAddressMode::XForwardedFor { - tracing::debug!("skipping X-Forwarded-For injection on h2 connection from {peer_ip}"); - SourceAddressMode::None - } else { - source_mode - }; - proxy_via_tls(tls_stream, destination, peer_addr, source_mode, &ctx).await + // Header injection (XFF / X-JA3) never touches an h2 stream: forward() + // gates it on the negotiated protocol (l7_http1), covering static and + // suspended-route configs alike. + proxy_via_tls(tls_stream, destination, sf, &ctx).await } Ok(Err(e)) => { tracing::debug!("TLS handshake error from {peer_ip}: {e}"); @@ -207,7 +213,7 @@ pub async fn handle(stream: TcpStream, peer_addr: SocketAddr, ctx: Arc, dest: &Destination, - peer_addr: SocketAddr, - source_mode: SourceAddressMode, + sf: SourceForwarding<'_>, ctx: &ConnContext, ) -> std::io::Result<()> { - let mut upstream = upstream::connect(dest, Some(peer_addr.ip()), ctx.upstream_connect_timeout) + let mut upstream = upstream::connect(dest, Some(sf.peer_addr.ip()), ctx.upstream_connect_timeout) .await .map_err(|e| std::io::Error::new(std::io::ErrorKind::Other, e.to_string()))?; + // HTTP-header injection is only valid for a plaintext HTTP/1 upstream. An h2-negotiated + // upstream receives binary frames, so text header insertion would corrupt them. + let l7_http1 = client.get_ref().1.alpn_protocol() != Some(b"h2".as_ref()); + match &mut upstream { - UpstreamStream::Tcp(ref mut up) => { - apply_source_header(&mut client, up, peer_addr, source_mode).await?; - copy_with_idle_timeout(ctx.idle_timeout, &mut client, up).await - } + UpstreamStream::Tcp(ref mut up) => forward(&mut client, up, &sf, l7_http1, ctx.idle_timeout).await, UpstreamStream::Uds { ref mut stream, .. } => { - apply_source_header(&mut client, stream, peer_addr, source_mode).await?; - copy_with_idle_timeout(ctx.idle_timeout, &mut client, stream).await + forward(&mut client, stream, &sf, l7_http1, ctx.idle_timeout).await } } } @@ -243,63 +248,142 @@ async fn proxy_via_tls( async fn proxy_raw( mut client: TcpStream, dest: &Destination, - peer_addr: SocketAddr, - source_mode: SourceAddressMode, + sf: SourceForwarding<'_>, ctx: &ConnContext, ) -> std::io::Result<()> { - let mut upstream = upstream::connect(dest, Some(peer_addr.ip()), ctx.upstream_connect_timeout) + let mut upstream = upstream::connect(dest, Some(sf.peer_addr.ip()), ctx.upstream_connect_timeout) .await .map_err(|e| std::io::Error::new(std::io::ErrorKind::Other, e.to_string()))?; + // Passthrough forwards raw TLS bytes — never a plaintext HTTP/1 stream, so header injection + // is disabled (only PROXY protocol carriers apply here). match &mut upstream { - UpstreamStream::Tcp(ref mut up) => { - apply_source_header(&mut client, up, peer_addr, source_mode).await?; - copy_with_idle_timeout(ctx.idle_timeout, &mut client, up).await - } + UpstreamStream::Tcp(ref mut up) => forward(&mut client, up, &sf, false, ctx.idle_timeout).await, UpstreamStream::Uds { ref mut stream, .. } => { - apply_source_header(&mut client, stream, peer_addr, source_mode).await?; - copy_with_idle_timeout(ctx.idle_timeout, &mut client, stream).await + forward(&mut client, stream, &sf, false, ctx.idle_timeout).await } } } -/// Bidirectional copy with an optional idle timeout. -/// A zero `idle` means no timeout. -async fn copy_with_idle_timeout(idle: Duration, a: &mut A, b: &mut B) -> std::io::Result<()> +/// Write the configured source-address prefix (PROXY v1/v2 header) to the upstream, then copy +/// bidirectionally under the idle timeout. On an HTTP/1 injection route the client→upstream half +/// is the per-request header rewriter (finding fixes: the header read now lives inside the idle +/// timeout, and every request — fragmented, pipelined, or keep-alive — is stripped and rewritten, +/// not just the first read). +async fn forward( + client: &mut C, + upstream: &mut U, + sf: &SourceForwarding<'_>, + l7_http1: bool, + idle: Duration, +) -> std::io::Result<()> where - A: tokio::io::AsyncRead + tokio::io::AsyncWrite + Unpin, - B: tokio::io::AsyncRead + tokio::io::AsyncWrite + Unpin, + C: tokio::io::AsyncRead + tokio::io::AsyncWrite + Unpin, + U: tokio::io::AsyncRead + tokio::io::AsyncWrite + Unpin, { + let body = async { + write_connection_prefix(upstream, sf).await?; + let rewrites = header_rewrites(sf, l7_http1); + if rewrites.is_empty() { + copy_bidirectional(client, upstream).await.map(|_| ()) + } else { + crate::http_proxy::proxy_http1_rewriting(client, upstream, &rewrites).await + } + }; if idle.is_zero() { - copy_bidirectional(a, b).await.map(|_| ()) + body.await } else { - timeout(idle, copy_bidirectional(a, b)) + timeout(idle, body) .await .map_err(|_| std::io::Error::new(std::io::ErrorKind::TimedOut, "idle timeout"))? - .map(|_| ()) } } -/// Apply the configured source address forwarding before bidirectional copy. -async fn apply_source_header( - client: &mut C, - upstream: &mut U, - peer_addr: SocketAddr, +/// Per-connection source-address + fingerprint forwarding parameters. All fields are `Copy` +/// (`ja3`/`ja4` borrow the connection's `PeekInfo`), so the struct is passed by value. +#[derive(Clone, Copy)] +struct SourceForwarding<'a> { mode: SourceAddressMode, -) -> std::io::Result<()> + fingerprint: ForwardFingerprint, + ja3: &'a str, + ja4: &'a str, + peer_addr: SocketAddr, + local_addr: Option, +} + +impl SourceForwarding<'_> { + /// The fingerprint string selected for forwarding ("" when none, or unparsed). + fn fingerprint_value(&self) -> &str { + match self.fingerprint { + ForwardFingerprint::None => "", + ForwardFingerprint::Ja3 => self.ja3, + ForwardFingerprint::Ja4 => self.ja4, + } + } + + /// The HTTP header name symphony owns for the configured fingerprint mode, if any. + fn fingerprint_header_name(&self) -> Option<&'static str> { + match self.fingerprint { + ForwardFingerprint::Ja3 => Some("X-JA3"), + ForwardFingerprint::Ja4 => Some("X-JA4"), + ForwardFingerprint::None => None, + } + } +} + +/// Write the one-shot connection prefix (PROXY v1/v2 header) the mode calls for. HTTP-header +/// carriers have no prefix — they rewrite the request stream instead (see `header_rewrites`). +async fn write_connection_prefix(upstream: &mut U, sf: &SourceForwarding<'_>) -> std::io::Result<()> where - C: tokio::io::AsyncRead + tokio::io::AsyncWrite + Unpin, - U: tokio::io::AsyncRead + tokio::io::AsyncWrite + Unpin, + U: tokio::io::AsyncWrite + Unpin, { - match mode { - SourceAddressMode::None => Ok(()), - SourceAddressMode::ProxyProtocol => { - upstream::write_proxy_v1_header(upstream, peer_addr).await + match sf.mode { + SourceAddressMode::None | SourceAddressMode::XForwardedFor => Ok(()), + SourceAddressMode::ProxyProtocol => upstream::write_proxy_v1_header(upstream, sf.peer_addr).await, + SourceAddressMode::ProxyProtocolV2 => { + let value = sf.fingerprint_value(); + let mut tlvs: Vec<(u8, &str)> = Vec::new(); + match sf.fingerprint { + ForwardFingerprint::Ja3 => tlvs.push((upstream::PP2_TYPE_JA3, value)), + ForwardFingerprint::Ja4 => tlvs.push((upstream::PP2_TYPE_JA4, value)), + ForwardFingerprint::None => {} + } + upstream::write_proxy_v2_header(upstream, sf.peer_addr, sf.local_addr, &tlvs).await } - SourceAddressMode::XForwardedFor => { - upstream::inject_x_forwarded_for(client, upstream, peer_addr).await + } +} + +/// The set of headers symphony owns end-to-end on the client→upstream request stream, applied to +/// every HTTP/1 request. Empty (→ a plain copy, no rewriting) unless the upstream is a plaintext +/// HTTP/1 stream and the mode injects HTTP headers. +/// +/// A configured header is *always* stripped from the client's request, even when symphony has no +/// authoritative value to substitute (`value: None`) — a client must never smuggle its own +/// `X-JA3`/`X-JA4`/`X-Forwarded-For` through precisely when we can't replace it. PROXY v2 carries +/// the fingerprint in a TLV, so it adds no header rewrite. +fn header_rewrites(sf: &SourceForwarding<'_>, l7_http1: bool) -> Vec { + use crate::http_proxy::HeaderRewrite; + if !l7_http1 { + return Vec::new(); + } + let mut rewrites: Vec = Vec::new(); + if matches!(sf.mode, SourceAddressMode::XForwardedFor) { + rewrites.push(HeaderRewrite { + name: "X-Forwarded-For", + value: Some(sf.peer_addr.ip().to_string()), + }); + } + // The fingerprint header rides every HTTP-header mode (None/v1/XFF); v2 uses a TLV instead. + if !matches!(sf.mode, SourceAddressMode::ProxyProtocolV2) { + if let Some(name) = sf.fingerprint_header_name() { + let value = sf.fingerprint_value(); + rewrites.push(HeaderRewrite { + name, + value: (!value.is_empty()).then(|| value.to_string()), + }); } } + rewrites } // ── Helpers ─────────────────────────────────────────────────────────────────── @@ -311,6 +395,7 @@ struct EffectiveRoute { tls_config: Option>, terminate_tls: bool, source_address_mode: SourceAddressMode, + forward_fingerprint: ForwardFingerprint, } struct ActiveGuard { diff --git a/src/router.rs b/src/router.rs index d13eadb..0a465f7 100644 --- a/src/router.rs +++ b/src/router.rs @@ -97,12 +97,28 @@ impl RouteTokenBucket { pub enum SourceAddressMode { /// No source address forwarding. None, - /// Send a PROXY protocol v1 header before any application data. + /// Send a PROXY protocol v1 (text) header before any application data. ProxyProtocol, + /// Send a PROXY protocol v2 (binary) header before any application data. The v2 + /// framing carries a TLV section, the carrier used for `ForwardFingerprint`. + ProxyProtocolV2, /// Parse the beginning of the HTTP request and insert an X-Forwarded-For header. XForwardedFor, } +/// Which client TLS fingerprint (if any) symphony forwards to the upstream so the backend +/// can act on it itself. Carrier depends on `SourceAddressMode`: a PROXY v2 TLV under +/// `ProxyProtocolV2`, otherwise an injected `X-JA3`/`X-JA4` HTTP header. +#[derive(Clone, Copy, Debug, PartialEq)] +pub enum ForwardFingerprint { + /// Do not forward a fingerprint. + None, + /// Forward the JA3 fingerprint. + Ja3, + /// Forward the JA4 fingerprint. + Ja4, +} + // ── Route destination ───────────────────────────────────────────────────────── #[derive(Clone)] @@ -128,6 +144,8 @@ pub struct Route { pub rate_limiter: Option>, /// How the real client IP is forwarded to the upstream. pub source_address_mode: SourceAddressMode, + /// Which client TLS fingerprint (if any) is forwarded to the upstream. + pub forward_fingerprint: ForwardFingerprint, } // ── Route table ─────────────────────────────────────────────────────────────── @@ -225,6 +243,8 @@ pub struct RouteSpec { pub burst: Option, /// How the real client IP is forwarded to the upstream. pub source_address_mode: SourceAddressMode, + /// Which client TLS fingerprint (if any) is forwarded to the upstream. + pub forward_fingerprint: ForwardFingerprint, /// Advertise h2 in ALPN so clients can negotiate HTTP/2. pub http2: bool, } @@ -426,6 +446,7 @@ fn build_route( suspend_timeout: Duration::from_millis(spec.suspend_timeout_ms.max(1)), rate_limiter, source_address_mode: spec.source_address_mode, + forward_fingerprint: spec.forward_fingerprint, }) } @@ -622,6 +643,7 @@ UlqL1DcgX6Szi9w/p7B4BZO9iA== max_cps: None, burst: None, source_address_mode: SourceAddressMode::None, + forward_fingerprint: ForwardFingerprint::None, http2: false, } } diff --git a/src/suspended.rs b/src/suspended.rs index 2bcbf86..9ec2de3 100644 --- a/src/suspended.rs +++ b/src/suspended.rs @@ -1,5 +1,5 @@ use crate::balancer::{UdsBalancer, UdsSlotSpec}; -use crate::router::{Destination, SourceAddressMode}; +use crate::router::{Destination, ForwardFingerprint, SourceAddressMode}; use dashmap::DashMap; use rustls::ServerConfig; use std::sync::atomic::{AtomicU64, Ordering}; @@ -12,6 +12,7 @@ pub struct ResolvedRoute { pub tls_config: Option>, pub terminate_tls: bool, pub source_address_mode: SourceAddressMode, + pub forward_fingerprint: ForwardFingerprint, } /// Registry of suspended connections waiting for `resolveConnection()`. @@ -69,6 +70,7 @@ pub struct ResolveSpec { pub mtls_ca_pem: Option>, pub require_client_cert: bool, pub source_address_mode: SourceAddressMode, + pub forward_fingerprint: ForwardFingerprint, pub http2: bool, } @@ -123,5 +125,6 @@ pub fn build_resolved_route(spec: &ResolveSpec) -> crate::error::Result( stream.write_all(header.as_bytes()).await } -/// Read the first chunk of HTTP data from `client`, insert an -/// `X-Forwarded-For` header after the request line, write the modified -/// data to `upstream`, then return so the caller can proceed with -/// bidirectional copy for the remaining data. +/// The 12-byte PROXY protocol v2 signature (`\r\n\r\n\0\r\nQUIT\n`). +const PROXY_V2_SIGNATURE: [u8; 12] = + [0x0D, 0x0A, 0x0D, 0x0A, 0x00, 0x0D, 0x0A, 0x51, 0x55, 0x49, 0x54, 0x0A]; + +/// PP2 TLV type carrying the JA3 fingerprint. HAProxy reserves the 0xE0–0xEF range for +/// private/experimental TLVs (`PP2_TYPE_MIN_CUSTOM`); there is no registered type for JA3/JA4. +pub const PP2_TYPE_JA3: u8 = 0xE0; +/// PP2 TLV type carrying the JA4 fingerprint. +pub const PP2_TYPE_JA4: u8 = 0xE1; + +/// Normalized address bytes for a PROXY v2 address block, IPv4-mapped IPv6 unwrapped to v4 +/// so the emitted family matches the v1 path's behaviour. +enum V2Addr { + V4([u8; 4]), + V6([u8; 16]), +} + +fn normalize_v2_addr(ip: IpAddr) -> V2Addr { + match ip { + IpAddr::V4(v4) => V2Addr::V4(v4.octets()), + IpAddr::V6(v6) => match v6.to_ipv4_mapped() { + Some(v4) => V2Addr::V4(v4.octets()), + None => V2Addr::V6(v6.octets()), + }, + } +} + +/// Write a PROXY protocol v2 (binary) header so the backend can recover the real client +/// address, plus any `tlvs` (type, value) appended to the TLV section — used to carry the +/// client TLS fingerprint downstream. TLVs with an empty value are skipped. /// -/// If the initial read contains no `\r\n` (not a valid HTTP request), -/// the header is prepended before the data as a best-effort fallback. -pub async fn inject_x_forwarded_for( - client: &mut C, - upstream: &mut U, +/// `local_addr` is the address the client connected to (the destination). When absent, or of a +/// different family than the source, a family-matched placeholder is used; the source address — +/// the field backends actually consume — is always accurate. +pub async fn write_proxy_v2_header( + stream: &mut W, peer_addr: SocketAddr, -) -> std::io::Result<()> -where - C: tokio::io::AsyncRead + Unpin, - U: tokio::io::AsyncWrite + Unpin, -{ - let mut buf = vec![0u8; 8192]; - let n = client.read(&mut buf).await?; - if n == 0 { - return Ok(()); - } - let data = &buf[..n]; + local_addr: Option, + tlvs: &[(u8, &str)], +) -> std::io::Result<()> { + let src = normalize_v2_addr(peer_addr.ip()); + let dst = local_addr.map(|a| normalize_v2_addr(a.ip())); + let dst_port = local_addr.map(|a| a.port()).unwrap_or(0); - let xff = format!("X-Forwarded-For: {}\r\n", peer_addr.ip()); + let mut out = Vec::with_capacity(28); + out.extend_from_slice(&PROXY_V2_SIGNATURE); + out.push(0x21); // version 2 (high nibble) | PROXY command (low nibble) + + // Address block. The family/proto byte is (family << 4) | transport; transport is + // STREAM (0x1). Source family wins; a differing/absent dst falls back to loopback. + let addr_block: Vec = match src { + V2Addr::V4(src_ip) => { + out.push(0x11); // AF_INET | STREAM + let dst_ip = match dst { + Some(V2Addr::V4(ip)) => ip, + _ => [127, 0, 0, 1], + }; + let mut b = Vec::with_capacity(12); + b.extend_from_slice(&src_ip); + b.extend_from_slice(&dst_ip); + b.extend_from_slice(&peer_addr.port().to_be_bytes()); + b.extend_from_slice(&dst_port.to_be_bytes()); + b + } + V2Addr::V6(src_ip) => { + out.push(0x21); // AF_INET6 | STREAM + let dst_ip = match dst { + Some(V2Addr::V6(ip)) => ip, + _ => std::net::Ipv6Addr::LOCALHOST.octets(), + }; + let mut b = Vec::with_capacity(36); + b.extend_from_slice(&src_ip); + b.extend_from_slice(&dst_ip); + b.extend_from_slice(&peer_addr.port().to_be_bytes()); + b.extend_from_slice(&dst_port.to_be_bytes()); + b + } + }; - // Find the end of the HTTP request line (first \r\n) - let insert_pos = data - .windows(2) - .position(|w| w == b"\r\n") - .map(|p| p + 2) // insert after the \r\n - .unwrap_or(0); // no \r\n found — prepend + // TLV section, appended after the address block. + let mut tlv_block = Vec::new(); + for (ty, value) in tlvs { + if value.is_empty() { + continue; + } + let bytes = value.as_bytes(); + // TLV and header length fields are u16; refuse to silently truncate an oversized value. + let tlv_len: u16 = bytes.len().try_into().map_err(|_| { + std::io::Error::new(std::io::ErrorKind::InvalidInput, "PROXY v2 TLV value exceeds 65535 bytes") + })?; + tlv_block.push(*ty); + tlv_block.extend_from_slice(&tlv_len.to_be_bytes()); + tlv_block.extend_from_slice(bytes); + } - upstream.write_all(&data[..insert_pos]).await?; - upstream.write_all(xff.as_bytes()).await?; - upstream.write_all(&data[insert_pos..]).await?; + // 16-bit length of everything after the 16-byte fixed header (address block + TLVs). + let len: u16 = (addr_block.len() + tlv_block.len()).try_into().map_err(|_| { + std::io::Error::new(std::io::ErrorKind::InvalidInput, "PROXY v2 header exceeds 65535 bytes") + })?; + out.extend_from_slice(&len.to_be_bytes()); + out.extend_from_slice(&addr_block); + out.extend_from_slice(&tlv_block); - Ok(()) + stream.write_all(&out).await } // ── tokio::io::AsyncRead + AsyncWrite impls via delegation ──────────────────── @@ -135,3 +202,78 @@ where // // This avoids the overhead of dynamic dispatch on every read/write call. // proxy_conn.rs handles the match and calls the appropriate copy_bidirectional. + +#[cfg(test)] +mod tests { + use super::*; + + // A JA3-shaped value (32 hex chars) for TLV assertions. + const JA3: &str = "0123456789abcdef0123456789abcdef"; + + #[tokio::test] + async fn v2_header_ipv4_with_ja3_tlv() { + let peer: SocketAddr = "192.168.1.5:51000".parse().unwrap(); + let local: SocketAddr = "10.0.0.1:443".parse().unwrap(); + let mut out: Vec = Vec::new(); + write_proxy_v2_header(&mut out, peer, Some(local), &[(PP2_TYPE_JA3, JA3)]) + .await + .unwrap(); + + // Fixed 16-byte header. + assert_eq!(&out[0..12], &PROXY_V2_SIGNATURE); + assert_eq!(out[12], 0x21, "version 2 | PROXY command"); + assert_eq!(out[13], 0x11, "AF_INET | STREAM"); + let declared_len = u16::from_be_bytes([out[14], out[15]]) as usize; + assert_eq!(out.len(), 16 + declared_len, "declared length covers the rest"); + + // IPv4 address block: src(4) dst(4) sport(2) dport(2). + assert_eq!(&out[16..20], &[192, 168, 1, 5]); + assert_eq!(&out[20..24], &[10, 0, 0, 1]); + assert_eq!(u16::from_be_bytes([out[24], out[25]]), 51000); + assert_eq!(u16::from_be_bytes([out[26], out[27]]), 443); + + // TLV: type(1) len(2) value. + assert_eq!(out[28], PP2_TYPE_JA3); + assert_eq!(u16::from_be_bytes([out[29], out[30]]) as usize, JA3.len()); + assert_eq!(&out[31..31 + JA3.len()], JA3.as_bytes()); + assert_eq!(out.len(), 31 + JA3.len(), "no trailing bytes"); + } + + #[tokio::test] + async fn v2_header_skips_empty_tlv_and_defaults_dst() { + let peer: SocketAddr = "203.0.113.9:1234".parse().unwrap(); + let mut out: Vec = Vec::new(); + // Empty fingerprint value (unparsed ClientHello) and no local_addr. + write_proxy_v2_header(&mut out, peer, None, &[(PP2_TYPE_JA3, "")]) + .await + .unwrap(); + + let declared_len = u16::from_be_bytes([out[14], out[15]]) as usize; + assert_eq!(declared_len, 12, "address block only, empty TLV skipped"); + assert_eq!(out.len(), 28); + assert_eq!(&out[20..24], &[127, 0, 0, 1], "dst defaults to loopback"); + assert_eq!(u16::from_be_bytes([out[26], out[27]]), 0, "dst port defaults to 0"); + } + + #[tokio::test] + async fn v2_header_unwraps_ipv4_mapped_ipv6() { + // ::ffff:1.2.3.4 must emit as AF_INET, matching the v1 path. + let peer: SocketAddr = "[::ffff:1.2.3.4]:9000".parse().unwrap(); + let mut out: Vec = Vec::new(); + write_proxy_v2_header(&mut out, peer, None, &[]).await.unwrap(); + assert_eq!(out[13], 0x11, "IPv4-mapped IPv6 collapses to AF_INET"); + assert_eq!(&out[16..20], &[1, 2, 3, 4]); + } + + #[tokio::test] + async fn v2_header_ipv6() { + let peer: SocketAddr = "[2001:db8::1]:8443".parse().unwrap(); + let mut out: Vec = Vec::new(); + write_proxy_v2_header(&mut out, peer, None, &[]).await.unwrap(); + assert_eq!(out[13], 0x21, "AF_INET6 | STREAM"); + let declared_len = u16::from_be_bytes([out[14], out[15]]) as usize; + assert_eq!(declared_len, 36, "IPv6 address block is 36 bytes"); + assert_eq!(&out[16..32], &std::net::Ipv6Addr::new(0x2001, 0xdb8, 0, 0, 0, 0, 0, 1).octets()); + } + +} diff --git a/ts/addon.d.ts b/ts/addon.d.ts index 4cb0032..5a90636 100644 --- a/ts/addon.d.ts +++ b/ts/addon.d.ts @@ -45,9 +45,16 @@ export interface JsRouteConfig { burst?: number /** * How the real client IP is forwarded to the upstream. - * "proxyProtocol" (default for UDS), "xForwardedFor", or "none" (default for TCP). + * "proxyProtocol" (v1, default for UDS), "proxyProtocolV2", "xForwardedFor", or + * "none" (default for TCP). */ sourceAddressHeader?: string + /** + * Which client TLS fingerprint to forward downstream: "ja3", "ja4", or "none" + * (default). Carried as a PROXY v2 TLV under "proxyProtocolV2", otherwise as an + * injected X-JA3/X-JA4 HTTP header. + */ + forwardFingerprint?: string /** Advertise h2 in ALPN so clients can negotiate HTTP/2. Default: false. */ http2?: boolean } @@ -105,6 +112,7 @@ export interface JsResolveRoute { cert?: JsCertConfig mtls?: JsMtlsConfig sourceAddressHeader?: string + forwardFingerprint?: string http2?: boolean } export declare class SymphonyProxyWrap { diff --git a/ts/proxy.ts b/ts/proxy.ts index 1362fb4..fbe37d2 100644 --- a/ts/proxy.ts +++ b/ts/proxy.ts @@ -102,6 +102,7 @@ function toJsRoute(r: RouteConfig): JsRouteConfig { maxConnectionsPerSecond: r.maxConnectionsPerSecond, burst: r.burst, sourceAddressHeader: r.sourceAddressHeader, + forwardFingerprint: r.forwardFingerprint, http2: r.http2, }; } @@ -256,6 +257,8 @@ export class SymphonyProxy extends EventEmitter { cert: route.cert ? toJsCert(route.cert) : undefined, mtls: route.mtls ? toJsMtls(route.mtls) : undefined, sourceAddressHeader: route.sourceAddressHeader, + forwardFingerprint: route.forwardFingerprint, + http2: route.http2, }); } } diff --git a/ts/types.ts b/ts/types.ts index 8ce0c37..9fe74df 100644 --- a/ts/types.ts +++ b/ts/types.ts @@ -89,14 +89,32 @@ export interface RouteConfig { /** * How the real client IP is forwarded to the upstream. * - * - `'proxyProtocol'` — Send a PROXY protocol v1 header before application data. + * - `'proxyProtocol'` — Send a PROXY protocol v1 (text) header before application data. * Default for UDS upstreams. + * - `'proxyProtocolV2'` — Send a PROXY protocol v2 (binary) header before application + * data. v2 carries a TLV section, which is the carrier for `forwardFingerprint`. + * Keep opt-in: consumers must speak v2 (HAProxy/nginx do; Harper core's UDS reader + * currently parses v1 only). * - `'xForwardedFor'` — Parse the beginning of the HTTP request and insert an * `X-Forwarded-For` header. Use this for backends (e.g. Bun) that do not * support the PROXY protocol. * - `'none'` — Do not forward source address information. Default for TCP upstreams. */ - sourceAddressHeader?: 'proxyProtocol' | 'xForwardedFor' | 'none'; + sourceAddressHeader?: 'proxyProtocol' | 'proxyProtocolV2' | 'xForwardedFor' | 'none'; + /** + * Forward the client's TLS fingerprint (computed from the ClientHello) downstream so the + * upstream can make its own bot/abuse decisions on it. + * + * - `'ja3'` / `'ja4'` — forward that fingerprint. + * - `'none'` (default) — do not forward. + * + * Carrier: a PROXY v2 TLV when `sourceAddressHeader` is `'proxyProtocolV2'` (works in + * passthrough too, since it prefixes the raw TLS bytes); otherwise an injected + * `X-JA3` / `X-JA4` HTTP header, which requires a plaintext HTTP/1 upstream + * (`terminateTls: true` and not `http2`) — it is skipped otherwise. Any client-supplied + * `X-JA3` / `X-JA4` is stripped so the injected value is authoritative. + */ + forwardFingerprint?: 'ja3' | 'ja4' | 'none'; /** * Advertise HTTP/2 (`h2`) in the TLS ALPN extension so clients can negotiate * HTTP/2. When true, symphony declares `['h2', 'http/1.1']` in ALPN and the @@ -221,7 +239,9 @@ export interface ResolveRoute { cert?: CertConfig; mtls?: MtlsConfig; /** How the real client IP is forwarded to the upstream. See RouteConfig.sourceAddressHeader. */ - sourceAddressHeader?: 'proxyProtocol' | 'xForwardedFor' | 'none'; + sourceAddressHeader?: 'proxyProtocol' | 'proxyProtocolV2' | 'xForwardedFor' | 'none'; + /** Which client TLS fingerprint to forward downstream. See RouteConfig.forwardFingerprint. */ + forwardFingerprint?: 'ja3' | 'ja4' | 'none'; /** Advertise h2 in ALPN for this resolved connection. See RouteConfig.http2. */ http2?: boolean; }