1// The access gate: lofi-node's public front listener. JazzServer binds2// loopback-only, so the gate is BOTH the enforcement point for app tickets3// and what makes the node reachable beyond localhost at all.4//5// Ticket mode routes requests shaped /t/<secret>/… — the secret arrives as a6// serverUrl base-path prefix (jazz clients preserve base paths through every7// WS connect and catalogue fetch), gets verified against the ticket store8// (timing-safe, digest vs digest), and is stripped before proxying to the9// internal Jazz port. WS is terminated and re-originated (safe: Jazz auth is10// an in-band frame, proven by the iroh tunnel doing the same); HTTP streams.1112import {13 type AppTicketStore,14 encodeAppTicket,15 isRevokedByLineage,16 SECRET_LENGTH,17} from "./appticket.ts";18import { forwardableHeaders, sanitizeCloseCode } from "./tunnel.ts";1920const DEBUG = Deno.env.get("LOFI_NODE_DEBUG") === "1";21function debug(...args: unknown[]): void {22 if (DEBUG) console.error("[gate]", ...args);23}2425/** WS close code sent when a ticket is revoked mid-session; the app should26 * surface re-enrollment (see docs/app-ticket.md). */27export const CLOSE_TICKET_REVOKED = 4001;2829const TICKET_PATH = new RegExp(`^/t/([A-Za-z0-9_-]{${SECRET_LENGTH}})(/.*)?$`);3031export interface GateOptions {32 port: number;33 hostname?: string;34 /** Internal Jazz http URL, read per-request so Jazz restarts (pair()) never35 * invalidate the gate. */36 target: () => string;37 mode: "open" | "ticket";38 store: AppTicketStore;39 /** The node's Jazz app id + admin secret, for provision-scoped injection40 * and the store-status endpoint. The secret never transits the gate inbound41 * — client-supplied X-Jazz-Admin-Secret headers are stripped in ticket42 * mode; the gate injects config's own for provision-scoped requests. */43 appId: string;44 adminSecret: string;45 /** Base URL embedded into tickets minted by the derive endpoint (the46 * node's publicUrl); defaults to the gate's own loopback URL — same47 * election as CLI-issued tickets. */48 publicBase?: string;49 /** The node's iroh EndpointTicket, carried on derived tickets exactly like50 * node-issued ones. */51 nodeTicket?: string;52}5354export interface Gate {55 port: number;56 url: string;57 /** Live gated WS connections per ticket id. */58 stats(): { ticketId: string; connections: number }[];59 close(): Promise<void>;60}6162function unauthorized(): Response {63 return new Response(JSON.stringify({ error: "invalid_ticket" }), {64 status: 401,65 headers: { "content-type": "application/json" },66 });67}6869/** Pump two OPEN (or opening) WebSockets both ways until either closes.70 * `backlog` holds client data that arrived before the upstream dial finished;71 * flushed before the listener attaches (same task — no interleave). */72export function pumpWebSockets(73 client: WebSocket,74 upstream: WebSocket,75 backlog: (string | ArrayBuffer)[] = [],76): void {77 client.binaryType = "arraybuffer";78 upstream.binaryType = "arraybuffer";79 let torndown = false;80 const teardown = (code: number, reason: string) => {81 if (torndown) return;82 torndown = true;83 for (const ws of [client, upstream]) {84 if (ws.readyState === WebSocket.OPEN || ws.readyState === WebSocket.CONNECTING) {85 try {86 ws.close(sanitizeCloseCode(code), reason.slice(0, 120));87 } catch {88 // already closing89 }90 }91 }92 };9394 const forward = (to: WebSocket) => (ev: MessageEvent) => {95 try {96 to.send(ev.data);97 } catch {98 teardown(1011, "gate forward failed");99 }100 };101 for (const data of backlog) upstream.send(data);102 backlog.length = 0;103 client.addEventListener("message", forward(upstream));104 upstream.addEventListener("message", forward(client));105 client.addEventListener("close", (ev) => teardown(ev.code || 1000, ev.reason ?? ""));106 upstream.addEventListener("close", (ev) => teardown(ev.code || 1000, ev.reason ?? ""));107 client.addEventListener("error", () => teardown(1011, "client ws error"));108 upstream.addEventListener("error", () => teardown(1011, "upstream ws error"));109}110111async function proxyHttp(112 req: Request,113 targetBase: string,114 path: string,115 headerOverrides?: Record<string, string | null>,116): Promise<Response> {117 try {118 const headers = forwardableHeaders(req.headers);119 for (const [name, value] of Object.entries(headerOverrides ?? {})) {120 if (value === null) delete headers[name];121 else headers[name] = value;122 }123 const res = await fetch(new URL(path, targetBase), {124 method: req.method,125 headers,126 body: req.method === "GET" || req.method === "HEAD" ? undefined : req.body,127 });128 return new Response(res.body, {129 status: res.status,130 headers: forwardableHeaders(res.headers),131 });132 } catch (e) {133 debug("http proxy error", (e as Error).message);134 return new Response("upstream unavailable", { status: 502 });135 }136}137138/** Node-served, metadata-only store preflight: lets a sync-scoped client139 * classify `no_schema` / hash-mismatch instead of hanging on writes (lofi#109140 * failure surface). Never returns schema contents, policies, or secrets. */141async function storeStatus(142 targetBase: string,143 appId: string,144 adminSecret: string,145): Promise<Response> {146 try {147 const admin = { "X-Jazz-Admin-Secret": adminSecret };148 const schemasRes = await fetch(new URL(`/apps/${appId}/schemas`, targetBase), {149 headers: admin,150 });151 if (!schemasRes.ok) {152 await schemasRes.body?.cancel();153 debug("store-status schemas fetch", schemasRes.status);154 return new Response(JSON.stringify({ error: "store_unavailable" }), {155 status: 502,156 headers: { "content-type": "application/json" },157 });158 }159 const body = await schemasRes.json() as {160 hashes?: string[];161 schemas?: { hash: string; publishedAt?: number | string }[];162 };163 // Head = newest published schema; the bare `hashes` array is NOT ordered.164 const entries = body.schemas ?? [];165 let headHash: string | undefined;166 let newest = -Infinity;167 for (const entry of entries) {168 const at = typeof entry.publishedAt === "string"169 ? Date.parse(entry.publishedAt)170 : entry.publishedAt ?? 0;171 if (at >= newest) {172 newest = at;173 headHash = entry.hash;174 }175 }176 const hashes = body.hashes ?? entries.map((e) => e.hash);177 headHash ??= hashes[hashes.length - 1];178 let permissionsHead: string | null = null;179 try {180 const headRes = await fetch(181 new URL(`/apps/${appId}/admin/permissions/head`, targetBase),182 { headers: admin },183 );184 if (headRes.ok) {185 permissionsHead = ((await headRes.json()) as { head?: string | null }).head ?? null;186 } else {187 await headRes.body?.cancel();188 }189 } catch {190 // permissions head stays null191 }192 const schema = hashes.length > 0193 ? { deployed: true, headHash, permissionsHead }194 : { deployed: false };195 return new Response(JSON.stringify({ v: 1, appId, schema }), {196 headers: { "content-type": "application/json" },197 });198 } catch (e) {199 debug("store-status error", (e as Error).message);200 return new Response(JSON.stringify({ error: "store_unavailable" }), {201 status: 502,202 headers: { "content-type": "application/json" },203 });204 }205}206207const ADMIN_PATH = /^\/apps\/[^/]+\/admin(\/|$)/;208209function proxyWebSocket(req: Request, targetBase: string, path: string): {210 response: Response;211 socket: WebSocket;212} {213 const requestedProtocols = req.headers.get("sec-websocket-protocol");214 const { socket, response } = Deno.upgradeWebSocket(215 req,216 requestedProtocols ? { protocol: requestedProtocols.split(",")[0].trim() } : {},217 );218 // Buffer client data while the upstream dial is in flight; pumpWebSockets219 // flushes before attaching its own listener (same task, no gap).220 const backlog: (string | ArrayBuffer)[] = [];221 const buffer = (ev: MessageEvent) => backlog.push(ev.data);222 socket.addEventListener("message", buffer);223 socket.addEventListener("open", () => {224 const wsUrl = new URL(path, targetBase);225 wsUrl.protocol = wsUrl.protocol === "https:" ? "wss:" : "ws:";226 const upstream = requestedProtocols227 ? new WebSocket(wsUrl, requestedProtocols.split(",").map((s) => s.trim()))228 : new WebSocket(wsUrl);229 upstream.addEventListener("open", () => {230 socket.removeEventListener("message", buffer);231 pumpWebSockets(socket, upstream, backlog);232 });233 upstream.addEventListener("error", () => {234 try {235 socket.close(1011, "upstream dial failed");236 } catch {237 // already closed238 }239 });240 });241 return { response, socket };242}243244export function startGate(options: GateOptions): Gate {245 const hostname = options.hostname ?? "0.0.0.0";246 const liveByTicket = new Map<string, Set<WebSocket>>();247248 const track = (ticketId: string, socket: WebSocket) => {249 let set = liveByTicket.get(ticketId);250 if (!set) liveByTicket.set(ticketId, set = new Set());251 set.add(socket);252 socket.addEventListener("close", () => {253 set.delete(socket);254 if (set.size === 0) liveByTicket.delete(ticketId);255 });256 };257258 // Revocation sweep: piggyback on the store's hot-reload — poll live tickets259 // and close sockets whose ticket flipped to revoked. Cheap (only runs while260 // gated sockets exist).261 const sweep = setInterval(async () => {262 if (liveByTicket.size === 0) return;263 const records = await options.store.list();264 for (const [ticketId, sockets] of liveByTicket) {265 const record = records.find((r) => r.id === ticketId);266 if (record && !isRevokedByLineage(record, records)) continue;267 for (const socket of [...sockets]) {268 try {269 socket.close(CLOSE_TICKET_REVOKED, "ticket revoked");270 } catch {271 // already closing272 }273 }274 liveByTicket.delete(ticketId);275 }276 }, 2000);277278 const server = Deno.serve(279 { hostname, port: options.port, onListen: () => {} },280 async (req) => {281 const url = new URL(req.url);282 const isUpgrade = req.headers.get("upgrade")?.toLowerCase() === "websocket";283 const targetBase = options.target();284285 if (options.mode === "open") {286 // Open nodes still serve the preflight so lofi's store classifier287 // works against dev setups.288 if (url.pathname === "/store-status" && !isUpgrade) {289 return storeStatus(targetBase, options.appId, options.adminSecret);290 }291 const path = url.pathname + url.search;292 return isUpgrade293 ? proxyWebSocket(req, targetBase, path).response294 : proxyHttp(req, targetBase, path);295 }296297 // Ticket mode.298 if (url.pathname === "/health" && !isUpgrade) {299 return proxyHttp(req, targetBase, "/health");300 }301 const match = url.pathname.match(TICKET_PATH);302 if (!match) return new Response("not found", { status: 404 });303 const [, secret, rest] = match;304 const verdict = await options.store.verify(secret);305 if (verdict.status !== "valid") {306 debug("rejected", verdict.status, url.pathname.slice(0, 24));307 return unauthorized();308 }309 const scope = verdict.record.scope ?? "sync";310 const strippedPath = rest || "/";311 if (strippedPath === "/store-status" && !isUpgrade) {312 return storeStatus(targetBase, options.appId, options.adminSecret);313 }314 // Scope-down exchange: a provision ticket mints a derived sync ticket315 // (the lofi app persists that one at rest, keeping the provision ticket316 // sealed or memory-only). The derived record carries parentId, so317 // revoking the provision ticket kills its derived tickets too. A sync318 // ticket gets the SAME 401 as an invalid one (nothing to enumerate).319 if (strippedPath === "/derive-sync-ticket" && !isUpgrade) {320 if (scope !== "provision") {321 debug("rejected sync-scope on derive", verdict.record.id);322 return unauthorized();323 }324 if (req.method !== "POST") {325 return new Response("method not allowed", { status: 405 });326 }327 let requested: string | undefined;328 try {329 const body = await req.json() as { label?: unknown };330 if (typeof body.label === "string") requested = body.label;331 } catch {332 // empty or non-JSON body: fall through to the default label333 }334 const parent = verdict.record;335 const label = requested ?? `${parent.label ?? parent.id} (sync)`;336 const derived = await options.store.issue(label, "sync", parent.id);337 const base = (options.publicBase ??338 `http://127.0.0.1:${(server.addr as Deno.NetAddr).port}`).replace(/\/+$/, "");339 const ticket = encodeAppTicket({340 v: 1,341 appId: options.appId,342 url: `${base}/t/${derived.secret}`,343 label,344 node: options.nodeTicket,345 });346 debug("derived sync ticket", derived.record.id, "from", parent.id);347 return new Response(JSON.stringify({ v: 1, id: derived.record.id, ticket }), {348 headers: { "content-type": "application/json" },349 });350 }351 // Admin/catalogue-mutating routes need provision scope; a sync ticket352 // gets the SAME 401 shape as an invalid ticket (nothing to enumerate).353 if (ADMIN_PATH.test(strippedPath) && scope !== "provision") {354 debug("rejected sync-scope on admin path", verdict.record.id);355 return unauthorized();356 }357 const path = strippedPath + url.search;358 debug(isUpgrade ? "ws" : req.method, path, "ticket", verdict.record.id, scope);359 if (isUpgrade) {360 const { response, socket } = proxyWebSocket(req, targetBase, path);361 track(verdict.record.id, socket);362 return response;363 }364 // The admin secret only ever originates from the node: inbound headers365 // are stripped; provision-scoped requests get config's secret injected366 // (on catalogue reads too — the merge flow fetches the head schema367 // verbatim without ever holding the secret client-side).368 return proxyHttp(req, targetBase, path, {369 "x-jazz-admin-secret": scope === "provision" ? options.adminSecret : null,370 });371 },372 );373374 const port = (server.addr as Deno.NetAddr).port;375 return {376 port,377 url: `http://127.0.0.1:${port}`,378 stats: () =>379 [...liveByTicket.entries()].map(([ticketId, sockets]) => ({380 ticketId,381 connections: sockets.size,382 })),383 close: async () => {384 clearInterval(sweep);385 await server.shutdown();386 },387 };388}