feat(#707): secure Discord service ingress (#716)

apps/gateway/src/chat/chat.gateway-auth.ts

@@ -1,3 +1,4 @@
+import { timingSafeEqual } from 'node:crypto';
 import type { IncomingHttpHeaders } from 'node:http';
 import { fromNodeHeaders } from 'better-auth/node';
 
@@ -12,6 +13,19 @@ export interface SessionAuth {
   };
 }
 
+export function validateDiscordServiceToken(
+  candidate: unknown,
+  expected: string | undefined,
+): boolean {
+  if (typeof candidate !== 'string' || !expected) return false;
+  const candidateBuffer = Buffer.from(candidate);
+  const expectedBuffer = Buffer.from(expected);
+  return (
+    candidateBuffer.length === expectedBuffer.length &&
+    timingSafeEqual(candidateBuffer, expectedBuffer)
+  );
+}
+
 export async function validateSocketSession(
   headers: IncomingHttpHeaders,
   auth: SessionAuth,

apps/gateway/src/chat/chat.gateway.ts

@@ -11,6 +11,11 @@ import {
 } from '@nestjs/websockets';
 import { Server, Socket } from 'socket.io';
 import type { AgentSessionEvent } from '@mariozechner/pi-coding-agent';
+import {
+  verifyDiscordIngressEnvelope,
+  type DiscordIngressEnvelope,
+  type DiscordIngressPayload,
+} from '@mosaicstack/discord-plugin';
 import type { Auth } from '@mosaicstack/auth';
 import type { Brain } from '@mosaicstack/brain';
 import type {
@@ -34,7 +39,8 @@ import { CommandExecutorService } from '../commands/command-executor.service.js'
 import { RoutingEngineService } from '../agent/routing/routing-engine.service.js';
 import { v4 as uuid } from 'uuid';
 import { ChatSocketMessageDto } from './chat.dto.js';
-import { validateSocketSession } from './chat.gateway-auth.js';
+import { validateDiscordServiceToken, validateSocketSession } from './chat.gateway-auth.js';
+import { DiscordReplayProtector } from '../plugin/discord-replay-protector.js';
 
 /** Per-client state tracking streaming accumulation for persistence. */
 interface ClientSession {
@@ -58,6 +64,37 @@ interface ClientSession {
  */
 const modelOverrides = new Map<string, string>();
 
+function isDiscordIngressEnvelope(value: unknown): value is DiscordIngressEnvelope {
+  if (typeof value !== 'object' || value === null) return false;
+  const envelope = value as { payload?: unknown; signature?: unknown };
+  if (
+    typeof envelope.signature !== 'string' ||
+    typeof envelope.payload !== 'object' ||
+    envelope.payload === null
+  ) {
+    return false;
+  }
+  const payload = envelope.payload as Record<string, unknown>;
+  return [
+    payload['correlationId'],
+    payload['messageId'],
+    payload['guildId'],
+    payload['channelId'],
+    payload['userId'],
+    payload['conversationId'],
+    payload['content'],
+  ].every((field: unknown): boolean => typeof field === 'string');
+}
+
+function isChatSocketMessage(value: unknown): value is ChatSocketMessageDto {
+  if (typeof value !== 'object' || value === null) return false;
+  const payload = value as { content?: unknown; conversationId?: unknown };
+  return (
+    typeof payload.content === 'string' &&
+    (payload.conversationId === undefined || typeof payload.conversationId === 'string')
+  );
+}
+
 @WebSocketGateway({
   cors: {
     origin: process.env['GATEWAY_CORS_ORIGIN'] ?? 'http://localhost:3000',
@@ -70,6 +107,7 @@ export class ChatGateway implements OnGatewayInit, OnGatewayConnection, OnGatewa
 
   private readonly logger = new Logger(ChatGateway.name);
   private readonly clientSessions = new Map<string, ClientSession>();
+  private readonly discordReplayProtector = new DiscordReplayProtector();
 
   constructor(
     @Inject(AgentService) private readonly agentService: AgentService,
@@ -85,6 +123,13 @@ export class ChatGateway implements OnGatewayInit, OnGatewayConnection, OnGatewa
   }
 
   async handleConnection(client: Socket): Promise<void> {
+    const serviceToken = client.handshake.auth['discordServiceToken'];
+    if (validateDiscordServiceToken(serviceToken, process.env['DISCORD_SERVICE_TOKEN'])) {
+      client.data.discordService = true;
+      this.logger.log(`Authenticated Discord service connected: ${client.id}`);
+      return;
+    }
+
     const session = await validateSocketSession(client.handshake.headers, this.auth);
     if (!session) {
       this.logger.warn(`Rejected unauthenticated WebSocket client: ${client.id}`);
@@ -95,8 +140,6 @@ export class ChatGateway implements OnGatewayInit, OnGatewayConnection, OnGatewa
     client.data.user = session.user;
     client.data.session = session.session;
     this.logger.log(`Client connected: ${client.id}`);
-
-    // Broadcast command manifest to the newly connected client
     client.emit('commands:manifest', { manifest: this.commandRegistry.getManifest() });
   }
 
@@ -131,17 +174,49 @@ export class ChatGateway implements OnGatewayInit, OnGatewayConnection, OnGatewa
   @SubscribeMessage('message')
   async handleMessage(
     @ConnectedSocket() client: Socket,
-    @MessageBody() data: ChatSocketMessageDto,
+    @MessageBody() rawData: unknown,
   ): Promise<void> {
+    let discordIngress: DiscordIngressPayload | null = null;
+    let data: ChatSocketMessageDto;
+    if (client.data.discordService) {
+      if (!isDiscordIngressEnvelope(rawData)) {
+        this.logger.warn(`Rejected malformed Discord ingress from ${client.id}`);
+        return;
+      }
+      discordIngress = this.resolveDiscordIngress(client, rawData);
+      if (!discordIngress) return;
+      data = { conversationId: discordIngress.conversationId, content: discordIngress.content };
+    } else {
+      if (!isChatSocketMessage(rawData)) {
+        this.logger.warn(`Rejected malformed chat message from ${client.id}`);
+        return;
+      }
+      data = rawData;
+    }
     const conversationId = data.conversationId ?? uuid();
-    const scope = this.getClientScope(client);
+    const discordServiceUserId = process.env['DISCORD_SERVICE_USER_ID'];
+    if (discordIngress && !discordServiceUserId) {
+      this.logger.warn(
+        `Rejected Discord ingress without configured service owner from ${client.id}`,
+      );
+      return;
+    }
+    const scope = discordIngress
+      ? {
+          userId: discordServiceUserId!,
+          tenantId: process.env['DISCORD_SERVICE_TENANT_ID'] ?? discordServiceUserId!,
+        }
+      : this.getClientScope(client);
     if (!scope) {
       client.emit('error', { conversationId, error: 'Authenticated user scope is required.' });
       return;
     }
     const userId = scope.userId;
+    const correlationId = discordIngress?.correlationId;
 
-    this.logger.log(`Message from ${client.id} in conversation ${conversationId}`);
+    this.logger.log(
+      `Message from ${client.id} in conversation ${conversationId}${correlationId ? ` correlation=${correlationId}` : ''}`,
+    );
 
     // Ensure agent session exists for this conversation
     let sessionRoutingDecision: RoutingDecisionInfo | undefined;
@@ -245,6 +320,13 @@ export class ChatGateway implements OnGatewayInit, OnGatewayConnection, OnGatewa
             content: data.content,
             metadata: {
               timestamp: new Date().toISOString(),
+              ...(correlationId
+                ? {
+                    correlationId,
+                    discordMessageId: discordIngress?.messageId,
+                    discordUserId: discordIngress?.userId,
+                  }
+                : {}),
             },
           },
           userId,
@@ -308,7 +390,17 @@ export class ChatGateway implements OnGatewayInit, OnGatewayConnection, OnGatewa
     }
 
     // Send acknowledgment
-    client.emit('message:ack', { conversationId, messageId: uuid() });
+    client.emit('message:ack', {
+      conversationId,
+      messageId: uuid(),
+      ...(correlationId
+        ? {
+            correlationId,
+            discordMessageId: discordIngress?.messageId,
+            discordUserId: discordIngress?.userId,
+          }
+        : {}),
+    });
 
     // Dispatch to agent
     try {
@@ -534,6 +626,39 @@ export class ChatGateway implements OnGatewayInit, OnGatewayConnection, OnGatewa
    * Creates it if absent — safe to call concurrently since a duplicate insert
    * would fail on the PK constraint and be caught here.
    */
+  private resolveDiscordIngress(
+    client: Socket,
+    envelope: DiscordIngressEnvelope,
+  ): DiscordIngressPayload | null {
+    const payload = verifyDiscordIngressEnvelope(
+      envelope,
+      process.env['DISCORD_SERVICE_TOKEN'] ?? '',
+      {
+        guildIds: this.readDiscordAllowlist('DISCORD_ALLOWED_GUILD_IDS'),
+        channelIds: this.readDiscordAllowlist('DISCORD_ALLOWED_CHANNEL_IDS'),
+        userIds: this.readDiscordAllowlist('DISCORD_ALLOWED_USER_IDS'),
+      },
+    );
+    if (!payload) {
+      this.logger.warn(`Rejected invalid Discord ingress envelope from ${client.id}`);
+      return null;
+    }
+    if (!this.discordReplayProtector.claim(payload.messageId)) {
+      this.logger.warn(
+        `Rejected replayed Discord message=${payload.messageId} correlation=${payload.correlationId}`,
+      );
+      return null;
+    }
+    return payload;
+  }
+
+  private readDiscordAllowlist(name: string): string[] {
+    return (process.env[name] ?? '')
+      .split(',')
+      .map((id: string): string => id.trim())
+      .filter((id: string): boolean => id.length > 0);
+  }
+
   private async ensureConversation(conversationId: string, userId: string): Promise<void> {
     try {
       const existing = await this.brain.conversations.findById(conversationId, userId);