fix(session-ingest): deny blocked users and stale token peppers

apps/web/src/routers/active-sessions-router.test.ts

@@ -1,5 +1,6 @@
 import { describe, expect, it, jest, beforeAll, afterEach } from '@jest/globals';
 import { TRPCError } from '@trpc/server';
+import jwt from 'jsonwebtoken';
 import { insertTestUser } from '@/tests/helpers/user.helper';
 import { db } from '@/lib/drizzle';
 import { organizations, organization_memberships } from '@kilocode/db/schema';
@@ -30,13 +31,16 @@ import type { createCallerForUser as CreateCallerForUser } from '@/routers/test-
 // `require()`.
 process.env.SESSION_INGEST_WORKER_URL = 'https://test-ingest.example.com';
 let createCallerForUser: typeof CreateCallerForUser;
+let JWT_TOKEN_VERSION: number;
+let TOKEN_EXPIRY: { oneHour: number };
 
 let regularUser: User;
 let testOrganization: Organization;
 
 describe('active-sessions-router', () => {
   beforeAll(async () => {
     ({ createCallerForUser } = await import('@/routers/test-utils'));
+    ({ JWT_TOKEN_VERSION, TOKEN_EXPIRY } = await import('@/lib/tokens'));
 
     regularUser = await insertTestUser({
       google_user_email: 'active-sessions-user@example.com',
@@ -72,6 +76,32 @@ describe('active-sessions-router', () => {
     jest.restoreAllMocks();
   });
 
+  describe('getToken', () => {
+    it('returns a one-hour pepper-bearing API token for the caller', async () => {
+      const knownPepper = 'active-sessions-get-token-pepper';
+      const tokenUser = await insertTestUser({
+        google_user_email: `active-sessions-token-${crypto.randomUUID()}@example.com`,
+        google_user_name: 'Active Sessions Token User',
+        api_token_pepper: knownPepper,
+      });
+
+      const caller = await createCallerForUser(tokenUser.id);
+      const result = await caller.activeSessions.getToken();
+      const payload = jwt.decode(result.token) as jwt.JwtPayload & {
+        kiloUserId: string;
+        apiTokenPepper: string;
+        version: number;
+      };
+
+      expect(payload.kiloUserId).toBe(tokenUser.id);
+      expect(payload.apiTokenPepper).toBe(knownPepper);
+      expect(payload.version).toBe(JWT_TOKEN_VERSION);
+      expect(payload.exp).toBeDefined();
+      expect(payload.iat).toBeDefined();
+      expect((payload.exp ?? 0) - (payload.iat ?? 0)).toBe(TOKEN_EXPIRY.oneHour);
+    });
+  });
+
   describe('listInstances', () => {
     it('returns the instances from the worker when the upstream call succeeds', async () => {
       const fetchSpy = jest.spyOn(global, 'fetch').mockResolvedValue(

apps/web/src/routers/active-sessions-router.ts

@@ -3,7 +3,7 @@ import { baseProcedure, createTRPCRouter } from '@/lib/trpc/init';
 import { TRPCError } from '@trpc/server';
 import { z } from 'zod';
 import { SESSION_INGEST_WORKER_URL } from '@/lib/config.server';
-import { generateInternalServiceToken } from '@/lib/tokens';
+import { generateApiToken, generateInternalServiceToken, TOKEN_EXPIRY } from '@/lib/tokens';
 import { db } from '@/lib/drizzle';
 import { cli_sessions_v2, cloud_agent_session_runs } from '@kilocode/db/schema';
 import { and, desc, eq, gt, inArray, isNotNull, isNull, or, sql } from 'drizzle-orm';
@@ -221,7 +221,9 @@ function throwOrgContextFailure(error: unknown): never {
 
 export const activeSessionsRouter = createTRPCRouter({
   getToken: baseProcedure.query(async ({ ctx }) => {
-    const token = generateInternalServiceToken(ctx.user.id);
+    const token = generateApiToken(ctx.user, undefined, {
+      expiresIn: TOKEN_EXPIRY.oneHour,
+    });
     return { token };
   }),
 

packages/worker-utils/src/kilo-token-auth.ts

@@ -61,6 +61,9 @@ export async function verifyKiloBearerAgainstCurrentPepper(params: {
       return null;
     }
 
+    // This shared helper intentionally normalizes legacy tokens without a
+    // pepper to null. Session-ingest has a separate internal-token class and
+    // must not reuse this normalization for its admission boundary.
     const tokenPepper = payload.apiTokenPepper ?? null;
     if (result.pepper !== tokenPepper) {
       return null;

services/session-ingest/src/app.ts

@@ -4,7 +4,7 @@ import { createMiddleware } from 'hono/factory';
 import type { Env } from './env';
 import { z } from 'zod';
 
-import { kiloJwtAuthMiddleware } from './middleware/kilo-jwt-auth';
+import { kiloJwtAuthMiddleware, USER_AUTH_CACHE_KEY_PREFIX } from './middleware/kilo-jwt-auth';
 import { api } from './routes/api';
 import { cloudAgentSessionScopeApi } from './routes/cloud-agent-session-scope';
 import { getSessionIngestDO } from './dos/SessionIngestDO';
@@ -18,6 +18,9 @@ const invalidateSessionAccessSchema = z.object({
   kiloUserId: z.string().min(1),
   organizationId: z.uuid(),
 });
+const invalidateUserAuthSchema = z.object({
+  kiloUserId: z.string().min(1),
+});
 
 async function hasValidInternalSecret(c: {
   req: { header(name: string): string | undefined };
@@ -44,7 +47,18 @@ const requireValidInternalSecret = createMiddleware<{
   Bindings: Env;
   Variables: { user_id: string };
 }>(async (c, next) => {
-  if (!(await hasValidInternalSecret(c))) {
+  let isValid: boolean;
+  try {
+    isValid = await hasValidInternalSecret(c);
+  } catch (error) {
+    console.error('Auth infrastructure failure', {
+      operation: 'internal-api-secret-get',
+      errorClass: error instanceof Error ? error.name : typeof error,
+      errorMessage: error instanceof Error ? error.message : String(error),
+    });
+    return c.json({ success: false, error: 'Service temporarily unavailable' }, 503);
+  }
+  if (!isValid) {
     return c.json({ success: false, error: 'Unauthorized' }, 401);
   }
   return next();
@@ -112,11 +126,7 @@ app.get('/session/:shareToken/metadata', async c => {
   );
 });
 
-app.post('/internal/session-access/invalidate', async c => {
-  if (!(await hasValidInternalSecret(c))) {
-    return c.json({ success: false, error: 'Unauthorized' }, 401);
-  }
-
+app.post('/internal/session-access/invalidate', requireValidInternalSecret, async c => {
   const parsed = invalidateSessionAccessSchema.safeParse(await c.req.json().catch(() => null));
   if (!parsed.success) {
     return c.json({ success: false, error: 'Invalid request', issues: parsed.error.issues }, 400);
@@ -131,12 +141,18 @@ app.post('/internal/session-access/invalidate', async c => {
   return c.body(null, 204);
 });
 
-// Internal route for service-binding HTTP fetch (secret-protected)
-app.get('/internal/session/:sessionId/export', async c => {
-  if (!(await hasValidInternalSecret(c))) {
-    return c.json({ success: false, error: 'Unauthorized' }, 401);
+app.post('/internal/user-auth/invalidate', requireValidInternalSecret, async c => {
+  const parsed = invalidateUserAuthSchema.safeParse(await c.req.json().catch(() => null));
+  if (!parsed.success) {
+    return c.json({ success: false, error: 'Invalid request', issues: parsed.error.issues }, 400);
   }
 
+  await c.env.USER_EXISTS_CACHE.delete(`${USER_AUTH_CACHE_KEY_PREFIX}${parsed.data.kiloUserId}`);
+  return c.body(null, 204);
+});
+
+// Internal route for service-binding HTTP fetch (secret-protected)
+app.get('/internal/session/:sessionId/export', requireValidInternalSecret, async c => {
   const kiloUserId = c.req.header('X-Kilo-User-Id');
   if (!kiloUserId) return c.json({ success: false, error: 'Missing X-Kilo-User-Id' }, 400);
 

services/session-ingest/src/index.test.ts

@@ -45,6 +45,7 @@ type TestBindings = {
   INTERNAL_API_SECRET_PROD: { get(): Promise<string> };
   SESSION_SHARE_JWT_SECRET_PROD: { get(): Promise<string> };
   SESSION_SHARE_TOKEN_MIN_IAT: string;
+  USER_EXISTS_CACHE?: { delete: ReturnType<typeof vi.fn<(key: string) => Promise<void>>> };
 };
 
 function makeDbFakes() {
@@ -167,6 +168,142 @@ describe('session access invalidation route', () => {
   });
 });
 
+describe('user auth invalidation route', () => {
+  beforeEach(() => {
+    vi.resetAllMocks();
+  });
+
+  it('rejects invalidation without the internal secret', async () => {
+    const cache = { delete: vi.fn(async () => undefined) };
+
+    const res = await app.request(
+      '/internal/user-auth/invalidate',
+      {
+        method: 'POST',
+        headers: { 'content-type': 'application/json' },
+        body: JSON.stringify({ kiloUserId: 'usr_blocked' }),
+      },
+      { ...defaultEnv, USER_EXISTS_CACHE: cache }
+    );
+
+    expect(res.status).toBe(401);
+    expect(cache.delete).not.toHaveBeenCalled();
+  });
+
+  it('rejects invalidation with an incorrect internal secret', async () => {
+    const cache = { delete: vi.fn(async () => undefined) };
+
+    const res = await app.request(
+      '/internal/user-auth/invalidate',
+      {
+        method: 'POST',
+        headers: {
+          'content-type': 'application/json',
+          'X-Internal-Secret': 'wrong-secret',
+        },
+        body: JSON.stringify({ kiloUserId: 'usr_blocked' }),
+      },
+      { ...defaultEnv, USER_EXISTS_CACHE: cache }
+    );
+
+    expect(res.status).toBe(401);
+    expect(cache.delete).not.toHaveBeenCalled();
+  });
+
+  it('rejects invalidation with a malformed body', async () => {
+    const cache = { delete: vi.fn(async () => undefined) };
+
+    const res = await app.request(
+      '/internal/user-auth/invalidate',
+      {
+        method: 'POST',
+        headers: {
+          'content-type': 'application/json',
+          'X-Internal-Secret': 'internal-secret',
+        },
+        body: JSON.stringify({ kiloUserId: 123 }),
+      },
+      { ...defaultEnv, USER_EXISTS_CACHE: cache }
+    );
+
+    expect(res.status).toBe(400);
+    expect(cache.delete).not.toHaveBeenCalled();
+  });
+
+  it('deletes the versioned user-auth cache key', async () => {
+    const cache = { delete: vi.fn(async () => undefined) };
+
+    const res = await app.request(
+      '/internal/user-auth/invalidate',
+      {
+        method: 'POST',
+        headers: {
+          'content-type': 'application/json',
+          'X-Internal-Secret': 'internal-secret',
+        },
+        body: JSON.stringify({ kiloUserId: 'usr_blocked' }),
+      },
+      { ...defaultEnv, USER_EXISTS_CACHE: cache }
+    );
+
+    expect(res.status).toBe(204);
+    expect(cache.delete).toHaveBeenCalledWith('user-auth:v1:usr_blocked');
+  });
+
+  it('protects the session export route with the shared internal-secret middleware', async () => {
+    const res = await app.request(
+      '/internal/session/ses_12345678901234567890123456/export',
+      { method: 'GET' },
+      defaultEnv
+    );
+
+    expect(res.status).toBe(401);
+  });
+
+  it('returns 503 when the Secrets Store cannot resolve the internal secret', async () => {
+    const error = vi.spyOn(console, 'error').mockImplementation(() => undefined);
+    const cache = { delete: vi.fn(async () => undefined) };
+    const suppliedSecret = 'caller-supplied-secret';
+
+    const res = await app.request(
+      '/internal/user-auth/invalidate',
+      {
+        method: 'POST',
+        headers: {
+          'content-type': 'application/json',
+          'X-Internal-Secret': suppliedSecret,
+        },
+        body: JSON.stringify({ kiloUserId: 'usr_blocked' }),
+      },
+      {
+        ...defaultEnv,
+        USER_EXISTS_CACHE: cache,
+        INTERNAL_API_SECRET_PROD: {
+          get: async () => {
+            throw new Error('secret store unavailable');
+          },
+        },
+      }
+    );
+
+    const body = await res.json();
+    expect(res.status).toBe(503);
+    expect(body).toEqual({ success: false, error: 'Service temporarily unavailable' });
+    expect(JSON.stringify(body)).not.toContain('secret store unavailable');
+    expect(JSON.stringify(body)).not.toContain(suppliedSecret);
+    expect(cache.delete).not.toHaveBeenCalled();
+    expect(error).toHaveBeenCalledWith(
+      'Auth infrastructure failure',
+      expect.objectContaining({
+        operation: 'internal-api-secret-get',
+        errorClass: 'Error',
+      })
+    );
+    expect(JSON.stringify(error.mock.calls)).not.toContain(suppliedSecret);
+    error.mockRestore();
+  });
+});
+
 describe('public session route', () => {
   beforeEach(() => {
     vi.resetAllMocks();