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 { type AppTicketStore, SECRET_LENGTH } from "./appticket.ts";13import { forwardableHeaders, sanitizeCloseCode } from "./tunnel.ts";1415const DEBUG = Deno.env.get("LOFI_NODE_DEBUG") === "1";16function debug(...args: unknown[]): void {17 if (DEBUG) console.error("[gate]", ...args);18}1920/** WS close code sent when a ticket is revoked mid-session; the app should21 * surface re-enrollment (see docs/app-ticket.md). */22export const CLOSE_TICKET_REVOKED = 4001;2324const TICKET_PATH = new RegExp(`^/t/([A-Za-z0-9_-]{${SECRET_LENGTH}})(/.*)?$`);2526export interface GateOptions {27 port: number;28 hostname?: string;29 /** Internal Jazz http URL, read per-request so Jazz restarts (pair()) never30 * invalidate the gate. */31 target: () => string;32 mode: "open" | "ticket";33 store: AppTicketStore;34 /** The node's Jazz app id + admin secret, for provision-scoped injection35 * and the store-status endpoint. The secret never transits the gate inbound36 * — client-supplied X-Jazz-Admin-Secret headers are stripped in ticket37 * mode; the gate injects config's own for provision-scoped requests. */38 appId: string;39 adminSecret: string;40}4142export interface Gate {43 port: number;44 url: string;45 /** Live gated WS connections per ticket id. */46 stats(): { ticketId: string; connections: number }[];47 close(): Promise<void>;48}4950function unauthorized(): Response {51 return new Response(JSON.stringify({ error: "invalid_ticket" }), {52 status: 401,53 headers: { "content-type": "application/json" },54 });55}5657/** Pump two OPEN (or opening) WebSockets both ways until either closes.58 * `backlog` holds client data that arrived before the upstream dial finished;59 * flushed before the listener attaches (same task — no interleave). */60export function pumpWebSockets(61 client: WebSocket,62 upstream: WebSocket,63 backlog: (string | ArrayBuffer)[] = [],64): void {65 client.binaryType = "arraybuffer";66 upstream.binaryType = "arraybuffer";67 let torndown = false;68 const teardown = (code: number, reason: string) => {69 if (torndown) return;70 torndown = true;71 for (const ws of [client, upstream]) {72 if (ws.readyState === WebSocket.OPEN || ws.readyState === WebSocket.CONNECTING) {73 try {74 ws.close(sanitizeCloseCode(code), reason.slice(0, 120));75 } catch {76 // already closing77 }78 }79 }80 };8182 const forward = (to: WebSocket) => (ev: MessageEvent) => {83 try {84 to.send(ev.data);85 } catch {86 teardown(1011, "gate forward failed");87 }88 };89 for (const data of backlog) upstream.send(data);90 backlog.length = 0;91 client.addEventListener("message", forward(upstream));92 upstream.addEventListener("message", forward(client));93 client.addEventListener("close", (ev) => teardown(ev.code || 1000, ev.reason ?? ""));94 upstream.addEventListener("close", (ev) => teardown(ev.code || 1000, ev.reason ?? ""));95 client.addEventListener("error", () => teardown(1011, "client ws error"));96 upstream.addEventListener("error", () => teardown(1011, "upstream ws error"));97}9899async function proxyHttp(100 req: Request,101 targetBase: string,102 path: string,103 headerOverrides?: Record<string, string | null>,104): Promise<Response> {105 try {106 const headers = forwardableHeaders(req.headers);107 for (const [name, value] of Object.entries(headerOverrides ?? {})) {108 if (value === null) delete headers[name];109 else headers[name] = value;110 }111 const res = await fetch(new URL(path, targetBase), {112 method: req.method,113 headers,114 body: req.method === "GET" || req.method === "HEAD" ? undefined : req.body,115 });116 return new Response(res.body, {117 status: res.status,118 headers: forwardableHeaders(res.headers),119 });120 } catch (e) {121 debug("http proxy error", (e as Error).message);122 return new Response("upstream unavailable", { status: 502 });123 }124}125126/** Node-served, metadata-only store preflight: lets a sync-scoped client127 * classify `no_schema` / hash-mismatch instead of hanging on writes (lofi#109128 * failure surface). Never returns schema contents, policies, or secrets. */129async function storeStatus(130 targetBase: string,131 appId: string,132 adminSecret: string,133): Promise<Response> {134 try {135 const admin = { "X-Jazz-Admin-Secret": adminSecret };136 const schemasRes = await fetch(new URL(`/apps/${appId}/schemas`, targetBase), {137 headers: admin,138 });139 if (!schemasRes.ok) {140 await schemasRes.body?.cancel();141 debug("store-status schemas fetch", schemasRes.status);142 return new Response(JSON.stringify({ error: "store_unavailable" }), {143 status: 502,144 headers: { "content-type": "application/json" },145 });146 }147 const body = await schemasRes.json() as {148 hashes?: string[];149 schemas?: { hash: string; publishedAt?: number | string }[];150 };151 // Head = newest published schema; the bare `hashes` array is NOT ordered.152 const entries = body.schemas ?? [];153 let headHash: string | undefined;154 let newest = -Infinity;155 for (const entry of entries) {156 const at = typeof entry.publishedAt === "string"157 ? Date.parse(entry.publishedAt)158 : entry.publishedAt ?? 0;159 if (at >= newest) {160 newest = at;161 headHash = entry.hash;162 }163 }164 const hashes = body.hashes ?? entries.map((e) => e.hash);165 headHash ??= hashes[hashes.length - 1];166 let permissionsHead: string | null = null;167 try {168 const headRes = await fetch(169 new URL(`/apps/${appId}/admin/permissions/head`, targetBase),170 { headers: admin },171 );172 if (headRes.ok) {173 permissionsHead = ((await headRes.json()) as { head?: string | null }).head ?? null;174 } else {175 await headRes.body?.cancel();176 }177 } catch {178 // permissions head stays null179 }180 const schema = hashes.length > 0181 ? { deployed: true, headHash, permissionsHead }182 : { deployed: false };183 return new Response(JSON.stringify({ v: 1, appId, schema }), {184 headers: { "content-type": "application/json" },185 });186 } catch (e) {187 debug("store-status error", (e as Error).message);188 return new Response(JSON.stringify({ error: "store_unavailable" }), {189 status: 502,190 headers: { "content-type": "application/json" },191 });192 }193}194195const ADMIN_PATH = /^\/apps\/[^/]+\/admin(\/|$)/;196197function proxyWebSocket(req: Request, targetBase: string, path: string): {198 response: Response;199 socket: WebSocket;200} {201 const requestedProtocols = req.headers.get("sec-websocket-protocol");202 const { socket, response } = Deno.upgradeWebSocket(203 req,204 requestedProtocols ? { protocol: requestedProtocols.split(",")[0].trim() } : {},205 );206 // Buffer client data while the upstream dial is in flight; pumpWebSockets207 // flushes before attaching its own listener (same task, no gap).208 const backlog: (string | ArrayBuffer)[] = [];209 const buffer = (ev: MessageEvent) => backlog.push(ev.data);210 socket.addEventListener("message", buffer);211 socket.addEventListener("open", () => {212 const wsUrl = new URL(path, targetBase);213 wsUrl.protocol = wsUrl.protocol === "https:" ? "wss:" : "ws:";214 const upstream = requestedProtocols215 ? new WebSocket(wsUrl, requestedProtocols.split(",").map((s) => s.trim()))216 : new WebSocket(wsUrl);217 upstream.addEventListener("open", () => {218 socket.removeEventListener("message", buffer);219 pumpWebSockets(socket, upstream, backlog);220 });221 upstream.addEventListener("error", () => {222 try {223 socket.close(1011, "upstream dial failed");224 } catch {225 // already closed226 }227 });228 });229 return { response, socket };230}231232export function startGate(options: GateOptions): Gate {233 const hostname = options.hostname ?? "0.0.0.0";234 const liveByTicket = new Map<string, Set<WebSocket>>();235236 const track = (ticketId: string, socket: WebSocket) => {237 let set = liveByTicket.get(ticketId);238 if (!set) liveByTicket.set(ticketId, set = new Set());239 set.add(socket);240 socket.addEventListener("close", () => {241 set.delete(socket);242 if (set.size === 0) liveByTicket.delete(ticketId);243 });244 };245246 // Revocation sweep: piggyback on the store's hot-reload — poll live tickets247 // and close sockets whose ticket flipped to revoked. Cheap (only runs while248 // gated sockets exist).249 const sweep = setInterval(async () => {250 if (liveByTicket.size === 0) return;251 const records = await options.store.list();252 for (const [ticketId, sockets] of liveByTicket) {253 const record = records.find((r) => r.id === ticketId);254 if (record && !record.revokedAt) continue;255 for (const socket of [...sockets]) {256 try {257 socket.close(CLOSE_TICKET_REVOKED, "ticket revoked");258 } catch {259 // already closing260 }261 }262 liveByTicket.delete(ticketId);263 }264 }, 2000);265266 const server = Deno.serve(267 { hostname, port: options.port, onListen: () => {} },268 async (req) => {269 const url = new URL(req.url);270 const isUpgrade = req.headers.get("upgrade")?.toLowerCase() === "websocket";271 const targetBase = options.target();272273 if (options.mode === "open") {274 // Open nodes still serve the preflight so lofi's store classifier275 // works against dev setups.276 if (url.pathname === "/store-status" && !isUpgrade) {277 return storeStatus(targetBase, options.appId, options.adminSecret);278 }279 const path = url.pathname + url.search;280 return isUpgrade281 ? proxyWebSocket(req, targetBase, path).response282 : proxyHttp(req, targetBase, path);283 }284285 // Ticket mode.286 if (url.pathname === "/health" && !isUpgrade) {287 return proxyHttp(req, targetBase, "/health");288 }289 const match = url.pathname.match(TICKET_PATH);290 if (!match) return new Response("not found", { status: 404 });291 const [, secret, rest] = match;292 const verdict = await options.store.verify(secret);293 if (verdict.status !== "valid") {294 debug("rejected", verdict.status, url.pathname.slice(0, 24));295 return unauthorized();296 }297 const scope = verdict.record.scope ?? "sync";298 const strippedPath = rest || "/";299 if (strippedPath === "/store-status" && !isUpgrade) {300 return storeStatus(targetBase, options.appId, options.adminSecret);301 }302 // Admin/catalogue-mutating routes need provision scope; a sync ticket303 // gets the SAME 401 shape as an invalid ticket (nothing to enumerate).304 if (ADMIN_PATH.test(strippedPath) && scope !== "provision") {305 debug("rejected sync-scope on admin path", verdict.record.id);306 return unauthorized();307 }308 const path = strippedPath + url.search;309 debug(isUpgrade ? "ws" : req.method, path, "ticket", verdict.record.id, scope);310 if (isUpgrade) {311 const { response, socket } = proxyWebSocket(req, targetBase, path);312 track(verdict.record.id, socket);313 return response;314 }315 // The admin secret only ever originates from the node: inbound headers316 // are stripped; provision-scoped requests get config's secret injected317 // (on catalogue reads too — the merge flow fetches the head schema318 // verbatim without ever holding the secret client-side).319 return proxyHttp(req, targetBase, path, {320 "x-jazz-admin-secret": scope === "provision" ? options.adminSecret : null,321 });322 },323 );324325 const port = (server.addr as Deno.NetAddr).port;326 return {327 port,328 url: `http://127.0.0.1:${port}`,329 stats: () =>330 [...liveByTicket.entries()].map(([ticketId, sockets]) => ({331 ticketId,332 connections: sockets.size,333 })),334 close: async () => {335 clearInterval(sweep);336 await server.shutdown();337 },338 };339}