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 '<continent>/<colo>' and, only behind a verified ingress, appends
'/<country>/<ingress name>'.

No user effect until the ingress vars/secret are set or the director calls verify.
This commit is contained in:
Neil
2026-11-14 18:14:55 -08:00
committed by GitHub
parent 7ff5252b03
commit d3e6f3270f
32 changed files with 2333 additions and 33 deletions
+5 -1
View File
@@ -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",
@@ -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 "<deployed>" # 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<string[]> {
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<void> {
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
})
@@ -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 <a hostname in the zone, e.g. the edge's>
// 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 <hostname in the zone>') }
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<T> = { success: boolean; result: T; errors?: { code: number; message: string }[] }
async function get<T>(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<T> | 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<void> {
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<Record<string, unknown>>(`${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<string, unknown>; action_parameters?: Record<string, unknown> }[] }>(`${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<string, number> = {}
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<unknown[]>(`${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
})
+17 -1
View File
@@ -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<typeof AssignBodySchema>
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<typeof VerifyBodySchema>
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<string> {
+60 -1
View File
@@ -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<HostRelay>
@@ -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<object, EdgeConfig>()
@@ -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<EdgeConfig, 'ingressSecrets' | 'ingressToken' | 'ingressProblems'> {
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 }
}
+13 -1
View File
@@ -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<VerifyResult | ReplayRefusal> {
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 } {
+38 -15
View File
@@ -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<string, string>,
target: 'host' | 'splice' = 'host',
@@ -29,8 +32,9 @@ function forward(
): Promise<Response> {
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)
}
+100
View File
@@ -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 <name> <origin>
// <token>` 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<Ingress | null> {
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<boolean> {
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<string>(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)
})
}
+14 -4
View File
@@ -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 '<continent>/<colo>' (e.g. 'AS/HKG'), '' where no client is involved. One index per event for sampling.
// client's ingress '<continent>/<colo>' (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
}
+13 -7
View File
@@ -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<boolean> {
export async function admitPreAuth(
env: Env,
request: Request,
kind: 'phone' | 'data' | 'health',
ingress?: Ingress | null
): Promise<boolean> {
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
+24
View File
@@ -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<VerifyResult> {
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 }
}
@@ -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<string, string> = 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, '<html>blocked</html>'))).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' }, '<html>'))).toBe('cf-zone')
expect(classify(a(429, cf, 'error code: 1015'))).toBe('cf-zone')
expect(classify(a(403, cf, '<title>Attention Required! | Cloudflare</title>'))).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')
})
})
@@ -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<string, string | string[] | undefined>; 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'
}
+306
View File
@@ -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://<front door hostname> G1_EDGE=https://<staging edge hostname> \
// G1_ADMIN_KEY_FILE=<file with the edge admin signing key> G1_ADMIN_ISSUER=<edge 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<void> {
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<void> {
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<string, string> = {}): Promise<Answer> {
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<T>(count: number, concurrency: number, task: (i: number) => Promise<T>): Promise<T[]> {
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<string, number> {
const counts: Record<string, number> = {}
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<void> {
// 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)
})
+131
View File
@@ -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<void> {
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<void>((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<number>((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)
}
}
+287
View File
@@ -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<number> {
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<void>((r) => (child.exitCode !== null || child.signalCode !== null ? r() : child.once('exit', () => r())))
async function main(): Promise<void> {
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<void>((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<typeof startProxy>): Promise<ReturnType<typeof startProxy>> {
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<typeof desktop>; 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
}
+265
View File
@@ -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<Env> = {}) {
const seen: { limiterKeys: string[]; forwarded: Record<string, string>[]; statuses: number[] } = { limiterKeys: [], forwarded: [], statuses: [] }
// Metrics as '<event>:<label>' (kept out of `seen`, which compares requests with today's).
const metrics: string[] = []
const capture = async (request: Request) => {
seen.forwarded.push(Object.fromEntries(request.headers))
return new Response(null, { status: 204 })
}
const namespace = (stub: object) => ({ idFromName: (name: string) => name, get: () => stub })
const env = {
EDGE_BACKEND: 'spike',
SPIKE_DEBUG: '1',
RELAY_JWKS: '{"keys":[]}',
RELAY_JWT_AUDIENCE: 'orca-relay',
PING_INTERVAL_MS: '30000',
SILENCE_TIMEOUT_MS: '75000',
ORPHAN_GRACE_MS: '30000',
HOST_RELAY: namespace({ fetch: capture, allocateConnId: async (id: string) => `${id}.0.conn` }),
SPLICE_RELAY: namespace({ fetch: capture }),
PREAUTH_LIMITER: { limit: async ({ key }: { key: string }) => (seen.limiterKeys.push(key), { success: true }) },
METRICS: { writeDataPoint: ({ blobs }: { blobs: string[] }) => metrics.push(`${blobs[0]}:${blobs[1]}`) },
...extra
} as unknown as Env
const send = async (headers: Record<string, string> = {}) => {
const base = { 'cf-connecting-ip': '203.0.113.1', upgrade: 'websocket', ...headers }
for (const path of [`/v1/connect/${HOST}`, `/__spike/host/${HOST}`]) {
seen.statuses.push((await worker.fetch(new Request(`https://edge.relay.example${path}`, { headers: base }), env)).status)
}
return seen
}
return { send, metrics, env }
}
describe('worker behind an ingress', () => {
it('without the header, configuring a secret changes nothing', async () => {
const today = await harness().send()
expect(today.limiterKeys).toEqual(['phone:203.0.113.1'])
expect(today.forwarded.map((h) => [h['x-edge-origin'], h['x-edge-place']])).toEqual([
['https://edge.relay.example', 'local'],
['https://edge.relay.example', 'local']
])
expect(await harness({ INGRESS_SECRETS: SECRET }).send()).toEqual(today)
})
it('a header that fails to verify is dropped and the request is treated as direct', async () => {
const today = await harness().send()
const forged = { [INGRESS_HEADER]: await signIngressHeader(OLD_SECRET, CLAIM) }
expect(await harness({ INGRESS_SECRETS: SECRET }).send(forged)).toEqual(today)
// No secret configured: even a validly signed header is ignored.
expect(await harness().send({ [INGRESS_HEADER]: await signIngressHeader(SECRET, CLAIM) })).toEqual(today)
})
it('a verified header supplies the relay origin, the pre-auth source and the place label', async () => {
const seen = await harness({ INGRESS_SECRETS: `${SECRET},${OLD_SECRET}` }).send({ [INGRESS_HEADER]: await signIngressHeader(SECRET, CLAIM) })
expect(seen.limiterKeys).toEqual(['phone:198.51.100.7'])
for (const headers of seen.forwarded) {
expect(headers['x-edge-origin']).toBe('https://cn.relay.example')
expect(headers['x-edge-place']).toBe('local/cn')
expect(headers[INGRESS_HEADER]).toBeUndefined()
}
})
it('a short secret switches the signed mode off, loudly, and the edge keeps serving', async () => {
const today = await harness().send()
const errors = vi.spyOn(console, 'error').mockImplementation(() => {})
const h = harness({ INGRESS_SECRETS: `${SECRET},short` })
expect(await h.send({ [INGRESS_HEADER]: await signIngressHeader(SECRET, CLAIM) })).toEqual(today)
expect(h.metrics).toContain('ingress_config_invalid:secret_too_short')
expect(h.metrics).toContain('ingress_refused:')
expect(JSON.stringify(errors.mock.calls)).toContain('relay_edge_ingress_config_invalid')
errors.mockRestore()
})
})
describe('token-mode ingress (a load balancer that cannot sign)', () => {
const TOKEN = 't'.repeat(40)
const tokenEnv = { INGRESS_TOKENS: `${TOKEN},${'u'.repeat(40)}`, INGRESS_ORIGINS: 'https://cn.relay.example', INGRESS_SOURCE_CIDRS: '34.96.0.0/20, 34.127.192.0/18, 2001:db8:a::/48' }
const lb = (over: Record<string, string> = {}) => ({
[INGRESS_HEADER]: `v1s cn https://cn.relay.example ${TOKEN}`,
'cf-connecting-ip': '34.96.3.9',
'x-orca-client-ip': '198.51.100.7',
'x-orca-client-country': 'CN',
...over
})
it('trusts the header only with token, allowlisted origin and an egress source; strips it all before objects', async () => {
const seen = await harness(tokenEnv).send(lb())
expect(seen.limiterKeys).toEqual(['phone:198.51.100.7'])
for (const headers of seen.forwarded) {
expect(headers['x-edge-origin']).toBe('https://cn.relay.example')
expect(headers['x-edge-place']).toBe('local/cn')
expect([headers[INGRESS_HEADER], headers['x-orca-client-ip'], headers['x-orca-client-country']]).toEqual([undefined, undefined, undefined])
}
expect((await harness(tokenEnv).send(lb({ 'cf-connecting-ip': '2001:db8:a:ffff::1' }))).limiterKeys).toEqual(['phone:198.51.100.7'])
})
it('from the front door ranges but failing the rule: 403 before anything is keyed or forwarded', async () => {
for (const headers of [
lb({ [INGRESS_HEADER]: `v1s cn https://cn.relay.example ${'x'.repeat(40)}` }),
lb({ [INGRESS_HEADER]: `v1s cn https://cn.relay.example ${TOKEN.slice(1)}` }),
lb({ [INGRESS_HEADER]: `v1s cn https://evil.example ${TOKEN}` }),
lb({ [INGRESS_HEADER]: `v1s cn https://cn.relay.example ${TOKEN} extra` }),
lb({ 'x-orca-client-ip': '' }),
lb({ 'x-orca-client-ip': '198.51.100.7, 203.0.113.5' }),
{ 'cf-connecting-ip': '34.127.200.1' }
]) {
const h = harness(tokenEnv)
const seen = await h.send(headers)
expect(seen.statuses).toEqual([403, 403])
expect(seen.limiterKeys).toEqual([])
expect(seen.forwarded).toEqual([])
expect(h.metrics).toEqual(['ingress_rejected:', 'ingress_rejected:'])
}
})
it('director admin calls are matched before the range rule: no 403 even from inside the ranges', async () => {
const h = harness({ ...tokenEnv, EDGE_BACKEND: 'object', ADMIN_SIGNING_KEY: 'k'.repeat(40), ADMIN_ISSUER: 'https://director.test.invalid' })
for (const op of ['assign', 'export', 'erase']) {
const response = await worker.fetch(new Request(`https://edge.relay.example/__admin/${op}`, { method: 'POST', headers: { 'cf-connecting-ip': '34.96.3.9' }, body: '{}' }), h.env)
// Unsigned, so the admin check refuses it (401); the range rule never ran.
expect(response.status).toBe(401)
}
expect(h.metrics.filter((m) => m.startsWith('ingress_rejected'))).toEqual([])
// Everything else from the ranges still meets the rule.
const spike = await worker.fetch(new Request(`https://edge.relay.example/__spike/host/${HOST}`, { headers: { 'cf-connecting-ip': '34.96.3.9' } }), h.env)
expect(spike.status).toBe(403)
})
it('outside the front door ranges, a token header is ignored and the request is direct, as today', async () => {
const today = await harness().send()
for (const ip of ['203.0.113.1', '34.96.16.1']) {
const seen = await harness(tokenEnv).send(lb({ 'cf-connecting-ip': ip }))
expect(seen.forwarded.map((h) => h['x-edge-origin'])).toEqual(today.forwarded.map((h) => h['x-edge-origin']))
expect(seen.forwarded.map((h) => h['x-edge-place'])).toEqual(['local', 'local'])
expect(seen.limiterKeys).toEqual([`phone:${ip}`])
}
// Token mode unset: even a perfect header from the ranges is ignored (no 403 either).
const unset = await harness().send(lb())
expect(unset.statuses).toEqual(today.statuses)
expect(unset.limiterKeys).toEqual(['phone:34.96.3.9'])
// Only the HMAC path is configured: a token header is not accepted by it.
expect((await harness({ INGRESS_SECRETS: SECRET }).send(lb())).limiterKeys).toEqual(['phone:34.96.3.9'])
})
it('a partial or malformed token setup switches token mode off, reports why, and the edge keeps serving', async () => {
const today = await harness().send()
const errors = vi.spyOn(console, 'error').mockImplementation(() => {})
for (const [bad, problem] of [
[{ INGRESS_TOKENS: TOKEN }, 'token_mode_partial'],
[{ ...tokenEnv, INGRESS_ORIGINS: '' }, 'token_mode_partial'],
[{ ...tokenEnv, INGRESS_TOKENS: 'short' }, 'token_too_short'],
[{ ...tokenEnv, INGRESS_ORIGINS: 'https://cn.relay.example/' }, 'origin_invalid'],
[{ ...tokenEnv, INGRESS_ORIGINS: 'http://cn.relay.example' }, 'origin_invalid'],
[{ ...tokenEnv, INGRESS_SOURCE_CIDRS: '34.96.0.0/33' }, 'cidr_invalid'],
[{ ...tokenEnv, INGRESS_SOURCE_CIDRS: '34.96.0.0/20,0.0.0.0/0' }, 'cidr_too_broad'],
[{ ...tokenEnv, INGRESS_SOURCE_CIDRS: '34.0.0.0/7' }, 'cidr_too_broad'],
[{ ...tokenEnv, INGRESS_SOURCE_CIDRS: '2001::/15' }, 'cidr_too_broad']
] as const) {
const h = harness(bad)
// A front-door request is now treated as direct (no 403: token mode is off).
const seen = await h.send(lb())
expect(seen.statuses).toEqual(today.statuses)
expect(seen.limiterKeys).toEqual(['phone:34.96.3.9'])
expect(h.metrics).toContain(`ingress_config_invalid:${problem}`)
expect(edgeConfig(h.env).ingressToken.tokens).toEqual([])
}
expect(edgeConfig(harness({ ...tokenEnv, INGRESS_SOURCE_CIDRS: '34.0.0.0/8,2001:db8::/16' }).env).ingressProblems).toEqual([])
errors.mockRestore()
})
it("a malformed ingress config does not stop the host object's alarm: connected desktops keep their pings", async () => {
const errors = vi.spyOn(console, 'error').mockImplementation(() => {})
const h = harness({ INGRESS_TOKENS: TOKEN, INGRESS_SECRETS: 'short', INGRESS_SOURCE_CIDRS: 'nonsense' })
const sent: string[] = []
let attachment: Record<string, unknown> = {
k: 'control',
phase: 'active',
claims: { exp: Date.now() + 3_600_000 },
lastPongAt: Date.now(),
lastPingAt: 0
}
const ws = {
deserializeAttachment: () => attachment,
serializeAttachment: (a: Record<string, unknown>) => (attachment = a),
send: (text: string) => sent.push(text)
}
const storage = { get: async () => undefined, put: async () => {}, getAlarm: async () => null, setAlarm: async () => {}, deleteAlarm: async () => {} }
// Outside the Workers runtime: the constructor's keepalive auto-response needs a stand-in.
vi.stubGlobal('WebSocketRequestResponsePair', class {})
const relay = new HostRelay({ getWebSockets: () => [ws], storage, setWebSocketAutoResponse: () => {}, getWebSocketAutoResponseTimestamp: () => null } as never, h.env)
vi.unstubAllGlobals()
await relay.alarm()
expect(sent.map((t) => JSON.parse(t).type)).toEqual(['ping'])
errors.mockRestore()
})
it('places the client country reported by the ingress', () => {
const request = new Request('https://edge.relay.example/v1/connect/x')
Object.defineProperty(request, 'cf', { value: { continent: 'EU', colo: 'AMS', country: 'NL' } })
expect(requestPlace(request, { name: 'cn', country: 'CN' })).toBe('EU/AMS/CN/cn')
expect(requestPlace(request, { name: 'cn' })).toBe('EU/AMS/NL/cn')
})
})
describe('address parsing', () => {
it('parses IPv4/IPv6 and matches CIDRs by family', () => {
const cidrs = ['34.96.0.0/20', '2001:db8::/32', '10.0.0.1/32'].map((c) => parseCidr(c)!)
expect(inCidrs('34.96.15.255', cidrs)).toBe(true)
expect(inCidrs('34.96.16.0', cidrs)).toBe(false)
expect(inCidrs('10.0.0.1', cidrs)).toBe(true)
expect(inCidrs('2001:DB8:ffff::1', cidrs)).toBe(true)
expect(inCidrs('2001:db9::1', cidrs)).toBe(false)
expect(inCidrs('::ffff:34.96.0.1', cidrs)).toBe(false)
for (const bad of ['', '1.2.3', '256.1.1.1', '1::2::3', '1:2:3:4:5:6:7:8:9', 'g::1', '1:2:3:4:5:6:7::8']) { expect(parseIp(bad)).toBeNull() }
expect(parseIp('::')).toEqual({ v6: true, value: 0n })
expect(parseCidr('34.96.0.0/x')).toBeNull()
})
})
+167
View File
@@ -0,0 +1,167 @@
// Shared by the local ingress end-to-end run and the G1 front-door check: the SHIPPED desktop
// control client and data transport, phone-like sockets, and the director's signed admin push.
import { createHash } from 'node:crypto'
import { get as httpsGet } from 'node:https'
import { SignJWT, type KeyLike } from 'jose'
import nacl from 'tweetnacl'
import WebSocket from 'ws'
import { RelayControlClient } from '../../../../src/main/runtime/mobile-cloud-relay/control-channel/relay-control-client'
import { CloudRelayTransport } from '../../../../src/main/runtime/rpc/transports/relay-transport'
export const sleep = (ms: number) => new Promise((r) => setTimeout(r, ms))
export const sha = (s: string) => createHash('sha256').update(s).digest('base64url')
export async function waitFor(cond: () => boolean, ms: number): Promise<boolean> {
const end = Date.now() + ms
while (Date.now() < end) {
if (cond()) { return true }
await sleep(10)
}
return cond()
}
export type Host = { relayHostId: string; keypair: { publicKey: Uint8Array; secretKey: Uint8Array; publicKeyB64: string } }
export function newHost(): Host {
const box = nacl.box.keyPair()
return {
relayHostId: createHash('sha256').update(box.publicKey).digest('base64url').slice(0, 16),
keypair: { publicKey: box.publicKey, secretKey: box.secretKey, publicKeyB64: Buffer.from(box.publicKey).toString('base64') }
}
}
export type Account = { user: string; epoch: number; jwtIssuer: string; jwtKid: string }
export function mint(account: Account, key: KeyLike, relayHostId: string): Promise<string> {
return new SignJWT({ purpose: 'host-control', prof: `prof-${account.user}`, relayHostId })
.setProtectedHeader({ alg: 'ES256', kid: account.jwtKid })
.setSubject(account.user)
.setIssuer(account.jwtIssuer)
.setAudience('orca-relay')
.setIssuedAt()
.setExpirationTime('30m')
.sign(key)
}
// The desktop: shipped control client + shipped data transport, echoing every phone frame.
// `dialOrigin` sends the control socket somewhere other than `cellUrl` (negative controls), and
// `extraHeaders` adds headers to it. `events.serverPings` counts the edge's control pings.
export function desktop(
account: Account,
cellUrl: string,
host: Host,
jwt: string,
opts: { dialOrigin?: string; extraHeaders?: Record<string, string>; appVersion?: string } = {}
) {
const events = { closes: [] as number[], lastConnId: '', serverPings: 0, socketPongs: 0 }
let transport: CloudRelayTransport | null = null
const client = new RelayControlClient({
cellUrl,
relayJwt: jwt,
relayHostId: host.relayHostId,
assignmentEpoch: account.epoch,
identity: { userId: account.user, profileId: `prof-${account.user}`, organizationId: '' },
keypair: host.keypair as never,
appVersion: opts.appVersion ?? 'ingress-e2e',
onConnectionOpen: (m) => {
events.lastConnId = m.connId
void transport?.openConnection(m).catch(() => {})
},
onDrain: () => {},
onClose: (code) => events.closes.push(code),
createSocket: (url, token) => {
const target = opts.dialOrigin ? url.replace(new URL(url).origin, opts.dialOrigin.replace(/^http/, 'ws')) : url
const ws = new WebSocket(target, {
headers: { authorization: `Bearer ${token}`, 'x-orca-host-capabilities': 'pending-conn-details', ...opts.extraHeaders },
perMessageDeflate: false,
maxPayload: 1024 * 1024 + 1024
})
ws.on('message', (raw) => {
if (raw.toString().startsWith('{"type":"ping"')) { events.serverPings++ }
})
ws.on('pong', () => events.socketPongs++)
return ws
}
})
return {
client,
events,
async connect() {
const ack = await client.connect()
transport = new CloudRelayTransport({ cellUrl, relayHostId: host.relayHostId, generation: ack.generation })
transport.onMessage(((m: string | Uint8Array, reply: (r: string | Uint8Array) => void) => reply(m)) as never)
return ack
}
}
}
// A phone-like socket: auth frame, then the relay hello, then raw frames.
export function phone(origin: string, relayHostId: string, credential: string) {
const ws = new WebSocket(`${origin.replace(/^http/, 'ws')}/v1/connect/${relayHostId}`, { perMessageDeflate: false, maxPayload: 8 * 1024 * 1024 })
const state = { hello: null as Record<string, unknown> | null, frames: [] as Buffer[], closed: null as number | null, closedAt: 0, reason: '' }
ws.on('open', () => ws.send(JSON.stringify({ type: 'relay-auth', v: 1, mode: 'connect', credential })))
ws.on('message', (raw) => {
if (state.hello === null) { state.hello = JSON.parse(raw.toString()) }
else { state.frames.push(raw as Buffer) }
})
ws.on('close', (code, reason) => {
state.closed = code
state.closedAt = Date.now()
state.reason = reason.toString()
})
ws.on('error', () => {})
return { ws, state }
}
export async function phoneHello(origin: string, relayHostId: string, credential: string) {
const p = phone(origin, relayHostId, credential)
await waitFor(() => p.state.hello !== null || p.state.closed !== null, 8000)
return p
}
// The director's push. It talks to the edge directly; front doors refuse admin paths.
export async function adminAssign(
edge: string,
admin: { key: string; issuer: string },
account: Account,
relayHostId: string,
devices: unknown[] = []
) {
const raw = JSON.stringify({ userId: account.user, relayHostId, cellId: 'edge', assignmentEpoch: account.epoch, devices })
const token = await new SignJWT({ purpose: 'edge-admin', op: 'assign', bodySha256: sha(raw) })
.setProtectedHeader({ alg: 'HS256' })
.setIssuer(admin.issuer)
.setAudience('orca-relay-edge-admin')
.setIssuedAt()
.setExpirationTime('1m')
.sign(new TextEncoder().encode(admin.key))
const r = await fetch(`${edge}/__admin/assign`, { method: 'POST', headers: { authorization: `Bearer ${token}`, 'content-type': 'application/json' }, body: raw })
return { status: r.status, ...((await r.json().catch(() => ({}))) as { outcome?: string }) }
}
// Pre-auth probe: a plain GET to the phone route is admitted (426) or limited (429).
export function preauthStatus(origin: string, relayHostId: string): Promise<number> {
return new Promise((resolve) => {
httpsGet(`${origin}/v1/connect/${relayHostId}`, (res) => {
res.resume()
resolve(res.statusCode ?? 0)
}).on('error', () => resolve(-1))
})
}
// A control socket with no usable JWT is refused before the upgrade.
export function unauthenticatedControl(origin: string): Promise<number> {
return new Promise((res) => {
const ws = new WebSocket(`${origin.replace(/^http/, 'ws')}/v1/host/control`)
ws.on('unexpected-response', (_q, r) => res(r.statusCode ?? 0))
ws.on('error', () => res(-1))
})
}
// A control frame over 1 MiB closes 1009 (the JWT is valid; the frame is checked before the hello).
export function oversizeControl(origin: string, jwt: string): Promise<number> {
const big = new WebSocket(`${origin.replace(/^http/, 'ws')}/v1/host/control`, { headers: { authorization: `Bearer ${jwt}` }, perMessageDeflate: false })
big.on('error', () => {})
return new Promise<number>((res) => {
big.on('open', () => big.send('x'.repeat(1024 * 1024 + 1)))
big.on('close', (c) => res(c))
setTimeout(() => res(-1), 8000)
})
}
+139
View File
@@ -0,0 +1,139 @@
// POST /__admin/verify on the host's own store: the director's check for phone credentials that
// exist only on the edge. Must answer correctly and change nothing.
import { DatabaseSync } from 'node:sqlite'
import { describe, expect, it } from 'vitest'
import { RelayCredentialStore, hashCredential } from '../../relay/src/credentials/credential-store'
import type { DeviceRow, VerifyBody } from '../src/admin'
import { VerifyBodySchema } from '../src/admin'
import { ObjectStore } from '../src/object-store'
import { verifyCredential } from '../src/verify'
import { nodeObjectSql } from './object-sqlite-fake'
const CELL = 'edge'
const HOST = 'edgecohorthost01'
const USER = 'user-1'
const GOOD = 'A'.repeat(43)
const ROTATED = 'C'.repeat(43)
function objectStore(): { store: ObjectStore; db: DatabaseSync } {
const db = new DatabaseSync(':memory:')
const sql = nodeObjectSql(db)
let open = false
const store = new ObjectStore(
sql,
async (operation) => {
if (open) { throw new Error('nested storage transaction') }
open = true
db.exec('BEGIN')
try {
const result = await operation()
db.exec('COMMIT')
return result
} catch (error) {
db.exec('ROLLBACK')
throw error
} finally {
open = false
}
},
CELL
)
return { store, db }
}
function device(id: string, token: string, extra: Partial<DeviceRow> = {}): DeviceRow {
return {
user_id: USER,
relay_host_id: HOST,
relay_device_id: id,
current_hash: hashCredential(token),
current_version: 1,
current_expires_at: Date.now() + 86_400_000,
grace_hash: null,
grace_version: null,
grace_expires_at: null,
revoked_at: null,
updated_at: Date.now(),
...extra
}
}
// Every row of every table, so a test can prove the check wrote nothing.
function snapshot(db: DatabaseSync): string {
const tables = db.prepare(`SELECT name FROM sqlite_master WHERE type = 'table' ORDER BY name`).all() as { name: string }[]
return JSON.stringify(tables.map(({ name }) => [name, db.prepare(`SELECT * FROM "${name}"`).all()]))
}
const body = (token: string, kind: VerifyBody['kind'] = 'resume'): VerifyBody => ({ relayHostId: HOST, kind, tokenHash: hashCredential(token) })
describe('edge verify (director phone credential check)', () => {
it('verifies a device the edge holds (paired here, or its grace token after a rotation)', async () => {
const { store } = objectStore()
await store.applyAssignment({ userId: USER, relayHostId: HOST, cellId: CELL, assignmentEpoch: 3, devices: [] })
const sql = store.database
await sql.query(
`INSERT INTO relay_devices (user_id, relay_host_id, relay_device_id, current_hash, current_version, current_expires_at,
grace_hash, grace_version, grace_expires_at, revoked_at, updated_at) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)`,
Object.values(device('phone-edge', ROTATED, { grace_hash: hashCredential(GOOD), grace_version: 1, current_version: 2, grace_expires_at: Date.now() + 60_000 }))
)
expect(await verifyCredential(store, CELL, body(ROTATED))).toEqual({ outcome: 'verified', userId: USER, relayDeviceId: 'phone-edge', assignmentEpoch: 3 })
expect(await verifyCredential(store, CELL, body(GOOD))).toMatchObject({ outcome: 'verified', relayDeviceId: 'phone-edge' })
})
it('refuses unknown, revoked and expired credentials', async () => {
const { store } = objectStore()
await store.applyAssignment({
userId: USER,
relayHostId: HOST,
cellId: CELL,
assignmentEpoch: 1,
devices: [device('phone-revoked', GOOD, { revoked_at: Date.now() - 1 }), device('phone-expired', ROTATED, { current_expires_at: Date.now() - 1 })]
})
expect(await verifyCredential(store, CELL, body('B'.repeat(43)))).toEqual({ outcome: 'none' })
expect(await verifyCredential(store, CELL, body(GOOD))).toEqual({ outcome: 'none' })
expect(await verifyCredential(store, CELL, body(ROTATED))).toEqual({ outcome: 'none' })
})
it('verifies an invite created on the edge, and only as an invite', async () => {
const { store } = objectStore()
await store.applyAssignment({ userId: USER, relayHostId: HOST, cellId: CELL, assignmentEpoch: 2, devices: [] })
const invite = await new RelayCredentialStore(store.database).createInvite({ userId: USER, relayHostId: HOST }, 'phone-new')
expect(await verifyCredential(store, CELL, body(invite.inviteToken, 'invite'))).toEqual({
outcome: 'verified',
userId: USER,
relayDeviceId: 'phone-new',
assignmentEpoch: 2
})
expect(await verifyCredential(store, CELL, body(invite.inviteToken, 'resume'))).toEqual({ outcome: 'none' })
})
it('answers not_here when the object does not hold the host here (never placed, moved away, frozen)', async () => {
const { store } = objectStore()
expect(await verifyCredential(store, CELL, body(GOOD))).toEqual({ outcome: 'not_here' })
await store.applyAssignment({ userId: USER, relayHostId: HOST, cellId: CELL, assignmentEpoch: 1, devices: [device('phone-a', GOOD)] })
expect(await verifyCredential(store, CELL, { ...body(GOOD), relayHostId: 'otherhost0000001' })).toEqual({ outcome: 'not_here' })
store.freezeForExport()
expect(await verifyCredential(store, CELL, body(GOOD))).toEqual({ outcome: 'not_here' })
await store.applyAssignment({ userId: USER, relayHostId: HOST, cellId: 'us-c1', assignmentEpoch: 2 })
expect(await verifyCredential(store, CELL, body(GOOD))).toEqual({ outcome: 'not_here' })
})
it('changes nothing in the object, whatever the answer', async () => {
const { store, db } = objectStore()
await store.applyAssignment({ userId: USER, relayHostId: HOST, cellId: CELL, assignmentEpoch: 1, devices: [device('phone-a', GOOD)] })
const invite = await new RelayCredentialStore(store.database).createInvite({ userId: USER, relayHostId: HOST }, 'phone-b')
const before = snapshot(db)
await verifyCredential(store, CELL, body(GOOD))
await verifyCredential(store, CELL, body('B'.repeat(43)))
await verifyCredential(store, CELL, body(invite.inviteToken, 'invite'))
expect(snapshot(db)).toBe(before)
expect(store.frozen()).toBe(false)
})
it('accepts only a hash-shaped body, never a raw token or extra fields', () => {
expect(VerifyBodySchema.safeParse(body(GOOD)).success).toBe(true)
expect(VerifyBodySchema.safeParse({ ...body(GOOD), tokenHash: `${hashCredential(GOOD)}=` }).success).toBe(false)
expect(VerifyBodySchema.safeParse({ ...body(GOOD), token: GOOD }).success).toBe(false)
expect(VerifyBodySchema.safeParse({ ...body(GOOD), kind: 'device' }).success).toBe(false)
})
})
+33
View File
@@ -0,0 +1,33 @@
# Relay ingress proxy (src/proxy.ts). Build from cloud/: docker build -f apps/relay-ingress/Dockerfile .
# Runtime config: see src/index.ts. Certificate, key and secret are mounted files, never baked in.
FROM node:24.22.2-alpine@sha256:8a32c8cf649a49915a66e81e7a609223c708d0adabf288ee3f4077d765542f85 AS build
WORKDIR /app
ENV PATH="/opt/pnpm/node_modules/.bin:$PATH"
COPY package.json pnpm-lock.yaml pnpm-workspace.yaml tsconfig.base.json ./
RUN npm install --prefix /opt/pnpm --ignore-scripts --no-audit --no-fund "$(node -p 'require("./package.json").packageManager.split("+")[0]')" \
&& npm audit signatures --prefix /opt/pnpm
COPY packages/relay-contract/package.json packages/relay-contract/package.json
COPY apps/relay-ingress/package.json apps/relay-ingress/package.json
RUN pnpm install --frozen-lockfile --filter @orca-cloud/relay-ingress...
COPY packages/relay-contract packages/relay-contract
COPY apps/relay-ingress apps/relay-ingress
RUN pnpm --filter @orca-cloud/relay-contract build && pnpm --filter @orca-cloud/relay-ingress build
FROM node:24.22.2-alpine@sha256:8a32c8cf649a49915a66e81e7a609223c708d0adabf288ee3f4077d765542f85 AS runtime
ENV NODE_ENV=production
WORKDIR /app
RUN apk upgrade --no-cache
# The proxy has no runtime dependencies beyond the contract's built output.
COPY --from=build /app/packages/relay-contract/package.json packages/relay-contract/package.json
COPY --from=build /app/packages/relay-contract/dist packages/relay-contract/dist
COPY --from=build /app/apps/relay-ingress/package.json apps/relay-ingress/package.json
COPY --from=build /app/apps/relay-ingress/dist apps/relay-ingress/dist
RUN mkdir -p apps/relay-ingress/node_modules/@orca-cloud \
&& ln -s ../../../../packages/relay-contract apps/relay-ingress/node_modules/@orca-cloud/relay-contract \
&& rm -f /usr/local/bin/npm /usr/local/bin/npx /usr/local/bin/corepack /usr/local/bin/yarn /usr/local/bin/yarnpkg
USER node
# 443 needs CAP_NET_BIND_SERVICE as non-root; the default listens on 8443 behind the load balancer.
ENV PORT=8443 HEALTH_PORT=8081 HEALTH_HOST=0.0.0.0
EXPOSE 8443 8081
STOPSIGNAL SIGTERM
CMD ["node", "apps/relay-ingress/dist/index.js"]
+27
View File
@@ -0,0 +1,27 @@
{
"name": "@orca-cloud/relay-ingress",
"private": true,
"version": "0.0.0",
"type": "module",
"main": "dist/index.js",
"scripts": {
"build": "pnpm clean && tsc -p tsconfig.build.json",
"clean": "node -e \"require('fs').rmSync('dist', { recursive: true, force: true })\"",
"lint": "tsc -p tsconfig.json --noEmit",
"pretest": "pnpm --filter @orca-cloud/relay-contract build",
"start": "node dist/index.js",
"test": "vitest run",
"typecheck": "tsc -p tsconfig.json --noEmit"
},
"dependencies": {
"@orca-cloud/relay-contract": "workspace:*"
},
"devDependencies": {
"@types/node": "^24.19.2",
"@types/ws": "^8.18.2",
"tsx": "^4.23.15",
"typescript": "^7.0.2",
"vitest": "^5.0.5",
"ws": "^8.22.2"
}
}
+61
View File
@@ -0,0 +1,61 @@
// Ingress proxy process: config from the environment, files for the secret and certificate.
// INGRESS_NAME (metric label, e.g. cn), INGRESS_ORIGIN (https://cn.relay.<zone>), UPSTREAM_URL
// (the edge Worker's https origin), INGRESS_SECRET_FILE (shared with the edge's INGRESS_SECRETS),
// TLS_CERT_FILE / TLS_KEY_FILE, PORT (443), HOST (::), HEALTH_PORT (8081, GET /healthz, /stats),
// HEALTH_HOST (127.0.0.1), IDLE_TIMEOUT_MS (300000, must exceed 120000), DRAIN_DELAY_MS (10000),
// DRAIN_MS (30000), UPSTREAM_CA_FILE (local tests only: replaces the system trust store).
// SIGTERM/SIGINT: drain (health 503, stop accepting, cut what is left after DRAIN_MS).
// SIGHUP: reload the certificate files (run by the certificate renewal hook).
import { readFileSync } from 'node:fs'
import { createIngressProxy } from './proxy.js'
function required(name: string): string {
const value = process.env[name]
if (!value) { throw new Error(`relay_ingress_config_${name.toLowerCase()}`) }
return value
}
const int = (name: string, fallback: number) => {
const value = Number(process.env[name] ?? fallback)
if (!Number.isSafeInteger(value) || value < 0) { throw new Error(`relay_ingress_config_${name.toLowerCase()}`) }
return value
}
const readTls = () => ({ key: readFileSync(required('TLS_KEY_FILE')), cert: readFileSync(required('TLS_CERT_FILE')) })
const publicOrigin = new URL(required('INGRESS_ORIGIN')).origin
const upstream = new URL(required('UPSTREAM_URL'))
const secret = readFileSync(required('INGRESS_SECRET_FILE'), 'utf8').trim()
if (secret.length < 32) { throw new Error('relay_ingress_config_secret') }
if (upstream.protocol !== 'https:') { throw new Error('relay_ingress_config_upstream_url') }
const idleTimeoutMs = int('IDLE_TIMEOUT_MS', 300_000)
// Edge control pings arrive every 30 s, phones may idle longer; 120 s is the floor.
if (idleTimeoutMs <= 120_000) { throw new Error('relay_ingress_config_idle_timeout_ms') }
const proxy = createIngressProxy({
name: required('INGRESS_NAME'),
publicOrigin,
upstream,
secret,
upstreamCa: process.env.UPSTREAM_CA_FILE ? readFileSync(process.env.UPSTREAM_CA_FILE) : undefined,
tls: readTls(),
idleTimeoutMs,
drainDelayMs: int('DRAIN_DELAY_MS', 10_000),
drainMs: int('DRAIN_MS', 30_000)
})
const port = int('PORT', 443)
const healthPort = int('HEALTH_PORT', 8081)
proxy.server.listen(port, process.env.HOST ?? '::', () => proxy.log({ event: 'listening', port, name: required('INGRESS_NAME') }))
proxy.health.listen(healthPort, process.env.HEALTH_HOST ?? '127.0.0.1')
const statsTimer = setInterval(() => proxy.log({ event: 'stats', ...proxy.stats }), 60_000)
statsTimer.unref()
for (const signal of ['SIGTERM', 'SIGINT'] as const) {
process.once(signal, () => void proxy.drain().then(() => process.exit(0)))
}
process.on('SIGHUP', () => {
try {
proxy.reloadTls(readTls())
proxy.log({ event: 'tls_reloaded' })
} catch {
proxy.log({ event: 'tls_reload_failed' })
}
})
+148
View File
@@ -0,0 +1,148 @@
import { execFileSync } from 'node:child_process'
import { mkdtempSync, readFileSync, rmSync } from 'node:fs'
import type { IncomingHttpHeaders } from 'node:http'
import { createServer, request, type Server } from 'node:https'
import type { AddressInfo } from 'node:net'
import { tmpdir } from 'node:os'
import { join } from 'node:path'
import { afterAll, beforeAll, describe, expect, it } from 'vitest'
import WebSocket, { WebSocketServer } from 'ws'
import { INGRESS_HEADER, verifyIngressHeader } from '@orca-cloud/relay-contract/ingress-header'
import { createIngressProxy } from './proxy.js'
const SECRET = 's'.repeat(40)
const ORIGIN = 'https://cn.relay.example'
const dir = mkdtempSync(join(tmpdir(), 'relay-ingress-test-'))
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 tls = { key: readFileSync(join(dir, 'key.pem')), cert: readFileSync(join(dir, 'cert.pem')) }
const seen: { headers: IncomingHttpHeaders[]; closes: number[] } = { headers: [], closes: [] }
const events: Record<string, unknown>[] = []
let upstream: Server
let proxy: ReturnType<typeof createIngressProxy>
let base = ''
function listen(server: { listen: (port: number, host: string, cb: () => void) => unknown; address: () => unknown }): Promise<number> {
return new Promise((resolve) => server.listen(0, '127.0.0.1', () => resolve((server.address() as AddressInfo).port)))
}
beforeAll(async () => {
// Stands in for the edge Worker: echoes frames, closes on request, refuses unauthenticated upgrades.
upstream = createServer(tls, (req, res) => {
seen.headers.push(req.headers)
res.writeHead(200, { 'content-type': 'text/plain' }).end('edge says hi')
})
const wss = new WebSocketServer({ noServer: true, maxPayload: 16 * 1024 * 1024 })
upstream.on('upgrade', (req, socket, head) => {
seen.headers.push(req.headers)
if (!req.headers.authorization) { return void socket.end('HTTP/1.1 401 Unauthorized\r\ncontent-length: 0\r\n\r\n') }
wss.handleUpgrade(req, socket, head, (ws) => {
ws.on('message', (data, binary) => (data.toString() === 'close-me' ? ws.close(4409, 'host not on this cell') : ws.send(data, { binary })))
ws.on('close', (code) => seen.closes.push(code))
})
})
const port = await listen(upstream)
proxy = createIngressProxy({
name: 'cn',
publicOrigin: ORIGIN,
upstream: new URL(`https://127.0.0.1:${port}`),
upstreamCa: tls.cert,
secret: SECRET,
tls,
idleTimeoutMs: 300_000,
drainDelayMs: 50,
drainMs: 400,
log: (event) => events.push(event)
})
base = `https://localhost:${await listen(proxy.server)}`
await listen(proxy.health)
})
afterAll(() => {
upstream.close()
rmSync(dir, { recursive: true, force: true })
})
function call(method: string, path: string, headers: Record<string, string> = {}): Promise<{ status: number; body: string }> {
return new Promise((resolve, reject) => {
request(`${base}${path}`, { method, ca: tls.cert, headers }, (res) => {
let body = ''
res.on('data', (d) => (body += d))
res.on('end', () => resolve({ status: res.statusCode ?? 0, body }))
}).on('error', reject).end()
})
}
const open = (path: string, headers: Record<string, string>) => {
const ws = new WebSocket(`${base.replace('https', 'wss')}${path}`, { ca: tls.cert, headers, perMessageDeflate: false, maxPayload: 16 * 1024 * 1024 })
ws.on('error', () => {})
return ws
}
describe('ingress proxy', () => {
it('forwards plain requests with a signed header and strips anything a client could forge', async () => {
const forged = { [INGRESS_HEADER]: 'v1 cn https://evil.example 1.2.3.4 0 x', 'cf-connecting-ip': '1.2.3.4', 'x-forwarded-for': '1.2.3.4', 'x-edge-origin': 'https://evil.example' }
const response = await call('GET', '/health', forged)
expect(response).toEqual({ status: 200, body: 'edge says hi' })
const headers = seen.headers.at(-1)!
expect(headers.host).toMatch(/^127\.0\.0\.1:\d+$/)
expect(headers['cf-connecting-ip']).toBeUndefined()
expect(headers['x-forwarded-for']).toBeUndefined()
expect(headers['x-edge-origin']).toBeUndefined()
expect(await verifyIngressHeader(headers[INGRESS_HEADER] as string, [SECRET])).toEqual({ name: 'cn', origin: ORIGIN, clientIp: '127.0.0.1' })
})
it('answers 404 itself for anything outside /v1 and /health (admin and spike routes never reach the edge)', async () => {
const before = seen.headers.length
const { status } = await call('POST', '/__admin/assign')
const ws = open('/__spike/abort/x', { authorization: 'Bearer t' })
const refused = await new Promise<number>((r) => ws.on('unexpected-response', (_q, res) => r(res.statusCode ?? 0)))
expect(refused).toBe(404)
expect(status).toBe(404)
expect(seen.headers.length).toBe(before)
})
it('pipes WebSocket frames (4 MiB) and close codes both ways, unchanged', async () => {
const ws = open('/v1/host/control', { authorization: 'Bearer token-value' })
await new Promise((r) => ws.once('open', r))
expect(seen.headers.at(-1)!.authorization).toBe('Bearer token-value')
const big = Buffer.alloc(4 * 1024 * 1024, 7)
ws.send(big)
const echoed = await new Promise<Buffer>((r) => ws.once('message', (d) => r(d as Buffer)))
expect(echoed.equals(big)).toBe(true)
ws.send('close-me')
const [code, reason] = await new Promise<[number, string]>((r) => ws.once('close', (c, why) => r([c, why.toString()])))
expect([code, reason]).toEqual([4409, 'host not on this cell'])
const mine = open('/v1/connect/AbCdEfGhIjKlMnOp', { authorization: 'Bearer x' })
await new Promise((r) => mine.once('open', r))
mine.close(4001, 'bye')
await new Promise((r) => mine.once('close', r))
await expect.poll(() => seen.closes.at(-1)).toBe(4001)
})
it("passes the edge's refusal of an upgrade through as-is", async () => {
const ws = open('/v1/host/control', {})
expect(await new Promise<number>((r) => ws.on('unexpected-response', (_q, res) => r(res.statusCode ?? 0)))).toBe(401)
})
it('drains: health fails, open tunnels are cut after the drain window, logs carry no address, path or token', async () => {
const ws = open('/v1/host/control', { authorization: 'Bearer secret-token' })
await new Promise((r) => ws.once('open', r))
const healthUrl = `http://127.0.0.1:${(proxy.health.address() as AddressInfo).port}/healthz`
expect((await fetch(healthUrl)).status).toBe(200)
const started = Date.now()
const drained = proxy.drain()
expect((await fetch(healthUrl)).status).toBe(503)
const code = await new Promise<number>((r) => ws.once('close', (c) => r(c)))
await drained
expect(code).toBe(1006)
expect(Date.now() - started).toBeGreaterThanOrEqual(400)
expect(events.map((e) => e.event)).toEqual(['drain_start', 'drain_done'])
expect(events.at(-1)).toMatchObject({ cut: 1 })
expect(proxy.stats.upgrades).toMatchObject({ 'control:101': 2, 'control:401': 1, 'connect:101': 1 })
const logged = JSON.stringify([events, proxy.stats])
for (const leak of ['127.0.0.1', '/v1/', 'Bearer', 'AbCdEfGhIjKlMnOp', 'secret-token']) { expect(logged).not.toContain(leak) }
})
})
+167
View File
@@ -0,0 +1,167 @@
// Stateless ingress for the relay edge: terminates TLS for one public hostname and re-originates
// every request to the edge Worker over TLS, adding the signed ingress header (origin the client
// dialled + client address). WebSocket upgrades become a raw byte pipe once the request head is
// sent, so frames, close codes and the edge's own refusals pass through unchanged.
// Logs carry counts only: never addresses, paths (they hold host ids) or headers (they hold tokens).
import { createServer as createHttpServer, type IncomingMessage, type ServerResponse } from 'node:http'
import { Agent, createServer as createHttpsServer, request as httpsRequest } from 'node:https'
import { isIP, type Socket } from 'node:net'
import { connect as tlsConnect } from 'node:tls'
import { INGRESS_HEADER, signIngressHeader } from '@orca-cloud/relay-contract/ingress-header'
export type IngressConfig = {
name: string
publicOrigin: string
upstream: URL
secret: string
// Extra trust for the upstream certificate (local tests only; production uses the system store).
upstreamCa?: string | Buffer
tls: { key: string | Buffer; cert: string | Buffer }
idleTimeoutMs: number
// SIGTERM: health fails at once; after drainDelayMs (load balancer notices) the listener closes;
// tunnels still open after drainMs more are cut.
drainDelayMs: number
drainMs: number
log?: (event: Record<string, unknown>) => void
}
// Inbound headers a client must never be able to set on the edge hop, and hop-by-hop headers.
const STRIPPED = /^(host|x-orca-ingress.*|x-edge-.*|cf-.*|x-forwarded-.*|forwarded|x-real-ip|true-client-ip)$/i
const HOP = /^(connection|keep-alive|proxy-connection|proxy-authorization|te|trailer|transfer-encoding|upgrade)$/i
const ROUTED = /^\/(health$|v1\/)/
const clientIp = (socket: Socket) => (socket.remoteAddress ?? '').replace(/^::ffff:(?=\d+\.)/, '')
const route = (path: string) => /^\/v1\/(host\/control|connect|host\/data)/.exec(path)?.[1]?.replace('host/', '') ?? 'other'
export function createIngressProxy(config: IngressConfig) {
const log = config.log ?? ((event) => console.log(JSON.stringify({ at: new Date().toISOString(), ...event })))
const upstreamPort = Number(config.upstream.port || 443)
const servername = isIP(config.upstream.hostname) ? undefined : config.upstream.hostname
const agent = new Agent({ keepAlive: true, servername, ca: config.upstreamCa })
const tunnels = new Set<Socket>()
const stats = { tunnels: 0, opened: 0, upstreamErrors: 0, upgrades: {} as Record<string, number>, requests: 0 }
let draining = false
async function outboundHeaders(req: IncomingMessage, keepHop: boolean): Promise<[string, string][]> {
const out: [string, string][] = []
for (let i = 0; i < req.rawHeaders.length; i += 2) {
const name = req.rawHeaders[i]!
if (STRIPPED.test(name) || (!keepHop && HOP.test(name))) { continue }
out.push([name, req.rawHeaders[i + 1]!])
}
out.push(['host', config.upstream.host])
out.push([INGRESS_HEADER, await signIngressHeader(config.secret, { name: config.name, origin: config.publicOrigin, clientIp: clientIp(req.socket) })])
return out
}
function idle(socket: Socket, onIdle: () => void): void {
socket.setTimeout(config.idleTimeoutMs, onIdle)
socket.setKeepAlive(true, 30_000)
socket.setNoDelay(true)
}
async function onUpgrade(req: IncomingMessage, client: Socket, head: Buffer): Promise<void> {
client.on('error', () => client.destroy())
if (draining || !ROUTED.test(req.url ?? '')) {
client.end(`HTTP/1.1 ${draining ? 503 : 404} ${draining ? 'Service Unavailable' : 'Not Found'}\r\nconnection: close\r\ncontent-length: 0\r\n\r\n`)
return
}
const kind = route(req.url ?? '')
const headers = await outboundHeaders(req, true)
const upstream = tlsConnect({ host: config.upstream.hostname, port: upstreamPort, servername, ca: config.upstreamCa, ALPNProtocols: ['http/1.1'] })
tunnels.add(client)
stats.opened++
stats.tunnels = tunnels.size
const both = () => {
client.destroy()
upstream.destroy()
}
idle(client, both)
idle(upstream, both)
upstream.on('error', () => {
stats.upstreamErrors++
if (!client.writable || client.bytesWritten > 0) { return both() }
client.end('HTTP/1.1 502 Bad Gateway\r\nconnection: close\r\ncontent-length: 0\r\n\r\n')
})
upstream.write(`${req.method} ${req.url} HTTP/1.1\r\n${headers.map(([k, v]) => `${k}: ${v}\r\n`).join('')}\r\n`)
if (head.length) { upstream.write(head) }
upstream.once('data', (chunk: Buffer) => {
const status = `${kind}:${chunk.subarray(9, 12).toString('latin1')}`
stats.upgrades[status] = (stats.upgrades[status] ?? 0) + 1
})
// Byte pipe both ways. A FIN on one side half-closes the other once its writes have flushed.
upstream.pipe(client)
client.pipe(upstream)
client.on('close', () => {
tunnels.delete(client)
stats.tunnels = tunnels.size
upstream.end()
setTimeout(() => upstream.destroy(), 5_000).unref()
})
upstream.on('close', () => {
client.end()
setTimeout(() => client.destroy(), 5_000).unref()
})
}
async function onRequest(req: IncomingMessage, res: ServerResponse): Promise<void> {
if (!ROUTED.test(req.url ?? '')) {
res.writeHead(404, { 'content-length': 0 }).end()
return
}
stats.requests++
const upstream = httpsRequest({
agent,
host: config.upstream.hostname,
port: upstreamPort,
method: req.method,
path: req.url,
headers: (await outboundHeaders(req, false)).flat()
})
upstream.on('response', (up) => {
const headers: string[] = []
for (let i = 0; i < up.rawHeaders.length; i += 2) {
if (!HOP.test(up.rawHeaders[i]!)) { headers.push(up.rawHeaders[i]!, up.rawHeaders[i + 1]!) }
}
res.writeHead(up.statusCode ?? 502, headers)
up.pipe(res)
})
upstream.on('error', () => {
stats.upstreamErrors++
if (!res.headersSent) { res.writeHead(502, { 'content-length': 0 }) }
res.end()
})
req.pipe(upstream)
}
const server = createHttpsServer({ ...config.tls, ALPNProtocols: ['http/1.1'] })
server.keepAliveTimeout = 65_000
server.on('request', (req, res) => void onRequest(req, res).catch(() => res.destroy()))
server.on('upgrade', (req, socket: Socket, head) => void onUpgrade(req, socket, head).catch(() => socket.destroy()))
server.on('tlsClientError', () => {})
// Load balancer health and counters, on a separate plain port that is never public.
const health = createHttpServer((req, res) => {
if (req.url === '/stats') { return void res.writeHead(200, { 'content-type': 'application/json' }).end(JSON.stringify({ draining, ...stats })) }
res.writeHead(draining ? 503 : 200, { 'content-length': 0 }).end()
})
async function drain(): Promise<void> {
if (draining) { return }
draining = true
log({ event: 'drain_start', tunnels: tunnels.size })
await new Promise((r) => setTimeout(r, config.drainDelayMs))
server.close()
server.closeIdleConnections()
const deadline = Date.now() + config.drainMs
while (tunnels.size > 0 && Date.now() < deadline) { await new Promise((r) => setTimeout(r, 200)) }
const cut = tunnels.size
for (const socket of tunnels) { socket.destroy() }
server.closeAllConnections()
health.close()
agent.destroy()
log({ event: 'drain_done', cut })
}
return { server, health, drain, stats, reloadTls: (tls: IngressConfig['tls']) => server.setSecureContext(tls), log }
}
@@ -0,0 +1,10 @@
{
"extends": "./tsconfig.json",
"compilerOptions": {
"declaration": true,
"noEmit": false,
"outDir": "dist",
"rootDir": "src"
},
"exclude": ["src/**/*.test.ts"]
}
+5
View File
@@ -0,0 +1,5 @@
{
"extends": "../../tsconfig.base.json",
"compilerOptions": { "noEmit": true },
"include": ["src/**/*.ts"]
}
@@ -286,7 +286,19 @@ export class RelayCredentialStore {
| { outcome: 'refused' }
| { outcome: 'resume'; reservation: CredentialReservation }
> {
const tokenHash = hashCredential(token)
return await this.lookupResumeHash(relayHostId, hashCredential(token), reservationId)
}
// As lookupResume, from the token's hash (the relay edge is asked with the hash, never the token).
async lookupResumeHash(
relayHostId: string,
tokenHash: string,
reservationId: string = randomUUID()
): Promise<
| { outcome: 'none' }
| { outcome: 'refused' }
| { outcome: 'resume'; reservation: CredentialReservation }
> {
const rows = await this.database.query(
`SELECT * FROM relay_devices
WHERE relay_host_id = ? AND (current_hash = ? OR grace_hash = ?)`,
@@ -687,7 +699,14 @@ export class RelayCredentialStore {
relayHostId: string,
token: string
): Promise<{ userId: string; relayDeviceId: string } | null> {
const found = await this.lookupResume(relayHostId, token)
return await this.resolveResumeHash(relayHostId, hashCredential(token))
}
async resolveResumeHash(
relayHostId: string,
tokenHash: string
): Promise<{ userId: string; relayDeviceId: string } | null> {
const found = await this.lookupResumeHash(relayHostId, tokenHash)
return found.outcome === 'resume'
? { userId: found.reservation.userId, relayDeviceId: found.reservation.relayDeviceId }
: null
@@ -701,7 +720,14 @@ export class RelayCredentialStore {
relayHostId: string,
token: string
): Promise<{ userId: string; relayDeviceId: string } | null> {
const tokenHash = hashCredential(token)
return await this.resolveInviteForMoveHash(relayHostId, hashCredential(token))
}
// Read-only, like resolveResumeHash: the relay edge answers the director's credential check with it.
async resolveInviteForMoveHash(
relayHostId: string,
tokenHash: string
): Promise<{ userId: string; relayDeviceId: string } | null> {
const rows = await this.database.query(
`SELECT user_id, relay_device_id, token_hash, state, attempt_count, max_attempts, expires_at
FROM relay_invites WHERE relay_host_id = ? AND token_hash = ?`,
@@ -81,6 +81,10 @@
"./control-messages": {
"types": "./dist/control-messages.d.ts",
"default": "./dist/control-messages.js"
},
"./ingress-header": {
"types": "./dist/ingress-header.d.ts",
"default": "./dist/ingress-header.js"
}
}
}
@@ -15,3 +15,4 @@ export * from './splice-state-machine.js'
export * from './wire-scalars.js'
export * from './region-correction.js'
export * from './idle-regional-rehome.js'
export * from './ingress-header.js'
@@ -0,0 +1,71 @@
// Signed front-door header. A stateless ingress proxy (one hostname per ingress, e.g. the Hong Kong
// one for mainland China hosts) re-originates requests to the relay edge and adds
// x-orca-ingress: v1 <name> <origin> <client-ip> <unix-seconds> <hmac-sha256 base64url>
// The MAC covers everything before it, under a secret shared by the edge and its ingresses. The
// edge trusts the fields only when the MAC verifies inside the skew window; otherwise it behaves as
// if the header were absent. Both sides use WebCrypto (Node 24 and Workers).
export const INGRESS_HEADER = 'x-orca-ingress'
export const INGRESS_MAX_SKEW_SECONDS = 120
// Secrets shorter than this are a misconfiguration, not a key.
export const INGRESS_SECRET_MIN_LENGTH = 32
export type IngressClaim = {
// Metric label for the ingress, e.g. 'cn'.
name: string
// The origin the client dialled (https://cn.relay.example). The host proof binds this origin.
origin: string
// The client's address as the ingress saw it. Pre-auth limits key on it.
clientIp: string
}
const NAME = /^[a-z0-9-]{1,32}$/
const IP = /^[0-9A-Fa-f:.]{2,45}$/
// Typed from the call so the module compiles against both the Node and the Workers globals.
type HmacKey = ReturnType<typeof crypto.subtle.importKey>
const keys = new Map<string, HmacKey>()
function hmacKey(secret: string): HmacKey {
let key = keys.get(secret)
if (!key) {
key = crypto.subtle.importKey('raw', new TextEncoder().encode(secret), { name: 'HMAC', hash: 'SHA-256' }, false, ['sign', 'verify'])
keys.set(secret, key)
}
return key
}
const toBase64Url = (bytes: ArrayBuffer) =>
btoa(String.fromCharCode(...new Uint8Array(bytes))).replace(/\+/g, '-').replace(/\//g, '_').replace(/=+$/, '')
const fromBase64Url = (text: string) => Uint8Array.from(atob(text.replace(/-/g, '+').replace(/_/g, '/')), (c) => c.charCodeAt(0))
function isOrigin(value: string): boolean {
try {
const url = new URL(value)
return (url.protocol === 'https:' || url.protocol === 'http:') && url.origin === value
} catch {
return false
}
}
export async function signIngressHeader(secret: string, claim: IngressClaim, nowMs = Date.now()): Promise<string> {
const signed = `v1 ${claim.name} ${claim.origin} ${claim.clientIp} ${Math.floor(nowMs / 1000)}`
const mac = await crypto.subtle.sign('HMAC', await hmacKey(secret), new TextEncoder().encode(signed))
return `${signed} ${toBase64Url(mac)}`
}
// secrets: every accepted secret (current first, then any being rotated out).
export async function verifyIngressHeader(value: string | null, secrets: readonly string[], nowMs = Date.now()): Promise<IngressClaim | null> {
if (!value || secrets.length === 0) { return null }
const parts = value.split(' ')
if (parts.length !== 6 || parts[0] !== 'v1') { return null }
const [, name, origin, clientIp, seconds, mac] = parts as [string, string, string, string, string, string]
if (!NAME.test(name) || !isOrigin(origin) || !IP.test(clientIp) || !/^\d{1,12}$/.test(seconds)) { return null }
if (Math.abs(nowMs / 1000 - Number(seconds)) > INGRESS_MAX_SKEW_SECONDS) { return null }
if (!/^[A-Za-z0-9_-]{43}$/.test(mac)) { return null }
const data = new TextEncoder().encode(parts.slice(0, 5).join(' '))
const signature = fromBase64Url(mac)
for (const secret of secrets) {
if (await crypto.subtle.verify('HMAC', await hmacKey(secret), signature, data)) { return { name, origin, clientIp } }
}
return null
}
+25
View File
@@ -348,6 +348,31 @@ importers:
specifier: ^5.0.5
version: 5.0.5(@types/node@24.19.2)(vite@8.3.4(@types/node@24.19.2)(esbuild@0.28.2)(tsx@4.23.15))
apps/relay-ingress:
dependencies:
'@orca-cloud/relay-contract':
specifier: workspace:*
version: link:../../packages/relay-contract
devDependencies:
'@types/node':
specifier: ^24.19.2
version: 24.19.2
'@types/ws':
specifier: ^8.18.2
version: 8.18.2
tsx:
specifier: ^4.23.15
version: 4.23.15
typescript:
specifier: ^7.0.2
version: 7.0.2
vitest:
specifier: ^5.0.5
version: 5.0.5(@types/node@24.19.2)(vite@8.3.4(@types/node@24.19.2)(esbuild@0.28.2)(tsx@4.23.15))
ws:
specifier: ^8.22.2
version: 8.22.2
packages/cloud-database:
dependencies:
'@orca-cloud/postgres-schema':
@@ -57,6 +57,7 @@ const NON_UNIT_TEST_RUNNERS = [
},
...[
'cloud/apps/relay-fence-broker',
'cloud/apps/relay-ingress',
'cloud/packages/cloud-database',
'cloud/packages/connector-contract',
'cloud/packages/postgres-schema',