feat(broker): device-code auth with PostgreSQL persistence

- Drizzle schema: device_code + cli_session tables in mesh pgSchema
- Broker endpoints: POST /cli/device-code, GET /cli/device-code/:code,
  POST /cli/device-code/:code/approve, GET /cli/sessions
- Web app API routes now proxy to broker (no in-memory state)
- Tracks devices per user: hostname, platform, arch, last_seen, token_hash
- JWT signed with CLI_SYNC_SECRET, 30-day expiry
- Session revocation support via revokedAt column

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

apps/broker/src/index.ts

@@ -645,6 +645,30 @@ function handleHttpRequest(req: IncomingMessage, res: ServerResponse): void {
     return;
   }
 
+  // --- CLI device-code auth ---
+
+  if (req.method === "POST" && req.url === "/cli/device-code") {
+    handleDeviceCodeNew(req, res, started);
+    return;
+  }
+
+  if (req.method === "GET" && req.url?.startsWith("/cli/device-code/")) {
+    const code = req.url.slice("/cli/device-code/".length).split("?")[0]!;
+    handleDeviceCodePoll(code, res, started);
+    return;
+  }
+
+  if (req.method === "POST" && req.url?.startsWith("/cli/device-code/") && req.url?.endsWith("/approve")) {
+    const code = req.url.slice("/cli/device-code/".length).replace("/approve", "");
+    handleDeviceCodeApprove(req, code, res, started);
+    return;
+  }
+
+  if (req.method === "GET" && req.url?.startsWith("/cli/sessions")) {
+    handleCliSessionsList(req, res, started);
+    return;
+  }
+
   // 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";
@@ -4556,6 +4580,237 @@ function main(): void {
   });
 }
 
