From ba745bb71a36360490dc053176d1e1c15d61a558 Mon Sep 17 00:00:00 2001 From: maaz519 Date: Mon, 6 Jul 2026 15:28:01 +0530 Subject: [PATCH] feat(iios): policy-enforced membership + threaded replies + dev IdP (generic) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Enriches the platform PLANE, not the kernel: - DevOpaPort: the dev OPA stub now evaluates a small policy table (behind the same opa.decide) — DM capped at 2, add-participant requires member/group-admin, governed self-join for membership threads. Real OPA swaps in unchanged. - Dev IdP login (POST /v1/dev/login) issues the same JWT claims a real IdP would. - PolicyDeniedFilter maps fail-closed denials to HTTP 403. Generic kernel additions (no chat vocabulary — 'dm'/'group' live only as OPA policy + an opaque thread attribute): - MessageService.addParticipant (governed membership by userId), governed openThread self-join (scoped to threads with a membership attribute), parentInteractionId on send (reply link), and a generic listThreads. - REST: GET /v1/threads, POST /v1/threads, POST /v1/threads/:id/participants; socket add_participant + membership/parentInteractionId. ensureParticipant gains a role. Tests: dev-opa.port.spec + message.spec (DM cap / group admin / governed join / listThreads / reply). smoke-membership.mjs; realtime smokes updated for governed join. 175 unit tests + all smokes green; kernel free of dm/group literals. Co-Authored-By: Claude Opus 4.8 (1M context) --- .../iios-service/scripts/smoke-membership.mjs | 61 +++++++++ .../scripts/smoke-realtime-cluster.mjs | 4 +- .../iios-service/scripts/smoke-realtime.mjs | 3 +- .../iios-service/src/dev/dev.controller.ts | 36 +++++- .../src/identity/actor.resolver.ts | 4 +- packages/iios-service/src/main.ts | 2 + .../src/messaging/message.gateway.ts | 14 +- .../src/messaging/message.service.ts | 120 +++++++++++++++++- .../src/messaging/message.spec.ts | 54 ++++++++ .../src/platform/dev-opa.port.spec.ts | 45 +++++++ .../iios-service/src/platform/dev-opa.port.ts | 44 +++++++ .../src/platform/platform-ports.ts | 9 +- .../src/platform/policy-denied.filter.ts | 16 +++ .../src/threads/send-message.dto.ts | 1 + .../src/threads/threads.controller.ts | 33 ++++- 15 files changed, 418 insertions(+), 28 deletions(-) create mode 100644 packages/iios-service/scripts/smoke-membership.mjs create mode 100644 packages/iios-service/src/platform/dev-opa.port.spec.ts create mode 100644 packages/iios-service/src/platform/dev-opa.port.ts create mode 100644 packages/iios-service/src/platform/policy-denied.filter.ts diff --git a/packages/iios-service/scripts/smoke-membership.mjs b/packages/iios-service/scripts/smoke-membership.mjs new file mode 100644 index 0000000..9389bba --- /dev/null +++ b/packages/iios-service/scripts/smoke-membership.mjs @@ -0,0 +1,61 @@ +// v1.1 smoke: dev IdP login, policy-enforced DM cap / group membership, listThreads, +// and threaded replies — all over REST. Requires the service with IIOS_DEV_TOKENS=1. +import 'dotenv/config'; + +const SERVICE = process.env.SMOKE_URL ?? 'http://localhost:3200'; +const assert = (c, m) => { if (!c) { console.error('✗', m); process.exit(1); } console.log('✓', m); }; + +async function login(username, password) { + const r = await fetch(`${SERVICE}/v1/dev/login`, { + method: 'POST', headers: { 'content-type': 'application/json' }, + body: JSON.stringify({ username, password }), + }); + return { status: r.status, body: r.ok ? await r.json() : null }; +} +function call(token, path, method = 'GET', body) { + return fetch(`${SERVICE}${path}`, { + method, + headers: { 'content-type': 'application/json', authorization: `Bearer ${token}` }, + body: body ? JSON.stringify(body) : undefined, + }); +} +async function json(token, path, method = 'GET', body) { + const r = await call(token, path, method, body); + if (!r.ok) throw new Error(`${method} ${path} ${r.status}: ${await r.text()}`); + return r.json(); +} + +// 1) dev IdP: correct password issues a token; wrong password is rejected. +const ok = await login('alice', 'alice'); +assert(ok.body?.token, 'dev IdP: alice logs in with password'); +const bad = await login('alice', 'nope'); +assert(bad.status === 401, 'dev IdP: wrong password → 401'); +const alice = ok.body.token; + +// 2) DM is capped at two (policy). +const dm = await json(alice, '/v1/threads', 'POST', { membership: 'dm' }); +assert(dm.threadId, `alice created a DM (${dm.threadId})`); +const add2 = await json(alice, `/v1/threads/${dm.threadId}/participants`, 'POST', { userId: 'bob' }); +assert(add2.participantCount === 2, 'added bob → 2 participants'); +const add3 = await call(alice, `/v1/threads/${dm.threadId}/participants`, 'POST', { userId: 'carol' }); +assert(add3.status === 403, `adding a 3rd to a DM → 403 policy denied (${add3.status})`); + +// 3) group: creator (ADMIN) can add several. +const grp = await json(alice, '/v1/threads', 'POST', { membership: 'group', creatorRole: 'ADMIN' }); +await json(alice, `/v1/threads/${grp.threadId}/participants`, 'POST', { userId: 'bob' }); +const g3 = await json(alice, `/v1/threads/${grp.threadId}/participants`, 'POST', { userId: 'carol' }); +assert(g3.participantCount === 3, 'group admin added bob + carol → 3 participants'); + +// 4) threaded reply round-trips. +const first = await json(alice, `/v1/threads/${grp.threadId}/messages`, 'POST', { content: 'question?' }); +const reply = await json(alice, `/v1/threads/${grp.threadId}/messages`, 'POST', { content: 'answer', parentInteractionId: first.id }); +assert(reply.parentInteractionId === first.id, 'a reply carries parentInteractionId'); + +// 5) listThreads shows the caller's threads with membership + last message. +const mine = await json(alice, '/v1/threads'); +const g = mine.find((t) => t.threadId === grp.threadId); +assert(g && g.membership === 'group' && g.participantCount === 3, 'GET /v1/threads lists the group with membership + count'); +assert(g.lastMessage === 'answer', 'thread summary carries the last message'); + +console.log('\nv1.1 membership + replies smoke: PASS'); +process.exit(0); diff --git a/packages/iios-service/scripts/smoke-realtime-cluster.mjs b/packages/iios-service/scripts/smoke-realtime-cluster.mjs index 0153f6b..525613c 100644 --- a/packages/iios-service/scripts/smoke-realtime-cluster.mjs +++ b/packages/iios-service/scripts/smoke-realtime-cluster.mjs @@ -29,10 +29,12 @@ try { await Promise.all([once(alice, 'connect'), once(bob, 'connect')]); assert(true, `Alice connected to A (${A_URL}), Bob to B (${B_URL})`); - // Alice opens a thread on instance A; Bob joins the SAME thread on instance B. + // Alice opens a thread on instance A; membership is governed so Alice ADDS Bob, who + // then opens the SAME thread on instance B. const opened = await alice.emitWithAck('open_thread', {}); const threadId = opened.threadId; assert(!!threadId, `Alice opened thread ${threadId} on instance A`); + await alice.emitWithAck('add_participant', { threadId, userId: 'bob-cluster' }); await bob.emitWithAck('open_thread', { threadId }); assert(true, 'Bob joined the same thread on instance B'); diff --git a/packages/iios-service/scripts/smoke-realtime.mjs b/packages/iios-service/scripts/smoke-realtime.mjs index f0d3767..24cda46 100644 --- a/packages/iios-service/scripts/smoke-realtime.mjs +++ b/packages/iios-service/scripts/smoke-realtime.mjs @@ -44,10 +44,11 @@ const alice = connect('alice'); const bob = connect('bob'); try { - // Alice creates a thread; Bob joins it. + // Alice creates a thread; membership is governed, so Alice ADDS Bob (he can't self-join). const opened = await alice.emitWithAck('open_thread', {}); const threadId = opened.threadId; assert(!!threadId, `Alice created thread ${threadId}`); + await alice.emitWithAck('add_participant', { threadId, userId: 'bob' }); await bob.emitWithAck('open_thread', { threadId }); // Alice sends; Bob should receive it live. diff --git a/packages/iios-service/src/dev/dev.controller.ts b/packages/iios-service/src/dev/dev.controller.ts index db58ec5..8e546c9 100644 --- a/packages/iios-service/src/dev/dev.controller.ts +++ b/packages/iios-service/src/dev/dev.controller.ts @@ -1,5 +1,5 @@ import { randomUUID } from 'node:crypto'; -import { BadRequestException, Body, Controller, ForbiddenException, Headers, Param, Post } from '@nestjs/common'; +import { BadRequestException, Body, Controller, ForbiddenException, Headers, Param, Post, UnauthorizedException } from '@nestjs/common'; import jwt from 'jsonwebtoken'; import { IIOS_EVENTS } from '@insignia/iios-contracts'; import { hmacSign } from '@insignia/iios-adapter-sdk'; @@ -27,20 +27,44 @@ export class DevController { @Post('token') token(@Body() body: { appId: string; userId: string; name?: string; orgId?: string }): { token: string } { this.assertDev(); + return { token: this.signToken(body.appId, body.userId, body.name, body.orgId) }; + } + + /** + * Dev IdP: a credentialed login that mints the SAME JWT claims a real IdP / Session + * Broker would issue (sub/name/appId/orgId). Credentials come from DEV_USERS (JSON + * `{username: password}`; defaults to a few demo users). SessionVerifier is the stable + * verify seam — swapping in a real IdP means issuing these same claims, nothing else. + */ + @Post('login') + login(@Body() body: { username: string; password: string; appId?: string }): { token: string; userId: string } { + this.assertDev(); + const appId = body.appId ?? 'portal-demo'; + let users: Record; + try { + users = JSON.parse(process.env.DEV_USERS ?? '{"alice":"alice","bob":"bob","carol":"carol"}'); + } catch { + users = {}; + } + const expected = users[body.username]; + if (expected === undefined || expected !== body.password) throw new UnauthorizedException('invalid username or password'); + return { token: this.signToken(appId, body.username, body.username), userId: body.username }; + } + + private signToken(appId: string, userId: string, name?: string, orgId?: string): string { let secrets: Record; try { secrets = JSON.parse(process.env.APP_SECRETS ?? '{}'); } catch { secrets = {}; } - const secret = secrets[body.appId]; - if (!secret) throw new BadRequestException(`unknown app: ${body.appId}`); - const token = jwt.sign( - { sub: body.userId, name: body.name ?? body.userId, appId: body.appId, orgId: body.orgId ?? `org_${body.appId}` }, + const secret = secrets[appId]; + if (!secret) throw new BadRequestException(`unknown app: ${appId}`); + return jwt.sign( + { sub: userId, name: name ?? userId, appId, orgId: orgId ?? `org_${appId}` }, secret, { algorithm: 'HS256', expiresIn: '2h' }, ); - return { token }; } /** Server signs a sample provider payload and feeds it to the inbound pipeline. */ diff --git a/packages/iios-service/src/identity/actor.resolver.ts b/packages/iios-service/src/identity/actor.resolver.ts index d07a5e7..6f94b39 100644 --- a/packages/iios-service/src/identity/actor.resolver.ts +++ b/packages/iios-service/src/identity/actor.resolver.ts @@ -68,10 +68,10 @@ export class ActorResolver { ); } - async ensureParticipant(threadId: string, actorId: string): Promise { + async ensureParticipant(threadId: string, actorId: string, participantRole = 'MEMBER'): Promise { await this.prisma.iiosThreadParticipant.upsert({ where: { threadId_actorId: { threadId, actorId } }, - create: { threadId, actorId }, + create: { threadId, actorId, participantRole }, update: {}, }); } diff --git a/packages/iios-service/src/main.ts b/packages/iios-service/src/main.ts index e3c6537..97ea751 100644 --- a/packages/iios-service/src/main.ts +++ b/packages/iios-service/src/main.ts @@ -5,6 +5,7 @@ import { ValidationPipe } from '@nestjs/common'; import { AppModule } from './app.module'; import { traceMiddleware } from './observability/trace.middleware'; import { RedisIoAdapter } from './realtime/redis-io.adapter'; +import { PolicyDeniedFilter } from './platform/policy-denied.filter'; async function bootstrap(): Promise { const app = await NestFactory.create(AppModule, { rawBody: true }); @@ -12,6 +13,7 @@ async function bootstrap(): Promise { app.useGlobalPipes( new ValidationPipe({ whitelist: true, forbidNonWhitelisted: true, transform: true }), ); + app.useGlobalFilters(new PolicyDeniedFilter()); // fail-closed policy denials → 403 app.enableCors({ origin: true, credentials: true }); // Multi-replica realtime: fan socket.io room emits across instances via Redis. diff --git a/packages/iios-service/src/messaging/message.gateway.ts b/packages/iios-service/src/messaging/message.gateway.ts index d0dd7c0..2967640 100644 --- a/packages/iios-service/src/messaging/message.gateway.ts +++ b/packages/iios-service/src/messaging/message.gateway.ts @@ -66,17 +66,23 @@ export class MessageGateway implements OnGatewayInit, OnGatewayConnection { } @SubscribeMessage('open_thread') - async openThread(@ConnectedSocket() client: Socket, @MessageBody() body: { threadId?: string }) { + async openThread(@ConnectedSocket() client: Socket, @MessageBody() body: { threadId?: string; membership?: string; creatorRole?: string }) { const { principal } = client.data as SocketState; - const result = await this.messages.openThread(body?.threadId ?? null, principal); + const result = await this.messages.openThread(body?.threadId ?? null, principal, { membership: body?.membership, creatorRole: body?.creatorRole }); await client.join(result.threadId); return result; } + @SubscribeMessage('add_participant') + async addParticipant(@ConnectedSocket() client: Socket, @MessageBody() body: { threadId: string; userId: string; role?: string }) { + const { principal } = client.data as SocketState; + return this.messages.addParticipant(body.threadId, principal, body.userId, body.role); + } + @SubscribeMessage('send_message') async sendMessage( @ConnectedSocket() client: Socket, - @MessageBody() body: { threadId: string; content: string; contentRef?: string }, + @MessageBody() body: { threadId: string; content: string; contentRef?: string; parentInteractionId?: string }, ) { const { principal } = client.data as SocketState; const msg = await this.messages.send( @@ -84,6 +90,8 @@ export class MessageGateway implements OnGatewayInit, OnGatewayConnection { principal, { content: body.content, contentRef: body.contentRef }, randomUUID(), + undefined, + body.parentInteractionId, ); this.emittedLocally.add(msg.id); this.server.to(body.threadId).emit('message', msg); diff --git a/packages/iios-service/src/messaging/message.service.ts b/packages/iios-service/src/messaging/message.service.ts index 5ac7dbc..6339c8f 100644 --- a/packages/iios-service/src/messaging/message.service.ts +++ b/packages/iios-service/src/messaging/message.service.ts @@ -15,10 +15,21 @@ export interface MessageDto { senderActorId: string; content: string; contentRef?: string; + parentInteractionId?: string; traceId: string; createdAt: Date; } +export interface ThreadSummary { + threadId: string; + subject: string | null; + membership?: string; + participantCount: number; + unread: number; + lastMessage?: string; + lastAt?: Date; +} + export interface OpenThreadResult { threadId: string; status: string; @@ -39,16 +50,28 @@ export class MessageService { private readonly actors: ActorResolver, ) {} - /** Open an existing thread, or create one when no id is given. Joins as participant. */ - async openThread(threadId: string | null, principal: MessagePrincipal): Promise { + /** + * Open an existing thread, or create one when no id is given. On create, an optional + * generic `membership` attribute is stamped on the thread's metadata (the app's DM/group + * hint — the kernel never branches on it; policy does), and the creator joins as ADMIN + * for a group. Opening an existing thread is a GOVERNED join: only an existing member may + * re-open it (policy `iios.thread.join`) — new members enter via addParticipant. + */ + async openThread(threadId: string | null, principal: MessagePrincipal, opts?: { membership?: string; creatorRole?: string }): Promise { if (!threadId) { await decideOrThrow(this.ports, { action: 'iios.thread.create', scope: principal }); const scope = await this.actors.resolveScope(principal); const actor = await this.actors.resolveActor(scope.id, principal); + // `membership` + `creatorRole` are generic, app-supplied thread attributes — the kernel + // stores/echoes them but never branches on their chat meaning (that lives in policy + app). const thread = await this.prisma.iiosThread.create({ - data: { scopeId: scope.id, createdByActorId: actor.id }, + data: { + scopeId: scope.id, + createdByActorId: actor.id, + metadata: opts?.membership ? ({ membership: opts.membership } as Prisma.InputJsonValue) : undefined, + }, }); - await this.actors.ensureParticipant(thread.id, actor.id); + await this.actors.ensureParticipant(thread.id, actor.id, opts?.creatorRole ?? 'MEMBER'); return { threadId: thread.id, status: thread.status, history: [] }; } @@ -56,16 +79,96 @@ export class MessageService { if (!thread) throw new NotFoundException('thread not found'); await decideOrThrow(this.ports, { action: 'iios.thread.read', threadId, scopeId: thread.scopeId }); const actor = await this.actors.resolveActor(thread.scopeId, principal); + const alreadyMember = + (await this.prisma.iiosThreadParticipant.findUnique({ where: { threadId_actorId: { threadId, actorId: actor.id } } })) !== null; + // Join is governed ONLY for threads that opted into a membership model (chat dm/group); + // support/inbox/generic threads (no membership attr) keep open-join, unchanged. + const membership = (thread.metadata as { membership?: string } | null)?.membership; + await decideOrThrow(this.ports, { action: 'iios.thread.join', threadId, scopeId: thread.scopeId, membership, alreadyMember }); await this.actors.ensureParticipant(threadId, actor.id); return { threadId, status: thread.status, history: await this.history(threadId) }; } + /** + * Governed thread membership (a member adds another user by userId). Fail-closed via + * policy: the DM cap / group-admin rule is enforced by OPA (dev stub now, real later); + * the kernel only supplies generic context and adds the participant. The target's actor + * is resolved-or-created so you can add someone who hasn't logged in yet. + */ + async addParticipant(threadId: string, principal: MessagePrincipal, targetUserId: string, role = 'MEMBER'): Promise<{ threadId: string; participantCount: number }> { + const thread = await this.prisma.iiosThread.findUnique({ where: { id: threadId } }); + if (!thread) throw new NotFoundException('thread not found'); + const caller = await this.actors.resolveActor(thread.scopeId, principal); + const callerP = await this.prisma.iiosThreadParticipant.findUnique({ where: { threadId_actorId: { threadId, actorId: caller.id } } }); + const participantCount = await this.prisma.iiosThreadParticipant.count({ where: { threadId } }); + const membership = (thread.metadata as { membership?: string } | null)?.membership; + + await decideOrThrow(this.ports, { + action: 'iios.thread.participant.add', + threadId, + scopeId: thread.scopeId, + membership, + participantCount, + callerRole: callerP?.participantRole, + targetUserId, + role, + }); + + const scope = await this.actors.resolveScope(principal); + const target = await this.actors.resolveActor(scope.id, { + userId: targetUserId, + appId: principal.appId, + orgId: principal.orgId, + tenantId: principal.tenantId, + displayName: targetUserId, + }); + await this.actors.ensureParticipant(threadId, target.id, role); + return { threadId, participantCount: participantCount + 1 }; + } + + /** Generic "my threads": every thread the caller participates in, with last message + unread. */ + async listThreads(principal: MessagePrincipal): Promise { + const scope = await this.actors.findScope(principal); + if (!scope) return []; + const actor = await this.actors.resolveActor(scope.id, principal); + const memberships = await this.prisma.iiosThreadParticipant.findMany({ where: { actorId: actor.id }, select: { threadId: true } }); + const threadIds = memberships.map((m) => m.threadId); + if (threadIds.length === 0) return []; + + const [threads, unreads] = await Promise.all([ + this.prisma.iiosThread.findMany({ where: { id: { in: threadIds } }, include: { _count: { select: { participants: true } } } }), + this.prisma.iiosUnreadCounter.findMany({ where: { threadId: { in: threadIds }, actorId: actor.id } }), + ]); + const unreadBy = new Map(unreads.map((u) => [u.threadId, u.unreadCount])); + + const summaries = await Promise.all( + threads.map(async (t) => { + const last = await this.prisma.iiosInteraction.findFirst({ + where: { threadId: t.id }, + orderBy: { occurredAt: 'desc' }, + include: { parts: { where: { kind: 'TEXT' }, take: 1 } }, + }); + return { + threadId: t.id, + subject: t.subject, + membership: (t.metadata as { membership?: string } | null)?.membership, + participantCount: t._count.participants, + unread: unreadBy.get(t.id) ?? 0, + lastMessage: last?.parts[0]?.bodyText ?? undefined, + lastAt: last?.occurredAt, + }; + }), + ); + return summaries.sort((a, b) => (b.lastAt?.getTime() ?? 0) - (a.lastAt?.getTime() ?? 0)); + } + async send( threadId: string, principal: MessagePrincipal, body: { content: string; contentRef?: string }, idempotencyKey: string, traceId: string = randomUUID(), + parentInteractionId?: string, ): Promise { const thread = await this.prisma.iiosThread.findUnique({ where: { id: threadId } }); if (!thread) throw new NotFoundException('thread not found'); @@ -75,6 +178,11 @@ export class MessageService { const actor = await this.actors.resolveActor(thread.scopeId, principal); await this.actors.ensureParticipant(threadId, actor.id); + // A reply links to its parent — but only if the parent is in the same thread (else ignored). + const parentRef = parentInteractionId + ? (await this.prisma.iiosInteraction.findFirst({ where: { id: parentInteractionId, threadId }, select: { id: true } }))?.id + : undefined; + // Idempotency: a repeat key returns the existing message (no re-increment). const existing = await this.prisma.iiosInteraction.findUnique({ where: { scopeId_idempotencyKey: { scopeId: thread.scopeId, idempotencyKey } }, @@ -93,6 +201,7 @@ export class MessageService { idempotencyKey, status: 'NORMALIZED', traceId, + parentInteractionId: parentRef, }, }); @@ -239,7 +348,7 @@ export class MessageService { // ─── helpers ──────────────────────────────────────────────────── private toDto( - interaction: { id: string; actorId: string | null; traceId: string | null; occurredAt: Date; parts: Array<{ kind: string; bodyText: string | null; contentRef: string | null }> }, + interaction: { id: string; actorId: string | null; parentInteractionId?: string | null; traceId: string | null; occurredAt: Date; parts: Array<{ kind: string; bodyText: string | null; contentRef: string | null }> }, threadId: string, ): MessageDto { const text = interaction.parts.find((p) => p.kind === 'TEXT'); @@ -250,6 +359,7 @@ export class MessageService { senderActorId: interaction.actorId ?? '', content: text?.bodyText ?? '', contentRef: file?.contentRef ?? undefined, + parentInteractionId: interaction.parentInteractionId ?? undefined, traceId: interaction.traceId ?? '', createdAt: interaction.occurredAt, }; diff --git a/packages/iios-service/src/messaging/message.spec.ts b/packages/iios-service/src/messaging/message.spec.ts index 7f0c161..5347ea1 100644 --- a/packages/iios-service/src/messaging/message.spec.ts +++ b/packages/iios-service/src/messaging/message.spec.ts @@ -1,15 +1,20 @@ import { describe, it, expect, beforeAll, afterAll, beforeEach } from 'vitest'; import { PrismaClient } from '@prisma/client'; +import { PolicyDeniedError, type IiosPlatformPorts } from '@insignia/iios-contracts'; import { resetDb } from '../test-utils/reset-db'; import { makeFakePorts } from '@insignia/iios-testkit'; import { MessageService, type MessagePrincipal } from './message.service'; import { ActorResolver } from '../identity/actor.resolver'; +import { DevOpaPort } from '../platform/dev-opa.port'; import type { PrismaService } from '../prisma/prisma.service'; const url = process.env.DATABASE_URL ?? 'postgresql://iios:iios@localhost:5434/iios?schema=public'; const prisma = new PrismaClient({ datasources: { db: { url } } }); const asService = prisma as unknown as PrismaService; const svc = () => new MessageService(asService, makeFakePorts(), new ActorResolver(asService)); +// A service whose OPA actually enforces the membership rules (real dev policy plane). +const gov = () => + new MessageService(asService, { ...makeFakePorts(), opa: new DevOpaPort() } as IiosPlatformPorts, new ActorResolver(asService)); const alice: MessagePrincipal = { userId: 'alice', orgId: 'org_demo', appId: 'portal-demo', displayName: 'Alice' }; const bob: MessagePrincipal = { userId: 'bob', orgId: 'org_demo', appId: 'portal-demo', displayName: 'Bob' }; @@ -101,3 +106,52 @@ describe('MessageService (P2 native messaging)', () => { expect(ce.insignia.correlationId).toBe(msg.traceId); }); }); + +describe('Governed membership + replies (v1.1, policy-enforced)', () => { + it('a direct message is capped at two people (OPA policy)', async () => { + const s = gov(); + const { threadId } = await s.openThread(null, alice, { membership: 'dm' }); + expect((await s.addParticipant(threadId, alice, 'bob')).participantCount).toBe(2); + await expect(s.addParticipant(threadId, alice, 'carol')).rejects.toBeInstanceOf(PolicyDeniedError); + expect(await prisma.iiosThreadParticipant.count({ where: { threadId } })).toBe(2); // unchanged + }); + + it('group: the creator is ADMIN and can add; a plain member cannot', async () => { + const s = gov(); + const { threadId } = await s.openThread(null, alice, { membership: 'group', creatorRole: 'ADMIN' }); + await s.addParticipant(threadId, alice, 'bob'); // admin adds a member + expect(await prisma.iiosThreadParticipant.count({ where: { threadId } })).toBe(2); + await expect(s.addParticipant(threadId, bob, 'carol')).rejects.toBeInstanceOf(PolicyDeniedError); // bob is MEMBER + }); + + it('self-join is governed: a non-member cannot open a thread by id; after being added, they can', async () => { + const s = gov(); + const { threadId } = await s.openThread(null, alice, { membership: 'group', creatorRole: 'ADMIN' }); + await expect(s.openThread(threadId, bob)).rejects.toBeInstanceOf(PolicyDeniedError); + await s.addParticipant(threadId, alice, 'bob'); + expect((await s.openThread(threadId, bob)).threadId).toBe(threadId); + }); + + it('listThreads returns the caller’s threads with membership, count, last message + unread', async () => { + const s = gov(); + const { threadId } = await s.openThread(null, alice, { membership: 'group', creatorRole: 'ADMIN' }); + await s.addParticipant(threadId, alice, 'bob'); + await s.send(threadId, alice, { content: 'hello team' }, 'k1'); + const mine = await s.listThreads(alice); + expect(mine).toHaveLength(1); + expect(mine[0]).toMatchObject({ membership: 'group', participantCount: 2, lastMessage: 'hello team' }); + expect((await s.listThreads(bob))[0]?.unread).toBe(1); + }); + + it('a reply stores + returns parentInteractionId; a cross-thread parent is ignored', async () => { + const s = gov(); + const { threadId } = await s.openThread(null, alice, { membership: 'dm' }); + const first = await s.send(threadId, alice, { content: 'question?' }, 'k1'); + const reply = await s.send(threadId, alice, { content: 'answer' }, 'k2', undefined, first.id); + expect(reply.parentInteractionId).toBe(first.id); + + const { threadId: other } = await s.openThread(null, alice, { membership: 'dm' }); + const cross = await s.send(other, alice, { content: 'x' }, 'k3', undefined, first.id); + expect(cross.parentInteractionId).toBeUndefined(); // parent not in this thread + }); +}); diff --git a/packages/iios-service/src/platform/dev-opa.port.spec.ts b/packages/iios-service/src/platform/dev-opa.port.spec.ts new file mode 100644 index 0000000..43ae472 --- /dev/null +++ b/packages/iios-service/src/platform/dev-opa.port.spec.ts @@ -0,0 +1,45 @@ +import { describe, it, expect } from 'vitest'; +import { DevOpaPort } from './dev-opa.port'; + +const opa = new DevOpaPort(); + +describe('DevOpaPort (dev policy plane — membership rules)', () => { + it('allows every existing action by default (thread create/read, message send)', async () => { + for (const action of ['iios.thread.create', 'iios.thread.read', 'iios.message.send', undefined]) { + expect((await opa.decide({ action })).allow).toBe(true); + } + }); + + it('DM is capped at two: adding a 2nd is allowed, a 3rd is denied', async () => { + const add = (participantCount: number) => + opa.decide({ action: 'iios.thread.participant.add', membership: 'dm', callerRole: 'MEMBER', participantCount }); + expect((await add(1)).allow).toBe(true); // 1 → adding the 2nd + const third = await add(2); // 2 → adding a 3rd + expect(third.allow).toBe(false); + expect(third.obligations[0]?.reason).toMatch(/two people/); + }); + + it('adding a participant requires the caller be a member', async () => { + const d = await opa.decide({ action: 'iios.thread.participant.add', membership: 'group', callerRole: 'NONE', participantCount: 3 }); + expect(d.allow).toBe(false); + expect(d.obligations[0]?.reason).toMatch(/member/); + }); + + it('group add/remove requires ADMIN', async () => { + const asMember = await opa.decide({ action: 'iios.thread.participant.add', membership: 'group', callerRole: 'MEMBER', participantCount: 3 }); + expect(asMember.allow).toBe(false); + expect(asMember.obligations[0]?.reason).toMatch(/admin/); + const asAdmin = await opa.decide({ action: 'iios.thread.participant.add', membership: 'group', callerRole: 'ADMIN', participantCount: 3 }); + expect(asAdmin.allow).toBe(true); + }); + + it('self-join is governed only on membership threads', async () => { + // generic / support thread (no membership attr) → open join, unchanged + expect((await opa.decide({ action: 'iios.thread.join', alreadyMember: false })).allow).toBe(true); + // a membership (chat) thread → existing member allowed, stranger denied + expect((await opa.decide({ action: 'iios.thread.join', membership: 'group', alreadyMember: true })).allow).toBe(true); + const stranger = await opa.decide({ action: 'iios.thread.join', membership: 'group', alreadyMember: false }); + expect(stranger.allow).toBe(false); + expect(stranger.obligations[0]?.reason).toMatch(/not a member/); + }); +}); diff --git a/packages/iios-service/src/platform/dev-opa.port.ts b/packages/iios-service/src/platform/dev-opa.port.ts new file mode 100644 index 0000000..1aca2cc --- /dev/null +++ b/packages/iios-service/src/platform/dev-opa.port.ts @@ -0,0 +1,44 @@ +import type { PolicyDecision } from '@insignia/iios-contracts'; + +/** + * Dev stand-in for the real OPA policy plane. It evaluates a small **policy table** + * against the generic `{ action, ...context }` the kernel passes to `opa.decide`. + * Everything defaults to ALLOW (so existing actions are unchanged); only the rules + * below deny. This is the *policy plane*, NOT the kernel — the chat meaning of "dm" + * vs "group" lives here as policy, and a real OPA drops in behind the same interface. + */ +export interface OpaInput { + action?: string; + membership?: string; // app-set generic thread attribute: 'dm' | 'group' + participantCount?: number; + callerRole?: string; // 'MEMBER' | 'ADMIN' + alreadyMember?: boolean; + [k: string]: unknown; +} + +const allow = (): PolicyDecision => ({ decisionRef: 'dev-allow', allow: true, obligations: [], ttlSeconds: 60 }); +const deny = (reason: string): PolicyDecision => ({ decisionRef: 'dev-deny', allow: false, obligations: [{ kind: 'AUDIT', reason }], ttlSeconds: 0 }); + +export class DevOpaPort { + async decide(input: unknown): Promise { + const i = (input ?? {}) as OpaInput; + switch (i.action) { + case 'iios.thread.participant.add': { + const count = i.participantCount ?? 0; + if (i.membership === 'dm' && count >= 2) return deny('a direct message is limited to two people'); + if (i.callerRole !== 'MEMBER' && i.callerRole !== 'ADMIN') return deny('only a member can add participants'); + if (i.membership === 'group' && i.callerRole !== 'ADMIN') return deny('only a group admin can add or remove members'); + return allow(); + } + case 'iios.thread.join': { + // Only threads that opted into a membership model (chat dm/group) are governed; + // generic/support threads (no membership attr) keep open-join. For a governed + // thread, new members must be added first (governed add-participant). + if (!i.membership) return allow(); + return i.alreadyMember ? allow() : deny('you are not a member of this thread'); + } + default: + return allow(); + } + } +} diff --git a/packages/iios-service/src/platform/platform-ports.ts b/packages/iios-service/src/platform/platform-ports.ts index d8ff652..5c6deab 100644 --- a/packages/iios-service/src/platform/platform-ports.ts +++ b/packages/iios-service/src/platform/platform-ports.ts @@ -1,10 +1,10 @@ import type { IiosPlatformPorts, PlatformPrincipal, - PolicyDecision, ContextDecisionBundle, SourceHandleResolution, } from '@insignia/iios-contracts'; +import { DevOpaPort } from './dev-opa.port'; /** DI token for the platform ports (Session/OPA/CMP/MDM/CRRE/SAS/Capability). */ export const PLATFORM_PORTS = Symbol('PLATFORM_PORTS'); @@ -22,11 +22,8 @@ export class LocalDevPorts implements IiosPlatformPorts { return { principalRef: 'local-dev', orgId: 'org_demo', appId: 'portal-demo', assurance: 'service' }; }, }; - opa = { - async decide(_input: unknown): Promise { - return { decisionRef: 'local-allow', allow: true, obligations: [], ttlSeconds: 60 }; - }, - }; + // Dev policy plane: default-allow + the membership rules (real OPA swaps in here). + opa = new DevOpaPort(); cmp = { async checkPurpose(_input: unknown): Promise<{ receiptRef?: string; status: 'ALLOW' | 'DENY' | 'NOT_REQUIRED' }> { return { status: 'NOT_REQUIRED' }; diff --git a/packages/iios-service/src/platform/policy-denied.filter.ts b/packages/iios-service/src/platform/policy-denied.filter.ts new file mode 100644 index 0000000..6552a32 --- /dev/null +++ b/packages/iios-service/src/platform/policy-denied.filter.ts @@ -0,0 +1,16 @@ +import { ArgumentsHost, Catch, HttpStatus, type ExceptionFilter } from '@nestjs/common'; +import { PolicyDeniedError } from '@insignia/iios-contracts'; +import type { Response } from 'express'; + +/** + * Maps a fail-closed `PolicyDeniedError` (from `decideOrThrow`) to HTTP 403 instead of a + * generic 500, so callers can tell "policy denied" (with the reason) from a server fault. + * Applies to every policy-gated REST endpoint (e.g. the DM-cap / membership rules). + */ +@Catch(PolicyDeniedError) +export class PolicyDeniedFilter implements ExceptionFilter { + catch(err: PolicyDeniedError, host: ArgumentsHost): void { + const res = host.switchToHttp().getResponse(); + res.status(HttpStatus.FORBIDDEN).json({ statusCode: 403, code: err.code, message: err.message }); + } +} diff --git a/packages/iios-service/src/threads/send-message.dto.ts b/packages/iios-service/src/threads/send-message.dto.ts index 0b8d8b0..9a330ce 100644 --- a/packages/iios-service/src/threads/send-message.dto.ts +++ b/packages/iios-service/src/threads/send-message.dto.ts @@ -3,4 +3,5 @@ import { IsOptional, IsString } from 'class-validator'; export class SendMessageDto { @IsString() content!: string; @IsOptional() @IsString() contentRef?: string; + @IsOptional() @IsString() parentInteractionId?: string; } diff --git a/packages/iios-service/src/threads/threads.controller.ts b/packages/iios-service/src/threads/threads.controller.ts index 22d5fdb..3e4c95e 100644 --- a/packages/iios-service/src/threads/threads.controller.ts +++ b/packages/iios-service/src/threads/threads.controller.ts @@ -22,6 +22,26 @@ export class ThreadsController { private readonly session: SessionVerifier, ) {} + /** Generic "my threads" — every thread the caller participates in (last message + unread). */ + @Get() + async listThreads(@Headers('authorization') auth?: string) { + return this.messages.listThreads(this.principal(auth)); + } + + /** Create a thread; `membership`/`creatorRole` are opaque, app-supplied thread attributes the kernel stores but never interprets. */ + @Post() + @HttpCode(201) + async createThread(@Body() body: { membership?: string; creatorRole?: string }, @Headers('authorization') auth?: string) { + return this.messages.openThread(null, this.principal(auth), { membership: body?.membership, creatorRole: body?.creatorRole }); + } + + /** Governed membership: add a user (by userId) to a thread — policy enforces DM cap / roles. */ + @Post(':id/participants') + @HttpCode(201) + async addParticipant(@Param('id') id: string, @Body() body: { userId: string; role?: string }, @Headers('authorization') auth?: string) { + return this.messages.addParticipant(id, this.principal(auth), body.userId, body.role); + } + @Get(':id/messages') async listMessages(@Param('id') id: string) { return this.threads.getMessages(id); @@ -36,14 +56,19 @@ export class ThreadsController { @Headers('authorization') authorization?: string, @Headers('idempotency-key') idempotencyKey?: string, ) { - const token = (authorization ?? '').replace(/^Bearer\s+/i, ''); - if (!token) throw new BadRequestException('Authorization bearer token is required'); - const principal = this.session.verify(token); return this.messages.send( id, - principal, + this.principal(authorization), { content: body.content, contentRef: body.contentRef }, idempotencyKey ?? randomUUID(), + undefined, + body.parentInteractionId, ); } + + private principal(authorization?: string) { + const token = (authorization ?? '').replace(/^Bearer\s+/i, ''); + if (!token) throw new BadRequestException('Authorization bearer token is required'); + return this.session.verify(token); + } }