From 8176cc77881be95f630c5fd3bc4823f03c807182 Mon Sep 17 00:00:00 2001 From: Demali-876 <90882773+Demali-876@users.noreply.github.com> Date: Sat, 18 Jul 2026 12:38:52 -0400 Subject: [PATCH 1/6] refactor(proxy): remove API-key dedupe scoping --- src/runtime/benchmarks/suites/composite-request.ts | 5 ++--- src/runtime/dedupe.ts | 9 +-------- src/runtime/dedupe.vectors.json | 11 ----------- src/runtime/proxy-serve.ts | 4 ++-- src/tests/request-ticket.test.ts | 2 +- 5 files changed, 6 insertions(+), 25 deletions(-) diff --git a/src/runtime/benchmarks/suites/composite-request.ts b/src/runtime/benchmarks/suites/composite-request.ts index 5ecd013..958152f 100644 --- a/src/runtime/benchmarks/suites/composite-request.ts +++ b/src/runtime/benchmarks/suites/composite-request.ts @@ -124,13 +124,12 @@ const TARGET_URL = "https://upstream.example.com/api/v1/resource?b=2&a=1"; // eviction is O(1) amortized — but rotation still keeps the measured cost steady.) const REPLAY_ROTATE_ITERATIONS = 50_000; -// Realistic scoped GET: `accept`/`content-type` exercise the semantic-header -// canonicalization, `x-api-key` forces the scope hash, the rest is passthrough. +// Realistic GET: `accept`/`content-type` exercise semantic-header +// canonicalization while user-agent remains non-semantic passthrough metadata. const REQUEST_HEADERS: Record = { accept: "application/json", "content-type": "application/json", "user-agent": "consensus-bench/1.0", - "x-api-key": "bench-scope-key", }; const STAGE_NAMES: CompositeStageName[] = [ diff --git a/src/runtime/dedupe.ts b/src/runtime/dedupe.ts index 2f3586c..63e0e79 100644 --- a/src/runtime/dedupe.ts +++ b/src/runtime/dedupe.ts @@ -85,18 +85,11 @@ export function computeBodyHash(body: RequestBody): string { return sha256Hex(stableStringify(body)); } -export function getScope(headers: Headers): string { - for (const k in headers) { - if (k.toLowerCase() === 'x-api-key') return sha256Hex(headers[k]!); - } - return 'global'; -} - export function generateDedupeKey({ target_url, method, headers = {}, body }: DedupeParams): string { const semanticHeaders = canonicalizeSemanticHeaders(headers); const canonical = { v: 1, - scope: getScope(headers), + scope: 'global', method: method.toUpperCase(), url: canonicalizeUrl(target_url), headers: semanticHeaders, diff --git a/src/runtime/dedupe.vectors.json b/src/runtime/dedupe.vectors.json index f2cf27a..e819517 100644 --- a/src/runtime/dedupe.vectors.json +++ b/src/runtime/dedupe.vectors.json @@ -33,17 +33,6 @@ }, "key": "7f0eb385c62d00117d8e0685e279731e535c3b4199ba6fb97ce79ac720d165a2" }, - { - "name": "api-key-scope", - "input": { - "target_url": "https://api.example.com/p", - "method": "GET", - "headers": { - "x-api-key": "secret" - } - }, - "key": "762942d3cea517745d2919b8fe98c45e8b0a5abcc9c1c94c3b2077014cc4cee9" - }, { "name": "semantic-headers-only", "input": { diff --git a/src/runtime/proxy-serve.ts b/src/runtime/proxy-serve.ts index ad85975..2311cc0 100644 --- a/src/runtime/proxy-serve.ts +++ b/src/runtime/proxy-serve.ts @@ -83,8 +83,8 @@ function buildHopByHopDenySet(headers: Record): Set { // target. On the direct data plane the client supplies the request headers and // the node serves them against the client's chosen upstream, so these are // stripped node-side as defense-in-depth — the orchestrator (relayed path) and -// the consensus-client both strip them too. Notably this keeps `x-api-key` (the -// caller's orchestrator scoping credential) from leaking upstream. +// the consensus-client both strip them too. The deprecated `x-api-key` remains +// only as a denylist entry and has no identity, routing, or cache semantics. // // Source of truth: STRIP_REQUEST_HEADERS in the consensus repo // (server/features/proxy/proxy.ts). This mirrors that list with ONE deliberate diff --git a/src/tests/request-ticket.test.ts b/src/tests/request-ticket.test.ts index 665bf7d..e9abe2b 100644 --- a/src/tests/request-ticket.test.ts +++ b/src/tests/request-ticket.test.ts @@ -13,7 +13,7 @@ const NODE = "node-1"; const request: DedupeParams = { target_url: "https://api.example.com/v1/data?b=2&a=1", method: "GET", - headers: { "content-type": "application/json", "x-api-key": "secret" }, + headers: { "content-type": "application/json" }, }; const dedupeKey = generateDedupeKey(request); const otherRequest: DedupeParams = { ...request, target_url: "https://api.example.com/v1/OTHER" }; From 357243e0221ba89bfd5a0e6ca7edf21b8a30a3bb Mon Sep 17 00:00:00 2001 From: Demali-876 <90882773+Demali-876@users.noreply.github.com> Date: Sun, 30 Aug 2026 23:07:46 -0400 Subject: [PATCH 2/6] feat(node): add profile-v1 execution plans Carries an anonymous execution profile through the node's request path: the profile rides the data-plane proxy request, participates in the dedupe key, and its hash comes back on the response. Mirrors the client-side plan in consensus-client, with shared vectors so both sides agree on the wire format. Co-Authored-By: Claude Opus 5 --- package.json | 1 + src/runtime/dedupe.ts | 12 +- src/runtime/dedupe.vectors.json | 12 ++ src/runtime/profile-v1.ts | 233 ++++++++++++++++++++++++++++ src/runtime/profile-v1.vectors.json | 54 +++++++ src/runtime/proxy-command.ts | 23 ++- src/runtime/proxy-serve.ts | 66 +++++++- src/runtime/proxy-worker.ts | 22 +-- src/tests/data-plane.test.ts | 53 ++++++- src/tests/dedupe.test.ts | 6 + src/tests/profile-v1.test.ts | 28 ++++ src/tests/proxy-serve.test.ts | 41 ++++- src/tests/request-ticket.test.ts | 24 +++ src/tunnel/data-plane.ts | 37 ++++- src/tunnel/messages.ts | 5 + 15 files changed, 581 insertions(+), 36 deletions(-) create mode 100644 src/runtime/profile-v1.ts create mode 100644 src/runtime/profile-v1.vectors.json create mode 100644 src/tests/profile-v1.test.ts diff --git a/package.json b/package.json index 4279a32..5d536d5 100644 --- a/package.json +++ b/package.json @@ -42,6 +42,7 @@ "test:ssrf": "bun src/tests/ssrf.test.ts", "test:tickets": "bun src/tests/tickets.test.ts", "test:dedupe": "bun src/tests/dedupe.test.ts", + "test:profile-v1": "bun src/tests/profile-v1.test.ts", "test:pin": "bun src/tests/pin.test.ts", "test:responder-auth": "bun src/tests/responder-auth.test.ts", "test:data-handshake": "bun src/tests/data-handshake.test.ts", diff --git a/src/runtime/dedupe.ts b/src/runtime/dedupe.ts index 63e0e79..5933532 100644 --- a/src/runtime/dedupe.ts +++ b/src/runtime/dedupe.ts @@ -14,9 +14,11 @@ export interface DedupeParams { method: string; headers?: Headers; body?: RequestBody; + profile_hash?: string; } const ALLOW_HEADERS = new Set(['accept', 'content-type']); +const HASH_HEADERS = new Set(['authorization', 'cookie']); const MULTI_SPACE = /\s+/g; export function sha256Hex(input: string | Buffer): string { @@ -65,16 +67,19 @@ export function canonicalizeUrl(raw: string): string { } export function canonicalizeSemanticHeaders(headers: Headers): Headers { - // Two-phase: collect the two allowed keys, then emit in fixed alphabetical order - // so the result is deterministic without a sort step ('accept' < 'content-type'). + // Collect public semantic values plus hashes of sensitive upstream credentials, + // then emit in fixed alphabetical order. Secrets never enter the canonical form. const result: Headers = {}; for (const [k, v] of Object.entries(headers)) { const lower = k.toLowerCase(); // HTTP names have no surrounding whitespace if (ALLOW_HEADERS.has(lower)) result[lower] = v.trim().replace(MULTI_SPACE, ' '); + else if (HASH_HEADERS.has(lower)) result[lower] = sha256Hex(v); } const ordered: Headers = {}; if (result['accept']) ordered['accept'] = result['accept']; + if (result['authorization']) ordered['authorization'] = result['authorization']; if (result['content-type']) ordered['content-type'] = result['content-type']; + if (result['cookie']) ordered['cookie'] = result['cookie']; return ordered; } @@ -85,7 +90,7 @@ export function computeBodyHash(body: RequestBody): string { return sha256Hex(stableStringify(body)); } -export function generateDedupeKey({ target_url, method, headers = {}, body }: DedupeParams): string { +export function generateDedupeKey({ target_url, method, headers = {}, body, profile_hash }: DedupeParams): string { const semanticHeaders = canonicalizeSemanticHeaders(headers); const canonical = { v: 1, @@ -94,6 +99,7 @@ export function generateDedupeKey({ target_url, method, headers = {}, body }: De url: canonicalizeUrl(target_url), headers: semanticHeaders, body_hash: computeBodyHash(body), + profile_hash: profile_hash || undefined, }; return sha256Hex(stableStringify(canonical)); diff --git a/src/runtime/dedupe.vectors.json b/src/runtime/dedupe.vectors.json index e819517..06681c1 100644 --- a/src/runtime/dedupe.vectors.json +++ b/src/runtime/dedupe.vectors.json @@ -46,6 +46,18 @@ }, "key": "2f112dcafa15f0548a1661bb31187a8d810066860c4a28ce9f21a612333c8dbc" }, + { + "name": "credential-headers-hashed", + "input": { + "target_url": "https://api.example.com/private", + "method": "GET", + "headers": { + "authorization": "Bearer secret", + "cookie": "session=secret" + } + }, + "key": "4d3c084eb4e3f72fd7770f28433cd04d83c158f8f9099aaccdc894497e556fcd" + }, { "name": "json-body-sorted", "input": { diff --git a/src/runtime/profile-v1.ts b/src/runtime/profile-v1.ts new file mode 100644 index 0000000..1238bdf --- /dev/null +++ b/src/runtime/profile-v1.ts @@ -0,0 +1,233 @@ +import crypto from 'node:crypto'; + +export const PROXY_PROFILE_PROTOCOL = 'consensus.proxy-profile' as const; +export const PROXY_PROFILE_VERSION = 1 as const; + +export interface ProxyExecutionProfileV1 { + protocol: typeof PROXY_PROFILE_PROTOCOL; + version: typeof PROXY_PROFILE_VERSION; + base_url: string; + allowed_methods: string[]; + allowed_paths: string[]; + cache_ttl: number; + verbose: boolean; + node_region?: string; + node_domain?: string; + node_exclude?: string; + direct: boolean; +} + +const PROFILE_KEYS = new Set([ + 'protocol', + 'version', + 'base_url', + 'allowed_methods', + 'allowed_paths', + 'cache_ttl', + 'verbose', + 'node_region', + 'node_domain', + 'node_exclude', + 'direct', +]); +const METHOD_ORDER = ['GET', 'HEAD', 'POST', 'PUT', 'PATCH', 'DELETE', 'OPTIONS']; +const METHODS = new Set(METHOD_ORDER); +const MAX_PATHS = 64; +const MAX_CACHE_TTL_SECONDS = 3_600; +const MAX_PREFERENCE_LENGTH = 256; +const PROFILE_CONTROL_HEADERS = new Set([ + 'x-cache-ttl', 'x-verbose', 'x-node-region', 'x-node-domain', 'x-node-exclude', 'x-direct', +]); + +function stableValue(value: unknown): unknown { + if (Array.isArray(value)) return value.map(stableValue); + if (value !== null && typeof value === 'object') { + return Object.fromEntries( + Object.entries(value as Record) + .map(([key, item]): [string, unknown] => [key, stableValue(item)]) + .sort(([a], [b]) => (a < b ? -1 : a > b ? 1 : 0)), + ); + } + return value; +} + +function normalizePath(value: unknown, field: string): string { + if (typeof value !== 'string' || !value.startsWith('/') || value.startsWith('//') || value.includes('\\')) { + throw new TypeError(`${field} must be an origin-relative path beginning with /`); + } + const parsed = new URL(value, 'http://profile.local'); + if (parsed.origin !== 'http://profile.local' || parsed.search || parsed.hash) { + throw new TypeError(`${field} must not contain an origin, query, or fragment`); + } + return parsed.pathname; +} + +function optionalBoolean(value: unknown, field: string): boolean | undefined { + if (value === undefined) return undefined; + if (typeof value !== 'boolean') throw new TypeError(`${field} must be a boolean`); + return value; +} + +function optionalPreference(value: unknown, field: string): string | undefined { + if (value === undefined) return undefined; + if (typeof value !== 'string') throw new TypeError(`${field} must be a string`); + const normalized = value.split(',').map((item) => item.trim()).filter(Boolean).join(','); + if (!normalized || normalized.length > MAX_PREFERENCE_LENGTH) { + throw new TypeError(`${field} must contain 1-${MAX_PREFERENCE_LENGTH} characters`); + } + return normalized; +} + +/** Validate and canonicalize the anonymous execution plan carried on the wire. */ +export function normalizeProxyProfileV1(input: unknown): ProxyExecutionProfileV1 { + if (!input || typeof input !== 'object' || Array.isArray(input)) { + throw new TypeError('proxy profile must be an object'); + } + const value = input as Record; + for (const key of Object.keys(value)) { + if (!PROFILE_KEYS.has(key)) throw new TypeError(`unsupported proxy profile field: ${key}`); + } + if (value.protocol !== PROXY_PROFILE_PROTOCOL || value.version !== PROXY_PROFILE_VERSION) { + throw new TypeError(`unsupported proxy profile protocol/version; expected ${PROXY_PROFILE_PROTOCOL}@${PROXY_PROFILE_VERSION}`); + } + + let base: URL; + try { + base = new URL(String(value.base_url ?? '')); + } catch { + throw new TypeError('proxy profile base_url is invalid'); + } + if (!['http:', 'https:'].includes(base.protocol) || base.username || base.password || base.search || base.hash) { + throw new TypeError('proxy profile base_url must be an http(s) URL without credentials, query, or fragment'); + } + base.pathname = base.pathname === '/' ? '/' : base.pathname.replace(/\/+$/, ''); + + if (!Array.isArray(value.allowed_methods) || value.allowed_methods.length === 0) { + throw new TypeError('proxy profile allowed_methods must be a non-empty array'); + } + const methodSet = new Set(value.allowed_methods.map((method) => String(method).toUpperCase())); + if ([...methodSet].some((method) => !METHODS.has(method))) { + throw new TypeError('proxy profile contains an unsupported HTTP method'); + } + const allowedMethods = METHOD_ORDER.filter((method) => methodSet.has(method)); + + if (!Array.isArray(value.allowed_paths) || value.allowed_paths.length === 0 || value.allowed_paths.length > MAX_PATHS) { + throw new TypeError(`proxy profile allowed_paths must contain 1-${MAX_PATHS} entries`); + } + const allowedPaths = [...new Set( + value.allowed_paths.map((path) => normalizePath(path, 'proxy profile allowed_paths entry')), + )].sort(); + + let cacheTtl = 300; + if (value.cache_ttl !== undefined) { + cacheTtl = Number(value.cache_ttl); + if (!Number.isInteger(cacheTtl) || cacheTtl < 1 || cacheTtl > MAX_CACHE_TTL_SECONDS) { + throw new TypeError(`proxy profile cache_ttl must be an integer from 1-${MAX_CACHE_TTL_SECONDS}`); + } + } + + return { + protocol: PROXY_PROFILE_PROTOCOL, + version: PROXY_PROFILE_VERSION, + base_url: base.toString(), + allowed_methods: allowedMethods, + allowed_paths: allowedPaths, + cache_ttl: cacheTtl, + verbose: optionalBoolean(value.verbose, 'proxy profile verbose') ?? false, + ...(value.node_region === undefined ? {} : { node_region: optionalPreference(value.node_region, 'proxy profile node_region')! }), + ...(value.node_domain === undefined ? {} : { node_domain: optionalPreference(value.node_domain, 'proxy profile node_domain')! }), + ...(value.node_exclude === undefined ? {} : { node_exclude: optionalPreference(value.node_exclude, 'proxy profile node_exclude')! }), + direct: optionalBoolean(value.direct, 'proxy profile direct') ?? true, + }; +} + +export function hashProxyProfileV1(input: unknown): string { + const profile = normalizeProxyProfileV1(input); + return hashNormalizedProfile(profile); +} + +function hashNormalizedProfile(profile: ProxyExecutionProfileV1): string { + return crypto.createHash('sha256').update(JSON.stringify(stableValue(profile))).digest('hex'); +} + +/** Enforce the profile independently at every execution boundary. */ +export function assertProxyProfileRequestV1( + input: unknown, + targetUrl: string, + method: string, +): ProxyExecutionProfileV1 { + const profile = normalizeProxyProfileV1(input); + let target: URL; + try { + target = new URL(targetUrl); + } catch { + throw new TypeError('proxy profile target_url is invalid'); + } + const base = new URL(profile.base_url); + if (target.protocol !== 'http:' && target.protocol !== 'https:') { + throw new TypeError('proxy profile target must use http(s)'); + } + if (target.username || target.password || target.hash) { + throw new TypeError('proxy profile target cannot contain credentials or a fragment'); + } + if (target.origin !== base.origin) throw new TypeError('proxy profile target is outside its configured origin'); + + const basePath = base.pathname === '/' ? '' : base.pathname.replace(/\/$/, ''); + if (basePath && target.pathname !== basePath && !target.pathname.startsWith(`${basePath}/`)) { + throw new TypeError('proxy profile target is outside its configured base path'); + } + const relativePath = target.pathname.slice(basePath.length) || '/'; + const allowedPath = profile.allowed_paths.some((prefix) => + prefix === '/' || relativePath === prefix || relativePath.startsWith(`${prefix}/`), + ); + if (!allowedPath) throw new TypeError('proxy profile target path is not allowed'); + + const normalizedMethod = method.toUpperCase(); + if (!profile.allowed_methods.includes(normalizedMethod)) { + throw new TypeError(`method ${normalizedMethod} is not allowed by the proxy profile`); + } + return profile; +} + +/** Existing control headers remain the v1 execution adapter for cache and routing. */ +export function proxyProfileControlHeadersV1(input: unknown): Record { + const profile = normalizeProxyProfileV1(input); + return controlHeadersForNormalizedProfile(profile); +} + +function controlHeadersForNormalizedProfile(profile: ProxyExecutionProfileV1): Record { + return { + ...(profile.cache_ttl === undefined ? {} : { 'x-cache-ttl': String(profile.cache_ttl) }), + ...(profile.verbose === true ? { 'x-verbose': 'true' } : {}), + ...(profile.node_region === undefined ? {} : { 'x-node-region': profile.node_region }), + ...(profile.node_domain === undefined ? {} : { 'x-node-domain': profile.node_domain }), + ...(profile.node_exclude === undefined ? {} : { 'x-node-exclude': profile.node_exclude }), + ...(profile.direct === true ? { 'x-direct': 'true' } : {}), + }; +} + +export interface PreparedProxyProfileV1 { + profile: ProxyExecutionProfileV1; + profile_hash: string; + headers: Record; +} + +/** Canonicalize, enforce, hash, and apply a profile in one operation. */ +export function prepareProxyProfileRequestV1( + input: unknown, + targetUrl: string, + method: string, + headers: Record = {}, +): PreparedProxyProfileV1 { + const profile = assertProxyProfileRequestV1(input, targetUrl, method); + const requestHeaders = Object.fromEntries( + Object.entries(headers) + .filter(([key]) => !PROFILE_CONTROL_HEADERS.has(key.toLowerCase())) + .map(([key, value]) => [key, String(value)]), + ); + return { + profile, + profile_hash: hashNormalizedProfile(profile), + headers: { ...requestHeaders, ...controlHeadersForNormalizedProfile(profile) }, + }; +} diff --git a/src/runtime/profile-v1.vectors.json b/src/runtime/profile-v1.vectors.json new file mode 100644 index 0000000..3667810 --- /dev/null +++ b/src/runtime/profile-v1.vectors.json @@ -0,0 +1,54 @@ +{ + "_comment": "Shared profile-v1 canonicalization vectors. This file must remain byte-for-byte identical across consensus, consensus-client, and consensus-node.", + "version": 1, + "vectors": [ + { + "name": "catalog", + "input": { + "protocol": "consensus.proxy-profile", + "version": 1, + "base_url": "https://API.example.com:443/v1/", + "allowed_methods": ["get", "HEAD", "GET"], + "allowed_paths": ["/search", "/products", "/products"], + "cache_ttl": 120, + "node_region": "us-east", + "direct": false + }, + "normalized": { + "protocol": "consensus.proxy-profile", + "version": 1, + "base_url": "https://api.example.com/v1", + "allowed_methods": ["GET", "HEAD"], + "allowed_paths": ["/products", "/search"], + "cache_ttl": 120, + "verbose": false, + "node_region": "us-east", + "direct": false + }, + "hash": "3ad74b9c2800b3ef39132e89f440a3e8847988351e2daa316ad527770ed52c12" + }, + { + "name": "root-defaults-explicit", + "input": { + "protocol": "consensus.proxy-profile", + "version": 1, + "base_url": "http://example.com:80/", + "allowed_methods": ["HEAD", "GET"], + "allowed_paths": ["/"], + "verbose": true, + "direct": true + }, + "normalized": { + "protocol": "consensus.proxy-profile", + "version": 1, + "base_url": "http://example.com/", + "allowed_methods": ["GET", "HEAD"], + "allowed_paths": ["/"], + "cache_ttl": 300, + "verbose": true, + "direct": true + }, + "hash": "0576497336ea50afd848a53928cb232e6ec6b5ac0f1c1a299a7137361d70d46a" + } + ] +} diff --git a/src/runtime/proxy-command.ts b/src/runtime/proxy-command.ts index f5972da..3358939 100644 --- a/src/runtime/proxy-command.ts +++ b/src/runtime/proxy-command.ts @@ -1,32 +1,29 @@ import type { ProxyRequestMessage, ProxyResponseMessage } from "../tunnel/messages"; import { MESSAGE_TYPE, nowSeconds } from "../tunnel/messages"; +import { serveProxyRequest } from "./proxy-serve"; export async function executeProxyCommand(message: ProxyRequestMessage): Promise { const method = (message.method || "GET").toUpperCase(); - const start = performance.now(); const body = decodeBody(message.body, message.body_encoding); - - const response = await fetch(message.target_url, { + const response = await serveProxyRequest({ + target_url: message.target_url, method, - headers: { - ...(message.headers || {}), - "user-agent": "Consensus-Node/0.1", - }, - body: method === "GET" || method === "HEAD" ? undefined : body, - signal: AbortSignal.timeout(30_000), + headers: message.headers, + body, + profile: message.profile, }); - const responseBody = Buffer.from(await response.arrayBuffer()); - return { type: MESSAGE_TYPE.PROXY_RESPONSE, timestamp: nowSeconds(), reply_to: message.id ?? "", status: response.status, status_text: response.statusText, - headers: Object.fromEntries(response.headers.entries()), - body: responseBody.toString("base64"), + headers: response.headers, + body: response.body.toString("base64"), body_encoding: "base64", + cached: response.cached, + profile_hash: response.profile_hash, }; } diff --git a/src/runtime/proxy-serve.ts b/src/runtime/proxy-serve.ts index 2311cc0..ebb27ec 100644 --- a/src/runtime/proxy-serve.ts +++ b/src/runtime/proxy-serve.ts @@ -14,6 +14,11 @@ import { checkServerIdentity, type PeerCertificate } from "node:tls"; import { resolveAndCheckTarget, type SafeResolution } from "./ssrf"; +import { generateDedupeKey } from "./dedupe"; +import { + prepareProxyProfileRequestV1, + type ProxyExecutionProfileV1, +} from "./profile-v1"; export type SsrfCheck = (url: string) => Promise; @@ -22,6 +27,7 @@ export interface ProxyServeRequest { method?: string; headers?: Record; body?: string | Buffer | null; + profile?: ProxyExecutionProfileV1; } export interface ProxyResult { @@ -29,6 +35,8 @@ export interface ProxyResult { statusText: string; headers: Record; body: Buffer; + cached?: boolean; + profile_hash?: string; } export interface ProxyServeOptions { @@ -49,6 +57,33 @@ interface BunFetchInit extends RequestInit { } const DEFAULT_TIMEOUT_MS = 30_000; +const MAX_PROFILE_CACHE_ENTRIES = 1_000; +const profileCache = new Map(); + +export function clearProxyProfileCache(): void { + profileCache.clear(); +} + +function cachedProfileResult(key: string): ProxyResult | null { + const entry = profileCache.get(key); + if (!entry) return null; + if (entry.expiresAt <= Date.now()) { + profileCache.delete(key); + return null; + } + return { ...entry.value, body: Buffer.from(entry.value.body), cached: true }; +} + +function storeProfileResult(key: string, value: ProxyResult, ttlSeconds: number): void { + if (profileCache.size >= MAX_PROFILE_CACHE_ENTRIES) { + const oldest = profileCache.keys().next().value as string | undefined; + if (oldest) profileCache.delete(oldest); + } + profileCache.set(key, { + value: { ...value, body: Buffer.from(value.body), cached: false }, + expiresAt: Date.now() + ttlSeconds * 1_000, + }); +} // Always-hop-by-hop headers (RFC 7230 §6.1); `host` we set ourselves. The // `Connection` header itself additionally NAMES further hop-by-hop headers that @@ -121,6 +156,24 @@ export async function serveProxyRequest( ): Promise { const ssrfCheck = opts.ssrfCheck ?? resolveAndCheckTarget; const method = (request.method ?? "GET").toUpperCase(); + const prepared = request.profile + ? prepareProxyProfileRequestV1(request.profile, request.target_url, method, request.headers) + : undefined; + const profile = prepared?.profile; + const profileHash = prepared?.profile_hash; + const cacheKey = profile?.cache_ttl + ? generateDedupeKey({ + target_url: request.target_url, + method, + headers: prepared?.headers ?? request.headers, + body: request.body, + profile_hash: profileHash, + }) + : undefined; + if (cacheKey) { + const cached = cachedProfileResult(cacheKey); + if (cached) return cached; + } // SSRF gate: throws TypeError for private/loopback/invalid targets. const resolution = await ssrfCheck(request.target_url); @@ -133,9 +186,10 @@ export async function serveProxyRequest( // Pin the connection to the verified IP — no second DNS lookup. url.hostname = resolution.family === 6 ? `[${resolution.ip}]` : resolution.ip; - const deny = buildHopByHopDenySet(request.headers ?? {}); + const effectiveHeaders = prepared?.headers ?? request.headers ?? {}; + const deny = buildHopByHopDenySet(effectiveHeaders); const headers = new Headers(); - for (const [key, value] of Object.entries(request.headers ?? {})) { + for (const [key, value] of Object.entries(effectiveHeaders)) { const lower = key.toLowerCase(); if (deny.has(lower) || CONSENSUS_CONTROL_HEADERS.has(lower)) continue; headers.set(key, value); @@ -163,10 +217,16 @@ export async function serveProxyRequest( const response = await fetch(url.toString(), init); - return { + const result: ProxyResult = { status: response.status, statusText: response.statusText, headers: Object.fromEntries(response.headers.entries()), body: Buffer.from(await response.arrayBuffer()), + cached: false, + ...(profileHash ? { profile_hash: profileHash } : {}), }; + if (cacheKey && profile?.cache_ttl && result.status >= 200 && result.status < 300) { + storeProfileResult(cacheKey, result, profile.cache_ttl); + } + return result; } diff --git a/src/runtime/proxy-worker.ts b/src/runtime/proxy-worker.ts index e4fdb6b..d07952e 100644 --- a/src/runtime/proxy-worker.ts +++ b/src/runtime/proxy-worker.ts @@ -1,4 +1,6 @@ import type { FastifyInstance } from "fastify"; +import { serveProxyRequest } from "./proxy-serve"; +import type { ProxyExecutionProfileV1 } from "./profile-v1"; export async function registerProxyRoutes(app: FastifyInstance): Promise { app.post("/proxy", async (request, reply) => { @@ -7,36 +9,36 @@ export async function registerProxyRoutes(app: FastifyInstance): Promise { method?: string; headers?: Record; body?: unknown; + profile?: ProxyExecutionProfileV1; }; if (!body?.target_url) return reply.code(400).send({ error: "Missing target_url" }); const method = (body.method || "GET").toUpperCase(); const start = performance.now(); - const response = await fetch(body.target_url, { + const response = await serveProxyRequest({ + target_url: body.target_url, method, - headers: { - ...(body.headers || {}), - "user-agent": "Consensus-Node/0.1" - }, + headers: body.headers, body: method === "GET" || method === "HEAD" ? undefined : typeof body.body === "string" ? body.body : JSON.stringify(body.body ?? null), - signal: AbortSignal.timeout(30_000) + profile: body.profile, }); - - const responseText = await response.text(); + const responseText = response.body.toString("utf8"); return reply.code(response.status).send({ status: response.status, statusText: response.statusText, - headers: Object.fromEntries(response.headers.entries()), + headers: response.headers, data: responseText, meta: { processing_ms: Math.round(performance.now() - start), - timestamp: new Date().toISOString() + timestamp: new Date().toISOString(), + cached: response.cached, + profile_hash: response.profile_hash, } }); }); diff --git a/src/tests/data-plane.test.ts b/src/tests/data-plane.test.ts index 3316b28..9b9211c 100644 --- a/src/tests/data-plane.test.ts +++ b/src/tests/data-plane.test.ts @@ -17,6 +17,11 @@ import { generateDedupeKey } from "../runtime/dedupe"; import { serveProxyRequest, type ProxyServeRequest, type SsrfCheck } from "../runtime/proxy-serve"; import type { SafeResolution } from "../runtime/ssrf"; import type { NodeIdentity } from "../crypto/identity"; +import { + PROXY_PROFILE_PROTOCOL, + PROXY_PROFILE_VERSION, + hashProxyProfileV1, +} from "../runtime/profile-v1"; // ---- fixtures ------------------------------------------------------------- function newIdentity(): NodeIdentity { @@ -133,6 +138,48 @@ try { } // 2) Ticket bound to a different request -> unauthorized (request binding). + { + const profile = { + protocol: PROXY_PROFILE_PROTOCOL, + version: PROXY_PROFILE_VERSION, + base_url: `http://127.0.0.1:${upstream.port}/`, + allowed_methods: ["POST"], + allowed_paths: ["/echo"], + cache_ttl: 60, + verbose: false, + direct: true, + }; + const request = { ...echoRequest, profile }; + const profileHash = hashProxyProfileV1(profile); + const dedupeKey = generateDedupeKey({ + target_url: upstreamUrl, + method: "POST", + headers: echoRequest.headers, + body: Buffer.from("hi"), + profile_hash: profileHash, + }); + const pipe = memoryPipe(); + const [, response] = await Promise.all([ + serveDataConnection(pipe.server, baseDeps(new JtiReplayCache())), + runDataRequest(pipe.client, clientParams(mint(dedupeKey, "j-profile"), request)), + ]); + assert.equal(response.type, "proxy_response"); + assert.equal((response as { profile_hash?: string }).profile_hash, profileHash); + + const tampered = memoryPipe(); + const [, rejected] = await Promise.all([ + serveDataConnection(tampered.server, baseDeps(new JtiReplayCache())), + runDataRequest(tampered.client, clientParams( + mint(dedupeKey, "j-profile-tampered"), + { ...echoRequest, profile: { ...profile, cache_ttl: 30 } }, + )), + ]); + assert.equal(rejected.type, "error"); + assert.equal((rejected as { code: string }).code, "unauthorized"); + checks += 4; + } + + // 3) Ticket bound to a different request -> unauthorized (request binding). { const { client, server } = memoryPipe(); const wrongToken = mint(generateDedupeKey({ target_url: "https://elsewhere.test/", method: "GET" }), "j-wrong"); @@ -145,7 +192,7 @@ try { checks += 2; } - // 3) Valid ticket but SSRF-blocked target -> upstream_error (default real guard). + // 4) Valid ticket but SSRF-blocked target -> upstream_error (default real guard). { const { client, server } = memoryPipe(); const blocked: ProxyServeRequest = { target_url: "http://127.0.0.1/", method: "GET" }; @@ -166,7 +213,7 @@ try { checks += 3; } - // 4) Replay: the same ticket spent twice across connections (shared cache). + // 5) Replay: the same ticket spent twice across connections (shared cache). { const replay = new JtiReplayCache(); const token = mint(echoDedupe, "j-replay"); @@ -186,7 +233,7 @@ try { checks += 3; } - // 5) Live WebSocket round-trip through the actual Fastify route. + // 6) Live WebSocket round-trip through the actual Fastify route. { const app = Fastify(); await app.register(websocket); diff --git a/src/tests/dedupe.test.ts b/src/tests/dedupe.test.ts index 9113f5b..26b1f47 100644 --- a/src/tests/dedupe.test.ts +++ b/src/tests/dedupe.test.ts @@ -31,4 +31,10 @@ assert.equal(generateDedupeKey(base), generateDedupeKey({ ...base, target_url: ' assert.equal(generateDedupeKey(base), generateDedupeKey({ ...base, method: 'get' })); checks += 3; +assert.notEqual( + generateDedupeKey({ ...base, headers: { authorization: 'Bearer a' } }), + generateDedupeKey({ ...base, headers: { authorization: 'Bearer b' } }), +); +checks++; + console.log(`dedupe.test.ts: ${checks} dedupe vectors verified under Bun — node matches the orchestrator`); diff --git a/src/tests/profile-v1.test.ts b/src/tests/profile-v1.test.ts new file mode 100644 index 0000000..8924365 --- /dev/null +++ b/src/tests/profile-v1.test.ts @@ -0,0 +1,28 @@ +import assert from 'node:assert/strict'; +import fs from 'node:fs'; +import path from 'node:path'; +import { fileURLToPath } from 'node:url'; + +import { + assertProxyProfileRequestV1, + hashProxyProfileV1, + normalizeProxyProfileV1, +} from '../runtime/profile-v1'; + +const here = path.dirname(fileURLToPath(import.meta.url)); +const fixture = JSON.parse(fs.readFileSync(path.join(here, '../runtime/profile-v1.vectors.json'), 'utf8')) as { + vectors: Array<{ input: unknown; normalized: unknown; hash: string }>; +}; + +let checks = 0; +for (const vector of fixture.vectors) { + assert.deepEqual(normalizeProxyProfileV1(vector.input), vector.normalized); + assert.equal(hashProxyProfileV1(vector.input), vector.hash); + checks += 2; +} +const profile = fixture.vectors[0].normalized; +assert.doesNotThrow(() => assertProxyProfileRequestV1(profile, 'https://api.example.com/v1/products/1', 'GET')); +assert.throws(() => assertProxyProfileRequestV1(profile, 'https://api.example.com/v1/private', 'GET'), /not allowed/); +checks += 2; + +console.log(`profile-v1.test.ts: ${checks} shared contract checks passed`); diff --git a/src/tests/proxy-serve.test.ts b/src/tests/proxy-serve.test.ts index 9074f65..39fbc8e 100644 --- a/src/tests/proxy-serve.test.ts +++ b/src/tests/proxy-serve.test.ts @@ -6,7 +6,8 @@ import os from "node:os"; import path from "node:path"; import zlib from "node:zlib"; -import { serveProxyRequest, type SsrfCheck } from "../runtime/proxy-serve"; +import { clearProxyProfileCache, serveProxyRequest, type SsrfCheck } from "../runtime/proxy-serve"; +import { PROXY_PROFILE_PROTOCOL, PROXY_PROFILE_VERSION } from "../runtime/profile-v1"; import type { SafeResolution } from "../runtime/ssrf"; const allow = @@ -52,9 +53,11 @@ for (const target of [ // A local HTTP upstream. The real guard blocks loopback, so tests inject a // permissive ssrfCheck (same pattern as the server suite's noSsrf helper). +let upstreamHits = 0; const server = Bun.serve({ port: 0, async fetch(req) { + upstreamHits++; const url = new URL(req.url); if (url.pathname === "/redirect") { return new Response(null, { status: 302, headers: { location: "http://169.254.169.254/" } }); @@ -96,6 +99,42 @@ try { checks += 4; } + // profile-v1 is enforced and cached by the node itself. + { + clearProxyProfileCache(); + const profile = { + protocol: PROXY_PROFILE_PROTOCOL, + version: PROXY_PROFILE_VERSION, + base_url: `http://127.0.0.1:${port}/`, + allowed_methods: ["GET"], + allowed_paths: ["/echo"], + cache_ttl: 60, + verbose: false, + direct: true, + }; + const before = upstreamHits; + const first = await serveProxyRequest( + { target_url: `http://127.0.0.1:${port}/echo`, method: "GET", profile }, + { ssrfCheck: allow("127.0.0.1") }, + ); + const second = await serveProxyRequest( + { target_url: `http://127.0.0.1:${port}/echo`, method: "GET", profile }, + { ssrfCheck: allow("127.0.0.1") }, + ); + assert.equal(first.cached, false); + assert.equal(second.cached, true); + assert.equal(upstreamHits - before, 1, "node profile cache serves the second request locally"); + assert.equal(first.profile_hash, second.profile_hash); + await assert.rejects( + () => serveProxyRequest( + { target_url: `http://127.0.0.1:${port}/redirect`, method: "GET", profile }, + { ssrfCheck: allow("127.0.0.1") }, + ), + /not allowed/, + ); + checks += 5; + } + // 3) IP-pin + Host preservation: a hostname mapped to loopback still reaches // the server and the upstream sees the ORIGINAL Host. { diff --git a/src/tests/request-ticket.test.ts b/src/tests/request-ticket.test.ts index e9abe2b..f0dfb00 100644 --- a/src/tests/request-ticket.test.ts +++ b/src/tests/request-ticket.test.ts @@ -59,6 +59,30 @@ assert.throws( ); checks++; +// The anonymous profile hash is part of the request binding; changing the plan +// invalidates a ticket even when URL/method/body are otherwise identical. +{ + const profiled = { ...request, profile_hash: "a".repeat(64) }; + const profiledKey = generateDedupeKey(profiled); + const token = mint({ dedupeKey: profiledKey, jti: "profile-jti" }); + assert.doesNotThrow(() => verifyRequestTicket({ + token, + nodeId: NODE, + publicKey, + request: profiled, + now: 1010, + replay: new JtiReplayCache(), + })); + assert.throws(() => verifyRequestTicket({ + token: mint({ dedupeKey: profiledKey, jti: "profile-jti-2" }), + nodeId: NODE, + publicKey, + request: { ...profiled, profile_hash: "b".repeat(64) }, + now: 1010, + }), /does not match/); + checks += 2; +} + // 3) A ticket minted for another node is rejected (aud + implicit assertion). assert.throws(() => verifyRequestTicket({ token: mint({ nodeId: "node-A" }), nodeId: "node-B", publicKey, request, now: 1010 }), diff --git a/src/tunnel/data-plane.ts b/src/tunnel/data-plane.ts index 8a64cb8..e7365c1 100644 --- a/src/tunnel/data-plane.ts +++ b/src/tunnel/data-plane.ts @@ -30,6 +30,10 @@ import { verifyRequestTicket } from "../tickets/request-ticket"; import type { JtiReplayCache } from "../tickets/replay"; import type { DedupeParams } from "../runtime/dedupe"; import { serveProxyRequest, type ProxyResult, type ProxyServeRequest } from "../runtime/proxy-serve"; +import { + prepareProxyProfileRequestV1, + type ProxyExecutionProfileV1, +} from "../runtime/profile-v1"; import type { NodeIdentity } from "../crypto/identity"; export const DATA_PLANE_PATH = "/connect"; @@ -49,6 +53,7 @@ export interface ProxyRequestPayload { headers?: Record; body?: string; // base64 when present body_encoding?: "base64"; + profile?: ProxyExecutionProfileV1; } export type ProxyResponsePayload = @@ -59,6 +64,8 @@ export type ProxyResponsePayload = headers: Record; body: string; // base64 body_encoding: "base64"; + cached?: boolean; + profile_hash?: string; } | { type: "error"; code: string; message: string }; @@ -95,7 +102,7 @@ export async function runDataRequest( nodeId: string; expectedNodePublicKeyPem: string; token: string; - request: { target_url: string; method?: string; headers?: Record; body?: string | Buffer | null }; + request: { target_url: string; method?: string; headers?: Record; body?: string | Buffer | null; profile?: ProxyExecutionProfileV1 }; }, ): Promise { const client = await createDataInit({ nodeId: params.nodeId }); @@ -118,6 +125,7 @@ export async function runDataRequest( headers: params.request.headers, body: body ? body.toString("base64") : undefined, body_encoding: body ? "base64" : undefined, + profile: params.request.profile, }; await transport.send(sealFrame(session.sendKey, FRAME_TYPE.DATA, 0n, encodeJson(payload))); @@ -145,7 +153,22 @@ async function resolveProxyResponse( const body = decodeBody(payload.body, payload.body_encoding); const method = (payload.method ?? "GET").toUpperCase(); - const dedupeParams: DedupeParams = { target_url: payload.target_url, method, headers: payload.headers, body }; + let preparedProfile: ReturnType | undefined; + try { + preparedProfile = payload.profile + ? prepareProxyProfileRequestV1(payload.profile, payload.target_url, method, payload.headers) + : undefined; + } catch (err) { + return { type: "error", code: "bad_request", message: errorMessage(err) }; + } + const profileHash = preparedProfile?.profile_hash; + const dedupeParams: DedupeParams = { + target_url: payload.target_url, + method, + headers: preparedProfile?.headers ?? payload.headers, + body, + profile_hash: profileHash, + }; try { verifyRequestTicket({ @@ -161,7 +184,13 @@ async function resolveProxyResponse( try { const serve = deps.serve ?? defaultServe; - const result = await serve({ target_url: payload.target_url, method, headers: payload.headers, body }); + const result = await serve({ + target_url: payload.target_url, + method, + headers: preparedProfile?.headers ?? payload.headers, + body, + profile: preparedProfile?.profile, + }); return { type: "proxy_response", status: result.status, @@ -169,6 +198,8 @@ async function resolveProxyResponse( headers: result.headers, body: result.body.toString("base64"), body_encoding: "base64", + cached: result.cached, + profile_hash: result.profile_hash, }; } catch (err) { return { type: "error", code: "upstream_error", message: errorMessage(err) }; diff --git a/src/tunnel/messages.ts b/src/tunnel/messages.ts index 2b2feef..2350591 100644 --- a/src/tunnel/messages.ts +++ b/src/tunnel/messages.ts @@ -1,3 +1,5 @@ +import type { ProxyExecutionProfileV1 } from "../runtime/profile-v1"; + export const TUNNEL_MODE = { EVAL: "eval", CONTROL: "control", @@ -138,6 +140,7 @@ export interface ProxyRequestMessage extends BaseMessage { headers?: Record; body?: string; body_encoding?: "utf8" | "base64"; + profile?: ProxyExecutionProfileV1; } export interface ProxyResponseMessage extends BaseMessage { @@ -148,6 +151,8 @@ export interface ProxyResponseMessage extends BaseMessage { headers?: Record; body?: string; body_encoding?: "utf8" | "base64"; + cached?: boolean; + profile_hash?: string; } export interface StreamOpenMessage extends BaseMessage { From 91904f34ca10ce2e0071355f72e2fdba27370093 Mon Sep 17 00:00:00 2001 From: Demali-876 <90882773+Demali-876@users.noreply.github.com> Date: Sun, 30 Aug 2026 23:08:02 -0400 Subject: [PATCH 3/6] refactor(node)!: replace run-node.sh with a bun supervisor MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit src/supervise.ts runs the runtime server and the control tunnel as one unit and exits when either does, so `update_apply` and crashes alike cycle the unit from the refreshed `current` symlink. Same contract as the shell script it replaces, including exit 70 when no release is installed. Why: run-node.sh needed `wait -n` and therefore bash >= 4.3. Stock macOS ships bash 3.2, so it exited 78 on any Mac whose operator had not run `brew install bash` — the script carried that workaround as a documented caveat. Three deliberate differences from the shell version: - Children are spawned detached and signalled by process group. `bun run