From ce6880347150860537d25206c408bcbadcf759df Mon Sep 17 00:00:00 2001 From: Ajay Ghanwat Date: Wed, 5 Aug 2026 20:10:20 +0530 Subject: [PATCH] fix(rpc): close the service-collision fail-open and the fix-wave gaps Critical: - router: fail loudly (WRN-SERVICE-COLLISION) when two app/services files scan to the same service name, instead of silently letting directory-walk order pick a winner. Important: - server.ts: wrap a throwing input schema so its raw message cannot escape invoke(); returns RPC_INVALID and logs server-side instead. - client.ts: race timeoutMs against transport.call so a stalled transport cannot hang the caller; rejects with a ServiceError(RPC_TRANSPORT). - client.ts: the proxy returns undefined for undeclared properties (incl. then/catch/finally) instead of a function that throws, closing the await-client thenable trap. - gateway.ts / rpc-dispatch.ts: import RPC_PATH_PREFIX / RPC_INTERNAL_HEADER from @wrnexus/rpc instead of hardcoding local copies. - gateway.test.ts: cover the RPC-prefix edge block and internal-header stripping across casing variants. - http.test.ts / client.test.ts: cover anonymous-call header omission, the internal marker, the retryable-status sweep, network/malformed/HTML failures, AbortSignal propagation, the timeout path, and timer cleanup. Minor: - transport.ts: Object.hasOwn for handler lookup; note the entry-only abort check. - client.ts: wrap a missing/invalid WRNEXUS_RPC_SECRET as a ServiceError (RPC_IDENTITY) instead of a bare Error. - rpc/package.json: drop the unused @wrnexus/authz dependency. - server.ts: implement() now throws at construction time if a declared procedure has no own handler. Verified: reverting the service-collision check and the client timeout race each make their new test fail, then restore green. Co-Authored-By: Claude Opus 5 --- packages/dev-server/src/gateway.ts | 15 +++- packages/dev-server/src/rpc-dispatch.ts | 9 +- packages/dev-server/test/gateway.test.ts | 25 ++++++ packages/router/src/index.ts | 14 +++ .../router/test/services-discovery.test.ts | 65 ++++++++++++-- packages/rpc/package.json | 1 - packages/rpc/src/client.ts | 47 ++++++++-- packages/rpc/src/server.ts | 22 ++++- packages/rpc/src/transport.ts | 10 ++- packages/rpc/test/client.test.ts | 88 +++++++++++++++++-- packages/rpc/test/http.test.ts | 82 ++++++++++++++++- 11 files changed, 347 insertions(+), 31 deletions(-) diff --git a/packages/dev-server/src/gateway.ts b/packages/dev-server/src/gateway.ts index 761d0dc0..6255cf53 100644 --- a/packages/dev-server/src/gateway.ts +++ b/packages/dev-server/src/gateway.ts @@ -13,11 +13,9 @@ import { spawn, type ChildProcess } from "node:child_process"; import { existsSync } from "node:fs"; import { dirname, join, resolve } from "node:path"; import { fileURLToPath } from "node:url"; +import { RPC_INTERNAL_HEADER, RPC_PATH_PREFIX } from "@wrnexus/rpc"; import { RESTART_EXIT_CODE } from "./restart.ts"; -const RPC_PATH_PREFIX = "/__wrnexus/rpc"; -const RPC_INTERNAL_HEADER = "x-wrnexus-internal"; - export type GatewayForwardAuth = ( | { url: string; @@ -442,6 +440,15 @@ export function stripUntrustedInternalHeaders(headers: Headers): Headers { return sanitized; } +/** + * The reserved inter-app RPC namespace is refused at the gateway edge, before + * any proxying — it is only ever mounted by a child app's own dev-server and + * must never be reachable from outside the workspace. + */ +export function isRpcGatewayPath(pathname: string): boolean { + return pathname === RPC_PATH_PREFIX || pathname.startsWith(`${RPC_PATH_PREFIX}/`); +} + /** Boot every app as a child process, then route by Host on one gateway port. */ export async function startGateway(opts: GatewayOptions): Promise { const port = opts.port ?? 3000; @@ -625,7 +632,7 @@ export async function startGateway(opts: GatewayOptions): Promise { + expect(isRpcGatewayPath(RPC_PATH_PREFIX)).toBe(true); + expect(isRpcGatewayPath(`${RPC_PATH_PREFIX}/billing/createInvoice`)).toBe(true); + expect(isRpcGatewayPath("/api/billing")).toBe(false); + expect(isRpcGatewayPath("/__wrnexus/rpcfoo")).toBe(false); +}); + +test("an inbound internal-marker header from outside is stripped regardless of casing", () => { + for (const name of [ + RPC_INTERNAL_HEADER, + RPC_INTERNAL_HEADER.toUpperCase(), + "X-WrNexus-Internal", + ]) { + const request = new Request("http://localhost:3000/path", { + headers: { [name]: "1" }, + }); + const headers = stripUntrustedInternalHeaders( + gatewayProxyHeaders(request, new URL(request.url), "127.0.0.1", true), + ); + expect(headers.has(RPC_INTERNAL_HEADER)).toBe(false); + } +}); + test("gateway respawns development apps after an HMR restart exit", () => { expect(gatewayRestartDelay("development", 97, null)).toBe(0); expect(gatewayRestartDelay("development", 1, null)).toBe(1200); diff --git a/packages/router/src/index.ts b/packages/router/src/index.ts index 1fcd3911..79dd26fa 100644 --- a/packages/router/src/index.ts +++ b/packages/router/src/index.ts @@ -305,6 +305,7 @@ export function buildRouter(appDir: string, opts: RouterOptions = {}): Router { } const services: ComponentRef[] = []; + const serviceFilesByName = new Map(); for (const f of scanDir(join(appDir, "services"), [".js"])) { if (!/\.(ts|js)$/.test(f.file) || /[.]gen[.](ts|js)$/.test(f.file)) continue; const name = basename(f.file).replace(/\.(ts|js)$/, ""); @@ -312,6 +313,19 @@ export function buildRouter(appDir: string, opts: RouterOptions = {}): Router { console.warn(`[wrnexus] skipping service with unsafe name: ${name}`); continue; } + // A service name is a routable identity (/__wrnexus/rpc//...), so a + // collision is a configuration error, not something to resolve by + // directory-walk precedence like `components` silently does. Fail loudly, + // naming both files, instead of letting whichever file is visited last + // win non-deterministically. + const existing = serviceFilesByName.get(name); + if (existing) { + throw new Error( + `WRN-SERVICE-COLLISION: two services are both named "${name}": ${existing} and ${f.file}. ` + + `Rename one of the files — service names must be unique across app/services.`, + ); + } + serviceFilesByName.set(name, f.file); services.push({ name, file: f.file }); } diff --git a/packages/router/test/services-discovery.test.ts b/packages/router/test/services-discovery.test.ts index bf9f660f..e178f029 100644 --- a/packages/router/test/services-discovery.test.ts +++ b/packages/router/test/services-discovery.test.ts @@ -6,16 +6,69 @@ import { buildRouter } from "../src/index.ts"; const root = join(import.meta.dir, ".tmp-services"); mkdirSync(root, { recursive: true }); +function makeApp(): string { + const base = mkdtempSync(join(root, "app-")); + mkdirSync(join(base, "app", "pages"), { recursive: true }); + return join(base, "app"); +} + describe("service discovery", () => { test("discovers source services and skips generated files", () => { - const base = mkdtempSync(join(root, "app-")); - const services = join(base, "app", "services"); + const appDir = makeApp(); + const services = join(appDir, "services"); mkdirSync(services, { recursive: true }); - mkdirSync(join(base, "app", "pages"), { recursive: true }); writeFileSync(join(services, "billing.ts"), "export default {};"); writeFileSync(join(services, "types.gen.ts"), "export type T = string;"); - expect(buildRouter(join(base, "app")).services.map((service) => service.name)).toEqual([ - "billing", - ]); + expect(buildRouter(appDir).services.map((service) => service.name)).toEqual(["billing"]); + }); + + test("throws naming both files when two services collide on name", () => { + const appDir = makeApp(); + const services = join(appDir, "services"); + const nested = join(services, "legacy"); + mkdirSync(nested, { recursive: true }); + const top = join(services, "billing.ts"); + const shadow = join(nested, "billing.ts"); + writeFileSync(top, "export default {};"); + writeFileSync(shadow, "export default {};"); + expect(() => buildRouter(appDir)).toThrow(/WRN-SERVICE-COLLISION/); + try { + buildRouter(appDir); + } catch (error) { + const message = (error as Error).message; + expect(message).toContain(top); + expect(message).toContain(shadow); + } + }); + + test("unsafe service names are skipped with a warning", () => { + const appDir = makeApp(); + const services = join(appDir, "services"); + mkdirSync(services, { recursive: true }); + writeFileSync(join(services, "bad name.ts"), "export default {};"); + const originalWarn = console.warn; + let warned = false; + console.warn = (...args: unknown[]) => { + if (String(args[0]).includes("skipping service with unsafe name")) warned = true; + }; + try { + expect(buildRouter(appDir).services).toEqual([]); + expect(warned).toBe(true); + } finally { + console.warn = originalWarn; + } + }); + + test("a missing services directory yields []", () => { + const appDir = makeApp(); + expect(buildRouter(appDir).services).toEqual([]); + }); + + test("discovers .js service files", () => { + const appDir = makeApp(); + const services = join(appDir, "services"); + mkdirSync(services, { recursive: true }); + writeFileSync(join(services, "reports.js"), "export default {};"); + expect(buildRouter(appDir).services.map((service) => service.name)).toEqual(["reports"]); }); }); diff --git a/packages/rpc/package.json b/packages/rpc/package.json index fef417e6..b2b2b4a5 100644 --- a/packages/rpc/package.json +++ b/packages/rpc/package.json @@ -19,7 +19,6 @@ "check": "bun run typecheck && bun run test" }, "dependencies": { - "@wrnexus/authz": "workspace:*", "@wrnexus/core": "workspace:*", "@wrnexus/jwt": "workspace:*", "@wrnexus/helpers": "workspace:*", diff --git a/packages/rpc/src/client.ts b/packages/rpc/src/client.ts index 457b2c5d..b4764e71 100644 --- a/packages/rpc/src/client.ts +++ b/packages/rpc/src/client.ts @@ -32,24 +32,53 @@ export function serviceClient( const timeoutMs = options.timeoutMs ?? DEFAULT_TIMEOUT_MS; return new Proxy({} as ServiceClient, { get(_target, property) { - if (typeof property !== "string") return undefined; + // Every declared procedure is a string key; anything else (including + // `then`/`catch`/`finally`) is not one of ours. Returning a function for + // those makes `await client` or `return client` from an async function + // read the proxy as thenable — the runtime then calls `then(resolve, + // reject)`, which throws "Unknown procedure". Returning undefined lets + // the caller be treated as a plain (non-thenable) object instead. + if (typeof property !== "string" || !Object.hasOwn(contract.procedures, property)) { + return undefined; + } return async (input: unknown) => { - if (!Object.hasOwn(contract.procedures, property)) { - throw new ServiceError(RPC_ERROR_CODES.unknown, "Unknown procedure"); + let identity: string | undefined; + if (options.as) { + try { + identity = await exportSubjectContext(options.as, app); + } catch (error) { + // A missing/misconfigured WRNEXUS_RPC_SECRET otherwise rejects with + // a bare Error, so a caller matching on ServiceError treats + // misconfiguration as a crash instead of a handled RPC failure. + // The operator-facing message names no secret value, so it is safe + // to preserve. + const message = error instanceof Error ? error.message : String(error); + throw new ServiceError(RPC_ERROR_CODES.identity, message); + } } - const identity = options.as ? await exportSubjectContext(options.as, app) : undefined; const controller = new AbortController(); const timer = setTimeout(() => controller.abort(), timeoutMs); + const timeout = new Promise((_, reject) => { + controller.signal.addEventListener("abort", () => { + reject(new ServiceError(RPC_ERROR_CODES.transport, "Call timed out")); + }); + }); + const callPromise = options.transport.call( + { app, service: contract.name, procedure: property }, + input, + { signal: controller.signal, ...(identity ? { identity } : {}) }, + ); try { - const result = await options.transport.call( - { app, service: contract.name, procedure: property }, - input, - { signal: controller.signal, ...(identity ? { identity } : {}) }, - ); + const result = await Promise.race([callPromise, timeout]); if (result.ok) return result.value; throw new ServiceError(result.code, result.message); } finally { clearTimeout(timer); + // If the timeout won the race, the transport call may still settle + // later — a transport that ignores the abort signal keeps running. + // Nothing awaits it again, so swallow a late rejection here rather + // than let it surface as an unhandled promise rejection. + callPromise.catch(() => {}); } }; }, diff --git a/packages/rpc/src/server.ts b/packages/rpc/src/server.ts index adda1c4b..bea8c50f 100644 --- a/packages/rpc/src/server.ts +++ b/packages/rpc/src/server.ts @@ -34,6 +34,16 @@ export function implement( handlers: ServiceHandlers, options: ImplementOptions, ): ServiceImplementation { + // A declared procedure with no own handler would otherwise only surface at + // invoke time as RPC_UNKNOWN — a silent, permanent 404. Catch it now. + for (const procedureName of Object.keys(contract.procedures)) { + if (!Object.hasOwn(handlers, procedureName)) { + throw new Error( + `WRN-RPC-HANDLER: service "${contract.name}" declares procedure "${procedureName}" ` + + `but implement() was not given a handler for it.`, + ); + } + } return { contract, async invoke(procedureName, payload, identity) { @@ -69,7 +79,17 @@ export function implement( let input: unknown = payload; if (definition.input) { - const parsed = definition.input.parse(payload as Record); + // InputSchema is structural: any custom or wrapped schema may throw + // instead of returning { ok: false }. A throw must not escape invoke() + // with its raw message — that text can carry internals — so it is + // caught the same way the permission check above is. + let parsed: { ok: boolean; value?: unknown }; + try { + parsed = definition.input.parse(payload as Record); + } catch (error) { + console.error(`[wrnexus] RPC input schema threw for ${String(procedureName)}`, error); + return failure(RPC_ERROR_CODES.invalid, "Invalid input"); + } if (!parsed.ok) return failure(RPC_ERROR_CODES.invalid, "Invalid input"); input = parsed.value; } diff --git a/packages/rpc/src/transport.ts b/packages/rpc/src/transport.ts index 19fab891..cda9ef8d 100644 --- a/packages/rpc/src/transport.ts +++ b/packages/rpc/src/transport.ts @@ -25,8 +25,16 @@ export type InProcessHandler = ( export function inProcessTransport(handlers: Record): Transport { return { async call(target, payload, options) { + // Checked only at entry: this in-process transport does no I/O, so + // nothing yields between here and the handler call below, and a signal + // that aborts mid-flight is never observed. In-process tests therefore + // cannot exercise a mid-call timeout path — that needs a real transport. if (options.signal?.aborted) return failure(RPC_ERROR_CODES.transport, "Call aborted"); - const handler = handlers[`${target.service}/${target.procedure}`]; + const key = `${target.service}/${target.procedure}`; + // Object.hasOwn, not plain indexing: a prototype-inherited key (e.g. + // "constructor/toString") would otherwise resolve to a function that + // is not one of our handlers. + const handler = Object.hasOwn(handlers, key) ? handlers[key] : undefined; if (!handler) return failure(RPC_ERROR_CODES.unknown, "Unknown procedure"); try { return await handler(payload, options.identity); diff --git a/packages/rpc/test/client.test.ts b/packages/rpc/test/client.test.ts index 76136e98..6aad19d3 100644 --- a/packages/rpc/test/client.test.ts +++ b/packages/rpc/test/client.test.ts @@ -76,7 +76,7 @@ describe("serviceClient", () => { expect(seenAnon).toBeUndefined(); }); - test("calling an undeclared procedure throws rather than issuing a call", async () => { + test("an undeclared procedure is undefined rather than a function that throws", async () => { let called = false; const client = serviceClient(billing, { transport: { @@ -86,13 +86,91 @@ describe("serviceClient", () => { }, }, }); - const proxy = client as unknown as Record Promise>; - await expect(proxy.deleteEverything!({})).rejects.toMatchObject({ - code: RPC_ERROR_CODES.unknown, - }); + const proxy = client as unknown as Record; + expect(proxy.deleteEverything).toBeUndefined(); expect(called).toBe(false); }); + test("the proxy has no `then` escape — await/return does not trigger a call", async () => { + let called = false; + const client = serviceClient(billing, { + transport: { + call: async () => { + called = true; + return success({ invoiceId: "inv_1" }); + }, + }, + }); + const proxy = client as unknown as Record; + expect(proxy.then).toBeUndefined(); + expect(proxy.catch).toBeUndefined(); + expect(proxy.finally).toBeUndefined(); + // `await client` must resolve to the proxy object itself, not reject. + const awaited = await client; + expect(awaited).toBe(client); + expect(called).toBe(false); + }); + + test("the signal arrives at the transport", async () => { + let seenSignal: AbortSignal | undefined; + const client = serviceClient(billing, { + transport: { + call: async (_target, _payload, options) => { + seenSignal = options.signal; + return success({ invoiceId: "inv_1" }); + }, + }, + }); + await client.createInvoice({ amountCents: 1 }); + expect(seenSignal).toBeInstanceOf(AbortSignal); + }); + + test("the call aborts at timeoutMs when the transport ignores the signal", async () => { + const client = serviceClient(billing, { + timeoutMs: 20, + transport: { + call: () => new Promise(() => {}), // never resolves; ignores the signal + }, + }); + let caught: unknown; + const start = Date.now(); + try { + await client.createInvoice({ amountCents: 1 }); + } catch (err) { + caught = err; + } + const elapsed = Date.now() - start; + expect(caught).toBeInstanceOf(ServiceError); + expect(caught).toMatchObject({ code: RPC_ERROR_CODES.transport }); + expect(elapsed).toBeLessThan(500); + }); + + test("the timer is cleared on both success and failure", async () => { + const originalClearTimeout = globalThis.clearTimeout; + let clearCount = 0; + globalThis.clearTimeout = ((...args: Parameters) => { + clearCount++; + return originalClearTimeout(...args); + }) as typeof clearTimeout; + try { + const okClient = serviceClient(billing, { + transport: { call: async () => success({ invoiceId: "inv_1" }) }, + }); + await okClient.createInvoice({ amountCents: 1 }); + expect(clearCount).toBe(1); + + const failClient = serviceClient(billing, { + transport: { call: async () => failure(RPC_ERROR_CODES.transport, "down") }, + }); + await expect(failClient.createInvoice({ amountCents: 1 })).rejects.toBeInstanceOf( + ServiceError, + ); + expect(clearCount).toBe(2); + } finally { + globalThis.clearTimeout = originalClearTimeout; + } + }); + test("the app defaults to the service name", async () => { let seenTarget: RpcTarget | undefined; const client = serviceClient(billing, { diff --git a/packages/rpc/test/http.test.ts b/packages/rpc/test/http.test.ts index 856e4ba5..4d6b6eb8 100644 --- a/packages/rpc/test/http.test.ts +++ b/packages/rpc/test/http.test.ts @@ -1,7 +1,7 @@ import { describe, expect, test } from "bun:test"; -import { RPC_ERROR_CODES } from "../src/errors.ts"; +import { RPC_ERROR_CODES, isRetryableStatus } from "../src/errors.ts"; import { RPC_IDENTITY_HEADER } from "../src/identity.ts"; -import { httpTransport, rpcPath } from "../src/http.ts"; +import { RPC_INTERNAL_HEADER, httpTransport, rpcPath } from "../src/http.ts"; const target = { app: "billing", service: "billing", procedure: "createInvoice" }; @@ -42,4 +42,82 @@ describe("httpTransport", () => { test("uses the stable reserved path", () => { expect(rpcPath("billing", "createInvoice")).toBe("/__wrnexus/rpc/billing/createInvoice"); }); + + test("omits the identity header entirely for an anonymous call", async () => { + const transport = transportWith(async (request) => { + expect(request.headers.has(RPC_IDENTITY_HEADER)).toBe(false); + return Response.json({ ok: true, value: {} }); + }); + await transport.call(target, {}, {}); + }); + + test("sets the internal-marker header", async () => { + const transport = transportWith(async (request) => { + expect(request.headers.get(RPC_INTERNAL_HEADER)).toBe("1"); + return Response.json({ ok: true, value: {} }); + }); + await transport.call(target, {}, {}); + }); + + test("the full retryable-status sweep matches isRetryableStatus", async () => { + // 600 is covered directly on isRetryableStatus in errors.test.ts — the + // Fetch API cannot construct a Response with a status outside 200–599. + const statuses = [200, 400, 403, 404, 408, 409, 429, 500, 503, 599]; + for (const status of statuses) { + if (status === 200) continue; // handled by the success-path test above + const transport = transportWith(() => new Response("x", { status })); + const result = await transport.call(target, {}, {}); + expect(result).toMatchObject({ + ok: false, + code: RPC_ERROR_CODES.transport, + retryable: isRetryableStatus(status), + }); + } + }); + + test("a network throw yields a structured failure with no host/address surviving", async () => { + const transport = httpTransport({ + resolveOrigin: () => "http://internal-billing-host.private:4821", + fetch: (async () => { + throw new TypeError("fetch failed: connect ECONNREFUSED 10.0.0.7:4821"); + }) as unknown as typeof fetch, + }); + const result = await transport.call(target, {}, {}); + expect(result).toMatchObject({ ok: false, code: RPC_ERROR_CODES.transport }); + if (!result.ok) { + expect(result.message).not.toContain("10.0.0.7"); + expect(result.message).not.toContain("internal-billing-host"); + } + }); + + test("a malformed JSON body yields a structured failure with no body content surviving", async () => { + const transport = transportWith(() => new Response("{not json", { status: 200 })); + const result = await transport.call(target, {}, {}); + expect(result).toMatchObject({ ok: false, code: RPC_ERROR_CODES.malformed }); + if (!result.ok) expect(result.message).not.toContain("{not json"); + }); + + test("an HTML error page yields a structured failure with no page content surviving", async () => { + const transport = transportWith( + () => + new Response("500 Internal Server Error at db-host-42", { + status: 200, + headers: { "content-type": "text/html" }, + }), + ); + const result = await transport.call(target, {}, {}); + expect(result).toMatchObject({ ok: false, code: RPC_ERROR_CODES.malformed }); + if (!result.ok) expect(result.message).not.toContain("db-host-42"); + }); + + test("the AbortSignal reaches fetch", async () => { + const controller = new AbortController(); + let seenSignal: AbortSignal | undefined; + const transport = transportWith(async (request) => { + seenSignal = request.signal; + return Response.json({ ok: true, value: {} }); + }); + await transport.call(target, {}, { signal: controller.signal }); + expect(seenSignal).toBeDefined(); + }); });