diff --git a/docs/public-api-0.8.json b/docs/public-api-0.8.json index cc8fdd3f..a043b7ed 100644 --- a/docs/public-api-0.8.json +++ b/docs/public-api-0.8.json @@ -2413,6 +2413,7 @@ "inProcessTransport", "isRetryableStatus", "procedure", + "resolveAppOrigin", "rpcPath", "rpcSecret", "serviceClient", diff --git a/packages/dev-server/src/gateway.ts b/packages/dev-server/src/gateway.ts index 6255cf53..491d91d2 100644 --- a/packages/dev-server/src/gateway.ts +++ b/packages/dev-server/src/gateway.ts @@ -461,6 +461,14 @@ export async function startGateway(opts: GatewayOptions): Promise [app.name, app.publicOrigin ?? `http://${app.domains[0]}:${port}`]), ), ); + // Loopback-only origins, computed up front (ports are assigned by index + // before any child spawns) so every child can reach every other child + // directly — bypassing the gateway, which 404s the RPC prefix by design. + const internalOriginsEnv: Readonly> = Object.freeze( + Object.fromEntries( + opts.apps.map((app, i) => [app.name, `http://127.0.0.1:${app.port ?? port + 1 + i}`]), + ), + ); const displayHost = hostname === "0.0.0.0" || hostname === "::" ? "localhost" : hostname; // When the CLI is executed directly from a framework checkout, keep child // apps on that same source tree. Resolving the package name from an external @@ -498,6 +506,7 @@ export async function startGateway(opts: GatewayOptions): Promise> | undefined; - const loadServices = () => - (servicesPromise ??= (async () => { - const services = new Map(); - for (const entry of router.services) { - const imported = await loadModule(entry.file); - const implementation = imported.default as ServiceImplementation | undefined; - if (!implementation || typeof implementation.invoke !== "function") { - throw new Error(`RPC service ${entry.file} must default-export implement(...)`); + const loadServices = (): Promise> => { + if (!servicesPromise) { + servicesPromise = (async () => { + const services = new Map(); + for (const entry of router.services) { + const imported = await loadModule(entry.file); + const implementation = imported.default as ServiceImplementation | undefined; + if (!implementation || typeof implementation.invoke !== "function") { + throw new Error(`RPC service ${entry.file} must default-export implement(...)`); + } + if (implementation.contract.name !== entry.name) { + throw new Error( + `RPC service file ${entry.file} is mounted as "${entry.name}" (its filename) ` + + `but its contract is named "${implementation.contract.name}". Rename the file to ` + + `match the contract, or rename the contract to match the file.`, + ); + } + services.set(entry.name, implementation); } - services.set(entry.name, implementation); - } - return services; - })()); + return services; + })().catch((error) => { + servicesPromise = undefined; + const app = process.env.WRNEXUS_APP_NAME ?? "app"; + console.error( + `[wrnexus] failed to load RPC services (${app}):`, + error instanceof Error ? (error.stack ?? error.message) : error, + ); + throw error; + }); + } + return servicesPromise; + }; // Server-side realtime room manager (shared by every `defineRoom` connection). const realtime = createRealtimeRegistry(); @@ -964,8 +987,24 @@ export function createHandlers(deps: RuntimeDeps): Handlers { const secure = (res: Response): Response => withSecurityHeaders(req, res, mode, runtimeSecurity, nonce); - const rpcResponse = await handleRpcRequest(req, url, await loadServices()); - if (rpcResponse) return secure(rpcResponse); + if (isRpcPath(url.pathname)) { + let services: Map; + try { + services = await loadServices(); + } catch (error) { + const app = process.env.WRNEXUS_APP_NAME ?? "app"; + const detail = error instanceof Error ? (error.stack ?? error.message) : String(error); + console.error(`[wrnexus] RPC request failed to load services (${app})\n${detail}`); + return secure( + Response.json( + { ok: false, code: "RPC_UNKNOWN", message: "Service unavailable", retryable: false }, + { headers: { "cache-control": "private, no-store" } }, + ), + ); + } + const rpcResponse = await handleRpcRequest(req, url, services); + if (rpcResponse) return secure(rpcResponse); + } const preflight = createCorsPreflightResponse(req, deps.security); if (preflight) return secure(preflight); diff --git a/packages/dev-server/test/rpc-services-loading.test.ts b/packages/dev-server/test/rpc-services-loading.test.ts new file mode 100644 index 00000000..b839e8fa --- /dev/null +++ b/packages/dev-server/test/rpc-services-loading.test.ts @@ -0,0 +1,170 @@ +import { afterEach, expect, test } from "bun:test"; +import { mkdirSync, mkdtempSync, rmSync, writeFileSync } from "node:fs"; +import { join } from "node:path"; +import { tmpdir } from "node:os"; +import { buildRouter } from "@wrnexus/router"; +import { defineService, implement, procedure } from "@wrnexus/rpc"; +import { v } from "@wrnexus/validation"; +import { createHandlers, type RuntimeDeps } from "../src/runtime.ts"; + +const roots: string[] = []; +afterEach(() => roots.splice(0).forEach((root) => rmSync(root, { recursive: true, force: true }))); + +const greeter = defineService({ + name: "greeter", + procedures: { + greet: procedure + .input(v.object({ name: v.string() })) + .output<{ message: string }>() + .build(), + }, +}); + +function makeHandlers(app: string, loadModule: RuntimeDeps["loadModule"]) { + return createHandlers({ + mode: "production", + hmr: false, + router: buildRouter(app), + loadModule, + getMiddleware: async () => [], + assets: { serve: async () => null }, + } satisfies RuntimeDeps); +} + +test("a bad file under app/services does not break unrelated routes", async () => { + const root = mkdtempSync(join(tmpdir(), "wrnexus-rpc-bad-service-")); + roots.push(root); + const app = join(root, "app"); + mkdirSync(join(app, "pages"), { recursive: true }); + mkdirSync(join(app, "services"), { recursive: true }); + writeFileSync(join(app, "pages/home.ts"), "export default () => '';\n"); + // A co-located helper with no default export — the natural thing a + // developer drops next to a real service. + writeFileSync(join(app, "services/types.ts"), "export type Foo = string;\n"); + + const handlers = makeHandlers(app, async (file) => { + const normalized = file.replace(/\\/g, "/"); + if (normalized.endsWith("services/types.ts")) return {}; // no default export + if (normalized.endsWith("pages/home.ts")) return { default: () => "