+// ---------------------------------------------------------------------------
+// CLI device-code auth handlers
+// ---------------------------------------------------------------------------
+
+import { deviceCode as deviceCodeTable, cliSession as cliSessionTable } from "@turbostarter/db/schema/mesh";
+
+function generateShortCode(len: number): string {
+  const chars = "ABCDEFGHJKLMNPQRSTUVWXYZ23456789";
+  const bytes = new Uint8Array(len);
+  crypto.getRandomValues(bytes);
+  return Array.from(bytes, (b) => chars[b % chars.length]).join("");
+}
+
+async function signCliJwt(payload: Record<string, unknown>): Promise<string> {
+  const secret = env.CLI_SYNC_SECRET;
+  if (!secret) throw new Error("CLI_SYNC_SECRET not configured");
+  const encoder = new TextEncoder();
+  const headerB64 = btoa(JSON.stringify({ alg: "HS256", typ: "JWT" }))
+    .replace(/\+/g, "-").replace(/\//g, "_").replace(/=+$/, "");
+  const payloadB64 = btoa(JSON.stringify(payload))
+    .replace(/\+/g, "-").replace(/\//g, "_").replace(/=+$/, "");
+  const key = await crypto.subtle.importKey(
+    "raw", encoder.encode(secret), { name: "HMAC", hash: "SHA-256" }, false, ["sign"],
+  );
+  const sig = await crypto.subtle.sign("HMAC", key, encoder.encode(`${headerB64}.${payloadB64}`));
+  const sigB64 = btoa(String.fromCharCode(...new Uint8Array(sig)))
+    .replace(/\+/g, "-").replace(/\//g, "_").replace(/=+$/, "");
+  return `${headerB64}.${payloadB64}.${sigB64}`;
+}
+
+async function hashToken(token: string): Promise<string> {
+  const data = new TextEncoder().encode(token);
+  const hash = await crypto.subtle.digest("SHA-256", data);
+  return Array.from(new Uint8Array(hash)).map(b => b.toString(16).padStart(2, "0")).join("");
+}
+
+/** POST /cli/device-code — create a new device code. */
+async function handleDeviceCodeNew(req: IncomingMessage, res: ServerResponse, started: number): Promise<void> {
+  let body: { hostname?: string; platform?: string; arch?: string } = {};
+  try {
+    const chunks: Buffer[] = [];
+    for await (const chunk of req) chunks.push(chunk as Buffer);
+    body = JSON.parse(Buffer.concat(chunks).toString()) as typeof body;
+  } catch {}
+
+  const dc = generateShortCode(16);
+  const uc = generateShortCode(4) + "-" + generateShortCode(4);
+  const expiresAt = new Date(Date.now() + 5 * 60 * 1000);
+  const clientIp = (req.headers["x-forwarded-for"] as string)?.split(",")[0]?.trim() ?? req.socket.remoteAddress ?? "unknown";
+
+  try {
+    await db.insert(deviceCodeTable).values({
+      deviceCode: dc,
+      userCode: uc,
+      hostname: body.hostname,
+      platform: body.platform,
+      arch: body.arch,
+      ipAddress: clientIp,
+      expiresAt,
+    });
+
+    const baseUrl = process.env.APP_URL || "https://claudemesh.com";
+
+    writeJson(res, 200, {
+      device_code: dc,
+      user_code: uc,
+      expires_at: expiresAt.toISOString(),
+      verification_url: `${baseUrl}/cli-auth`,
+    });
+    log.info("device-code", { route: "POST /cli/device-code", user_code: uc, latency_ms: Date.now() - started });
+  } catch (e) {
+    log.error("device-code", { error: e instanceof Error ? e.message : String(e) });
+    writeJson(res, 500, { error: "Failed to create device code" });
+  }
+}
+
+/** GET /cli/device-code/:code — poll device code status. */
+async function handleDeviceCodePoll(code: string, res: ServerResponse, started: number): Promise<void> {
+  try {
+    const [entry] = await db.select().from(deviceCodeTable).where(eq(deviceCodeTable.deviceCode, code)).limit(1);
+
+    if (!entry) {
+      writeJson(res, 200, { status: "expired" });
+      return;
+    }
+
+    if (new Date() > entry.expiresAt && entry.status === "pending") {
+      await db.update(deviceCodeTable).set({ status: "expired" }).where(eq(deviceCodeTable.id, entry.id));
+      writeJson(res, 200, { status: "expired" });
+      return;
+    }
+
+    if (entry.status === "approved" && entry.sessionToken && entry.userId) {
+      // Mark as consumed so it can't be polled again
+      await db.update(deviceCodeTable).set({ status: "consumed" }).where(eq(deviceCodeTable.id, entry.id));
+
+      // Look up user info
+      const [u] = await db.select().from(user).where(eq(user.id, entry.userId)).limit(1);
+
+      writeJson(res, 200, {
+        status: "approved",
+        session_token: entry.sessionToken,
+        user: {
+          id: entry.userId,
+          display_name: u?.name ?? u?.email ?? "User",
+          email: u?.email ?? "",
+        },
+      });
+      log.info("device-code-poll", { route: "GET /cli/device-code/:code", status: "approved", latency_ms: Date.now() - started });
+      return;
+    }
+
+    writeJson(res, 200, { status: entry.status });
+  } catch (e) {
+    log.error("device-code-poll", { error: e instanceof Error ? e.message : String(e) });
+    writeJson(res, 500, { error: "Failed to poll device code" });
+  }
+}
+
+/** POST /cli/device-code/:code/approve — approve from browser (requires sync token). */
+async function handleDeviceCodeApprove(req: IncomingMessage, code: string, res: ServerResponse, started: number): Promise<void> {
+  let body: { user_id: string; email: string; name?: string };
+  try {
+    const chunks: Buffer[] = [];
+    for await (const chunk of req) chunks.push(chunk as Buffer);
+    body = JSON.parse(Buffer.concat(chunks).toString()) as typeof body;
+  } catch {
+    writeJson(res, 400, { error: "Invalid body" });
+    return;
+  }
+
+  if (!body.user_id || !body.email) {
+    writeJson(res, 400, { error: "user_id and email required" });
+    return;
+  }
+
+  try {
+    // Find device code by user_code (browser sends user_code, not device_code)
+    const [entry] = await db.select().from(deviceCodeTable)
+      .where(and(eq(deviceCodeTable.userCode, code), eq(deviceCodeTable.status, "pending")))
+      .limit(1);
+
+    if (!entry) {
+      writeJson(res, 404, { error: "Code not found or expired" });
+      return;
+    }
+
+    if (new Date() > entry.expiresAt) {
+      await db.update(deviceCodeTable).set({ status: "expired" }).where(eq(deviceCodeTable.id, entry.id));
+      writeJson(res, 410, { error: "Code expired" });
+      return;
+    }
+
+    // Sign a CLI session JWT (30 days)
+    const now = Math.floor(Date.now() / 1000);
+    const token = await signCliJwt({
+      sub: body.user_id,
+      email: body.email,
+      name: body.name,
+      type: "cli-session",
+      jti: crypto.randomUUID(),
+      iat: now,
+      exp: now + 30 * 24 * 60 * 60,
+    });
+
+    // Update device code as approved
+    await db.update(deviceCodeTable).set({
+      status: "approved",
+      userId: body.user_id,
+      sessionToken: token,
+      approvedAt: new Date(),
+    }).where(eq(deviceCodeTable.id, entry.id));
+
+    // Create CLI session record
+    await db.insert(cliSessionTable).values({
+      userId: body.user_id,
+      deviceCodeId: entry.id,
+      hostname: entry.hostname,
+      platform: entry.platform,
+      arch: entry.arch,
+      tokenHash: await hashToken(token),
+    });
+
+    writeJson(res, 200, { ok: true });
+    log.info("device-code-approve", {
+      route: "POST /cli/device-code/:code/approve",
+      user_id: body.user_id,
+      hostname: entry.hostname,
+      platform: entry.platform,
+      latency_ms: Date.now() - started,
+    });
+  } catch (e) {
+    log.error("device-code-approve", { error: e instanceof Error ? e.message : String(e) });
+    writeJson(res, 500, { error: "Failed to approve device code" });
+  }
+}
+
+/** GET /cli/sessions?user_id=... — list CLI sessions for a user. */
+async function handleCliSessionsList(req: IncomingMessage, res: ServerResponse, started: number): Promise<void> {
+  const url = new URL(req.url!, "http://localhost");
+  const userId = url.searchParams.get("user_id");
+
+  if (!userId) {
+    writeJson(res, 400, { error: "user_id required" });
+    return;
+  }
+
+  try {
+    const sessions = await db.select().from(cliSessionTable)
+      .where(and(eq(cliSessionTable.userId, userId), isNull(cliSessionTable.revokedAt)))
+      .orderBy(cliSessionTable.createdAt);
+
+    writeJson(res, 200, {
+      sessions: sessions.map(s => ({
+        id: s.id,
+        hostname: s.hostname,
+        platform: s.platform,
+        arch: s.arch,
+        last_seen_at: s.lastSeenAt?.toISOString(),
+        created_at: s.createdAt.toISOString(),
+      })),
+    });
+    log.info("cli-sessions", { route: "GET /cli/sessions", user_id: userId, count: sessions.length, latency_ms: Date.now() - started });
+  } catch (e) {
+    log.error("cli-sessions", { error: e instanceof Error ? e.message : String(e) });
+    writeJson(res, 500, { error: "Failed to list sessions" });
+  }
+}
+
+// ---------------------------------------------------------------------------
+
 // Skip starting the HTTP/WS server when running under vitest — tests import
 // claimInviteV2Core() directly and must not bind ports on module load.
 if (!process.env.VITEST) {

apps/web/src/app/api/auth/cli/device-code/[code]/route.ts

@@ -1,33 +1,15 @@
 import { NextResponse } from "next/server";
-import { deviceCodes } from "../new/route";
+
+const BROKER_URL = (process.env.BROKER_HTTP_URL || "https://ic.claudemesh.com").replace(/\/$/, "");
 
 export async function GET(
   _request: Request,
   { params }: { params: Promise<{ code: string }> },
 ) {
   const { code } = await params;
-  const entry = deviceCodes.get(code);
 
-  if (!entry) {
+  const brokerRes = await fetch(`${BROKER_URL}/cli/device-code/${code}`);
-    return NextResponse.json({ status: "expired" });
+  const brokerBody = await brokerRes.json().catch(() => ({ status: "expired" }));
-  }
 
-  if (Date.now() > entry.expires_at) {
+  return NextResponse.json(brokerBody as Record<string, unknown>, { status: brokerRes.status });
-    entry.status = "expired";
-    deviceCodes.delete(code);
-    return NextResponse.json({ status: "expired" });
-  }
-
-  if (entry.status === "approved") {
-    // Return token once, then clean up
-    const response = {
-      status: "approved",
-      session_token: entry.session_token,
-      user: entry.user,
-    };
-    deviceCodes.delete(code);
-    return NextResponse.json(response);
-  }
-
-  return NextResponse.json({ status: "pending" });
 }

apps/web/src/app/api/auth/cli/device-code/approve-by-user-code/route.ts

@@ -1,10 +1,10 @@
 import { NextResponse } from "next/server";
 import { headers } from "next/headers";
 import { auth } from "@turbostarter/auth/server";
-import { deviceCodes } from "../new/route";
+
+const BROKER_URL = (process.env.BROKER_HTTP_URL || "https://ic.claudemesh.com").replace(/\/$/, "");
 
 export async function POST(request: Request) {
-  // Verify the user is authenticated
   const reqHeaders = new Headers(await headers());
   reqHeaders.set("x-client-platform", "web-server");
   const session = await auth.api.getSession({ headers: reqHeaders });
@@ -24,65 +24,18 @@ export async function POST(request: Request) {
     return NextResponse.json({ error: "user_code required" }, { status: 400 });
   }
 
-  // Find the device code entry by user_code
+  // Proxy approve to the broker
-  let deviceCode: string | null = null;
+  const brokerRes = await fetch(`${BROKER_URL}/cli/device-code/${body.user_code}/approve`, {
-  let entry: (typeof deviceCodes extends Map<string, infer V> ? V : never) | null = null;
+    method: "POST",
+    headers: { "Content-Type": "application/json" },
+    body: JSON.stringify({
+      user_id: session.user.id,
+      email: session.user.email,
+      name: session.user.name,
+    }),
+  });
 
-  for (const [dc, e] of deviceCodes) {
+  const brokerBody = await brokerRes.json().catch(() => ({ error: "Broker error" }));
-    if (e.user_code === body.user_code && e.status === "pending") {
-      deviceCode = dc;
-      entry = e;
-      break;
-    }
-  }
 
-  if (!deviceCode || !entry) {
+  return NextResponse.json(brokerBody as Record<string, unknown>, { status: brokerRes.status });
-    return NextResponse.json({ error: "Code not found or expired" }, { status: 404 });
-  }
-
-  if (Date.now() > entry.expires_at) {
-    deviceCodes.delete(deviceCode);
-    return NextResponse.json({ error: "Code expired" }, { status: 410 });
-  }
-
-  // Sign a CLI session JWT
-  const secret = process.env.CLI_SYNC_SECRET;
-  if (!secret) {
-    return NextResponse.json({ error: "Server not configured" }, { status: 500 });
-  }
-
-  const now = Math.floor(Date.now() / 1000);
-  const payload = {
-    sub: session.user.id,
-    email: session.user.email,
-    name: session.user.name,
-    type: "cli-session",
-    jti: crypto.randomUUID(),
-    iat: now,
-    exp: now + 30 * 24 * 60 * 60,
-  };
-
-  const encoder = new TextEncoder();
-  const headerB64 = btoa(JSON.stringify({ alg: "HS256", typ: "JWT" }))
-    .replace(/\+/g, "-").replace(/\//g, "_").replace(/=+$/, "");
-  const payloadB64 = btoa(JSON.stringify(payload))
-    .replace(/\+/g, "-").replace(/\//g, "_").replace(/=+$/, "");
-  const key = await crypto.subtle.importKey(
-    "raw", encoder.encode(secret), { name: "HMAC", hash: "SHA-256" }, false, ["sign"],
-  );
-  const sig = await crypto.subtle.sign("HMAC", key, encoder.encode(`${headerB64}.${payloadB64}`));
-  const sigB64 = btoa(String.fromCharCode(...new Uint8Array(sig)))
-    .replace(/\+/g, "-").replace(/\//g, "_").replace(/=+$/, "");
-  const token = `${headerB64}.${payloadB64}.${sigB64}`;
-
-  // Mark as approved
-  entry.status = "approved";
-  entry.session_token = token;
-  entry.user = {
-    id: session.user.id,
-    display_name: session.user.name ?? session.user.email ?? "User",
-    email: session.user.email ?? "",
-  };
-
-  return NextResponse.json({ ok: true });
 }

apps/web/src/app/api/auth/cli/device-code/new/route.ts

@@ -1,65 +1,19 @@
 import { NextResponse } from "next/server";
 
-// In-memory store for device codes (production would use Redis/DB)
+const BROKER_URL = (process.env.BROKER_HTTP_URL || "https://ic.claudemesh.com").replace(/\/$/, "");
-// Exported so poll + approve routes can access it
-export const deviceCodes = new Map<
-  string,
-  {
-    user_code: string;
-    status: "pending" | "approved" | "expired";
-    session_token?: string;
-    user?: { id: string; display_name: string; email: string };
-    hostname: string;
-    platform: string;
-    arch: string;
-    created_at: number;
-    expires_at: number;
-  }
->();
-
-function generateCode(len: number): string {
-  const chars = "ABCDEFGHJKLMNPQRSTUVWXYZ23456789";
-  const bytes = new Uint8Array(len);
-  crypto.getRandomValues(bytes);
-  return Array.from(bytes, (b) => chars[b % chars.length]).join("");
-}
-
-// Clean expired codes every 5 min
-setInterval(() => {
-  const now = Date.now();
-  for (const [key, val] of deviceCodes) {
-    if (now > val.expires_at) deviceCodes.delete(key);
-  }
-}, 5 * 60 * 1000);
 
 export async function POST(request: Request) {
-  let body: { hostname?: string; platform?: string; arch?: string };
+  const body = await request.text();
-  try {
-    body = (await request.json()) as typeof body;
-  } catch {
-    body = {};
-  }
 
-  const device_code = generateCode(16);
+  const brokerRes = await fetch(`${BROKER_URL}/cli/device-code`, {
-  const user_code = generateCode(4) + "-" + generateCode(4);
+    method: "POST",
-  const expires_at = Date.now() + 5 * 60 * 1000;
+    headers: {
-
+      "Content-Type": "application/json",
-  const baseUrl = process.env.NEXT_PUBLIC_APP_URL || "https://claudemesh.com";
+      "X-Forwarded-For": request.headers.get("x-forwarded-for") ?? "",
-
+    },
-  deviceCodes.set(device_code, {
+    body,
-    user_code,
-    status: "pending",
-    hostname: body.hostname ?? "unknown",
-    platform: body.platform ?? "unknown",
-    arch: body.arch ?? "unknown",
-    created_at: Date.now(),
-    expires_at,
   });
 
-  return NextResponse.json({
+  const brokerBody = await brokerRes.json().catch(() => ({ error: "Broker error" }));
-    device_code,
+  return NextResponse.json(brokerBody as Record<string, unknown>, { status: brokerRes.status });
-    user_code,
-    expires_at: new Date(expires_at).toISOString(),
-    verification_url: `${baseUrl}/cli-auth`,
-  });
 }

packages/db/src/schema/mesh.ts

@@ -1017,3 +1017,95 @@ export const selectTelegramBridgeSchema = createSelectSchema(telegramBridge);
 export const insertTelegramBridgeSchema = createInsertSchema(telegramBridge);
 export type SelectTelegramBridge = typeof telegramBridge.$inferSelect;
 export type InsertTelegramBridge = typeof telegramBridge.$inferInsert;
+
+// ---------------------------------------------------------------------------
+// CLI device-code authentication
+// ---------------------------------------------------------------------------
+
+export const deviceCodeStatusEnum = meshSchema.enum("device_code_status", [
+  "pending",
+  "approved",
+  "consumed",
+  "expired",
+]);
+
+/**
+ * Device codes for CLI → browser → CLI OAuth flow.
+ * CLI creates a code, browser approves it, CLI polls until approved.
+ */
+export const deviceCode = meshSchema.table("device_code", {
+  id: text().primaryKey().notNull().$defaultFn(generateId),
+  /** Random 16-char code used by CLI to poll. */
+  deviceCode: text().notNull().unique(),
+  /** Human-readable code shown in browser (ABCD-EFGH). */
+  userCode: text().notNull(),
+  status: deviceCodeStatusEnum().notNull().default("pending"),
+  /** Filled on approve — the authenticated user. */
+  userId: text().references(() => user.id, { onDelete: "cascade" }),
+  /** Device info from CLI request. */
+  hostname: text(),
+  platform: text(),
+  arch: text(),
+  ipAddress: text(),
+  /** Signed JWT session token — filled on approve. */
+  sessionToken: text(),
+  createdAt: timestamp().defaultNow().notNull(),
+  approvedAt: timestamp(),
+  expiresAt: timestamp().notNull(),
+}, (table) => [
+  index("device_code_status_idx").on(table.status),
+  index("device_code_user_code_idx").on(table.userCode),
+]);
+
+export const deviceCodeRelations = relations(deviceCode, ({ one }) => ({
+  user: one(user, {
+    fields: [deviceCode.userId],
+    references: [user.id],
+  }),
+}));
+
+export const selectDeviceCodeSchema = createSelectSchema(deviceCode);
+export const insertDeviceCodeSchema = createInsertSchema(deviceCode);
+export type SelectDeviceCode = typeof deviceCode.$inferSelect;
+export type InsertDeviceCode = typeof deviceCode.$inferInsert;
+
+/**
+ * Persistent CLI session records — one per authenticated device.
+ * Enables dashboard "Signed in on N devices" view and per-device revocation.
+ */
+export const cliSession = meshSchema.table("cli_session", {
+  id: text().primaryKey().notNull().$defaultFn(generateId),
+  userId: text()
+    .references(() => user.id, { onDelete: "cascade" })
+    .notNull(),
+  /** Which device-code auth created this session. */
+  deviceCodeId: text().references(() => deviceCode.id),
+  hostname: text(),
+  platform: text(),
+  arch: text(),
+  /** SHA-256 hash of the JWT for revocation lookup. */
+  tokenHash: text().notNull(),
+  lastSeenAt: timestamp().defaultNow(),
+  createdAt: timestamp().defaultNow().notNull(),
+  /** NULL until user revokes from dashboard. */
+  revokedAt: timestamp(),
+}, (table) => [
+  index("cli_session_user_idx").on(table.userId),
+  index("cli_session_token_hash_idx").on(table.tokenHash),
+]);
+
+export const cliSessionRelations = relations(cliSession, ({ one }) => ({
+  user: one(user, {
+    fields: [cliSession.userId],
+    references: [user.id],
+  }),
+  deviceCodeEntry: one(deviceCode, {
+    fields: [cliSession.deviceCodeId],
+    references: [deviceCode.id],
+  }),
+}));
+
+export const selectCliSessionSchema = createSelectSchema(cliSession);
+export const insertCliSessionSchema = createInsertSchema(cliSession);
+export type SelectCliSession = typeof cliSession.$inferSelect;
+export type InsertCliSession = typeof cliSession.$inferInsert;