security(broker): harden telegram bridge for production

- Validate JWT signature + expiry in /start (was only decoding, not verifying)
- Constant-time signature comparison in telegram-token.ts (prevent timing attacks)
- Rate limit /tg/token endpoint: 10 requests/hour per IP
- Grammy bot.catch() error handler (prevent unhandled rejections crashing broker)
- Cap WS reconnect attempts at 20 (prevent infinite retry loop)
- Expire stale pendingDMs entries (prevent memory leak)

Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>

apps/broker/src/index.ts

@@ -142,6 +142,10 @@ interface PeerConn {
 const connections = new Map<string, PeerConn>();
 const connectionsPerMesh = new Map<string, number>();
 
+// Rate limiter for /tg/token endpoint (IP → count, cleared hourly)
+const tgTokenRateLimit = new Map<string, number>();
+setInterval(() => tgTokenRateLimit.clear(), 60 * 60_000).unref();
+
 // --- URL Watch engine ---
 interface WatchEntry {
   id: string;
@@ -630,8 +634,16 @@ function handleHttpRequest(req: IncomingMessage, res: ServerResponse): void {
     return;
   }
 
-  // Telegram connect token
+  // Telegram connect token (rate-limited: 10 requests/hour per IP)
   if (req.method === "POST" && req.url === "/tg/token") {
+    const clientIp = (req.headers["x-forwarded-for"] as string)?.split(",")[0]?.trim() ?? req.socket.remoteAddress ?? "unknown";
+    const tgRateBucket = `tg-token:${clientIp}`;
+    const tgRateCount = (tgTokenRateLimit.get(tgRateBucket) ?? 0) + 1;
+    tgTokenRateLimit.set(tgRateBucket, tgRateCount);
+    if (tgRateCount > 10) {
+      writeJson(res, 429, { error: "Rate limit exceeded. Max 10 tokens per hour." });
+      return;
+    }
     const chunks: Buffer[] = [];
     req.on("data", (c: Buffer) => chunks.push(c));
     req.on("end", () => {

apps/broker/src/telegram-bridge.ts

@@ -12,6 +12,7 @@
 import { Bot, InputFile } from "grammy";
 import WebSocket from "ws";
 import sodium from "libsodium-wrappers";
+import { validateTelegramConnectToken } from "./telegram-token";
 
 // ---------------------------------------------------------------------------
 // Types
@@ -314,12 +315,20 @@ class MeshConnection {
         this.connected = false;
         this.ws = null;
         if (this.reconnectTimer) return;
+        const MAX_RECONNECT_ATTEMPTS = 20;
+        if (this.reconnectAttempt >= MAX_RECONNECT_ATTEMPTS) {
+          console.error(
+            `[tg-bridge] mesh ${this.creds.meshId.slice(0, 8)} giving up after ${MAX_RECONNECT_ATTEMPTS} attempts`,
+          );
+          meshConnections.delete(this.creds.meshId);
+          return;
+        }
         const delays = [1000, 2000, 4000, 8000, 16000, 30000];
         const delay =
           delays[Math.min(this.reconnectAttempt, delays.length - 1)]!;
         this.reconnectAttempt++;
         console.log(
-          `[tg-bridge] mesh ${this.creds.meshId.slice(0, 8)} reconnecting in ${delay}ms (attempt ${this.reconnectAttempt})`,
+          `[tg-bridge] mesh ${this.creds.meshId.slice(0, 8)} reconnecting in ${delay}ms (attempt ${this.reconnectAttempt}/${MAX_RECONNECT_ATTEMPTS})`,
         );
         this.reconnectTimer = setTimeout(() => {
           this.reconnectTimer = null;
@@ -672,31 +681,20 @@ function setupBotCommands(
       return;
     }
 
-    // Decode JWT token (3-part base64url)
+    // Validate JWT signature, expiry, and claims
-    let payload: any;
+    const encKey = process.env.BROKER_ENCRYPTION_KEY;
-    try {
+    if (!encKey) {
-      const parts = token.split(".");
+      await ctx.reply("❌ Broker not configured for token validation.");
-      if (parts.length !== 3) throw new Error("not a JWT");
+      return;
-      payload = JSON.parse(
+    }
-        Buffer.from(parts[1]!, "base64url").toString("utf-8"),
+
-      );
+    const payload = validateTelegramConnectToken(token, encKey);
-    } catch {
+    if (!payload) {
-      await ctx.reply("❌ Invalid or expired token. Request a new link.");
+      await ctx.reply("❌ Invalid, expired, or tampered token. Request a new link.");
       return;
     }
 
-    // Validate required fields
     const { meshId, memberId, pubkey, secretKey, meshSlug } = payload;
-    if (!meshId || !memberId || !pubkey || !secretKey) {
-      await ctx.reply("❌ Malformed token — missing credentials.");
-      return;
-    }
-
-    // Check expiry
-    if (payload.expiresAt && Date.now() > payload.expiresAt) {
-      await ctx.reply("❌ Token expired. Request a new connect link.");
-      return;
-    }
 
     const chatId = ctx.chat.id;
     const chatType = ctx.chat.type;
@@ -1394,6 +1392,20 @@ export async function bootTelegramBridge(
     }
   }
 
+  // Grammy global error handler — prevents unhandled rejections from crashing broker
+  bot.catch((err) => {
+    console.error("[tg-bridge] Grammy error:", err.message ?? err);
+  });
+
+  // Expire stale pendingDMs entries every 5 minutes (prevent memory leak)
+  setInterval(() => {
+    // pendingDMs has no timestamp, so we just cap size — clear all if > 1000
+    if (pendingDMs.size > 1000) {
+      console.warn(`[tg-bridge] clearing ${pendingDMs.size} stale pendingDMs`);
+      pendingDMs.clear();
+    }
+  }, 5 * 60_000).unref();
+
   // Wire up bot commands
   setupBotCommands(
     bot,

apps/broker/src/telegram-token.ts

@@ -94,7 +94,12 @@ export function validateTelegramConnectToken(
     // Verify signature
     const signingInput = `${headerB64}.${payloadB64}`;
     const expectedSignature = sign(signingInput, secret);
-    if (signatureB64 !== expectedSignature) return null;
+    // Constant-time comparison to prevent timing attacks
+    const a = Buffer.from(signatureB64);
+    const b = Buffer.from(expectedSignature);
+    if (a.length !== b.length) return null;
+    const { timingSafeEqual } = require("node:crypto");
+    if (!timingSafeEqual(a, b)) return null;
 
     // Verify header algorithm
     const header = JSON.parse(base64urlDecode(headerB64));