diff --git a/apps/gateway/src/chat/chat.gateway.ts b/apps/gateway/src/chat/chat.gateway.ts index 15a36d1..9e5cd29 100644 --- a/apps/gateway/src/chat/chat.gateway.ts +++ b/apps/gateway/src/chat/chat.gateway.ts @@ -30,6 +30,7 @@ import type { AbortPayload, } from '@mosaicstack/types'; import { AgentService, type ConversationHistoryMessage } from '../agent/agent.service.js'; +import { RuntimeProviderService } from '../agent/runtime-provider-registry.service.js'; import { AUTH } from '../auth/auth.tokens.js'; import { scopeFromUser, @@ -128,6 +129,9 @@ export class ChatGateway implements OnGatewayInit, OnGatewayConnection, OnGatewa @Optional() @Inject(CommandAuthorizationService) private readonly commandAuthorization: CommandAuthorizationService | null = null, + @Optional() + @Inject(RuntimeProviderService) + private readonly runtimeRegistry: RuntimeProviderService | null = null, ) {} afterInit(): void { @@ -678,6 +682,42 @@ export class ChatGateway implements OnGatewayInit, OnGatewayConnection, OnGatewa }); } + @SubscribeMessage('discord:stop') + async handleDiscordStop( + @ConnectedSocket() client: Socket, + @MessageBody() envelope: DiscordIngressEnvelope, + ): Promise { + if (!client.data.discordService) return; + const ingress = this.resolveDiscordIngress(client, envelope, 'approve'); + const actionParts = ingress?.content.match(/^\/stop\s+([^\s]+)\s+([^\s]+)\s+([^\s]+)$/i); + const serviceUserId = process.env['DISCORD_SERVICE_USER_ID']; + if (!ingress || !actionParts || !serviceUserId || !this.runtimeRegistry) return; + const binding = resolveDiscordInteractionBinding( + parseDiscordInteractionBindings(process.env['DISCORD_INTERACTION_BINDINGS']), + ingress.guildId, + ingress.channelId, + ingress.userId, + 'approve', + ); + if (!binding) return; + + try { + // RuntimeProviderService consumes the durable approval exactly once using this + // server-derived context before it invokes the provider termination. + await this.runtimeRegistry.terminate(actionParts[1]!, actionParts[2]!, actionParts[3]!, { + actorScope: { + userId: serviceUserId, + tenantId: process.env['DISCORD_SERVICE_TENANT_ID'] ?? serviceUserId, + }, + channelId: ingress.channelId, + correlationId: ingress.correlationId, + }); + client.emit('discord:stop', { correlationId: ingress.correlationId, success: true }); + } catch { + client.emit('discord:stop', { correlationId: ingress.correlationId, success: false }); + } + } + private resolveDiscordIngress( client: Socket, envelope: DiscordIngressEnvelope, diff --git a/apps/gateway/src/plugin/discord-ingress.security.spec.ts b/apps/gateway/src/plugin/discord-ingress.security.spec.ts index bebc106..438c52c 100644 --- a/apps/gateway/src/plugin/discord-ingress.security.spec.ts +++ b/apps/gateway/src/plugin/discord-ingress.security.spec.ts @@ -1,14 +1,119 @@ -import { describe, expect, it } from 'vitest'; +import { afterEach, describe, expect, it, vi } from 'vitest'; import { createDiscordIngressEnvelope, verifyDiscordIngressEnvelope, + DiscordPlugin, type DiscordIngressPayload, resolveDiscordInteractionBinding, } from '@mosaicstack/discord-plugin'; +import { ChatGateway } from '../chat/chat.gateway.js'; +import { CommandAuthorizationService } from '../commands/command-authorization.service.js'; import { validateDiscordServiceToken } from '../chat/chat.gateway-auth.js'; import { DiscordReplayProtector } from './discord-replay-protector.js'; const SERVICE_TOKEN = 'test-service-token'; +const ENV_KEYS = [ + 'DISCORD_SERVICE_TOKEN', + 'DISCORD_SERVICE_USER_ID', + 'DISCORD_SERVICE_TENANT_ID', + 'DISCORD_INTERACTION_BINDINGS', + 'DISCORD_ALLOWED_GUILD_IDS', + 'DISCORD_ALLOWED_CHANNEL_IDS', + 'DISCORD_ALLOWED_USER_IDS', + 'MOSAIC_AGENT_NAME', +] as const; +const savedEnv = new Map(); + +function configureDiscordEnv(role: 'admin' | 'member' = 'admin'): void { + for (const key of ENV_KEYS) savedEnv.set(key, process.env[key]); + process.env['DISCORD_SERVICE_TOKEN'] = SERVICE_TOKEN; + process.env['DISCORD_SERVICE_USER_ID'] = 'discord-service'; + process.env['DISCORD_SERVICE_TENANT_ID'] = 'tenant-discord'; + process.env['MOSAIC_AGENT_NAME'] = 'Nova'; + process.env['DISCORD_ALLOWED_GUILD_IDS'] = 'guild-001'; + process.env['DISCORD_ALLOWED_CHANNEL_IDS'] = 'channel-001'; + process.env['DISCORD_ALLOWED_USER_IDS'] = 'user-001'; + process.env['DISCORD_INTERACTION_BINDINGS'] = JSON.stringify([ + { + instanceId: 'Nova', + guildId: 'guild-001', + channelId: 'channel-001', + pairedUsers: { 'user-001': role === 'admin' ? 'admin' : 'operator' }, + }, + ]); +} + +afterEach((): void => { + for (const key of ENV_KEYS) { + const value = savedEnv.get(key); + if (value === undefined) delete process.env[key]; + else process.env[key] = value; + } + savedEnv.clear(); +}); + +function commandAuthorization(role: 'admin' | 'member'): CommandAuthorizationService { + const entries = new Map(); + const db = { + select: () => ({ from: () => ({ where: () => ({ limit: async () => [{ role }] }) }) }), + }; + const redis = { + get: async (key: string) => entries.get(key) ?? null, + set: async (key: string, value: string) => entries.set(key, value), + del: async (key: string) => Number(entries.delete(key)), + }; + return new CommandAuthorizationService(db as never, redis); +} + +function discordGateway(role: 'admin' | 'member'): { + gateway: ChatGateway; + client: { data: { discordService: boolean }; emit: ReturnType }; +} { + const authorization = commandAuthorization(role); + const runtimeRegistry = { + terminate: async ( + providerId: string, + sessionId: string, + approvalId: string, + context: { + actorScope: { userId: string; tenantId: string }; + channelId: string; + correlationId: string; + }, + ): Promise => { + const approved = await authorization.consumeRuntimeTerminationApproval(approvalId, { + providerId, + sessionId, + actorId: context.actorScope.userId, + tenantId: context.actorScope.tenantId, + channelId: context.channelId, + correlationId: context.correlationId, + agentName: 'Nova', + }); + if (!approved) throw new Error('approval denied'); + }, + }; + return { + gateway: new ChatGateway( + {} as never, + {} as never, + {} as never, + {} as never, + {} as never, + {} as never, + authorization, + runtimeRegistry as never, + ), + client: { data: { discordService: true }, emit: vi.fn() }, + }; +} + +function ingressEnvelope( + content: string, + messageId: string, +): ReturnType { + return createDiscordIngressEnvelope(createPayload({ content, messageId }), SERVICE_TOKEN); +} function createPayload(overrides: Partial = {}): DiscordIngressPayload { return { @@ -106,4 +211,121 @@ describe('Discord ingress security', () => { expect(replayProtector.claim('discord-message-003')).toBe(true); expect(replayProtector.size).toBe(2); }); + + it('mints a Discord approval and consumes it for the exact stop action', async () => { + configureDiscordEnv(); + const { gateway, client } = discordGateway('admin'); + await gateway.handleDiscordApproval( + client as never, + ingressEnvelope('/approve fleet runtime-1', 'approve-message'), + ); + const approval = client.emit.mock.calls.find( + ([event]) => event === 'discord:approval', + )?.[1] as { + approvalId: string; + success: boolean; + }; + expect(approval.success).toBe(true); + + await gateway.handleDiscordStop( + client as never, + ingressEnvelope(`/stop fleet runtime-1 ${approval.approvalId}`, 'stop-message'), + ); + expect(client.emit).toHaveBeenCalledWith('discord:stop', { + correlationId: 'correlation-001', + success: true, + }); + }); + + it('rejects unpaired and non-admin Discord users for approval and stop', async () => { + configureDiscordEnv(); + const { gateway, client } = discordGateway('member'); + await gateway.handleDiscordApproval( + client as never, + ingressEnvelope('/approve fleet runtime-1', 'member-approve'), + ); + expect(client.emit).toHaveBeenCalledWith('discord:approval', { + correlationId: 'correlation-001', + success: false, + approvalId: undefined, + expiresAt: undefined, + }); + + process.env['DISCORD_INTERACTION_BINDINGS'] = JSON.stringify([]); + await gateway.handleDiscordStop( + client as never, + ingressEnvelope('/stop fleet runtime-1 forged', 'unpaired-stop'), + ); + expect(client.emit).not.toHaveBeenCalledWith('discord:stop', expect.anything()); + }); + + it('rejects replaying a Discord-created termination approval', async () => { + configureDiscordEnv(); + const { gateway, client } = discordGateway('admin'); + await gateway.handleDiscordApproval( + client as never, + ingressEnvelope('/approve fleet runtime-1', 'replay-approve'), + ); + const approval = client.emit.mock.calls.find( + ([event]) => event === 'discord:approval', + )?.[1] as { + approvalId: string; + }; + await gateway.handleDiscordStop( + client as never, + ingressEnvelope(`/stop fleet runtime-1 ${approval.approvalId}`, 'replay-stop-one'), + ); + await gateway.handleDiscordStop( + client as never, + ingressEnvelope(`/stop fleet runtime-1 ${approval.approvalId}`, 'replay-stop-two'), + ); + const stopResults = client.emit.mock.calls.filter(([event]) => event === 'discord:stop'); + expect(stopResults.map(([, result]) => (result as { success: boolean }).success)).toEqual([ + true, + false, + ]); + }); + + it('accepts a thread message through its allowed bound parent channel', () => { + const emitted = vi.fn(); + const plugin = new DiscordPlugin({ + token: 'unused', + gatewayUrl: 'http://unused', + serviceToken: SERVICE_TOKEN, + allowedGuildIds: ['guild-001'], + allowedChannelIds: ['channel-001'], + allowedUserIds: ['user-001'], + interactionBindings: [ + { + instanceId: 'Nova', + guildId: 'guild-001', + channelId: 'channel-001', + pairedUsers: { 'user-001': 'operator' }, + }, + ], + }); + const internals = plugin as unknown as { + client: { user: { id: string } }; + socket: { connected: boolean; emit: ReturnType }; + handleDiscordMessage(message: unknown): void; + }; + internals.client = { user: { id: 'bot-001' } }; + internals.socket = { connected: true, emit: emitted }; + internals.handleDiscordMessage({ + id: 'thread-message', + guildId: 'guild-001', + channelId: 'thread-001', + author: { id: 'user-001', bot: false }, + mentions: { has: () => true }, + content: '<@bot-001> hello from thread', + channel: { parentId: 'channel-001' }, + attachments: new Map(), + }); + + const [, envelope] = emitted.mock.calls[0] as [ + string, + ReturnType, + ]; + expect(verifyDiscordIngressEnvelope(envelope, SERVICE_TOKEN)?.channelId).toBe('channel-001'); + }); }); diff --git a/plugins/discord/src/index.ts b/plugins/discord/src/index.ts index eaf0bd9..28d835d 100644 --- a/plugins/discord/src/index.ts +++ b/plugins/discord/src/index.ts @@ -16,7 +16,7 @@ export interface DiscordPluginConfig { interactionBindings?: readonly DiscordInteractionBinding[]; } -export type DiscordInteractionOperation = 'bind' | 'attach' | 'send' | 'approve'; +export type DiscordInteractionOperation = 'bind' | 'attach' | 'send' | 'approve' | 'stop'; export type DiscordInteractionRole = 'viewer' | 'operator' | 'admin'; export interface DiscordInteractionBinding { @@ -34,6 +34,7 @@ const operationRoles: Readonly< attach: ['operator', 'admin'], send: ['operator', 'admin'], approve: ['admin'], + stop: ['admin'], }; /** Resolves a configuration-owned binding and applies pairing/RBAC before ingress. */ @@ -327,7 +328,14 @@ export class DiscordPlugin { }, this.config.serviceToken, ); - this.socket.emit(content.startsWith('/approve ') ? 'discord:approve' : 'message', envelope); + this.socket.emit( + content.startsWith('/approve ') + ? 'discord:approve' + : content.startsWith('/stop ') + ? 'discord:stop' + : 'message', + envelope, + ); } private isAllowedMessage(message: DiscordMessage): boolean {