From d3e6f3270f549b69263dc5ea72d0cf4cb9080cab Mon Sep 17 00:00:00 2001 From: Neil <4138956+nwparker@users.noreply.github.com> Date: Sat, 14 Nov 2026 18:14:55 -0800 Subject: [PATCH] relay-edge: ingress mode and read-only credential verify (dark), on main's edge (#42566) Ports the edge halves of two release-branch changes onto main, which production edge deploys come from (prod edge = ba3f991): - Ingress mode (#41986): signed ingress header (INGRESS_SECRETS) and token mode for a cloud load balancer (INGRESS_TOKENS secret, INGRESS_ORIGINS, INGRESS_SOURCE_CIDRS). All or nothing: a partial or malformed setup turns the mode off and logs relay_edge_ingress_config_invalid; it never throws. Requests from the configured egress ranges that fail the trust rule get 403. Admin calls are matched first. Adds cloud/apps/relay-ingress (stateless proxy) and the relay-contract ingress-header module. - POST /__admin/verify (#42538, edge half): the director's check for a phone resume token or invite that exists only on the edge. Hash only, read-only: an object that never held the host (or no longer holds it here) answers not_here before anything is written; otherwise the token's jti is its only write. Replay and jti_full are 409. Adds hash-input variants to RelayCredentialStore (shared with the edge; existing methods delegate, behaviour unchanged). Kept from main: drain, single-use admin tokens (jti), signing-key rotation, JWKS-outage tolerance, keepalive auto-response, health probe. The place label keeps main's '/' and, only behind a verified ingress, appends '//'. No user effect until the ingress vars/secret are set or the director calls verify. --- cloud/apps/relay-edge/package.json | 6 +- .../scripts/ingress-source-cidrs.ts | 46 +++ .../relay-edge/spike/zone-source-ip-audit.ts | 84 +++++ cloud/apps/relay-edge/src/admin.ts | 18 +- cloud/apps/relay-edge/src/env.ts | 61 +++- cloud/apps/relay-edge/src/host-relay.ts | 14 +- cloud/apps/relay-edge/src/index.ts | 53 ++- cloud/apps/relay-edge/src/ingress.ts | 100 ++++++ cloud/apps/relay-edge/src/metrics.ts | 18 +- cloud/apps/relay-edge/src/preauth.ts | 20 +- cloud/apps/relay-edge/src/verify.ts | 24 ++ .../relay-edge/test/front-door-answer.test.ts | 24 ++ .../apps/relay-edge/test/front-door-answer.ts | 18 ++ cloud/apps/relay-edge/test/g1-front-door.ts | 306 ++++++++++++++++++ cloud/apps/relay-edge/test/g1-local.ts | 131 ++++++++ cloud/apps/relay-edge/test/ingress-e2e.ts | 287 ++++++++++++++++ cloud/apps/relay-edge/test/ingress.test.ts | 265 +++++++++++++++ cloud/apps/relay-edge/test/relay-clients.ts | 167 ++++++++++ cloud/apps/relay-edge/test/verify.test.ts | 139 ++++++++ cloud/apps/relay-ingress/Dockerfile | 33 ++ cloud/apps/relay-ingress/package.json | 27 ++ cloud/apps/relay-ingress/src/index.ts | 61 ++++ cloud/apps/relay-ingress/src/proxy.test.ts | 148 +++++++++ cloud/apps/relay-ingress/src/proxy.ts | 167 ++++++++++ cloud/apps/relay-ingress/tsconfig.build.json | 10 + cloud/apps/relay-ingress/tsconfig.json | 5 + .../relay/src/credentials/credential-store.ts | 32 +- cloud/packages/relay-contract/package.json | 4 + cloud/packages/relay-contract/src/index.ts | 1 + .../relay-contract/src/ingress-header.ts | 71 ++++ cloud/pnpm-lock.yaml | 25 ++ .../check-test-file-runners.mjs | 1 + 32 files changed, 2333 insertions(+), 33 deletions(-) create mode 100644 cloud/apps/relay-edge/scripts/ingress-source-cidrs.ts create mode 100644 cloud/apps/relay-edge/spike/zone-source-ip-audit.ts create mode 100644 cloud/apps/relay-edge/src/ingress.ts create mode 100644 cloud/apps/relay-edge/src/verify.ts create mode 100644 cloud/apps/relay-edge/test/front-door-answer.test.ts create mode 100644 cloud/apps/relay-edge/test/front-door-answer.ts create mode 100644 cloud/apps/relay-edge/test/g1-front-door.ts create mode 100644 cloud/apps/relay-edge/test/g1-local.ts create mode 100644 cloud/apps/relay-edge/test/ingress-e2e.ts create mode 100644 cloud/apps/relay-edge/test/ingress.test.ts create mode 100644 cloud/apps/relay-edge/test/relay-clients.ts create mode 100644 cloud/apps/relay-edge/test/verify.test.ts create mode 100644 cloud/apps/relay-ingress/Dockerfile create mode 100644 cloud/apps/relay-ingress/package.json create mode 100644 cloud/apps/relay-ingress/src/index.ts create mode 100644 cloud/apps/relay-ingress/src/proxy.test.ts create mode 100644 cloud/apps/relay-ingress/src/proxy.ts create mode 100644 cloud/apps/relay-ingress/tsconfig.build.json create mode 100644 cloud/apps/relay-ingress/tsconfig.json create mode 100644 cloud/packages/relay-contract/src/ingress-header.ts diff --git a/cloud/apps/relay-edge/package.json b/cloud/apps/relay-edge/package.json index 946df722bad..76433414f56 100644 --- a/cloud/apps/relay-edge/package.json +++ b/cloud/apps/relay-edge/package.json @@ -9,7 +9,11 @@ "spike": "tsx spike/run-all.ts", "test": "vitest run", "test:schema": "tsx test/apply-schema.ts", - "test:mvp": "tsx test/mvp.ts" + "test:mvp": "tsx test/mvp.ts", + "ingress-cidrs": "tsx scripts/ingress-source-cidrs.ts", + "test:g1": "tsx test/g1-front-door.ts", + "test:g1-local": "tsx test/g1-local.ts", + "test:ingress-e2e": "tsx test/ingress-e2e.ts" }, "dependencies": { "@types/pg": "^8.23.1", diff --git a/cloud/apps/relay-edge/scripts/ingress-source-cidrs.ts b/cloud/apps/relay-edge/scripts/ingress-source-cidrs.ts new file mode 100644 index 00000000000..952377b29a0 --- /dev/null +++ b/cloud/apps/relay-edge/scripts/ingress-source-cidrs.ts @@ -0,0 +1,46 @@ +// The egress ranges a global load balancer uses to reach an internet NEG endpoint, for the edge's +// INGRESS_SOURCE_CIDRS. Published as an SPF-style TXT record; read here, never at request time. +// npx tsx scripts/ingress-source-cidrs.ts # print the comma-separated list +// npx tsx scripts/ingress-source-cidrs.ts --check "" # exit 1 on drift (scheduled check) +import { Resolver } from 'node:dns/promises' +import { parseCidr } from '../src/ingress' + +const RECORD = '_cloud-eoips.googleusercontent.com' + +async function ranges(name: string, resolver: Resolver, depth = 0): Promise { + if (depth > 5) { throw new Error(`include depth exceeded at ${name}`) } + const txt = (await resolver.resolveTxt(name)).map((chunks) => chunks.join('')).filter((t) => t.startsWith('v=spf1 ')) + if (txt.length !== 1) { throw new Error(`expected one spf record at ${name}, got ${txt.length}`) } + const out: string[] = [] + for (const term of txt[0]!.split(/\s+/).slice(1)) { + if (term.startsWith('ip4:') || term.startsWith('ip6:')) { out.push(term.slice(4)) } + else if (term.startsWith('include:')) { out.push(...(await ranges(term.slice(8), resolver, depth + 1))) } + } + return out +} + +const normalize = (list: string[]) => [...new Set(list.map((c) => c.trim().toLowerCase()).filter(Boolean))].sort() + +async function main(): Promise { + const resolver = new Resolver() + resolver.setServers(['8.8.8.8', '8.8.4.4']) + const published = normalize(await ranges(RECORD, resolver)) + if (published.length === 0 || published.some((c) => !parseCidr(c))) { throw new Error(`unusable ranges: ${published.join(',')}`) } + const checkAt = process.argv.indexOf('--check') + if (checkAt === -1) { + console.log(published.join(',')) + return + } + const deployed = normalize((process.argv[checkAt + 1] ?? '').split(',')) + const added = published.filter((c) => !deployed.includes(c)) + const removed = deployed.filter((c) => !published.includes(c)) + if (added.length || removed.length) { + console.error(JSON.stringify({ event: 'ingress_source_cidrs_drift', added, removed })) + process.exitCode = 1 + } else { console.log(`ok: ${published.length} ranges match`) } +} + +main().catch((error) => { + console.error(error instanceof Error ? error.message : error) + process.exitCode = 1 +}) diff --git a/cloud/apps/relay-edge/spike/zone-source-ip-audit.ts b/cloud/apps/relay-edge/spike/zone-source-ip-audit.ts new file mode 100644 index 00000000000..cdfcd6992bb --- /dev/null +++ b/cloud/apps/relay-edge/spike/zone-source-ip-audit.ts @@ -0,0 +1,84 @@ +// Read-only: the zone features that could act on traffic concentrated on a few source addresses +// (a load-balancer front door reaches the edge from a handful of egress IPs). GET requests only, +// with the logged-in wrangler session; prints settings and rule shapes, never ids, tokens, rule +// expressions' literal values or IP lists. +// npx tsx spike/zone-source-ip-audit.ts +// The wrangler login's `zone (read)` scope lists zones only; settings, bot management and rulesets +// need a token with Zone Settings Read, Bot Management Read and Zone WAF Read (CLOUDFLARE_API_TOKEN). +import { execFileSync } from 'node:child_process' + +const hostname = process.argv[2] +if (!hostname) { throw new Error('usage: zone-source-ip-audit.ts ') } +const token = process.env.CLOUDFLARE_API_TOKEN ?? execFileSync('npx', ['wrangler', 'auth', 'token'], { encoding: 'utf8', stdio: ['ignore', 'pipe', 'ignore'] }) + .split('\n') + .map((l) => l.trim()) + .find((l) => /^[A-Za-z0-9._-]{30,}$/.test(l)) +if (!token) { throw new Error('no wrangler token (run `npx wrangler login`)') } + +type Api = { success: boolean; result: T; errors?: { code: number; message: string }[] } +async function get(path: string): Promise<{ ok: true; result: T } | { ok: false; why: string }> { + const r = await fetch(`https://api.cloudflare.com/client/v4${path}`, { headers: { authorization: `Bearer ${token}` } }) + const body = (await r.json().catch(() => null)) as Api | null + if (r.ok && body?.success) { return { ok: true, result: body.result } } + return { ok: false, why: `${r.status} ${body?.errors?.map((e) => `${e.code}`).join(',') ?? ''}`.trim() } +} + +const show = (label: string, value: unknown) => console.log(`${label.padEnd(44)} ${typeof value === 'string' ? value : JSON.stringify(value)}`) +// Shape of a rule expression without its literals (IPs, hostnames, paths). +const fields = (expression = '') => [...new Set(expression.match(/\b(ip\.src[\w.]*|cf\.[\w.]+|http\.[\w.]+|ip\.geoip\.[\w.]+)/g) ?? [])].join(' ') + +async function main(): Promise { + const zones = await get<{ id: string; name: string; plan?: { name?: string } }[]>('/zones?per_page=50') + if (!zones.ok) { throw new Error(`zones: ${zones.why}`) } + const zone = zones.result.filter((z) => hostname === z.name || hostname.endsWith(`.${z.name}`)).sort((a, b) => b.name.length - a.name.length)[0] + if (!zone) { throw new Error('no zone visible to this session contains that hostname') } + show('zone plan', zone.plan?.name ?? '?') + const z = `/zones/${zone.id}` + + console.log('\n# Zone settings that act per source or per client') + const settings = await get<{ id: string; value: unknown }[]>(`${z}/settings`) + if (settings.ok) { + const wanted = ['security_level', 'browser_check', 'challenge_ttl', 'waf', 'websockets', 'ip_geolocation', 'privacy_pass', 'replace_insecure_js', 'opportunistic_onion', 'max_upload', 'http2', 'http3', 'min_tls_version'] + for (const s of settings.result.filter((s) => wanted.includes(s.id))) { show(s.id, s.value) } + } else { show('settings', `unreadable (${settings.why})`) } + + console.log('\n# Bot features') + const bots = await get>(`${z}/bot_management`) + if (bots.ok) { + for (const [k, v] of Object.entries(bots.result)) { + if (!/id$/i.test(k)) { show(`bot_management.${k}`, v) } + } + } else { show('bot_management', `unreadable (${bots.why})`) } + + console.log('\n# Rules (phase entrypoints): action, enabled, fields used; no literals') + for (const phase of ['http_request_firewall_custom', 'http_ratelimit', 'http_request_firewall_managed', 'http_request_sbfm', 'ddos_l7', 'http_request_dynamic_redirect', 'http_config_settings']) { + const ep = await get<{ rules?: { action?: string; enabled?: boolean; description?: string; expression?: string; ratelimit?: Record; action_parameters?: Record }[] }>(`${z}/rulesets/phases/${phase}/entrypoint`) + if (!ep.ok) { + show(phase, ep.why.startsWith('404') ? 'none' : `unreadable (${ep.why})`) + continue + } + const rules = ep.result.rules ?? [] + show(phase, `${rules.length} rule(s)`) + for (const rule of rules) { + const rl = rule.ratelimit + show(` - ${rule.action}${rule.enabled === false ? ' (disabled)' : ''}`, `${rule.description ?? ''} | fields: ${fields(rule.expression)}${rl ? ` | ratelimit: by ${JSON.stringify(rl.characteristics)} ${rl.requests_per_period}/${rl.period}s mitigate ${rl.mitigation_timeout}s` : ''}${rule.action_parameters?.overrides ? ` | overrides: ${JSON.stringify(rule.action_parameters.overrides)}` : ''}`) + } + } + + console.log('\n# Legacy per-IP controls (counts only)') + const access = await get<{ mode: string; configuration: { target: string } }[]>(`${z}/firewall/access_rules/rules?per_page=100`) + if (access.ok) { + const counts: Record = {} + for (const r of access.result) { counts[`${r.mode}:${r.configuration.target}`] = (counts[`${r.mode}:${r.configuration.target}`] ?? 0) + 1 } + show('IP access rules (zone)', counts) + } else { show('IP access rules (zone)', `unreadable (${access.why})`) } + for (const [label, path] of [['zone lockdown rules', 'firewall/lockdowns'], ['user-agent blocking rules', 'firewall/ua_rules']] as const) { + const r = await get(`${z}/${path}?per_page=100`) + show(label, r.ok ? `${r.result.length}` : `unreadable (${r.why})`) + } +} + +main().catch((error) => { + console.error(error instanceof Error ? error.message : error) + process.exitCode = 1 +}) diff --git a/cloud/apps/relay-edge/src/admin.ts b/cloud/apps/relay-edge/src/admin.ts index 1dc40580deb..072237741a8 100644 --- a/cloud/apps/relay-edge/src/admin.ts +++ b/cloud/apps/relay-edge/src/admin.ts @@ -4,6 +4,8 @@ // POST /__admin/assign { userId, relayHostId, cellId, assignmentEpoch, devices? } // POST /__admin/export { relayHostId } freezes the host for the move away (409 while connected) // POST /__admin/erase { userId, relayHostId } +// POST /__admin/verify { relayHostId, kind, tokenHash } read-only: is this phone credential valid +// here? (devices paired or rotated on the edge are not in Postgres) // POST /__admin/drain { relayHostId } sends the host's live control back to the director // Replay: a token with a `jti` (16-128 chars) is single use, enforced by the target host's object // (admin-jti.ts; a repeat is 409 `replay`). Transitional: a token without one is still accepted @@ -21,7 +23,7 @@ import type { AdminReplayGuard } from './admin-jti' export const ADMIN_AUDIENCE = 'orca-relay-edge-admin' export const ADMIN_PURPOSE = 'edge-admin' -export type AdminOp = 'assign' | 'export' | 'erase' | 'drain' +export type AdminOp = 'assign' | 'export' | 'erase' | 'drain' | 'verify' const HOST_ID = /^[A-Za-z0-9_-]{16}$/ const ADMIN_BODY_MAX = 256 * 1024 @@ -64,6 +66,20 @@ export type AssignBody = z.infer export const ExportBodySchema = z.object({ relayHostId: z.string().regex(HOST_ID) }).strict() export const DrainBodySchema = z.object({ relayHostId: z.string().regex(HOST_ID) }).strict() +// The credential's SHA-256 (base64url), as stored: the director never sends the token itself. +export const VerifyBodySchema = z + .object({ + relayHostId: z.string().regex(HOST_ID), + kind: z.enum(['resume', 'invite']), + tokenHash: z.string().regex(/^[A-Za-z0-9_-]{43}$/) + }) + .strict() +export type VerifyBody = z.infer +export type VerifyResult = + | { outcome: 'verified'; userId: string; relayDeviceId: string; assignmentEpoch: number } + | { outcome: 'none' } + | { outcome: 'not_here' } + export const EraseBodySchema = z.object({ userId: z.string().min(1), relayHostId: z.string().regex(HOST_ID) }).strict() async function sha256B64url(bytes: ArrayBuffer): Promise { diff --git a/cloud/apps/relay-edge/src/env.ts b/cloud/apps/relay-edge/src/env.ts index 88cb6a0ca06..f4260e3dd49 100644 --- a/cloud/apps/relay-edge/src/env.ts +++ b/cloud/apps/relay-edge/src/env.ts @@ -1,5 +1,8 @@ import type { HostRelay } from './host-relay' import type { SpliceRelay } from './splice-relay' +import { INGRESS_SECRET_MIN_LENGTH } from '../../../packages/relay-contract/src/ingress-header' +import { parseCidr, type Cidr } from './ingress' +import { metric } from './metrics' export type Env = { HOST_RELAY: DurableObjectNamespace @@ -54,6 +57,18 @@ export type Env = { ERASE_SIGNING_KEY?: string ERASE_SIGNING_KEY_NEXT?: string ERASE_ISSUER?: string + // Secret(s) shared with the ingress proxies (secret binding; comma-separated while rotating, + // current first). Unset or malformed (see ingressConfig): ingress headers are ignored and every + // request is treated as direct. + INGRESS_SECRETS?: string + // Token-mode ingress (a cloud load balancer; see ingress.ts). All three or none; a partial or + // malformed setup switches token mode off and reports why, it never fails requests. INGRESS_TOKENS is + // a secret (comma-separated while rotating); the origins (comma-separated, exact https origins) + // and the load balancer's egress CIDRs (comma-separated, from scripts/ingress-source-cidrs.ts) + // are plain vars. Unset: `v1s` headers are ignored. + INGRESS_TOKENS?: string + INGRESS_ORIGINS?: string + INGRESS_SOURCE_CIDRS?: string } export type EdgeClaims = { @@ -73,6 +88,11 @@ export type EdgeConfig = { sweepIntervalMs: number leftRetentionMs: number clientKeepalive: boolean + // Empty when that mode is off (unset, or switched off by a problem below). + ingressSecrets: string[] + ingressToken: { tokens: string[]; origins: string[]; sourceCidrs: Cidr[] } + // Why an ingress mode is off despite being configured (labels only, never values). + ingressProblems: string[] } const configs = new WeakMap() @@ -106,6 +126,7 @@ export function edgeConfig(env: Env): EdgeConfig { if (!env.RELAY_JWKS_URL && !env.RELAY_JWKS) { throw new Error('relay_edge_config_jwks') } if (env.RELAY_JWKS_URL && !env.RELAY_AUTH_ISSUER) { throw new Error('relay_edge_config_issuer') } if (!env.RELAY_JWT_AUDIENCE) { throw new Error('relay_edge_config_audience') } + const ingress = ingressConfig(env) const config: EdgeConfig = { backend, cellId: env.EDGE_CELL_ID || 'edge', @@ -114,12 +135,50 @@ export function edgeConfig(env: Env): EdgeConfig { orphanGraceMs: positiveInt('orphan_grace', env.ORPHAN_GRACE_MS), sweepIntervalMs: positiveInt('sweep_interval', env.SWEEP_INTERVAL_MS ?? String(24 * 60 * 60 * 1000)), leftRetentionMs: positiveInt('left_retention', env.LEFT_RETENTION_MS ?? String(30 * 24 * 60 * 60 * 1000)), - clientKeepalive: env.CLIENT_KEEPALIVE !== '0' + clientKeepalive: env.CLIENT_KEEPALIVE !== '0', + ...ingress } // The offer carries the ping interval, which the contract bounds to 1..60 s. config.clientKeepalive &&= config.pingIntervalMs >= 1_000 && config.pingIntervalMs <= 60_000 // Shipped desktops close a control that is silent for 75 s; a ping must arrive well inside it. if (config.pingIntervalMs >= config.silenceTimeoutMs) { throw new Error('relay_edge_config_ping_interval') } configs.set(env, config) + if (config.ingressProblems.length > 0) { + // Never thrown: this runs on every request and in every object's alarms. A secret landing + // before its vars (or one bad range) switches that ingress mode off, loudly, and nothing else. + console.error(JSON.stringify({ event: 'relay_edge_ingress_config_invalid', problems: config.ingressProblems })) + for (const problem of config.ingressProblems) { metric(env, 'ingress_config_invalid', 1, problem) } + } return config } + +const list = (value: string | undefined) => (value ?? '').split(',').map((s) => s.trim()).filter(Boolean) +// Narrower than this is a typo, not a load balancer's egress set. +const MIN_PREFIX = { v4: 8, v6: 16 } +const TOKEN_OFF: EdgeConfig['ingressToken'] = { tokens: [], origins: [], sourceCidrs: [] } + +// Each mode is all or nothing: a partial or malformed setup turns that mode off (requests are then +// treated as direct) and reports why; it never throws. +function ingressConfig(env: Env): Pick { + const problems: string[] = [] + let ingressSecrets = list(env.INGRESS_SECRETS) + if (ingressSecrets.some((s) => s.length < INGRESS_SECRET_MIN_LENGTH)) { + problems.push('secret_too_short') + ingressSecrets = [] + } + const tokens = list(env.INGRESS_TOKENS) + const origins = list(env.INGRESS_ORIGINS) + const cidrs = list(env.INGRESS_SOURCE_CIDRS) + if (tokens.length + origins.length + cidrs.length === 0) { return { ingressSecrets, ingressToken: TOKEN_OFF, ingressProblems: problems } } + const sourceCidrs = cidrs.map(parseCidr) + const originOk = (o: string) => URL.canParse(o) && new URL(o).protocol === 'https:' && new URL(o).origin === o + const tokenProblems = [ + tokens.length === 0 || origins.length === 0 || cidrs.length === 0 ? 'token_mode_partial' : '', + tokens.some((t) => t.length < INGRESS_SECRET_MIN_LENGTH || /\s/.test(t)) ? 'token_too_short' : '', + origins.every(originOk) ? '' : 'origin_invalid', + sourceCidrs.some((c) => c === null) ? 'cidr_invalid' : '', + sourceCidrs.some((c) => c !== null && c.bits < (c.v6 ? MIN_PREFIX.v6 : MIN_PREFIX.v4)) ? 'cidr_too_broad' : '' + ].filter(Boolean) + if (tokenProblems.length > 0) { return { ingressSecrets, ingressToken: TOKEN_OFF, ingressProblems: problems.concat(tokenProblems) } } + return { ingressSecrets, ingressToken: { tokens, origins, sourceCidrs: sourceCidrs as Cidr[] }, ingressProblems: problems } +} diff --git a/cloud/apps/relay-edge/src/host-relay.ts b/cloud/apps/relay-edge/src/host-relay.ts index bc6095bc33a..158f304b07e 100644 --- a/cloud/apps/relay-edge/src/host-relay.ts +++ b/cloud/apps/relay-edge/src/host-relay.ts @@ -12,7 +12,8 @@ import { RELAY_PROTOCOL_LIMITS } from '../../../packages/relay-contract/src/prot import { CONTROL_CONTINUITY_LIMITS } from '../../../packages/relay-contract/src/control-continuity' import { metric } from './metrics' import { RelayCredentialStore } from '../../relay/src/credentials/credential-store' -import type { AssignBody, DeviceRow } from './admin' +import type { AssignBody, DeviceRow, VerifyBody, VerifyResult } from './admin' +import { verifiesHere, verifyCredential } from './verify' import { ADMIN_JTI_KEY, admitJti, type AdminReplayGuard, type ReplayRefusal, type SeenJtis } from './admin-jti' import type { AssignResult, StoredAssignment } from './object-store' import type { EdgeClaims } from './env' @@ -100,6 +101,17 @@ export class HostRelay extends HostRelayPhones { return this.ctx.getWebSockets('control').filter((ws) => this.attachment(ws) !== null) } + // The director's phone credential check (/v1/resolve, the invite move): read-only (verify.ts). An + // object that never held a host (pushed() reads sqlite_master, so no schema is created) or does + // not hold this one answers before the jti is recorded and writes nothing; otherwise the jti is + // its only write. + async adminVerify(body: VerifyBody, replay?: AdminReplayGuard | null): Promise { + if (!this.pushed() || !verifiesHere(this.store, this.config.cellId, body.relayHostId)) { return { outcome: 'not_here' } } + const refused = await this.claimAdminJti(replay) + if (refused) { return refused } + return await verifyCredential(this.store, this.config.cellId, body) + } + // Health probe (GET /__health/object, health.ts), only ever called on the canary object: one // read of the object's SQLite. No writes, no alarm, no storage created. healthCheck(): { ok: true } { diff --git a/cloud/apps/relay-edge/src/index.ts b/cloud/apps/relay-edge/src/index.ts index c12f4c1979f..efbda0d92cd 100644 --- a/cloud/apps/relay-edge/src/index.ts +++ b/cloud/apps/relay-edge/src/index.ts @@ -8,11 +8,13 @@ import { SPLICE_SLOTS, makeConnId, parseConnId } from './splice-slot' import { metric, requestPlace } from './metrics' import { admitPreAuth, notOnEdge } from './preauth' import { objectHealth } from './health' -import { AssignBodySchema, DrainBodySchema, EraseBodySchema, ExportBodySchema, verifyAdminRequest, type AdminOp } from './admin' +import { INGRESS_HEADER } from '../../../packages/relay-contract/src/ingress-header' +import { fromTokenSource, readIngress, type Ingress } from './ingress' +import { AssignBodySchema, DrainBodySchema, EraseBodySchema, ExportBodySchema, VerifyBodySchema, verifyAdminRequest, type AdminOp } from './admin' export { HostRelay, SpliceRelay } -const INTERNAL_HEADERS = ['x-edge-claims', 'x-edge-origin', 'x-edge-kind', 'x-edge-conn-id', 'x-edge-place'] +const INTERNAL_HEADERS = ['x-edge-claims', 'x-edge-origin', 'x-edge-kind', 'x-edge-conn-id', 'x-edge-place', INGRESS_HEADER, 'x-orca-client-ip', 'x-orca-client-country'] function reject(status: number, retryAfter?: number): Response { const headers = retryAfter === undefined ? undefined : { 'retry-after': String(retryAfter) } @@ -22,6 +24,7 @@ function reject(status: number, retryAfter?: number): Response { function forward( env: Env, request: Request, + ingress: Ingress | null, relayHostId: string, extra: Record, target: 'host' | 'splice' = 'host', @@ -29,8 +32,9 @@ function forward( ): Promise { const headers = new Headers(request.headers) for (const name of INTERNAL_HEADERS) { headers.delete(name) } - headers.set('x-edge-origin', new URL(request.url).origin) - headers.set('x-edge-place', requestPlace(request)) + // The origin the client dialled: bound into the host proof transcript. + headers.set('x-edge-origin', ingress?.origin ?? new URL(request.url).origin) + headers.set('x-edge-place', requestPlace(request, ingress)) for (const [name, value] of Object.entries(extra)) { headers.set(name, value) } const init = new Request(request.url, { method: 'GET', headers }) if (target === 'splice') { return env.SPLICE_RELAY.get(env.SPLICE_RELAY.idFromName(spliceObject)).fetch(init) } @@ -82,6 +86,13 @@ async function handleAdminCall(env: Env, request: Request, op: AdminOp): Promise const result = await host(body.data.relayHostId).adminDrain(replay) return Response.json(result, { status: result.outcome === 'replay' || result.outcome === 'jti_full' ? 409 : 200 }) } + if (op === 'verify') { + const body = VerifyBodySchema.safeParse(verified.body) + if (!body.success) { return reject(400) } + // none and not_here are decisions (200); only a replayed or overflowing jti is 409. + const result = await host(body.data.relayHostId).adminVerify(body.data, replay) + return Response.json(result, { status: result.outcome === 'replay' || result.outcome === 'jti_full' ? 409 : 200 }) + } const body = EraseBodySchema.safeParse(verified.body) if (!body.success) { return reject(400) } const { outcome } = await host(body.data.relayHostId).adminErase(body.data.userId, replay) @@ -98,12 +109,24 @@ export default { } if (url.search) { return reject(400) } if (url.pathname === '/__health/object') { return objectHealth(env, request) } - const admin = url.pathname.match(/^\/__admin\/(assign|export|erase|drain)$/) + // Director admin calls are signature-authenticated and never go through a front door: matched + // before the ingress rules, so an egress overlap with a front door's ranges cannot 403 them. + const admin = url.pathname.match(/^\/__admin\/(assign|export|erase|drain|verify)$/) if (admin) { return handleAdmin(env, request, admin[1] as AdminOp) } + // A trusted ingress hop (ingress.ts) stands in for the Worker's own URL and the caller's address; + // anything else, including a header that fails to verify, is treated as absent. + const ingress = await readIngress(edgeConfig(env), request) + if (!ingress && fromTokenSource(edgeConfig(env), request)) { + // From a front door's egress ranges but not trusted (missing or wrong token, origin or client + // address): refused, never keyed on the shared egress address as if it were a direct client. + metric(env, 'ingress_rejected', 1, '', [], '', requestPlace(request)) + return reject(403) + } + if (!ingress && request.headers.has(INGRESS_HEADER)) { metric(env, 'ingress_refused', 1, '', [], '', requestPlace(request)) } // Spike-only: force the object to reset (simulates eviction / runtime restart). const abort = url.pathname.match(/^\/__spike\/abort\/([^/]+)$/) if (abort && env.SPIKE_ADMIN === '1') { - return forward(env, request, abort[1], { 'x-edge-kind': 'abort' }) + return forward(env, request, ingress, abort[1], { 'x-edge-kind': 'abort' }) } const abortSplice = url.pathname.match(/^\/__spike\/abort-splice\/([^/]+)$/) if (abortSplice && env.SPIKE_ADMIN === '1') { @@ -111,7 +134,7 @@ export default { const parsed = parseConnId(connId) if (!parsed) { return reject(429) } // Resets the whole slot object: every splice currently in that slot. - return forward(env, request, '', { 'x-edge-kind': 'abort', 'x-edge-conn-id': connId }, 'splice', parsed.object) + return forward(env, request, ingress, '', { 'x-edge-kind': 'abort', 'x-edge-conn-id': connId }, 'splice', parsed.object) } if (env.SPIKE_DEBUG === '1') { if (url.pathname === '/__spike/info') { @@ -141,7 +164,7 @@ export default { return Response.json({ colo: cf?.colo ?? 'local', ms }) } const info = url.pathname.match(/^\/__spike\/host\/([A-Za-z0-9_-]{16})$/) - if (info) { return forward(env, request, info[1], { 'x-edge-kind': 'info' }) } + if (info) { return forward(env, request, ingress, info[1], { 'x-edge-kind': 'info' }) } } const isUpgrade = request.headers.get('upgrade')?.toLowerCase() === 'websocket' @@ -154,12 +177,12 @@ export default { // desktops treat any refused upgrade the same (a network error, retried through the // director), so this changes no client behavior; it keeps an outage out of the 401s. if (verified?.reason === 'keys_unavailable') { - metric(env, 'control_upgrade', 1, 'keys_unavailable', [], '', requestPlace(request)) + metric(env, 'control_upgrade', 1, 'keys_unavailable', [], '', requestPlace(request, ingress)) return reject(503, 5) } return reject(401) } - return forward(env, request, verified.claims.relayHostId, { + return forward(env, request, ingress, verified.claims.relayHostId, { 'x-edge-kind': 'control', 'x-edge-claims': JSON.stringify(verified.claims) }) @@ -167,7 +190,7 @@ export default { const phone = url.pathname.match(/^\/v1\/connect\/([^/]+)$/) if (phone) { - if (!(await admitPreAuth(env, request, 'phone'))) { return reject(429) } + if (!(await admitPreAuth(env, request, 'phone', ingress))) { return reject(429) } if (!isUpgrade) { return reject(426) } // 429 for malformed ids matches the GCE cells. if (!HOST_ID.test(phone[1])) { return reject(429) } @@ -178,21 +201,21 @@ export default { .allocateConnId(phone[1]) .catch(() => makeConnId(phone[1], Math.floor(Math.random() * SPLICE_SLOTS))) if (connId === null) { - metric(env, 'not_on_edge', 1, '', [], '', requestPlace(request)) + metric(env, 'not_on_edge', 1, '', [], '', requestPlace(request, ingress)) return notOnEdge() } const parsed = parseConnId(connId)! - return forward(env, request, phone[1], { 'x-edge-kind': 'phone', 'x-edge-conn-id': connId }, 'splice', parsed.object) + return forward(env, request, ingress, phone[1], { 'x-edge-kind': 'phone', 'x-edge-conn-id': connId }, 'splice', parsed.object) } const data = url.pathname.match(/^\/v1\/host\/data\/([^/]+)$/) if (data) { - if (!(await admitPreAuth(env, request, 'data'))) { return reject(429) } + if (!(await admitPreAuth(env, request, 'data', ingress))) { return reject(429) } if (!isUpgrade) { return reject(426) } const connId = decodeURIComponent(data[1]) const parsed = parseConnId(connId) if (!parsed) { return reject(429) } - return forward(env, request, parsed.relayHostId, { 'x-edge-kind': 'data', 'x-edge-conn-id': connId }, 'splice', parsed.object) + return forward(env, request, ingress, parsed.relayHostId, { 'x-edge-kind': 'data', 'x-edge-conn-id': connId }, 'splice', parsed.object) } return reject(404) } diff --git a/cloud/apps/relay-edge/src/ingress.ts b/cloud/apps/relay-edge/src/ingress.ts new file mode 100644 index 00000000000..493c235dc7f --- /dev/null +++ b/cloud/apps/relay-edge/src/ingress.ts @@ -0,0 +1,100 @@ +// Front doors that are not Cloudflare. Two ways to be trusted; anything else is a direct request. +// - Signed (our ingress proxy): `x-orca-ingress: v1 …` with an HMAC (relay-contract ingress-header.ts). +// - Token (a cloud load balancer, which cannot compute a MAC): `x-orca-ingress: v1s +// ` plus `x-orca-client-ip` / `x-orca-client-country`, all set by the load balancer +// (it overwrites same-named client headers). Trusted only when the token is one of +// INGRESS_TOKENS, the origin is one of INGRESS_ORIGINS, AND cf-connecting-ip (the hop that +// reached Cloudflare) is inside INGRESS_SOURCE_CIDRS, the load balancer's published egress +// ranges, pinned in config (scripts/ingress-source-cidrs.ts refreshes and checks them). +// Those ranges are shared by every customer's load balancers: the token is the discriminator. +// A request from those ranges that fails the rule is refused (403, ingress_rejected) rather +// than treated as direct, which would key pre-auth limits on the shared egress address. +import { INGRESS_HEADER, verifyIngressHeader, type IngressClaim } from '../../../packages/relay-contract/src/ingress-header' +import type { EdgeConfig } from './env' + +export type Ingress = IngressClaim & { country?: string } +export type Cidr = { v6: boolean; base: bigint; bits: number } + +const COUNTRY = /^[A-Z]{2}$/ + +export async function readIngress(config: EdgeConfig, request: Request): Promise { + const header = request.headers.get(INGRESS_HEADER) + if (!header) { return null } + if (!header.startsWith('v1s ')) { return verifyIngressHeader(header, config.ingressSecrets) } + const { tokens, origins, sourceCidrs } = config.ingressToken + const parts = header.split(' ') + if (tokens.length === 0 || parts.length !== 4) { return null } + const [, name, origin, token] = parts as [string, string, string, string] + if (!/^[a-z0-9-]{1,32}$/.test(name) || !origins.includes(origin)) { return null } + if (!fromTokenSource(config, request)) { return null } + if (!(await tokenMatches(tokens, token))) { return null } + // Expanded by the load balancer; a missing value (empty) is refused rather than keyed on its egress. + const clientIp = request.headers.get('x-orca-client-ip')?.trim() ?? '' + if (!parseIp(clientIp)) { return null } + const country = request.headers.get('x-orca-client-country')?.trim().toUpperCase() ?? '' + return { name, origin, clientIp, ...(COUNTRY.test(country) ? { country } : {}) } +} + +// The hop that reached Cloudflare is inside the token-mode front door's egress ranges. Those ranges +// are shared by every customer's load balancers, so this alone proves nothing; the token does. +export function fromTokenSource(config: EdgeConfig, request: Request): boolean { + const { sourceCidrs } = config.ingressToken + return sourceCidrs.length > 0 && inCidrs(request.headers.get('cf-connecting-ip') ?? '', sourceCidrs) +} + +// Compares SHA-256 digests, so neither the length nor the content of a token leaks through timing. +// crypto.subtle.timingSafeEqual is the Workers runtime's; the loop is the same check elsewhere. +async function tokenMatches(tokens: readonly string[], presented: string): Promise { + const digest = async (text: string) => new Uint8Array(await crypto.subtle.digest('SHA-256', new TextEncoder().encode(text))) + const mine = await digest(presented) + const subtle = crypto.subtle as SubtleCrypto & { timingSafeEqual?: (a: ArrayBufferView, b: ArrayBufferView) => boolean } + let match = false + for (const token of tokens) { + const theirs = await digest(token) + let equal: boolean + if (subtle.timingSafeEqual) { equal = subtle.timingSafeEqual(mine, theirs) } + else { + let diff = 0 + for (let i = 0; i < mine.length; i++) { diff |= mine[i]! ^ theirs[i]! } + equal = diff === 0 + } + match ||= equal + } + return match +} + +// IPv4 or IPv6 text → family and 32/128-bit value; null for anything else. +export function parseIp(text: string): { v6: boolean; value: bigint } | null { + if (/^\d{1,3}(\.\d{1,3}){3}$/.test(text)) { + const octets = text.split('.').map(Number) + if (octets.some((o) => o > 255)) { return null } + return { v6: false, value: octets.reduce((acc, o) => (acc << 8n) | BigInt(o), 0n) } + } + if (!/^[0-9a-fA-F:]{2,39}$/.test(text) || text.split('::').length > 2) { return null } + const [head, tail] = text.split('::') as [string, string | undefined] + const left = head ? head.split(':') : [] + const right = tail ? tail.split(':') : [] + const fill = tail === undefined ? 0 : 8 - left.length - right.length + const groups = [...left, ...new Array(Math.max(fill, 0)).fill('0'), ...right] + if (groups.length !== 8 || (tail !== undefined && fill < 1) || groups.some((g) => !/^[0-9a-fA-F]{1,4}$/.test(g))) { return null } + return { v6: true, value: groups.reduce((acc, g) => (acc << 16n) | BigInt(Number.parseInt(g, 16)), 0n) } +} + +export function parseCidr(text: string): Cidr | null { + const [address, bitsText] = text.trim().split('/') + const ip = parseIp(address ?? '') + const width = ip?.v6 ? 128 : 32 + const bits = Number(bitsText) + if (!ip || !/^\d{1,3}$/.test(bitsText ?? '') || bits > width) { return null } + const mask = ((1n << BigInt(bits)) - 1n) << BigInt(width - bits) + return { v6: ip.v6, base: ip.value & mask, bits } +} + +export function inCidrs(text: string, cidrs: readonly Cidr[]): boolean { + const ip = parseIp(text.trim()) + if (!ip) { return false } + return cidrs.some((c) => { + const width = c.v6 ? 128 : 32 + return c.v6 === ip.v6 && ip.value >> BigInt(width - c.bits) === c.base >> BigInt(width - c.bits) + }) +} diff --git a/cloud/apps/relay-edge/src/metrics.ts b/cloud/apps/relay-edge/src/metrics.ts index c49284a7b06..8acaf349193 100644 --- a/cloud/apps/relay-edge/src/metrics.ts +++ b/cloud/apps/relay-edge/src/metrics.ts @@ -41,9 +41,17 @@ export type MetricEvent = // Pre-auth socket refused by the per-source limiter: label 'phone' | 'data' | 'health'. Unpushed-host phones: 'not_on_edge'. | 'preauth_limited' | 'not_on_edge' + // A request carried an ingress header that failed to verify (wrong secret, skew, malformed). + | 'ingress_refused' + // A request from a token-mode front door's egress ranges that failed the trust rule: answered 403. + | 'ingress_rejected' + // An ingress mode is configured but malformed, so it is off: label names the problem (per isolate). + | 'ingress_config_invalid' // blobs: [event, label, version, reason, place]; doubles: [value, ...extra]. place is the -// client's ingress '/' (e.g. 'AS/HKG'), '' where no client is involved. One index per event for sampling. +// client's ingress '/' (e.g. 'AS/HKG'), '' where no client is involved. A verified +// ingress hop appends the client's country (as the ingress reports it, else the hop's) and the hop's +// name: 'AS/HKG/CN/cn' (continent/colo are then the hop's). One index per event for sampling. // Heartbeat events (alarm_lag_ms, pong_rtt_ms) fire per host every ping interval; only 1 in // HEARTBEAT_FACTOR is written, and Analytics Engine's _sample_interval does not know that. Those // two events carry the factor as their last double (double2, as they pass no extra), so a count @@ -65,7 +73,9 @@ export function metric(env: Env, event: MetricEvent, value = 1, label = '', extr } // From the Worker's request (cf is absent in local dev). -export function requestPlace(request: Request): string { - const cf = (request as Request & { cf?: { colo?: string; continent?: string } }).cf - return cf ? `${cf.continent ?? '?'}/${cf.colo ?? '?'}` : 'local' +export function requestPlace(request: Request, ingress?: { name: string; country?: string } | null): string { + const cf = (request as Request & { cf?: { colo?: string; continent?: string; country?: string } }).cf + if (!cf) { return ingress ? `local/${ingress.name}` : 'local' } + const place = `${cf.continent ?? '?'}/${cf.colo ?? '?'}` + return ingress ? `${place}/${ingress.country ?? cf.country ?? '?'}/${ingress.name}` : place } diff --git a/cloud/apps/relay-edge/src/preauth.ts b/cloud/apps/relay-edge/src/preauth.ts index 2f7e50a94e8..0041f0114c1 100644 --- a/cloud/apps/relay-edge/src/preauth.ts +++ b/cloud/apps/relay-edge/src/preauth.ts @@ -5,12 +5,13 @@ import { RELAY_CLOSE_CODE } from '../../../packages/relay-contract/src/close-codes' import { RELAY_PROTOCOL_LIMITS } from '../../../packages/relay-contract/src/protocol-limits' import type { Env } from './env' +import type { Ingress } from './ingress' import { metric, requestPlace } from './metrics' -// The caller's address as Cloudflare saw it (clients cannot set this header). IPv6 callers are -// keyed by their /64, which one subscriber usually holds whole. -export function sourceKey(request: Request): string { - const ip = request.headers.get('cf-connecting-ip')?.trim() ?? '' +// The caller's address as Cloudflare saw it (clients cannot set this header), or as a verified +// ingress saw it. IPv6 callers are keyed by their /64, which one subscriber usually holds whole. +export function sourceKey(request: Request, ingress?: Ingress | null): string { + const ip = (ingress?.clientIp ?? request.headers.get('cf-connecting-ip'))?.trim() ?? '' if (!ip.includes(':')) { return ip || 'unknown' } const [head, tail = ''] = ip.split('::') const left = head ? head.split(':') : [] @@ -24,11 +25,16 @@ export function sourceKey(request: Request): string { // Counted per source per colo by the Rate Limiting binding. Fails open: a limiter outage must not // refuse live users. Without a binding (local spike runs) nothing is limited. -export async function admitPreAuth(env: Env, request: Request, kind: 'phone' | 'data' | 'health'): Promise { +export async function admitPreAuth( + env: Env, + request: Request, + kind: 'phone' | 'data' | 'health', + ingress?: Ingress | null +): Promise { if (!env.PREAUTH_LIMITER) { return true } try { - const { success } = await env.PREAUTH_LIMITER.limit({ key: `${kind}:${sourceKey(request)}` }) - if (!success) { metric(env, 'preauth_limited', 1, kind, [], '', requestPlace(request)) } + const { success } = await env.PREAUTH_LIMITER.limit({ key: `${kind}:${sourceKey(request, ingress)}` }) + if (!success) { metric(env, 'preauth_limited', 1, kind, [], '', requestPlace(request, ingress)) } return success } catch { return true diff --git a/cloud/apps/relay-edge/src/verify.ts b/cloud/apps/relay-edge/src/verify.ts new file mode 100644 index 00000000000..33768bb6f25 --- /dev/null +++ b/cloud/apps/relay-edge/src/verify.ts @@ -0,0 +1,24 @@ +// The director's phone credential check (POST /__admin/verify), on the host's own store. Read-only: +// it never freezes, fences or marks the host busy. Devices paired or rotated on the edge, and its +// invites, are only in this object; Postgres gets the devices back when the host leaves. +import { RelayCredentialStore } from '../../relay/src/credentials/credential-store' +import type { VerifyBody, VerifyResult } from './admin' +import type { ObjectStore } from './object-store' + +// Whether this object answers for the host at all: only its current placement here, not frozen. +export function verifiesHere(store: ObjectStore, cellId: string, relayHostId: string): boolean { + const assignment = store.assignment() + return assignment !== null && assignment.relayHostId === relayHostId && assignment.cellId === cellId && !store.frozen() +} + +export async function verifyCredential(store: ObjectStore, cellId: string, body: VerifyBody): Promise { + const assignment = store.assignment() + if (!assignment || !verifiesHere(store, cellId, body.relayHostId)) { return { outcome: 'not_here' } } + const credentials = new RelayCredentialStore(store.database) + const found = + body.kind === 'invite' + ? await credentials.resolveInviteForMoveHash(body.relayHostId, body.tokenHash) + : await credentials.resolveResumeHash(body.relayHostId, body.tokenHash) + if (!found || found.userId !== assignment.userId) { return { outcome: 'none' } } + return { outcome: 'verified', userId: found.userId, relayDeviceId: found.relayDeviceId, assignmentEpoch: assignment.assignmentEpoch } +} diff --git a/cloud/apps/relay-edge/test/front-door-answer.test.ts b/cloud/apps/relay-edge/test/front-door-answer.test.ts new file mode 100644 index 00000000000..44af48f6537 --- /dev/null +++ b/cloud/apps/relay-edge/test/front-door-answer.test.ts @@ -0,0 +1,24 @@ +import { describe, expect, it } from 'vitest' +import { classify } from './front-door-answer' + +const cf = { server: 'cloudflare', 'cf-ray': 'x' } +const a = (status: number, headers: Record = cf, body = '') => ({ status, headers, body, ms: 1 }) + +describe('G4 answer classification', () => { + it('keeps our refusals apart from the zone and the load balancer', () => { + expect(classify(a(101, {}))).toBe('upgraded') + expect(classify(a(426))).toBe('edge') + expect(classify(a(401))).toBe('edge') + expect(classify(a(403))).toBe('edge') + expect(classify(a(403, cf, 'blocked'))).toBe('other') + expect(classify(a(200, cf, '{"ok":true}'))).toBe('edge') + expect(classify(a(429))).toBe('edge-preauth-limiter') + expect(classify(a(403, { ...cf, 'cf-mitigated': 'challenge' }, ''))).toBe('cf-zone') + expect(classify(a(429, cf, 'error code: 1015'))).toBe('cf-zone') + expect(classify(a(403, cf, 'Attention Required! | Cloudflare'))).toBe('cf-zone') + expect(classify(a(502, { via: '1.1 google' }, 'Error: Server Error'))).toBe('load-balancer') + expect(classify(a(-1, {}))).toBe('network') + expect(classify(a(500))).toBe('other') + expect(classify(a(426, {}), '')).toBe('edge') + }) +}) diff --git a/cloud/apps/relay-edge/test/front-door-answer.ts b/cloud/apps/relay-edge/test/front-door-answer.ts new file mode 100644 index 00000000000..afaf5c67cb8 --- /dev/null +++ b/cloud/apps/relay-edge/test/front-door-answer.ts @@ -0,0 +1,18 @@ +// G4: who answered a handshake through the front door. Our Worker answers its refusals with an +// empty body (index.ts reject()); Cloudflare's zone features (bot fight mode, security level, +// rate/custom rules) set `cf-mitigated` or serve an HTML error page; the load balancer's own errors +// come without the edge's `server` header. +export type Answer = { status: number; headers: Record; body: string; ms: number } +export type Source = 'upgraded' | 'edge' | 'edge-preauth-limiter' | 'cf-zone' | 'load-balancer' | 'network' | 'other' + +export function classify(a: Answer, edgeServer = 'cloudflare'): Source { + if (a.status === 101) { return 'upgraded' } + if (a.status < 0) { return 'network' } + if (a.headers['cf-mitigated'] !== undefined) { return 'cf-zone' } + if (/error code: 1\d{3}|cf-error|challenge-platform|cf-chl|Just a moment|Attention Required|Sorry, you have been blocked/i.test(a.body)) { return 'cf-zone' } + if (String(a.headers.server ?? '').toLowerCase() !== edgeServer) { return 'load-balancer' } + if (a.status === 429 && a.body === '') { return 'edge-preauth-limiter' } + // 403 is the edge refusing a front-door request that failed its trust rule (ingress_rejected). + if (a.status === 200 || ([400, 401, 403, 404, 426].includes(a.status) && a.body === '')) { return 'edge' } + return 'other' +} diff --git a/cloud/apps/relay-edge/test/g1-front-door.ts b/cloud/apps/relay-edge/test/g1-front-door.ts new file mode 100644 index 00000000000..a7d29bd2460 --- /dev/null +++ b/cloud/apps/relay-edge/test/g1-front-door.ts @@ -0,0 +1,306 @@ +// G1 (China research, architecture.md §9.8): the staging front door (global external Application +// Load Balancer + global internet NEG) in front of a phase-2 staging edge Worker. Drives the +// SHIPPED desktop control client and data transport, and phone-like sockets, through it; then holds +// a control socket and an idle phone splice for G1_HOLD_MS (default 10 min). +// +// G1_FRONT_DOOR=https:// G1_EDGE=https:// \ +// G1_ADMIN_KEY_FILE= G1_ADMIN_ISSUER= \ +// npx tsx test/g1-front-door.ts +// Optional: G1_JWT_KEYS (default spike/.keys.json: the test JWKS the staging edge trusts), +// G1_JWT_ISSUER, G1_INGRESS_NAME (default cn), G1_HOLD_MS, G1_PREAUTH_PROBES (default 0: skip), +// G1_OUT (write the results as JSON), G1_PHASES (default g1,g4). Prints no secrets or tokens. +// G4 (folded in): a burst of G4_HANDSHAKES (300) handshakes, G4_CONCURRENCY (25) at a time, then +// G4_SOCKETS (50) parallel control sockets held G4_HOLD_MS (60 s). Every request reaches Cloudflare +// from the load balancer's few egress IPs; any challenge or block from the zone's own features +// (bot fight mode, security level, rate or custom rules) is reported apart from our pre-auth limiter. +// +// The staging edge must have INGRESS_TOKENS / INGRESS_ORIGINS (incl. the front door origin) / +// INGRESS_SOURCE_CIDRS set; otherwise G1.5 fails exactly like the negative control G1.4. +import { randomBytes } from 'node:crypto' +import { readFileSync, writeFileSync } from 'node:fs' +import { request as httpsRequest } from 'node:https' +import { classify, type Answer } from './front-door-answer' +import { importJWK, type JWK, type KeyLike } from 'jose' +import { + adminAssign, + desktop, + mint, + newHost, + oversizeControl, + phoneHello, + preauthStatus, + sha, + sleep, + unauthenticatedControl, + waitFor, + type Account +} from './relay-clients' + +const required = (name: string) => { + const value = process.env[name]?.replace(/\/$/, '') + if (!value) { throw new Error(`${name} is required`) } + return value +} +const FRONT = required('G1_FRONT_DOOR') +const EDGE = required('G1_EDGE') +const ADMIN = { key: readFileSync(required('G1_ADMIN_KEY_FILE'), 'utf8').trim(), issuer: required('G1_ADMIN_ISSUER') } +const KEYS = process.env.G1_JWT_KEYS ?? new URL('../spike/.keys.json', import.meta.url).pathname +const INGRESS_NAME = process.env.G1_INGRESS_NAME ?? 'cn' +const HOLD_MS = Number(process.env.G1_HOLD_MS ?? 10 * 60_000) +const PREAUTH_PROBES = Number(process.env.G1_PREAUTH_PROBES ?? 0) +const ACCOUNT: Account = { + user: `user-g1-${randomBytes(4).toString('hex')}`, + epoch: 1, + jwtIssuer: process.env.G1_JWT_ISSUER ?? 'https://auth.test.invalid', + jwtKid: 'spike' +} + +const results: { name: string; ok: boolean; detail: string }[] = [] +function record(name: string, ok: boolean, detail = ''): void { + results.push({ name, ok, detail }) + console.log(`${ok ? 'PASS' : 'FAIL'} ${name} ${detail}`) +} +const errorText = (value: unknown) => (value instanceof Error ? value.message.slice(0, 120) : '') + +// The `server` header on the edge's own responses (local dry runs have none). +const EDGE_SERVER = process.env.G1_EDGE_SERVER_HEADER ?? 'cloudflare' +const PHASES = new Set((process.env.G1_PHASES ?? 'g1,g4').split(',').map((p) => p.trim())) +const G4 = { + handshakes: Number(process.env.G4_HANDSHAKES ?? 300), + concurrency: Number(process.env.G4_CONCURRENCY ?? 25), + sockets: Number(process.env.G4_SOCKETS ?? 50), + holdMs: Number(process.env.G4_HOLD_MS ?? 60_000) +} + +async function main(): Promise { + const key = (await importJWK((JSON.parse(readFileSync(KEYS, 'utf8')) as { privateJwk: JWK }).privateJwk, 'ES256')) as KeyLike + if (PHASES.has('g1')) { await g1(key) } + if (PHASES.has('g4')) { await g4(key) } +} + +async function g1(key: KeyLike): Promise { + const host = newHost() + const id = host.relayHostId + const jwt = await mint(ACCOUNT, key, id) + console.log(`front door ${new URL(FRONT).host} → edge ${new URL(EDGE).host}; hold ${HOLD_MS / 1000} s`) + + // G1.1 The front door answers and is the load balancer (not a direct path to Cloudflare). + const health = await fetch(`${FRONT}/health`) + const body = (await health.json().catch(() => null)) as { ok?: boolean } | null + const via = health.headers.get('via') ?? '' + record('G1.1 /health through the front door', health.status === 200 && body?.ok === true && /google/i.test(via), `status=${health.status} via=${via || 'none'}`) + + // G1.2 Admin and test routes are refused by the URL map, never forwarded. + const adminViaFront = await fetch(`${FRONT}/__admin/assign`, { method: 'POST', body: '{}' }) + record('G1.2 /__admin/* through the front door is 404 at the load balancer', adminViaFront.status === 404, `status=${adminViaFront.status}`) + + const pushed = await adminAssign(EDGE, ADMIN, ACCOUNT, id) + record('G1.3 director push straight to the edge', pushed.outcome === 'applied', `status=${pushed.status} outcome=${pushed.outcome}`) + + // G1.4 Negative control: the same headers straight to the edge, with a made-up token, are not + // trusted; the edge binds its own origin and the desktop refuses the proof. + const forged = desktop(ACCOUNT, FRONT, host, jwt, { + dialOrigin: EDGE, + extraHeaders: { + 'x-orca-ingress': `v1s ${INGRESS_NAME} ${FRONT} ${randomBytes(32).toString('base64url')}`, + 'x-orca-client-ip': '198.51.100.7', + 'x-orca-client-country': 'CN' + } + }) + const forgedAck = await forged.connect().catch((e: Error) => e) + record( + 'G1.4 negative control: forged token straight to the edge is ignored; proof refused', + forgedAck instanceof Error && /relayOrigin/.test(forgedAck.message), + forgedAck instanceof Error ? errorText(forgedAck) : 'connected?!' + ) + forged.client.closeNow() + await sleep(500) + + // G1.5 Upgrade + host proof through the front door (the edge trusted the load balancer's headers). + const started = Date.now() + const main = desktop(ACCOUNT, FRONT, host, jwt, { appVersion: 'g1-front-door' }) + const ack = await main.connect().catch((e: Error) => e) + record('G1.5 control upgrade and host proof over the front door origin', !(ack instanceof Error), ack instanceof Error ? errorText(ack) : `generation=${ack.generation} in ${Date.now() - started} ms`) + if (ack instanceof Error) { return } + + // G1.6 Pairing and data through two hops. + const invite = await main.client.createInvite('g1-phone') + const invitePhone = await phoneHello(FRONT, id, invite.inviteToken) + const resumeToken = randomBytes(32).toString('base64url') + const installed = await main.client + .installCredential({ reqId: 'g1', relayDeviceId: 'g1-phone', newResumeTokenHash: sha(resumeToken), authorization: { mode: 'relay-basis', basisConnId: main.events.lastConnId } }) + .catch((e: Error) => e) + const sizes = [32, 64 * 1024, 1024 * 1024] + const echoStart = Date.now() + for (const size of sizes) { invitePhone.ws.send(randomBytes(size)) } + await waitFor(() => invitePhone.state.frames.length === sizes.length, 20_000) + record( + 'G1.6 phone invite, credential install, frames 32 B / 64 KiB / 1 MiB echo unchanged', + invitePhone.state.hello?.ok === true && !(installed instanceof Error) && invitePhone.state.frames.map((f) => f.length).join() === sizes.join(), + `hello=${invitePhone.state.hello?.ok} install=${installed instanceof Error ? errorText(installed) : 'ok'} echoed=${invitePhone.state.frames.map((f) => f.length).join('/')} in ${Date.now() - echoStart} ms` + ) + invitePhone.ws.close(4000, 'g1 invite done') + + // G1.7 Close codes and pre-upgrade refusals pass through unchanged. + const unknown = await phoneHello(FRONT, id, randomBytes(32).toString('base64url')) + await waitFor(() => unknown.state.closed !== null, 5000) + const stray = await phoneHello(FRONT, newHost().relayHostId, randomBytes(32).toString('base64url')) + await waitFor(() => stray.state.closed !== null, 5000) + const big = await oversizeControl(FRONT, jwt) + const noJwt = await unauthenticatedControl(FRONT) + record( + 'G1.7 close codes: unknown credential 4401, unpushed host 4409, control frame > 1 MiB 1009; 401 before upgrade', + unknown.state.closed === 4401 && stray.state.closed === 4409 && big === 1009 && noJwt === 401, + `${unknown.state.closed}/${stray.state.closed}/${big}/${noJwt}` + ) + + // G1.8 Hold: the control stays up on the edge's pings; an idle phone splice stays open with no + // traffic at all (the backend timeout, not the 24 h cap, governs idle sockets). + // One frame first: the shipped desktop drops a data socket whose phone never speaks within its + // attach deadline (a real phone starts its handshake at once). Then nothing at all. + const idlePhone = await phoneHello(FRONT, id, resumeToken) + idlePhone.ws.send(randomBytes(32)) + await waitFor(() => idlePhone.state.frames.length > 0, 10_000) + const attached = idlePhone.state.frames.length + const holdStart = Date.now() + const pingsBefore = main.events.serverPings + while (Date.now() - holdStart < HOLD_MS && main.events.closes.length === 0 && idlePhone.state.closed === null) { + await sleep(Math.min(60_000, HOLD_MS - (Date.now() - holdStart))) + console.log(` hold +${Math.round((Date.now() - holdStart) / 1000)} s: control pings=${main.events.serverPings - pingsBefore} socket pongs=${main.events.socketPongs} control closes=${main.events.closes.join(',') || 'none'} phone=${idlePhone.state.closed ?? 'open'}`) + } + const held = Date.now() - holdStart + const pings = main.events.serverPings - pingsBefore + idlePhone.ws.send(randomBytes(64 * 1024)) + await waitFor(() => idlePhone.state.frames.length > attached, 10_000) + record( + `G1.8 hold ${Math.round(HOLD_MS / 1000)} s: control and idle phone splice stay open; keepalives flow; data after idle`, + idlePhone.state.hello?.ok === true && held >= HOLD_MS && main.events.closes.length === 0 && idlePhone.state.closed === null && pings >= Math.floor(HOLD_MS / 60_000) && idlePhone.state.frames.length > attached, + `held=${held} ms control pings=${pings} socket pongs=${main.events.socketPongs} closes=${main.events.closes.join(',') || 'none'} phone=${idlePhone.state.closed ?? 'open'} echo-after-idle=${idlePhone.state.frames.length > attached}` + ) + + // G1.9 Close codes after the hold: the desktop leaves, the edge closes the phone's splice with a + // relay code (not a bare 1006); then a client-initiated close completes cleanly. + main.client.closeNow() + await waitFor(() => idlePhone.state.closed !== null, 45_000) + record('G1.9 after the hold, the edge close code reaches the phone (4408 peer dropped, not 1006)', idlePhone.state.closed === 4408, `phone close=${idlePhone.state.closed} reason=${idlePhone.state.reason}`) + + // G1.10 Optional: pre-auth limiting through the front door (keyed on this client's address). + if (PREAUTH_PROBES > 0) { + const probeHost = newHost().relayHostId + const statuses: number[] = [] + for (let i = 0; i < PREAUTH_PROBES; i++) { statuses.push(await preauthStatus(FRONT, probeHost)) } + record('G1.10 pre-auth probes through the front door (informational)', statuses.includes(426), `first=${statuses[0]} limitedFrom=${statuses.indexOf(429)} last=${statuses.at(-1)}`) + } +} + +// ---- G4: the zone's view of traffic concentrated on the front door's few egress IPs ---- + +// One fresh TLS connection per handshake (no agent), so the load balancer sees many clients' +// worth of connections; it still reaches Cloudflare from its few egress IPs. +function handshake(url: string, upgrade: boolean, headers: Record = {}): Promise { + const started = Date.now() + return new Promise((resolve) => { + const req = httpsRequest(url, { + agent: false, + headers: upgrade + ? { connection: 'Upgrade', upgrade: 'websocket', 'sec-websocket-version': '13', 'sec-websocket-key': randomBytes(16).toString('base64'), ...headers } + : headers + }) + req.setTimeout(15_000, () => req.destroy(new Error('timeout'))) + req.on('upgrade', (res, socket) => { + socket.destroy() + resolve({ status: 101, headers: res.headers, body: '', ms: Date.now() - started }) + }) + req.on('response', (res) => { + let body = '' + res.setEncoding('utf8') + res.on('data', (d: string) => (body = (body + d).slice(0, 4096))) + res.on('end', () => resolve({ status: res.statusCode ?? 0, headers: res.headers, body, ms: Date.now() - started })) + res.on('error', () => resolve({ status: res.statusCode ?? 0, headers: res.headers, body, ms: Date.now() - started })) + }) + req.on('error', () => resolve({ status: -1, headers: {}, body: '', ms: Date.now() - started })) + req.end() + }) +} + +async function pool(count: number, concurrency: number, task: (i: number) => Promise): Promise { + const out: T[] = new Array(count) + let next = 0 + await Promise.all(Array.from({ length: Math.min(concurrency, count) }, async () => { + while (next < count) { + const i = next++ + out[i] = await task(i) + } + })) + return out +} + +function tally(answers: Answer[]): Record { + const counts: Record = {} + for (const a of answers) { + const k = `${classify(a, EDGE_SERVER)}:${a.status}` + counts[k] = (counts[k] ?? 0) + 1 + } + return counts +} + +async function g4(key: KeyLike): Promise { + // G4.1 Burst of handshakes, three kinds in rotation: a plain request, a control upgrade with no + // JWT (401 before upgrade), and a phone upgrade for a host nobody pushed (101, or 429 from OUR + // pre-auth limiter once this client passes it; that is expected and kept apart). + const stray = newHost().relayHostId + const kinds = [ + () => handshake(`${FRONT}/health`, false), + () => handshake(`${FRONT}/v1/host/control`, true), + () => handshake(`${FRONT}/v1/connect/${stray}`, true) + ] + const burstStart = Date.now() + const answers = await pool(G4.handshakes, G4.concurrency, (i) => kinds[i % kinds.length]!()) + const counts = tally(answers) + const zone = answers.filter((a) => classify(a, EDGE_SERVER) === 'cf-zone') + const bad = answers.filter((a) => ['cf-zone', 'load-balancer', 'network', 'other'].includes(classify(a, EDGE_SERVER))) + const ms = answers.map((a) => a.ms).sort((x, y) => x - y) + const rayShare = answers.filter((a) => a.headers['cf-ray'] !== undefined).length / Math.max(1, answers.length) + record( + `G4.1 ${G4.handshakes} handshakes (${G4.concurrency} at a time) through the front door: no zone challenge/block, no LB error`, + bad.length === 0, + `in ${Date.now() - burstStart} ms; ${JSON.stringify(counts)}; p50=${ms[Math.floor(ms.length / 2)]} ms p95=${ms[Math.floor(ms.length * 0.95)]} ms; cf-ray on ${Math.round(rayShare * 100)}%${ + zone.length ? `; zone sample: status=${zone[0]!.status} cf-mitigated=${zone[0]!.headers['cf-mitigated'] ?? '-'} body=${JSON.stringify(zone[0]!.body.slice(0, 120))}` : '' + }` + ) + + // G4.2 Parallel long-lived sockets: G4_SOCKETS desktops (distinct hosts) hold their control + // sockets through the front door at once. + const hosts = Array.from({ length: G4.sockets }, () => newHost()) + const pushes = await pool(hosts.length, 10, (i) => adminAssign(EDGE, ADMIN, ACCOUNT, hosts[i]!.relayHostId)) + const desktops = await Promise.all(hosts.map(async (h) => desktop(ACCOUNT, FRONT, h, await mint(ACCOUNT, key, h.relayHostId), { appVersion: 'g4-front-door' }))) + const openStart = Date.now() + const acks = await Promise.all(desktops.map((d) => d.connect().then(() => null, (e: Error) => e))) + const opened = acks.filter((a) => a === null).length + const openMs = Date.now() - openStart + let failure = '' + const firstFail = acks.findIndex((a) => a !== null) + if (firstFail !== -1) { + // Classify one failure with a raw handshake carrying the same JWT. + const answer = await handshake(`${FRONT}/v1/host/control`, true, { authorization: `Bearer ${await mint(ACCOUNT, key, hosts[firstFail]!.relayHostId)}` }) + failure = `; first failure: ${errorText(acks[firstFail])} (raw retry: ${classify(answer, EDGE_SERVER)}:${answer.status})` + } + const holdStart = Date.now() + while (Date.now() - holdStart < G4.holdMs && desktops.every((d) => d.events.closes.length === 0)) { await sleep(1000) } + const closed = desktops.filter((d) => d.events.closes.length > 0) + record( + `G4.2 ${G4.sockets} parallel control sockets through the front door, held ${Math.round(G4.holdMs / 1000)} s`, + pushes.every((p) => p.outcome === 'applied') && opened === G4.sockets && closed.length === 0, + `pushed=${pushes.filter((p) => p.outcome === 'applied').length} opened=${opened} in ${openMs} ms; closed during hold=${closed.length}${closed.length ? ` codes=${closed.map((d) => d.events.closes[0]).join(',')}` : ''}${failure}` + ) + for (const d of desktops) { d.client.closeNow() } +} + +main() + .catch((error) => record('G1 run', false, errorText(error))) + .finally(() => { + const failed = results.filter((r) => !r.ok) + console.log(`\n${results.length - failed.length}/${results.length} passed`) + if (process.env.G1_OUT) { writeFileSync(process.env.G1_OUT, JSON.stringify({ at: new Date().toISOString(), holdMs: HOLD_MS, results }, null, 2)) } + process.exit(failed.length ? 1 : 0) + }) diff --git a/cloud/apps/relay-edge/test/g1-local.ts b/cloud/apps/relay-edge/test/g1-local.ts new file mode 100644 index 00000000000..2149337313b --- /dev/null +++ b/cloud/apps/relay-edge/test/g1-local.ts @@ -0,0 +1,131 @@ +// Dry run of test/g1-front-door.ts on localhost: wrangler dev over https with token-mode ingress +// config, behind a tiny emulation of the load balancer (static token header, client address and +// country headers, Host override to the edge, 404 for admin routes, `via: 1.1 google`). Proves the +// G1 script and the token path end to end before anyone touches staging. +// npx tsx test/g1-local.ts (G1_HOLD_MS defaults to 20 s here) +import { execFileSync, spawn } from 'node:child_process' +import { randomBytes } from 'node:crypto' +import { mkdtempSync, rmSync, writeFileSync, readFileSync } from 'node:fs' +import { createServer, request } from 'node:https' +import type { Socket } from 'node:net' +import { tmpdir } from 'node:os' +import { join } from 'node:path' +import { connect } from 'node:tls' +import { jwksOf, loadOrCreateKeys } from '../spike/lib' + +const EDGE_PORT = 8789 +const FRONT_PORT = 8463 +const EDGE = `https://127.0.0.1:${EDGE_PORT}` +const FRONT = `https://localhost:${FRONT_PORT}` +const TOKEN = randomBytes(32).toString('base64url') +const ADMIN_KEY = randomBytes(36).toString('base64url') +const ADMIN_ISSUER = 'https://director.test.invalid' +const LB_EGRESS = '192.0.2.10' + +if (!process.env.G1_LOCAL_DIR) { + const dir = mkdtempSync(join(tmpdir(), 'relay-g1-local-')) + execFileSync('openssl', ['req', '-x509', '-newkey', 'ec', '-pkeyopt', 'ec_paramgen_curve:prime256v1', '-nodes', '-days', '1', '-subj', '/CN=localhost', + '-addext', 'subjectAltName=DNS:localhost,IP:127.0.0.1', '-keyout', join(dir, 'key.pem'), '-out', join(dir, 'cert.pem')], { stdio: 'ignore' }) + const child = spawn(process.execPath, process.execArgv.concat(process.argv.slice(1)), { + stdio: 'inherit', + env: { ...process.env, G1_LOCAL_DIR: dir, NODE_EXTRA_CA_CERTS: join(dir, 'cert.pem') } + }) + child.on('exit', (code) => { + rmSync(dir, { recursive: true, force: true }) + process.exit(code ?? 1) + }) +} else { + void run() +} + +// What the load balancer adds. Client copies of these names are overwritten, as the LB does. +function lbHeaders(raw: string[], clientIp: string): [string, string][] { + const out: [string, string][] = [] + for (let i = 0; i < raw.length; i += 2) { + if (!/^(host|x-orca-.*|cf-connecting-ip)$/i.test(raw[i]!)) { out.push([raw[i]!, raw[i + 1]!]) } + } + return out.concat([ + ['host', `127.0.0.1:${EDGE_PORT}`], + // Local dev has no Cloudflare in front: the emulated egress address, which wrangler dev passes + // through, stands in for the load balancer's egress range. Direct callers keep 127.0.0.1. + ['cf-connecting-ip', LB_EGRESS], + ['x-orca-ingress', `v1s cn ${FRONT} ${TOKEN}`], + ['x-orca-client-ip', clientIp.replace(/^::ffff:/, '')], + ['x-orca-client-country', 'CN'] + ]) +} + +function frontDoor(dir: string) { + const tls = { key: readFileSync(join(dir, 'key.pem')), cert: readFileSync(join(dir, 'cert.pem')) } + const blocked = (path = '') => /^\/(__admin|__spike)\//.test(path) + const server = createServer(tls, (req, res) => { + if (blocked(req.url)) { return void res.writeHead(404, { 'content-length': 0 }).end() } + const up = request({ host: '127.0.0.1', port: EDGE_PORT, method: req.method, path: req.url, headers: lbHeaders(req.rawHeaders, req.socket.remoteAddress ?? '').flat() }, (r) => { + res.writeHead(r.statusCode ?? 502, { ...r.headers, via: '1.1 google' }) + r.pipe(res) + }) + up.on('error', () => res.writeHead(502).end()) + req.pipe(up) + }) + server.on('upgrade', (req, client: Socket, head: Buffer) => { + if (blocked(req.url)) { return void client.end('HTTP/1.1 404 Not Found\r\ncontent-length: 0\r\n\r\n') } + const upstream = connect({ host: '127.0.0.1', port: EDGE_PORT }) + upstream.write(`${req.method} ${req.url} HTTP/1.1\r\n${lbHeaders(req.rawHeaders, req.socket.remoteAddress ?? '').map(([k, v]) => `${k}: ${v}\r\n`).join('')}\r\n`) + if (head.length) { upstream.write(head) } + upstream.pipe(client) + client.pipe(upstream) + upstream.on('error', () => client.destroy()) + client.on('error', () => upstream.destroy()) + upstream.on('close', () => client.end()) + client.on('close', () => upstream.end()) + }) + return server +} + +async function run(): Promise { + const dir = process.env.G1_LOCAL_DIR! + const keys = await loadOrCreateKeys() + writeFileSync( + new URL('../.dev.vars', import.meta.url), + [ + 'EDGE_BACKEND=object', + `ADMIN_SIGNING_KEY=${ADMIN_KEY}`, + `ADMIN_ISSUER=${ADMIN_ISSUER}`, + `RELAY_JWKS='${jwksOf(keys)}'`, + `INGRESS_TOKENS=${TOKEN}`, + `INGRESS_ORIGINS=${FRONT}`, + // The emulated load balancer's egress (documentation range); direct calls come from 127.0.0.1. + 'INGRESS_SOURCE_CIDRS=192.0.2.0/24', + 'PING_INTERVAL_MS=3000', + 'SILENCE_TIMEOUT_MS=9000', + 'ORPHAN_GRACE_MS=4000' + ].join('\n') + ) + writeFileSync(join(dir, 'admin-key'), ADMIN_KEY) + const stateDir = mkdtempSync(join(tmpdir(), 'relay-g1-local-state-')) + const dev = spawn('npx', ['wrangler', 'dev', '--port', String(EDGE_PORT), '--ip', '127.0.0.1', '--persist-to', stateDir, '--local-protocol', 'https', + '--https-key-path', join(dir, 'key.pem'), '--https-cert-path', join(dir, 'cert.pem')], { cwd: new URL('..', import.meta.url).pathname, stdio: 'ignore' }) + const front = frontDoor(dir) + await new Promise((r) => front.listen(FRONT_PORT, '127.0.0.1', () => r())) + let code = 1 + try { + for (let i = 0; ; i++) { + try { + if ((await fetch(`${EDGE}/health`)).ok) { break } + } catch {} + if (i > 180) { throw new Error('wrangler dev did not start') } + await new Promise((r) => setTimeout(r, 500)) + } + const g1 = spawn(process.execPath, ['--import', 'tsx', new URL('./g1-front-door.ts', import.meta.url).pathname], { + stdio: 'inherit', + env: { ...process.env, G1_FRONT_DOOR: FRONT, G1_EDGE: EDGE, G1_ADMIN_KEY_FILE: join(dir, 'admin-key'), G1_ADMIN_ISSUER: ADMIN_ISSUER, G1_HOLD_MS: process.env.G1_HOLD_MS ?? '20000', G1_EDGE_SERVER_HEADER: '', G4_HANDSHAKES: process.env.G4_HANDSHAKES ?? '90', G4_SOCKETS: process.env.G4_SOCKETS ?? '50', G4_HOLD_MS: process.env.G4_HOLD_MS ?? '10000' } + }) + code = await new Promise((r) => g1.on('exit', (c) => r(c ?? 1))) + } finally { + front.close() + dev.kill('SIGINT') + await new Promise((r) => setTimeout(r, 500)) + rmSync(stateDir, { recursive: true, force: true }) + process.exit(code) + } +} diff --git a/cloud/apps/relay-edge/test/ingress-e2e.ts b/cloud/apps/relay-edge/test/ingress-e2e.ts new file mode 100644 index 00000000000..f293f2d1821 --- /dev/null +++ b/cloud/apps/relay-edge/test/ingress-e2e.ts @@ -0,0 +1,287 @@ +// Local end to end: the edge Worker (wrangler dev over https, object backend) behind the ingress +// proxy (apps/relay-ingress) on localhost with a throwaway self-signed certificate. Drives the +// SHIPPED desktop control client and data transport, and phone-like sockets, through the proxy. +// pnpm --filter @orca-cloud/relay-contract build && npx tsx test/ingress-e2e.ts +import { spawn, execFileSync, type ChildProcess } from 'node:child_process' +import { randomBytes } from 'node:crypto' +import { createServer } from 'node:http' +import { mkdtempSync, rmSync, writeFileSync } from 'node:fs' +import { tmpdir } from 'node:os' +import { join } from 'node:path' +import { exportJWK, generateKeyPair, type KeyLike } from 'jose' +import { RelayRetrySchedule } from '../../../../src/main/runtime/mobile-cloud-relay/authentication/relay-retry-schedule' +import { + adminAssign as assignOnEdge, + desktop as desktopClient, + mint as mintJwt, + newHost, + oversizeControl, + phoneHello, + preauthStatus, + sha, + sleep, + unauthenticatedControl, + waitFor, + type Account, + type Host +} from './relay-clients' + +const EDGE_PORT = 8789 +const PROXY_PORT = 8443 +const EDGE = `https://127.0.0.1:${EDGE_PORT}` +// A different origin from the edge's: the host proof only verifies if the edge binds this one. +const INGRESS = `https://localhost:${PROXY_PORT}` +const SECRET = randomBytes(32).toString('base64url') +const SIGNING_KEY = randomBytes(36).toString('base64url') +const DIRECTOR = 'https://director.test.invalid' +const ISSUER = 'https://auth.test.invalid' +const EPOCH = 3 +const USER = 'user-ingress' + +// NODE_EXTRA_CA_CERTS is read at start-up: make the certificate, then run again trusting it. +if (!process.env.INGRESS_E2E_DIR) { + const dir = mkdtempSync(join(tmpdir(), 'relay-ingress-e2e-')) + execFileSync('openssl', ['req', '-x509', '-newkey', 'ec', '-pkeyopt', 'ec_paramgen_curve:prime256v1', '-nodes', '-days', '1', '-subj', '/CN=localhost', + '-addext', 'subjectAltName=DNS:localhost,IP:127.0.0.1,IP:::1', '-keyout', join(dir, 'key.pem'), '-out', join(dir, 'cert.pem')], { stdio: 'ignore' }) + const child = spawn(process.execPath, process.execArgv.concat(process.argv.slice(1)), { + stdio: 'inherit', + env: { ...process.env, INGRESS_E2E_DIR: dir, NODE_EXTRA_CA_CERTS: join(dir, 'cert.pem') } + }) + child.on('exit', (code) => { + rmSync(dir, { recursive: true, force: true }) + process.exit(code ?? 1) + }) +} else { + void main().catch((error) => { + console.error(error) + process.exitCode = 1 + }) +} + +const DIR = process.env.INGRESS_E2E_DIR ?? '' +const results: { name: string; ok: boolean; detail: string }[] = [] +function record(name: string, ok: boolean, detail = ''): void { + results.push({ name, ok, detail }) + console.log(`${ok ? 'PASS' : 'FAIL'} ${name} ${detail}`) +} +const ACCOUNT: Account = { user: USER, epoch: EPOCH, jwtIssuer: ISSUER, jwtKid: 'e2e' } +const mint = (key: KeyLike, relayHostId: string) => mintJwt(ACCOUNT, key, relayHostId) +const desktop = (cellUrl: string, host: Host, jwt: string) => desktopClient(ACCOUNT, cellUrl, host, jwt) +const adminAssign = (relayHostId: string, devices: unknown[] = []) => assignOnEdge(EDGE, { key: SIGNING_KEY, issuer: DIRECTOR }, ACCOUNT, relayHostId, devices) + +function startProxy(port: number, secret: string): { child: ChildProcess; log: () => string } { + writeFileSync(join(DIR, `secret-${port}`), secret) + const child = spawn(process.execPath, ['--import', 'tsx', new URL('../../relay-ingress/src/index.ts', import.meta.url).pathname], { + env: { + ...process.env, + INGRESS_NAME: 'cn', + INGRESS_ORIGIN: `https://localhost:${port}`, + UPSTREAM_URL: EDGE, + UPSTREAM_CA_FILE: join(DIR, 'cert.pem'), + INGRESS_SECRET_FILE: join(DIR, `secret-${port}`), + TLS_KEY_FILE: join(DIR, 'key.pem'), + TLS_CERT_FILE: join(DIR, 'cert.pem'), + PORT: String(port), + HEALTH_PORT: String(port + 1), + DRAIN_DELAY_MS: '500', + DRAIN_MS: '2000' + }, + stdio: ['ignore', 'pipe', 'pipe'] + }) + let out = '' + child.stdout?.on('data', (d) => (out += d)) + child.stderr?.on('data', (d) => (out += d)) + return { child, log: () => out } +} +async function ready(url: string, ms = 60_000): Promise { + const started = Date.now() + while (Date.now() - started < ms) { + try { + if ((await fetch(url)).ok) { return Date.now() - started } + } catch {} + await sleep(20) + } + throw new Error(`not ready: ${url}`) +} +const exited = (child: ChildProcess) => new Promise((r) => (child.exitCode !== null || child.signalCode !== null ? r() : child.once('exit', () => r()))) + +async function main(): Promise { + const { publicKey, privateKey } = await generateKeyPair('ES256', { extractable: true }) + const jwk = { ...(await exportJWK(publicKey)), kid: 'e2e', alg: 'ES256', use: 'sig' } + const jwks = createServer((_req, res) => res.setHeader('content-type', 'application/json').end(JSON.stringify({ keys: [jwk] }))) + await new Promise((r) => jwks.listen(0, '127.0.0.1', () => r())) + writeFileSync( + new URL('../.dev.vars', import.meta.url), + [ + 'EDGE_BACKEND=object', + `ADMIN_SIGNING_KEY=${SIGNING_KEY}`, + `ADMIN_ISSUER=${DIRECTOR}`, + `RELAY_JWKS_URL=http://127.0.0.1:${(jwks.address() as { port: number }).port}/jwks`, + `RELAY_AUTH_ISSUER=${ISSUER}`, + `INGRESS_SECRETS=${SECRET}`, + 'PING_INTERVAL_MS=3000', + 'SILENCE_TIMEOUT_MS=9000', + 'ORPHAN_GRACE_MS=8000' + ].join('\n') + ) + const stateDir = mkdtempSync(join(tmpdir(), 'relay-edge-ingress-state-')) + const dev = spawn( + 'npx', + ['wrangler', 'dev', '--port', String(EDGE_PORT), '--ip', '127.0.0.1', '--persist-to', stateDir, '--local-protocol', 'https', + '--https-key-path', join(DIR, 'key.pem'), '--https-cert-path', join(DIR, 'cert.pem')], + { cwd: new URL('..', import.meta.url).pathname, stdio: ['ignore', 'pipe', 'pipe'] } + ) + let devLog = '' + dev.stdout?.on('data', (d) => (devLog += d)) + dev.stderr?.on('data', (d) => (devLog += d)) + let proxy = startProxy(PROXY_PORT, SECRET) + const wrong = startProxy(PROXY_PORT + 10, randomBytes(32).toString('base64url')) + try { + await ready(`${EDGE}/health`, 90_000) + await ready(`${INGRESS}/health`) + await ready(`https://localhost:${PROXY_PORT + 10}/health`) + proxy = await scenarios(privateKey, proxy) + } catch (error) { + console.error(error) + console.error(devLog.split('\n').filter((l) => /error|exception/i.test(l)).slice(-20).join('\n')) + console.error(proxy.log()) + process.exitCode = 1 + } finally { + proxy.child.kill('SIGKILL') + wrong.child.kill('SIGKILL') + dev.kill('SIGINT') + jwks.close() + await sleep(500) + rmSync(stateDir, { recursive: true, force: true }) + } + const logs = proxy.log() + wrong.log() + record('proxy logs carry no address, path or token', !/127\.0\.0\.1|::1|\/v1\/|Bearer|eyJ/.test(logs), `${logs.split('\n').filter(Boolean).length} lines`) + const failed = results.filter((r) => !r.ok) + console.log(`\n${results.length - failed.length}/${results.length} passed`) + if (failed.length) { process.exitCode = 1 } +} + +async function scenarios(key: KeyLike, proxy: ReturnType): Promise> { + const host = newHost() + const id = host.relayHostId + const jwt = await mint(key, id) + record('director push straight to the edge', (await adminAssign(id)).outcome === 'applied') + + // 1. Host proof binds the origin the desktop dialled. + const direct = desktop(EDGE, host, jwt) + const directAck = await direct.connect().catch((e: Error) => e) + record('direct to the edge (no ingress header): host proof ok, as today', !(directAck instanceof Error), directAck instanceof Error ? directAck.message : `generation=${directAck.generation}`) + direct.client.closeNow() + const unsigned = desktop(`https://localhost:${PROXY_PORT + 10}`, host, jwt) + const unsignedAck = await unsigned.connect().catch((e: Error) => e) + record( + 'via an ingress whose secret the edge does not hold: header ignored, edge binds its own origin, desktop refuses the proof', + unsignedAck instanceof Error && /invalid host challenge/.test(unsignedAck.message), + unsignedAck instanceof Error ? unsignedAck.message.slice(0, 90) : 'connected?!' + ) + unsigned.client.closeNow() + await sleep(300) + const main = desktop(INGRESS, host, jwt) + const ack = await main.connect() + record('via the ingress: host proof over the ingress origin succeeds (shipped client)', ack.generation >= 1, `generation=${ack.generation}`) + + // 2. Data flows: pair a phone through the ingress, then frames both ways through two hops. + const invite = await main.client.createInvite('phone-a') + const invitePhone = await phoneHello(INGRESS, id, invite.inviteToken) + record('phone invite via the ingress: relay-hello ok', invitePhone.state.hello?.ok === true, JSON.stringify(invitePhone.state.hello)) + const resumeToken = randomBytes(32).toString('base64url') + const installed = await main.client + .installCredential({ reqId: 'i1', relayDeviceId: 'phone-a', newResumeTokenHash: sha(resumeToken), authorization: { mode: 'relay-basis', basisConnId: main.events.lastConnId } }) + .catch((e: Error) => e) + record('credential install over the ingress control socket', !(installed instanceof Error), installed instanceof Error ? installed.message : '') + // The shipped desktop data socket accepts frames up to 1 MiB. + const sizes = [32, 64 * 1024, 1024 * 1024] + const started = Date.now() + for (const size of sizes) { invitePhone.ws.send(randomBytes(size)) } + await waitFor(() => invitePhone.state.frames.length === sizes.length, 15_000) + record( + 'phone ↔ desktop frames echo through ingress + edge splice unchanged (32 B, 64 KiB, 1 MiB)', + invitePhone.state.frames.map((f) => f.length).join() === sizes.join(), + `echoed=${invitePhone.state.frames.map((f) => f.length).join('/')} in ${Date.now() - started} ms` + ) + invitePhone.ws.close(4000, 'phone done') + await waitFor(() => invitePhone.state.closed !== null, 3000) + + // 3. Close codes from the edge reach the client unchanged. + const unknown = await phoneHello(INGRESS, id, randomBytes(32).toString('base64url')) + await waitFor(() => unknown.state.closed !== null, 3000) + const stray = await phoneHello(INGRESS, newHost().relayHostId, randomBytes(32).toString('base64url')) + await waitFor(() => stray.state.closed !== null, 3000) + const bigCode = await oversizeControl(INGRESS, jwt) + const noJwt = await unauthenticatedControl(INGRESS) + record( + 'close codes through the ingress: unknown credential 4401, unpushed host 4409, control frame > 1 MiB 1009; 401 before upgrade', + unknown.state.closed === 4401 && stray.state.closed === 4409 && bigCode === 1009 && noJwt === 401, + `${unknown.state.closed}/${stray.state.closed}/${bigCode}/${noJwt}` + ) + + // 4. Pre-auth limits key on the client the ingress saw (127.0.0.1 vs ::1), not on the proxy. + const probeHost = newHost().relayHostId + const fromV4: number[] = [] + for (let i = 0; i < 65; i++) { fromV4.push(await preauthStatus(`https://127.0.0.1:${PROXY_PORT}`, probeHost)) } + const fromV6 = await preauthStatus(`https://[::1]:${PROXY_PORT}`, probeHost) + record( + 'pre-auth: one client past the limit gets 429, another client behind the same ingress is still admitted', + fromV4.includes(426) && fromV4.at(-1) === 429 && fromV6 === 426, + `v4 first=${fromV4[0]} limitedFrom=${fromV4.indexOf(429)} last=${fromV4.at(-1)}; v6=${fromV6}` + ) + + // 5. Proxy restart: graceful (SIGTERM drain), crash (SIGKILL), and crash with the box down 6 s + // (a VM replacement is slower still). Desktop reconnects with the shipped retry schedule; the + // phone retries every 500 ms. + let current = main + for (const [mode, downMs] of [['SIGTERM', 0], ['SIGKILL', 0], ['SIGKILL', 6000]] as const) { + const live = await phoneHello(INGRESS, id, resumeToken) + const hadPhone = live.state.hello?.ok === true + const closesBefore = current.events.closes.length + const killedAt = Date.now() + proxy.child.kill(mode) + await waitFor(() => current.events.closes.length > closesBefore && (!hadPhone || live.state.closed !== null), 10_000) + const desktopSaw = current.events.closes.at(-1) + const cutAfter = Date.now() - killedAt + console.log(` ${mode}: desktop saw ${desktopSaw}, phone saw ${live.state.closed} at +${cutAfter} ms`) + await exited(proxy.child) + // Dials made while the box is down fail fast (connection refused) and back off. + const schedule = new RelayRetrySchedule() + let attempts = 0 + const back = new Promise<{ client: ReturnType; ms: number }>((resolve, reject) => { + const attempt = () => { + if (Date.now() - killedAt > 90_000) { return reject(new Error(`desktop did not recover after ${attempts} dials`)) } + attempts++ + const next = desktop(INGRESS, host, jwt) + next.connect().then( + () => resolve({ client: next, ms: Date.now() - killedAt }), + () => schedule.schedule(0, attempt) + ) + } + schedule.schedule(0, attempt) + }) + await sleep(downMs) + const restartAt = Date.now() + proxy = startProxy(PROXY_PORT, SECRET) + const readyIn = await ready(`${INGRESS}/health`) + const recovered = await back + current.client.closeNow() + current = recovered.client + let phoneBack = -1 + for (let i = 0; i < 40 && phoneBack < 0; i++) { + const p = await phoneHello(INGRESS, id, resumeToken) + if (p.state.hello?.ok === true) { + phoneBack = Date.now() - killedAt + p.ws.close() + } else { await sleep(500) } + } + record( + `proxy ${mode}${downMs ? ` + ${downMs / 1000} s down` : ''}: clients see a close, then recover through the restarted proxy`, + desktopSaw !== undefined && recovered.ms > 0 && phoneBack > 0, + `phone open before=${hadPhone}; desktop close=${desktopSaw} phone close=${live.state.closed} at +${cutAfter} ms; proxy started at +${restartAt - killedAt} ms, ready in ${readyIn} ms; desktop active again at +${recovered.ms} ms (${attempts} dials); phone admitted at +${phoneBack} ms` + ) + } + current.client.closeNow() + return proxy +} diff --git a/cloud/apps/relay-edge/test/ingress.test.ts b/cloud/apps/relay-edge/test/ingress.test.ts new file mode 100644 index 00000000000..ed0958da930 --- /dev/null +++ b/cloud/apps/relay-edge/test/ingress.test.ts @@ -0,0 +1,265 @@ +import { describe, expect, it, vi } from 'vitest' +import { INGRESS_HEADER, signIngressHeader, verifyIngressHeader } from '../../../packages/relay-contract/src/ingress-header' +import type { Env } from '../src/env' +import { inCidrs, parseCidr, parseIp } from '../src/ingress' +import { requestPlace } from '../src/metrics' + +vi.mock('cloudflare:workers', () => ({ + DurableObject: class { + constructor( + readonly ctx: unknown, + readonly env: unknown + ) {} + } +})) +const { default: worker, HostRelay } = await import('../src/index') +const { edgeConfig } = await import('../src/env') + +const SECRET = 'a'.repeat(40) +const OLD_SECRET = 'b'.repeat(40) +const CLAIM = { name: 'cn', origin: 'https://cn.relay.example', clientIp: '198.51.100.7' } +const HOST = 'AbCdEfGhIjKlMnOp' + +describe('ingress header', () => { + it('round-trips, and accepts a secret being rotated out', async () => { + expect(await verifyIngressHeader(await signIngressHeader(SECRET, CLAIM), [SECRET])).toEqual(CLAIM) + expect(await verifyIngressHeader(await signIngressHeader(OLD_SECRET, CLAIM), [SECRET, OLD_SECRET])).toEqual(CLAIM) + }) + + it('refuses a wrong secret, any edited field, skew past the window, and junk', async () => { + const now = 1_800_000_000_000 + const header = await signIngressHeader(SECRET, CLAIM, now) + expect(await verifyIngressHeader(header, [OLD_SECRET], now)).toBeNull() + expect(await verifyIngressHeader(header, [], now)).toBeNull() + expect(await verifyIngressHeader(header.replace('198.51.100.7', '198.51.100.8'), [SECRET], now)).toBeNull() + expect(await verifyIngressHeader(header.replace('https://cn.', 'https://xx.'), [SECRET], now)).toBeNull() + expect(await verifyIngressHeader(header, [SECRET], now + 121_000)).toBeNull() + expect(await verifyIngressHeader(header, [SECRET], now - 121_000)).toBeNull() + expect(await verifyIngressHeader(header, [SECRET], now + 119_000)).toEqual(CLAIM) + for (const junk of ['', 'v1', `${header} extra`, header.replace('v1', 'v2'), header.slice(0, -1)]) { + expect(await verifyIngressHeader(junk, [SECRET], now)).toBeNull() + } + const path = await signIngressHeader(SECRET, { ...CLAIM, origin: 'https://cn.relay.example/x' }, now) + expect(await verifyIngressHeader(path, [SECRET], now)).toBeNull() + }) +}) + +// Drives the Worker's router with stub objects and records what reaches them. +function harness(extra: Partial = {}) { + const seen: { limiterKeys: string[]; forwarded: Record[]; statuses: number[] } = { limiterKeys: [], forwarded: [], statuses: [] } + // Metrics as ':