Feat/s3 storage #2

Closed
maaz519 wants to merge 30 commits from feat/s3-storage into main
4 changed files with 106 additions and 0 deletions
Showing only changes of commit 3298401772 - Show all commits
@@ -8,6 +8,9 @@ export interface MessagePrincipal {
appId: string;
tenantId?: string;
displayName?: string;
/** The token's audience (`aud`). Used to enforce realtime delegation: a socket-scoped
* token (`iios-message`) must not be a full REST/actor token (`iios-core`), and vice-versa. */
audience?: string;
}
/**
@@ -0,0 +1,93 @@
import { describe, it, expect, afterEach, beforeAll, afterAll } from 'vitest';
import jwt from 'jsonwebtoken';
import type { Socket } from 'socket.io';
import { MessageGateway } from './message.gateway';
import type { MessageService, MessagePrincipal } from './message.service';
import { SessionVerifier } from '../platform/session.verifier';
import type { OutboxBus } from '../outbox/outbox.bus';
import type { PresenceService } from '../notifications/presence.service';
// Realtime delegation: the browser opens the socket with a short-lived token minted for the
// realtime audience (iios-message). When IIOS_REALTIME_AUDIENCE is set, the gateway accepts ONLY
// that audience — a full REST/actor token (iios-core) or an un-scoped token can't open the stream.
const APP = 'crm-web';
const SECRET = 'dev-crm-secret';
// ─── the real SessionVerifier must expose `aud` so the gateway can enforce it ───
describe('SessionVerifier — app-token audience surfacing', () => {
let v: SessionVerifier;
beforeAll(async () => {
delete process.env.AUTH_ISSUERS;
delete process.env.SUPABASE_URL;
process.env.APP_SECRETS = JSON.stringify({ [APP]: SECRET });
v = new SessionVerifier();
await v.onModuleInit();
});
afterAll(() => { delete process.env.APP_SECRETS; });
it('surfaces the aud claim of a realtime-scoped token', () => {
const tok = jwt.sign({ appId: APP, aud: 'iios-message' }, SECRET, { algorithm: 'HS256', subject: 'pp_1', expiresIn: '5m' });
expect(v.verify(tok).audience).toBe('iios-message');
});
it('leaves audience undefined for an un-scoped token (a REST/actor token)', () => {
const tok = jwt.sign({ appId: APP }, SECRET, { algorithm: 'HS256', subject: 'pp_1', expiresIn: '5m' });
expect(v.verify(tok).audience).toBeUndefined();
});
});
// ─── the gateway enforces the realtime audience on connect ───
function fakeSocket(): { sock: Socket; disconnected: () => boolean } {
let disconnected = false;
const sock = {
handshake: { auth: { token: 'tok' } },
disconnect: () => { disconnected = true; },
data: undefined as unknown,
} as unknown as Socket;
return { sock, disconnected: () => disconnected };
}
function gatewayReturning(principal: MessagePrincipal): MessageGateway {
const session = { verify: () => principal } as unknown as SessionVerifier;
return new MessageGateway(
undefined as unknown as MessageService,
session,
undefined as unknown as OutboxBus,
undefined as unknown as PresenceService,
);
}
const principal = (audience?: string): MessagePrincipal => ({ userId: 'u', appId: APP, orgId: 'org', audience });
describe('MessageGateway — realtime audience enforcement', () => {
afterEach(() => { delete process.env.IIOS_REALTIME_AUDIENCE; });
it('accepts any valid token when enforcement is OFF (env unset)', () => {
const { sock, disconnected } = fakeSocket();
gatewayReturning(principal(undefined)).handleConnection(sock);
expect(disconnected()).toBe(false);
expect((sock.data as { principal: MessagePrincipal }).principal.userId).toBe('u');
});
it('accepts a token whose aud matches the required realtime audience', () => {
process.env.IIOS_REALTIME_AUDIENCE = 'iios-message';
const { sock, disconnected } = fakeSocket();
gatewayReturning(principal('iios-message')).handleConnection(sock);
expect(disconnected()).toBe(false);
});
it('rejects a full REST/actor token (aud=iios-core) on the socket', () => {
process.env.IIOS_REALTIME_AUDIENCE = 'iios-message';
const { sock, disconnected } = fakeSocket();
gatewayReturning(principal('iios-core')).handleConnection(sock);
expect(disconnected()).toBe(true);
});
it('rejects an un-scoped token (no aud) when enforcement is on', () => {
process.env.IIOS_REALTIME_AUDIENCE = 'iios-message';
const { sock, disconnected } = fakeSocket();
gatewayReturning(principal(undefined)).handleConnection(sock);
expect(disconnected()).toBe(true);
});
});
@@ -61,6 +61,14 @@ export class MessageGateway implements OnGatewayInit, OnGatewayConnection, OnGat
try {
const token = String(client.handshake.auth?.token ?? '');
const principal = this.session.verify(token);
// Realtime delegation (least privilege): when IIOS_REALTIME_AUDIENCE is set, the socket
// accepts ONLY a token minted for that audience (a short-lived `iios-message` delegate) —
// a full REST/actor token (`iios-core`) or an un-scoped token cannot open the stream. So a
// leaked socket token can't drive privileged REST, and vice-versa. Unset → no enforcement.
const required = process.env.IIOS_REALTIME_AUDIENCE?.trim();
if (required && principal.audience !== required) {
throw new Error(`realtime audience "${principal.audience ?? '(none)'}" != required "${required}"`);
}
client.data = { principal } satisfies SocketState;
} catch (err) {
this.logger.warn(`rejecting socket: ${(err as Error).message}`);
@@ -135,6 +135,7 @@ export class SessionVerifier implements OnModuleInit {
orgId: entry.orgId,
tenantId: undefined,
displayName: meta.full_name ?? meta.name ?? email ?? userId,
audience: typeof payload.aud === 'string' ? payload.aud : entry.audience,
};
}
@@ -159,6 +160,7 @@ export class SessionVerifier implements OnModuleInit {
orgId: payload.orgId ? String(payload.orgId) : `org_${appId}`,
tenantId: payload.tenantId ? String(payload.tenantId) : undefined,
displayName: payload.name ? String(payload.name) : undefined,
audience: typeof payload.aud === 'string' ? payload.aud : undefined,
};
}