crossmate

room-worker.js (17253B)

---

 // Coarse debounce for the durable `lastSeenAt` write on the message hot path.
 // The room TTL only needs second-ish granularity, so a write + alarm reschedule
 // on every keystroke broadcast is pure churn. In-memory, so it resets to a
 // guaranteed write on each hibernation wake — which is fine.
 const TOUCH_DEBOUNCE_MS = 30 * 1000;
 
 export class EngagementRegisterLimiter {
   constructor(state, env) {
     this.state = state;
     this.env = env;
   }
 
   async fetch(request) {
     const url = new URL(request.url);
     const roomID = url.searchParams.get("roomID") || "";
     if (!roomID) {
       return new Response("Missing room ID", { status: 400 });
     }
 
     const result = await this.checkRoomRegisterRateLimit(request, roomID);
     return result.ok ? new Response(null, { status: 204 }) : rateLimitedResponse(result);
   }
 
   async checkRoomRegisterRateLimit(request, roomID) {
     const ip = request.headers.get("CF-Connecting-IP") || "unknown";
     const config = rateLimitConfig(this.env, "ROOM_REGISTER_IP", 60, 60 * 60);
     const now = Date.now();
     const windowMillis = config.windowSeconds * 1000;
     const cutoff = now - windowMillis;
     const keySecret = rateLimitKeySecret(this.env);
     const key = await rateLimitStorageKey("room-register:ip", ip, keySecret);
     const roomToken = await rateLimitToken(roomID, keySecret);
     const stored = await this.state.storage.get(key);
     const recent = Array.isArray(stored)
       ? stored.filter((entry) => entry && typeof entry.at === "number" && entry.at > cutoff && typeof entry.room === "string")
       : [];
     recent.sort((left, right) => left.at - right.at);
 
     const existing = recent.find((entry) => entry.room === roomToken);
     if (existing) {
       existing.at = now;
       await this.state.storage.put(key, recent);
       await this.ensureSweepScheduled(windowMillis);
       return { ok: true };
     }
 
     if (recent.length >= config.limit) {
       const retryAfterSeconds = Math.max(1, Math.ceil((recent[0].at + windowMillis - now) / 1000));
       await this.state.storage.put(key, recent);
       await this.ensureSweepScheduled(windowMillis);
       return { ok: false, retryAfterSeconds };
     }
 
     recent.push({ room: roomToken, at: now });
     await this.state.storage.put(key, recent);
     await this.ensureSweepScheduled(windowMillis);
     return { ok: true };
   }
 
   // Rate keys are written per client IP and would otherwise persist forever
   // once an IP stops registering. Arm-if-unarmed keeps the sweep at most one
   // per window even under constant traffic.
   async ensureSweepScheduled(windowMillis) {
     const scheduled = await this.state.storage.getAlarm();
     if (scheduled === null) {
       await this.state.storage.setAlarm(Date.now() + windowMillis);
     }
   }
 
   async alarm() {
     const config = rateLimitConfig(this.env, "ROOM_REGISTER_IP", 60, 60 * 60);
     const windowMillis = config.windowSeconds * 1000;
     const cutoff = Date.now() - windowMillis;
     const entries = await this.state.storage.list({ prefix: "rate:" });
     let liveKeys = 0;
     for (const [key, stored] of entries) {
       const newest = Array.isArray(stored)
         ? stored.reduce((max, entry) => (entry && typeof entry.at === "number" && entry.at > max ? entry.at : max), 0)
         : 0;
       if (newest <= cutoff) {
         await this.state.storage.delete(key);
       } else {
         liveKeys += 1;
       }
     }
     if (liveKeys > 0) {
       await this.state.storage.setAlarm(Date.now() + windowMillis);
     }
   }
 }
 
 export class EngagementRoom {
   constructor(state, env) {
     this.state = state;
     this.env = env;
     // Answer keepalive pings without waking the Durable Object, so an otherwise
     // idle foreground socket stays warm through NAT / edge idle timeouts. Pairs
     // with the client's periodic "ping" (EngagementHost.startPing). Auto-answered
     // frames never reach `webSocketMessage`, so they cost nothing here.
     this.state.setWebSocketAutoResponse(new WebSocketRequestResponsePair("ping", "pong"));
   }
 
   async fetch(request) {
     const url = new URL(request.url);
     const route = roomRouteFromPath(url.pathname);
     if (!route) {
       return new Response("Missing room ID", { status: 400 });
     }
 
     if (route.endpoint === "register") {
       if (request.method !== "POST") {
         return new Response("Method not allowed", { status: 405 });
       }
       return this.register(request);
     }
 
     const upgrade = request.headers.get("Upgrade");
     if (upgrade !== "websocket") {
       return new Response("Expected WebSocket upgrade", { status: 426 });
     }
 
     const auth = await this.authenticate(route.roomID, url.searchParams);
     if (!auth.ok) {
       return new Response(auth.message, { status: auth.status });
     }
 
     const pair = new WebSocketPair();
     const [client, server] = Object.values(pair);
     server.serializeAttachment({
       authorID: auth.authorID,
       deviceID: auth.deviceID,
       connectedAt: Date.now()
     });
     // Supersede any earlier socket from the same device. A reconnect would
     // otherwise leave the stale one lingering until its TCP dies, so the room
     // fans every broadcast at a zombie and can echo the device's own prior
     // frames back to it. `server` is not accepted yet, so it is not in the set.
     for (const existing of this.state.getWebSockets()) {
       const attachment = existing.deserializeAttachment();
       if (attachment && attachment.authorID === auth.authorID && attachment.deviceID === auth.deviceID) {
         try {
           existing.close(1000, "Superseded by a newer connection");
         } catch {
           // Already closing; the runtime will reap it.
         }
       }
     }
     this.state.acceptWebSocket(server);
 
     return new Response(null, {
       status: 101,
       webSocket: client
     });
   }
 
   // Registers the room secret ahead of a socket connect. First write wins:
   // the secret is created if absent, confirmed if it matches, and refused if
   // it differs. The secret arrives in the request body, never in a URL.
   //
   // Every legitimate holder of the room creds (distributed via the shared
   // CloudKit Game record) registers idempotently before each connect, so an
   // idle-expired room is resurrected by whichever peer returns first and
   // there is no ordering race between the minter and its peers. An attacker
   // who learns a room ID (it appears in URL paths) but not the secret can
   // neither register over a live room nor sign a connect.
   async register(request) {
     let body;
     try {
       body = await request.json();
     } catch {
       return new Response("Invalid JSON body", { status: 400 });
     }
     const secret = typeof body.secret === "string" ? body.secret : "";
     if (!isAcceptableSecret(secret)) {
       return new Response("Invalid secret", { status: 400 });
     }
 
     const stored = await this.state.storage.get("secret");
     if (stored) {
       if (!timingSafeEqual(stored, secret)) {
         return new Response("Room secret mismatch", { status: 409 });
       }
     } else {
       await this.state.storage.put("secret", secret);
       await this.state.storage.put("createdAt", Date.now());
       // TOFU-era rooms stored only a connect-time hash; the registered
       // secret supersedes it.
       await this.state.storage.delete("secretHash");
     }
 
     await this.state.storage.put("lastSeenAt", Date.now());
     await this.scheduleExpiry();
     return new Response(null, { status: stored ? 204 : 201 });
   }
 
   async authenticate(roomID, params) {
     const authorID = params.get("authorID") || "";
     const deviceID = params.get("deviceID") || "";
     const timestamp = params.get("timestamp") || "";
     const nonce = params.get("nonce") || "";
     const signature = params.get("signature") || "";
 
     if (!authorID || !deviceID || !timestamp || !nonce || !signature) {
       return { ok: false, status: 401, message: "Missing auth parameters" };
     }
 
     // The secret never travels on a connect; it must have been registered
     // via `register` (clients re-register idempotently before each connect,
     // which also resurrects a room whose idle expiry wiped this storage).
     const secret = await this.state.storage.get("secret");
     if (!secret) {
       return { ok: false, status: 403, message: "Room not registered" };
     }
 
     const nowSeconds = Math.floor(Date.now() / 1000);
     const timestampSeconds = Number(timestamp);
     const maxSkewSeconds = Number(this.env.MAX_AUTH_SKEW_SECONDS || "120");
     if (!Number.isFinite(timestampSeconds) || Math.abs(nowSeconds - timestampSeconds) > maxSkewSeconds) {
       return { ok: false, status: 401, message: "Stale auth timestamp" };
     }
 
     const nonceKey = `nonce:${nonce}`;
     if (await this.state.storage.get(nonceKey)) {
       return { ok: false, status: 401, message: "Nonce already used" };
     }
 
     const payload = [roomID, authorID, deviceID, timestamp, nonce].join("|");
     const expectedSignature = await hmacSHA256(secret, payload);
     if (!timingSafeEqual(signature, expectedSignature)) {
       return { ok: false, status: 401, message: "Invalid signature" };
     }
 
     await this.state.storage.put("lastSeenAt", Date.now());
     await this.state.storage.put(nonceKey, Date.now());
     await this.pruneNonces();
     await this.scheduleExpiry();
     this.lastTouchAt = Date.now();
 
     return { ok: true, authorID, deviceID };
   }
 
   async webSocketMessage(ws, message) {
     // Bound the relay before any storage or fanout work. The client enforces
     // the same limit (EngagementMessage.maxEncodedFrameBytes) on send and
     // receive, so only a hostile or badly broken peer trips this; closing the
     // socket (1009: message too big) is cheaper for the room than repeatedly
     // dropping its frames. String frames are measured in UTF-16 code units,
     // a lower bound on their UTF-8 size — real payloads travel as binary.
     const frameBytes = typeof message === "string" ? message.length : message.byteLength;
     if (frameBytes > maxFrameBytes(this.env)) {
       try {
         ws.close(1009, "Frame exceeds size limit");
       } catch {
         // Already closing; the runtime will reap it.
       }
       return;
     }
     const data = typeof message === "string" ? message : message.slice(0);
     await this.touch();
     for (const peer of this.state.getWebSockets()) {
       if (peer === ws) continue;
       try {
         peer.send(data);
       } catch {
         // A peer caught mid-close throws here; skip it so the rest of the room
         // still receives this frame. The dead socket is reaped via
         // `webSocketClose` / runtime cleanup.
       }
     }
   }
 
   async webSocketClose(ws, code, reason) {
     await this.state.storage.put("lastSeenAt", Date.now());
     await this.scheduleExpiry();
     ws.close(code, reason);
   }
 
   async alarm() {
     await this.pruneNonces();
     const ttlMs = Number(this.env.ROOM_TTL_SECONDS || "600") * 1000;
     const sockets = this.state.getWebSockets();
     if (sockets.length > 0) {
       // Still connected — the room is alive regardless of how long since the
       // last broadcast (keepalive pings are auto-answered and don't advance
       // `lastSeenAt`). Re-arm from now, not the stale deadline, or a long idle
       // session would refire the alarm in a tight loop.
       await this.state.storage.setAlarm(Date.now() + ttlMs);
       return;
     }
 
     const lastSeenAt = await this.state.storage.get("lastSeenAt");
     const idleMs = Date.now() - (lastSeenAt || Date.now());
     if (idleMs > ttlMs) {
       await this.state.storage.deleteAll();
       return;
     }
     await this.scheduleExpiry();
   }
 
   async touch() {
     const now = Date.now();
     // Debounce the durable write + alarm reschedule off the message hot path.
     if (this.lastTouchAt && now - this.lastTouchAt < TOUCH_DEBOUNCE_MS) return;
     this.lastTouchAt = now;
     await this.state.storage.put("lastSeenAt", now);
     await this.scheduleExpiry();
   }
 
   async pruneNonces() {
     const maxAgeMs = Number(this.env.NONCE_TTL_SECONDS || "300") * 1000;
     const cutoff = Date.now() - maxAgeMs;
     const nonces = await this.state.storage.list({ prefix: "nonce:" });
     for (const [key, createdAt] of nonces) {
       if (createdAt < cutoff) {
         await this.state.storage.delete(key);
       }
     }
   }
 
   async scheduleExpiry() {
     const ttlMs = Number(this.env.ROOM_TTL_SECONDS || "600") * 1000;
     const lastSeenAt = (await this.state.storage.get("lastSeenAt")) || Date.now();
     await this.state.storage.setAlarm(lastSeenAt + ttlMs);
   }
 }
 
 export default {
   async fetch(request, env) {
     const url = new URL(request.url);
     if (url.pathname === "/health") {
       return new Response("ok");
     }
     const route = roomRouteFromPath(url.pathname);
     if (!route) {
       return new Response("Not found", { status: 404 });
     }
     if (route.endpoint === "register" && request.method === "POST") {
       const limited = await checkRegisterRateLimit(request, env, route.roomID);
       if (limited) return limited;
     }
     const id = env.ENGAGEMENT_ROOMS.idFromName(route.roomID);
     return env.ENGAGEMENT_ROOMS.get(id).fetch(request);
   }
 };
 
 function roomRouteFromPath(pathname) {
   const match = pathname.match(/^\/rooms\/([^/]+)\/(socket|register)$/);
   return match ? { roomID: match[1], endpoint: match[2] } : null;
 }
 
 async function checkRegisterRateLimit(request, env, roomID) {
   // Fail closed: without the limiter binding, register would be unmetered
   // room creation. A misconfigured deploy should refuse registration (clients
   // re-register idempotently and durable Moves sync backstops realtime), not
   // silently drop the gate.
   if (!env.ENGAGEMENT_REGISTER_LIMITER) {
     return new Response("Rate limiter unavailable", { status: 503 });
   }
   const id = env.ENGAGEMENT_REGISTER_LIMITER.idFromName("register");
   const url = new URL(request.url);
   url.searchParams.set("roomID", roomID);
   const response = await env.ENGAGEMENT_REGISTER_LIMITER.get(id).fetch(
     new Request(url.toString(), {
       method: "POST",
       headers: request.headers
     })
   );
   return response.status === 204 ? null : response;
 }
 
 // Mirrors the client's EngagementMessage.maxEncodedFrameBytes (512 KiB);
 // Cloudflare's own per-message cap (1 MiB) is the outer backstop.
 function maxFrameBytes(env) {
   const parsed = Number(env.ROOM_MAX_FRAME_BYTES || "524288");
   return Number.isFinite(parsed) && parsed > 0 ? Math.floor(parsed) : 524288;
 }
 
 function rateLimitConfig(env, prefix, defaultLimit, defaultWindowSeconds) {
   const limit = Number(env[`${prefix}_LIMIT`] || String(defaultLimit));
   const windowSeconds = Number(env[`${prefix}_WINDOW_SECONDS`] || String(defaultWindowSeconds));
   return {
     limit: Number.isFinite(limit) && limit > 0 ? Math.floor(limit) : defaultLimit,
     windowSeconds: Number.isFinite(windowSeconds) && windowSeconds > 0
       ? Math.floor(windowSeconds)
       : defaultWindowSeconds
   };
 }
 
 function rateLimitedResponse(result) {
   return new Response("Rate limit exceeded", {
     status: 429,
     headers: {
       "Retry-After": String(result.retryAfterSeconds)
     }
   });
 }
 
 async function rateLimitStorageKey(bucket, identity, secret) {
   return `rate:${bucket}:${await rateLimitToken(identity, secret)}`;
 }
 
 async function rateLimitToken(identity, secret) {
   const key = await crypto.subtle.importKey(
     "raw",
     new TextEncoder().encode(secret),
     { name: "HMAC", hash: "SHA-256" },
     false,
     ["sign"]
   );
   const signature = await crypto.subtle.sign("HMAC", key, new TextEncoder().encode(String(identity || "")));
   return base64URLEncode(new Uint8Array(signature));
 }
 
 function rateLimitKeySecret(env) {
   return env.RATE_LIMIT_HASH_KEY || "crossmate-room-rate-limit-v1";
 }
 
 // The secret doubles as the HMAC key for connect signatures, so a registered
 // value must decode to at least 32 key bytes (clients mint exactly 32).
 function isAcceptableSecret(secret) {
   if (!secret) return false;
   let bytes;
   try {
     bytes = base64URLDecode(secret);
   } catch {
     return false;
   }
   return bytes.length >= 32;
 }
 
 async function hmacSHA256(secret, payload) {
   const key = await crypto.subtle.importKey(
     "raw",
     base64URLDecode(secret),
     { name: "HMAC", hash: "SHA-256" },
     false,
     ["sign"]
   );
   const signature = await crypto.subtle.sign("HMAC", key, new TextEncoder().encode(payload));
   return base64URLEncode(new Uint8Array(signature));
 }
 
 function base64URLDecode(value) {
   const base64 = value.replace(/-/g, "+").replace(/_/g, "/").padEnd(Math.ceil(value.length / 4) * 4, "=");
   const binary = atob(base64);
   return Uint8Array.from(binary, (char) => char.charCodeAt(0));
 }
 
 function base64URLEncode(bytes) {
   let binary = "";
   for (const byte of bytes) {
     binary += String.fromCharCode(byte);
   }
   return btoa(binary).replace(/\+/g, "-").replace(/\//g, "_").replace(/=+$/g, "");
 }
 
 function timingSafeEqual(a, b) {
   const left = new TextEncoder().encode(a);
   const right = new TextEncoder().encode(b);
   if (left.length !== right.length) return false;
   let diff = 0;
   for (let index = 0; index < left.length; index += 1) {
     diff |= left[index] ^ right[index];
   }
   return diff === 0;
 }