home

" }; + throw new Error(`unexpected module: ${file}`); + }); + + const res = await handlers.fetch(new Request("https://example.test/home"), { + upgrade: () => false, + }); + expect(await res!.text()).toContain("home"); +}); + +test("the RPC path returns a structured failure instead of throwing when service load fails", async () => { + const root = mkdtempSync(join(tmpdir(), "wrnexus-rpc-load-fail-")); + roots.push(root); + const app = join(root, "app"); + mkdirSync(join(app, "services"), { recursive: true }); + writeFileSync(join(app, "services/types.ts"), "export type Foo = string;\n"); + + const handlers = makeHandlers(app, async () => ({})); // no default export + + const res = await handlers.fetch( + new Request("https://example.test/__wrnexus/rpc/types/greet", { + method: "POST", + headers: { "content-type": "application/json", "x-wrnexus-internal": "1" }, + body: "{}", + }), + { upgrade: () => false }, + ); + expect(res).toBeDefined(); + expect(res!.status).toBeLessThan(500); + const body = await res!.json(); + expect(body).toMatchObject({ ok: false, code: "RPC_UNKNOWN" }); +}); + +test("a request after a load failure re-attempts rather than serving a cached rejection", async () => { + const root = mkdtempSync(join(tmpdir(), "wrnexus-rpc-retry-")); + roots.push(root); + const app = join(root, "app"); + mkdirSync(join(app, "services"), { recursive: true }); + writeFileSync(join(app, "services/greeter.ts"), "export default {};\n"); + + let attempt = 0; + const handlers = makeHandlers(app, async () => { + attempt += 1; + if (attempt === 1) return {}; // fails: no default export + return { + default: implement( + greeter, + { greet: async ({ name }) => ({ message: `Hi ${name}` }) }, + { + selfApp: "greeter", + }, + ), + }; + }); + + const call = () => + handlers.fetch( + new Request("https://example.test/__wrnexus/rpc/greeter/greet", { + method: "POST", + headers: { "content-type": "application/json", "x-wrnexus-internal": "1" }, + body: JSON.stringify({ name: "Ada" }), + }), + { upgrade: () => false }, + ); + + const first = await call(); + expect(await first!.json()).toMatchObject({ ok: false, code: "RPC_UNKNOWN" }); + + const second = await call(); + expect(await second!.json()).toEqual({ ok: true, value: { message: "Hi Ada" } }); + expect(attempt).toBe(2); +}); + +test("a contract name that does not match its mounted filename fails loudly, naming both", async () => { + const root = mkdtempSync(join(tmpdir(), "wrnexus-rpc-name-mismatch-")); + roots.push(root); + const app = join(root, "app"); + mkdirSync(join(app, "services"), { recursive: true }); + writeFileSync(join(app, "services/greeter.ts"), "export default {};\n"); + + const billing = defineService({ + name: "billing", + procedures: { + createInvoice: procedure + .input(v.object({ amountCents: v.number() })) + .output<{ invoiceId: string }>() + .build(), + }, + }); + const service = implement( + billing, + { createInvoice: async ({ amountCents }) => ({ invoiceId: `inv_${amountCents}` }) }, + { selfApp: "billing" }, + ); + + const handlers = makeHandlers(app, async () => ({ default: service })); + + // The typed client would call /__wrnexus/rpc/billing/... (contract name), + // but the file mounts as "greeter" — the client's request 404s as unknown. + const res = await handlers.fetch( + new Request("https://example.test/__wrnexus/rpc/billing/createInvoice", { + method: "POST", + headers: { "content-type": "application/json", "x-wrnexus-internal": "1" }, + body: JSON.stringify({ amountCents: 5 }), + }), + { upgrade: () => false }, + ); + const body = await res!.json(); + expect(body).toMatchObject({ ok: false, code: "RPC_UNKNOWN" }); + + // And the actual mounted name ("greeter") also fails: the implementation's + // contract does not match the filename it was mounted under. + const res2 = await handlers.fetch( + new Request("https://example.test/__wrnexus/rpc/greeter/createInvoice", { + method: "POST", + headers: { "content-type": "application/json", "x-wrnexus-internal": "1" }, + body: JSON.stringify({ amountCents: 5 }), + }), + { upgrade: () => false }, + ); + const body2 = await res2!.json(); + expect(body2).toMatchObject({ ok: false, code: "RPC_UNKNOWN" }); +}); diff --git a/packages/rpc/README.md b/packages/rpc/README.md index 9cf468d5..ca6744f6 100644 --- a/packages/rpc/README.md +++ b/packages/rpc/README.md @@ -42,3 +42,32 @@ Service files default-export `implement(...)` from `app/services`. The developme Pass `{ as: ctx }` to `serviceClient` to propagate the subject. The signed token contains only subject and tenant identifiers; permissions are always checked by the callee. Set `WRNEXUS_RPC_SECRET` in every app, use at least 32 characters, and never reuse the session secret. Calls time out by default. Retrying is intentionally deferred; when introduced, only procedures marked `.idempotent()` may be retried. + +## Deployment requirement: apps must be unreachable except through the gateway + +`/__wrnexus/rpc/*` is authenticated by TWO signals together: a marker header +(`x-wrnexus-internal: 1`) AND the absence of any `X-Forwarded-*` header. The +WrNexus gateway satisfies this by construction — it strips any inbound +marker header from the public request, and it always adds `X-Forwarded-*` +when proxying to an app. A direct loopback call from a sibling app process +carries the marker and no forwarded headers, so it passes; anything that +came through the gateway carries forwarded headers, so it's rejected even if +it also carries the marker. + +**This check only works if the app process is unreachable except through the +gateway.** If an app's port is exposed directly, or if a reverse proxy sits +in front of it WITHOUT setting `X-Forwarded-*` (a bare `proxy_pass` with no +`proxy_set_header X-Forwarded-For`/`X-Forwarded-Host`/`X-Forwarded-Proto`), +then an external caller can set the marker header itself, arrive with no +forwarded headers, and reach `/__wrnexus/rpc/*` as if it were an internal +call — bypassing the gateway's edge block entirely. + +Requirements for any deployment: + +- App processes must bind to a private/loopback interface and be reachable + ONLY through the gateway (or an equivalent trusted front door) — never + exposed directly to the internet or an untrusted network. +- Any reverse proxy placed in front of an app (nginx, a load balancer, etc.) + MUST set `X-Forwarded-For`, `X-Forwarded-Host`, and `X-Forwarded-Proto` on + every request it forwards. Omitting these silently reopens the private RPC + namespace to anyone who can reach the proxy. diff --git a/packages/rpc/src/client.ts b/packages/rpc/src/client.ts index b4764e71..246a4f67 100644 --- a/packages/rpc/src/client.ts +++ b/packages/rpc/src/client.ts @@ -71,7 +71,7 @@ export function serviceClient( try { const result = await Promise.race([callPromise, timeout]); if (result.ok) return result.value; - throw new ServiceError(result.code, result.message); + throw new ServiceError(result.code, result.message, result.retryable); } finally { clearTimeout(timer); // If the timeout won the race, the transport call may still settle diff --git a/packages/rpc/src/errors.ts b/packages/rpc/src/errors.ts index 02f7f709..303a1e90 100644 --- a/packages/rpc/src/errors.ts +++ b/packages/rpc/src/errors.ts @@ -59,11 +59,19 @@ export class ServiceError extends Error { readonly code: string; readonly retryable: boolean; - constructor(code: string, message: string) { + /** + * `retryable` defaults to the code-derived value for callers that + * construct a `ServiceError` directly. Pass it explicitly when relaying a + * wire result: the transport already computed the authoritative value + * (e.g. a bounded HTTP-status check), and recomputing it here from the + * code alone would silently flip it — `RPC_TRANSPORT` derives to `true`, + * even for a non-retryable 403. + */ + constructor(code: string, message: string, retryable?: boolean) { super(message); this.name = "ServiceError"; this.code = code; - this.retryable = retryableFor(code); + this.retryable = retryable ?? retryableFor(code); } /** diff --git a/packages/rpc/src/http.ts b/packages/rpc/src/http.ts index d4f72bf7..a473975b 100644 --- a/packages/rpc/src/http.ts +++ b/packages/rpc/src/http.ts @@ -11,6 +11,41 @@ export function rpcPath(service: string, procedure: string): string { return `${RPC_PATH_PREFIX}/${service}/${procedure}`; } +function parseOriginMap(value: string | undefined): Record { + if (!value) return {}; + try { + const parsed: unknown = JSON.parse(value); + if (!parsed || typeof parsed !== "object" || Array.isArray(parsed)) return {}; + return parsed as Record; + } catch { + return {}; + } +} + +/** + * Resolve the origin an RPC call to `app` should target. + * + * Prefer `WRNEXUS_INTERNAL_ORIGINS` (loopback origins the gateway hands each + * child before spawning it) over `appOrigin`, which resolves the app's + * PUBLIC origin. The public origin is the wrong target for RPC: the gateway + * unconditionally 404s the reserved `/__wrnexus/rpc` prefix on anything that + * arrives at a public origin — that block is the whole point, it is what + * keeps inter-app calls off the public internet. Falling back to `appOrigin` + * when no internal-origin map is present keeps single-app and test setups + * (which only set `WRNEXUS_WORKSPACE_ORIGINS`) working. + */ +export function resolveAppOrigin(app: string): string { + const internalOrigin = parseOriginMap(process.env.WRNEXUS_INTERNAL_ORIGINS)[app]; + if (internalOrigin) { + try { + return new URL(internalOrigin).origin; + } catch { + // Malformed internal-origin entry — fall through to the public origin. + } + } + return appOrigin(app); +} + export interface HttpTransportOptions { resolveOrigin?: (app: string) => string; fetch?: typeof fetch; @@ -29,7 +64,7 @@ function isServiceResult(value: unknown): value is ServiceResult { } export function httpTransport(options: HttpTransportOptions = {}): Transport { - const resolveOrigin = options.resolveOrigin ?? appOrigin; + const resolveOrigin = options.resolveOrigin ?? resolveAppOrigin; const doFetch = options.fetch ?? fetch; return { async call(target: RpcTarget, payload: unknown, callOptions: CallOptions) { diff --git a/packages/rpc/src/index.ts b/packages/rpc/src/index.ts index b4a854b3..15d15c72 100644 --- a/packages/rpc/src/index.ts +++ b/packages/rpc/src/index.ts @@ -42,5 +42,11 @@ export type { } from "./server.ts"; export { serviceClient } from "./client.ts"; export type { ServiceClient, ServiceClientOptions } from "./client.ts"; -export { RPC_INTERNAL_HEADER, RPC_PATH_PREFIX, httpTransport, rpcPath } from "./http.ts"; +export { + RPC_INTERNAL_HEADER, + RPC_PATH_PREFIX, + httpTransport, + resolveAppOrigin, + rpcPath, +} from "./http.ts"; export type { HttpTransportOptions } from "./http.ts"; diff --git a/packages/rpc/test/client.test.ts b/packages/rpc/test/client.test.ts index 0ea3412b..4d3ea68d 100644 --- a/packages/rpc/test/client.test.ts +++ b/packages/rpc/test/client.test.ts @@ -189,4 +189,29 @@ describe("serviceClient", () => { expect(seenTarget?.app).toBe("billing"); expect(seenTarget?.service).toBe("billing"); }); + + test("a wire-level non-retryable failure (e.g. a 403) surfaces at the client as retryable: false", async () => { + // RPC_TRANSPORT recomputes to retryable: true from the code alone — that + // is correct for a 5xx, but the transport already determined a 403 is + // NOT retryable via the bounded status check. The client must pass that + // wire value through rather than recompute it from the code. + const client = serviceClient(billing, { + transport: { + call: async () => ({ + ok: false, + code: RPC_ERROR_CODES.transport, + message: "Service returned 403", + retryable: false, + }), + }, + }); + let caught: unknown; + try { + await client.createInvoice({ amountCents: 1 }); + } catch (err) { + caught = err; + } + expect(caught).toBeInstanceOf(ServiceError); + expect(caught).toMatchObject({ code: RPC_ERROR_CODES.transport, retryable: false }); + }); }); diff --git a/packages/rpc/test/integration.test.ts b/packages/rpc/test/integration.test.ts index a807809f..c89e1ec3 100644 --- a/packages/rpc/test/integration.test.ts +++ b/packages/rpc/test/integration.test.ts @@ -114,4 +114,55 @@ describe("RPC integration", () => { code: RPC_ERROR_CODES.invalid, }); }); + + test("the default httpTransport reaches the callee over a real socket via the internal origin", async () => { + const service = implement( + billing, + { + createInvoice: async ({ amountCents }, ctx) => ({ + invoiceId: `inv_${amountCents}`, + forSubject: ctx.subject?.subjectId ?? "anon", + }), + }, + { selfApp: "billing", checkPermission: async () => true }, + ); + const server = Bun.serve({ + port: 0, + hostname: "127.0.0.1", + async fetch(req) { + const url = new URL(req.url); + const res = await handleRpcRequest(req, url, new Map([["billing", service]])); + return res ?? new Response("Not found", { status: 404 }); + }, + }); + const originalWorkspace = process.env.WRNEXUS_WORKSPACE_ORIGINS; + const originalInternal = process.env.WRNEXUS_INTERNAL_ORIGINS; + try { + // The workspace (public) origin deliberately points somewhere that + // cannot serve the RPC — the gateway 404s the RPC prefix on any + // request that arrives at a public origin. Only the internal-origin + // map points at the real server. If httpTransport() ever falls back + // to the public origin by default again, this call fails. + process.env.WRNEXUS_WORKSPACE_ORIGINS = JSON.stringify({ + billing: "http://127.0.0.1:1", // unroutable — nothing listens here + }); + process.env.WRNEXUS_INTERNAL_ORIGINS = JSON.stringify({ + billing: `http://127.0.0.1:${server.port}`, + }); + const client = serviceClient(billing, { + app: "billing", + transport: httpTransport(), // no resolveOrigin override — uses the real default + }); + expect(await client.createInvoice({ amountCents: 42 })).toEqual({ + invoiceId: "inv_42", + forSubject: "anon", + }); + } finally { + server.stop(true); + if (originalWorkspace === undefined) delete process.env.WRNEXUS_WORKSPACE_ORIGINS; + else process.env.WRNEXUS_WORKSPACE_ORIGINS = originalWorkspace; + if (originalInternal === undefined) delete process.env.WRNEXUS_INTERNAL_ORIGINS; + else process.env.WRNEXUS_INTERNAL_ORIGINS = originalInternal; + } + }); });