mirror of
https://github.com/bookorbit/bookorbit.git
synced 2026-11-03 00:11:37 -06:00
fix(server): restore socket subscriptions behind the global guards
Nest 12 runs global guards on gateway messages, and the HTTP JWT guard crashed every subscribe on a missing Authorization header. Skip it for sockets, which authenticate in handleConnection, and make subscribe handlers wait for that handshake. Library rooms now require library access, and a real-socket test runs the app's global guards. Closes #1743
This commit is contained in:
Generated
+3
@@ -577,6 +577,9 @@ importers:
|
||||
prettier:
|
||||
specifier: ^3.9.9
|
||||
version: 3.9.9
|
||||
socket.io-client:
|
||||
specifier: ^4.8.4
|
||||
version: 4.8.4(supports-color@10.2.2)
|
||||
source-map-support:
|
||||
specifier: ^0.5.21
|
||||
version: 0.5.21
|
||||
|
||||
@@ -136,6 +136,7 @@
|
||||
"globals": "^17.13.0",
|
||||
"pino-pretty": "^13.2.0",
|
||||
"prettier": "^3.9.9",
|
||||
"socket.io-client": "^4.8.4",
|
||||
"source-map-support": "^0.5.21",
|
||||
"supertest": "^7.3.1",
|
||||
"ts-loader": "^9.6.2",
|
||||
|
||||
@@ -24,10 +24,7 @@ import { loggerConfig } from './common/logger.config';
|
||||
import { CommonModule } from './common/common.module';
|
||||
import { SeriesIdentityModule } from './common/series-identity.module';
|
||||
import { AuditInterceptor } from './common/interceptors/audit.interceptor';
|
||||
import { JwtAuthGuard } from './common/guards/jwt-auth.guard';
|
||||
import { LibraryAccessGuard } from './common/guards/library-access.guard';
|
||||
import { PermissionGuard } from './common/guards/permission.guard';
|
||||
import { SensitiveEndpointThrottlerGuard } from './common/guards/sensitive-endpoint-throttler.guard';
|
||||
import { GLOBAL_GUARDS } from './common/guards/global-guards';
|
||||
import { DbModule } from './db/db.module';
|
||||
import { AnnotationModule } from './modules/annotation/annotation.module';
|
||||
import { DashboardModule } from './modules/dashboard/dashboard.module';
|
||||
@@ -202,12 +199,6 @@ import { PodcastModule } from './modules/podcast/podcast.module';
|
||||
WatchDownloadModule,
|
||||
PodcastModule,
|
||||
],
|
||||
providers: [
|
||||
{ provide: APP_INTERCEPTOR, useClass: AuditInterceptor },
|
||||
{ provide: APP_GUARD, useClass: SensitiveEndpointThrottlerGuard },
|
||||
{ provide: APP_GUARD, useClass: JwtAuthGuard },
|
||||
{ provide: APP_GUARD, useClass: PermissionGuard },
|
||||
{ provide: APP_GUARD, useClass: LibraryAccessGuard },
|
||||
],
|
||||
providers: [{ provide: APP_INTERCEPTOR, useClass: AuditInterceptor }, ...GLOBAL_GUARDS.map((useClass) => ({ provide: APP_GUARD, useClass }))],
|
||||
})
|
||||
export class AppModule {}
|
||||
|
||||
@@ -0,0 +1,10 @@
|
||||
import { JwtAuthGuard } from './jwt-auth.guard';
|
||||
import { LibraryAccessGuard } from './library-access.guard';
|
||||
import { PermissionGuard } from './permission.guard';
|
||||
import { SensitiveEndpointThrottlerGuard } from './sensitive-endpoint-throttler.guard';
|
||||
|
||||
/**
|
||||
* Every guard the app installs with APP_GUARD, in the order Nest runs them. Since Nest 12 these also
|
||||
* run on gateway messages, so the gateway socket tests install this exact list.
|
||||
*/
|
||||
export const GLOBAL_GUARDS = [SensitiveEndpointThrottlerGuard, JwtAuthGuard, PermissionGuard, LibraryAccessGuard] as const;
|
||||
@@ -27,8 +27,9 @@ function makeUser(overrides: Partial<RequestUser> = {}): RequestUser {
|
||||
};
|
||||
}
|
||||
|
||||
function makeContext() {
|
||||
function makeContext(type: 'http' | 'ws' = 'http') {
|
||||
return {
|
||||
getType: () => type,
|
||||
getHandler: vi.fn(),
|
||||
getClass: vi.fn(),
|
||||
switchToHttp: () => ({
|
||||
@@ -38,6 +39,32 @@ function makeContext() {
|
||||
}
|
||||
|
||||
describe('JwtAuthGuard', () => {
|
||||
/** The passport half of the guard, which reads a bearer token from HTTP headers. */
|
||||
const passportCanActivate = () => vi.spyOn(Object.getPrototypeOf(JwtAuthGuard.prototype) as { canActivate: () => unknown }, 'canActivate');
|
||||
|
||||
afterEach(() => {
|
||||
vi.restoreAllMocks();
|
||||
});
|
||||
|
||||
it('lets gateway messages through without asking passport, which would read headers a socket does not have', () => {
|
||||
const passport = passportCanActivate().mockReturnValue(false);
|
||||
const reflector = { getAllAndOverride: vi.fn() };
|
||||
const guard = new JwtAuthGuard(reflector as never);
|
||||
|
||||
expect(guard.canActivate(makeContext('ws'))).toBe(true);
|
||||
expect(passport).not.toHaveBeenCalled();
|
||||
});
|
||||
|
||||
it('still authenticates every non-public HTTP request through passport', () => {
|
||||
const passport = passportCanActivate().mockReturnValue(false);
|
||||
const reflector = { getAllAndOverride: vi.fn().mockReturnValue(undefined) };
|
||||
const guard = new JwtAuthGuard(reflector as never);
|
||||
const context = makeContext('http');
|
||||
|
||||
expect(guard.canActivate(context)).toBe(false);
|
||||
expect(passport).toHaveBeenCalledWith(context);
|
||||
});
|
||||
|
||||
it('returns true in canActivate for @Public routes', () => {
|
||||
const reflector = {
|
||||
getAllAndOverride: vi.fn((key: string) => {
|
||||
|
||||
@@ -14,6 +14,10 @@ export class JwtAuthGuard extends AuthGuard('jwt') {
|
||||
}
|
||||
|
||||
canActivate(context: ExecutionContext) {
|
||||
// Since Nest 12, global guards also run on gateway messages. A socket.io message carries no
|
||||
// Authorization header for passport to read; gateways authenticate the socket once, in
|
||||
// handleConnection, and their handlers read that user through resolveSocketUser.
|
||||
if (context.getType() === 'ws') return true;
|
||||
const isPublic = this.reflector.getAllAndOverride<boolean>(IS_PUBLIC_KEY, [context.getHandler(), context.getClass()]);
|
||||
if (isPublic) return true;
|
||||
return super.canActivate(context);
|
||||
|
||||
@@ -1,7 +1,8 @@
|
||||
import { BadRequestException, ForbiddenException, UnauthorizedException } from '@nestjs/common';
|
||||
import type { Socket } from 'socket.io';
|
||||
|
||||
import { rejectSocketConnection, WS_UNAUTHORIZED_EVENT } from './ws-auth.utils';
|
||||
import type { RequestUser } from '../types/request-user';
|
||||
import { rejectSocketConnection, resolveSocketUser, trackSocketHandshake, WS_UNAUTHORIZED_EVENT } from './ws-auth.utils';
|
||||
|
||||
function makeClient() {
|
||||
const order: string[] = [];
|
||||
@@ -65,3 +66,58 @@ describe('rejectSocketConnection', () => {
|
||||
expect(client.disconnect).toHaveBeenCalledTimes(2);
|
||||
});
|
||||
});
|
||||
|
||||
describe('resolveSocketUser', () => {
|
||||
const reader = { id: 42, username: 'reader', isSuperuser: false } as RequestUser;
|
||||
|
||||
function socket(data: Record<string, unknown> = {}): Socket {
|
||||
return { data } as unknown as Socket;
|
||||
}
|
||||
|
||||
function deferred<T>() {
|
||||
let resolve!: (value: T) => void;
|
||||
const promise = new Promise<T>((done) => {
|
||||
resolve = done;
|
||||
});
|
||||
return { promise, resolve };
|
||||
}
|
||||
|
||||
it('waits for a handshake that is still checking the token instead of reporting an anonymous socket', async () => {
|
||||
const client = socket();
|
||||
const handshake = deferred<RequestUser | null>();
|
||||
void trackSocketHandshake(client, handshake.promise);
|
||||
|
||||
let settled = false;
|
||||
const resolved = resolveSocketUser(client).then((user) => {
|
||||
settled = true;
|
||||
return user;
|
||||
});
|
||||
await Promise.resolve();
|
||||
expect(settled).toBe(false);
|
||||
|
||||
handshake.resolve(reader);
|
||||
await expect(resolved).resolves.toBe(reader);
|
||||
});
|
||||
|
||||
it('reports no user when the handshake refused the socket', async () => {
|
||||
const client = socket();
|
||||
await trackSocketHandshake(client, Promise.resolve(null));
|
||||
|
||||
await expect(resolveSocketUser(client)).resolves.toBeNull();
|
||||
});
|
||||
|
||||
it('keeps each socket to its own handshake', async () => {
|
||||
const accepted = socket();
|
||||
const refused = socket();
|
||||
void trackSocketHandshake(accepted, Promise.resolve(reader));
|
||||
void trackSocketHandshake(refused, Promise.resolve(null));
|
||||
|
||||
await expect(resolveSocketUser(accepted)).resolves.toBe(reader);
|
||||
await expect(resolveSocketUser(refused)).resolves.toBeNull();
|
||||
});
|
||||
|
||||
it('falls back to the user stored on the socket when no handshake was tracked', async () => {
|
||||
await expect(resolveSocketUser(socket({ user: reader }))).resolves.toBe(reader);
|
||||
await expect(resolveSocketUser(socket())).resolves.toBeNull();
|
||||
});
|
||||
});
|
||||
|
||||
@@ -1,6 +1,8 @@
|
||||
import { UnauthorizedException } from '@nestjs/common';
|
||||
import type { Socket } from 'socket.io';
|
||||
|
||||
import type { RequestUser } from '../types/request-user';
|
||||
|
||||
export const WS_UNAUTHORIZED_EVENT = 'unauthorized';
|
||||
|
||||
/**
|
||||
@@ -33,3 +35,28 @@ export function rejectSocketConnection(client: Socket, cause: unknown): void {
|
||||
}
|
||||
client.disconnect();
|
||||
}
|
||||
|
||||
const handshakes = new WeakMap<Socket, Promise<RequestUser | null>>();
|
||||
|
||||
/**
|
||||
* Records a gateway's handshake so its message handlers can wait for it.
|
||||
*
|
||||
* socket.io tells the client it is connected before Nest calls `handleConnection`, and Nest starts
|
||||
* delivering the socket's messages without waiting for that call to finish. A client that subscribes
|
||||
* the moment it connects therefore reaches a handler while its token is still being checked, and
|
||||
* reading `client.data.user` there sees an anonymous socket and drops the subscription. Call this
|
||||
* synchronously at the top of `handleConnection`, before anything is awaited.
|
||||
*
|
||||
* The handshake must resolve to null rather than reject when it refuses the socket.
|
||||
*/
|
||||
export function trackSocketHandshake(client: Socket, handshake: Promise<RequestUser | null>): Promise<RequestUser | null> {
|
||||
handshakes.set(client, handshake);
|
||||
return handshake;
|
||||
}
|
||||
|
||||
/** The socket's authenticated user, once its handshake has finished; null when it was refused. */
|
||||
export async function resolveSocketUser(client: Socket): Promise<RequestUser | null> {
|
||||
const handshake = handshakes.get(client);
|
||||
if (handshake) return handshake;
|
||||
return (client.data as { user?: RequestUser } | undefined)?.user ?? null;
|
||||
}
|
||||
|
||||
@@ -179,6 +179,38 @@ describe('MigrationProgressGateway', () => {
|
||||
expect(client.emit).not.toHaveBeenCalled();
|
||||
});
|
||||
|
||||
it('holds a subscription that arrives mid-handshake until the session check finishes', async () => {
|
||||
const { gateway, jwtService, authService, repo } = makeGateway();
|
||||
jwtService.verify.mockReturnValue({ sub: 7, ver: 1, sid: 3 });
|
||||
let finishSessionCheck!: (user: RequestUser) => void;
|
||||
authService.validateSessionUser.mockReturnValue(new Promise((done) => (finishSessionCheck = done)));
|
||||
repo.findRunById.mockResolvedValue(null);
|
||||
const client = makeClient();
|
||||
|
||||
const connected = gateway.handleConnection(client as never);
|
||||
const subscribed = gateway.handleSubscribeRun(client as never, 12);
|
||||
await Promise.resolve();
|
||||
expect(client.join).not.toHaveBeenCalled();
|
||||
|
||||
finishSessionCheck(makeUser({ permissions: [Permission.ManageAppSettings] }));
|
||||
await Promise.all([connected, subscribed]);
|
||||
|
||||
expect(client.join).toHaveBeenCalledWith('run:12');
|
||||
expect(client.disconnect).not.toHaveBeenCalled();
|
||||
});
|
||||
|
||||
it('drops a mid-handshake subscription from a user the handshake turns away', async () => {
|
||||
const { gateway, jwtService, authService } = makeGateway();
|
||||
jwtService.verify.mockReturnValue({ sub: 7, ver: 1, sid: 3 });
|
||||
authService.validateSessionUser.mockResolvedValue(makeUser());
|
||||
const client = makeClient();
|
||||
|
||||
await Promise.all([gateway.handleConnection(client as never), gateway.handleSubscribeRun(client as never, 12)]);
|
||||
|
||||
expect(client.disconnect).toHaveBeenCalledTimes(1);
|
||||
expect(client.join).not.toHaveBeenCalled();
|
||||
});
|
||||
|
||||
it('emits progress updates to room-scoped subscribers', () => {
|
||||
const { gateway } = makeGateway();
|
||||
const emit = vi.fn();
|
||||
|
||||
@@ -10,7 +10,7 @@ import type { RequestUser } from '../../common/types/request-user';
|
||||
import { AuthService } from '../auth/auth.service';
|
||||
import { MigrationRepository } from './migration.repository';
|
||||
import { sanitizeRunForApi } from './core/api-sanitizers';
|
||||
import { rejectSocketConnection } from '../../common/utils/ws-auth.utils';
|
||||
import { rejectSocketConnection, resolveSocketUser, trackSocketHandshake } from '../../common/utils/ws-auth.utils';
|
||||
import { appConfig } from '../../config/config';
|
||||
|
||||
@WebSocketGateway({ namespace: '/migration', cors: { credentials: true } })
|
||||
@@ -38,6 +38,10 @@ export class MigrationProgressGateway implements OnGatewayInit, OnGatewayConnect
|
||||
}
|
||||
|
||||
async handleConnection(client: Socket): Promise<void> {
|
||||
await trackSocketHandshake(client, this.authenticate(client));
|
||||
}
|
||||
|
||||
private async authenticate(client: Socket): Promise<RequestUser | null> {
|
||||
try {
|
||||
const token = client.handshake.auth?.token as string | undefined;
|
||||
if (!token) throw new UnauthorizedException('No token provided');
|
||||
@@ -49,11 +53,13 @@ export class MigrationProgressGateway implements OnGatewayInit, OnGatewayConnect
|
||||
this.assertCanViewMigrationProgress(user);
|
||||
(client.data as Record<string, unknown>).user = user;
|
||||
this.logger.debug(`[migration.ws_connection] [start] userId=${user.id} socketId=${client.id} - websocket connected`);
|
||||
return user;
|
||||
} catch (err) {
|
||||
this.logger.warn(
|
||||
`[migration.ws_connection] [fail] socketId=${client.id} errorClass=${err instanceof Error ? err.name : 'Error'} error="${sanitizeLogValue(err instanceof Error ? err.message : String(err))}" - websocket rejected`,
|
||||
);
|
||||
rejectSocketConnection(client, err);
|
||||
return null;
|
||||
}
|
||||
}
|
||||
|
||||
@@ -63,7 +69,7 @@ export class MigrationProgressGateway implements OnGatewayInit, OnGatewayConnect
|
||||
|
||||
@SubscribeMessage('subscribe:run')
|
||||
async handleSubscribeRun(client: Socket, runId: number): Promise<void> {
|
||||
const user = (client.data as { user?: RequestUser }).user;
|
||||
const user = await resolveSocketUser(client);
|
||||
if (!user) return;
|
||||
void client.join(`run:${runId}`);
|
||||
|
||||
|
||||
@@ -58,6 +58,33 @@ describe('PodcastGateway', () => {
|
||||
expect(client.emit).toHaveBeenCalledWith('podcast:subscription:error', { libraryId: 8 });
|
||||
});
|
||||
|
||||
it('holds a subscription that arrives mid-handshake until the session check finishes', async () => {
|
||||
const { gateway, jwtService, authService, libraries } = makeGateway();
|
||||
jwtService.verify.mockReturnValue({ sub: 42, ver: 3, sid: 5 });
|
||||
let finishSessionCheck!: (user: object) => void;
|
||||
authService.validateSessionUser.mockReturnValue(new Promise((done) => (finishSessionCheck = done)));
|
||||
libraries.verifyUserAccess.mockResolvedValue(undefined);
|
||||
const client = {
|
||||
id: 'podcast-2',
|
||||
handshake: { auth: { token: 'valid' } },
|
||||
data: {},
|
||||
join: vi.fn().mockResolvedValue(undefined),
|
||||
emit: vi.fn(),
|
||||
disconnect: vi.fn(),
|
||||
};
|
||||
|
||||
const connected = gateway.handleConnection(client as never);
|
||||
const subscribed = gateway.handleSubscribeLibrary(client as never, 7);
|
||||
await Promise.resolve();
|
||||
expect(libraries.verifyUserAccess).not.toHaveBeenCalled();
|
||||
|
||||
finishSessionCheck({ id: 42, username: 'listener', isSuperuser: false });
|
||||
await Promise.all([connected, subscribed]);
|
||||
|
||||
expect(libraries.verifyUserAccess).toHaveBeenCalledWith(42, 7, false);
|
||||
expect(client.join).toHaveBeenCalledWith('library:7');
|
||||
});
|
||||
|
||||
it('emits podcast events only to the owning library room', () => {
|
||||
const { gateway } = makeGateway();
|
||||
const emit = vi.fn();
|
||||
|
||||
@@ -18,7 +18,7 @@ import type { RequestUser } from '../../common/types/request-user';
|
||||
import { AuthService } from '../auth/auth.service';
|
||||
import { LibraryService } from '../library/library.service';
|
||||
import { PodcastEventsService } from './podcast-events.service';
|
||||
import { rejectSocketConnection } from '../../common/utils/ws-auth.utils';
|
||||
import { rejectSocketConnection, resolveSocketUser, trackSocketHandshake } from '../../common/utils/ws-auth.utils';
|
||||
import { appConfig } from '../../config/config';
|
||||
|
||||
@WebSocketGateway({ namespace: '/podcasts', cors: { credentials: true } })
|
||||
@@ -47,6 +47,10 @@ export class PodcastGateway implements OnGatewayInit, OnGatewayConnection, OnGat
|
||||
}
|
||||
|
||||
async handleConnection(client: Socket): Promise<void> {
|
||||
await trackSocketHandshake(client, this.authenticate(client));
|
||||
}
|
||||
|
||||
private async authenticate(client: Socket): Promise<RequestUser | null> {
|
||||
try {
|
||||
const token = client.handshake.auth?.token as string | undefined;
|
||||
if (!token) throw new UnauthorizedException('No token provided');
|
||||
@@ -57,11 +61,13 @@ export class PodcastGateway implements OnGatewayInit, OnGatewayConnection, OnGat
|
||||
if (!user) throw new UnauthorizedException('User not found or token revoked');
|
||||
(client.data as Record<string, unknown>).user = user;
|
||||
this.logger.debug(`[podcast.ws_connection] [start] userId=${user.id} socketId=${client.id} - websocket connected`);
|
||||
return user;
|
||||
} catch (error) {
|
||||
this.logger.warn(
|
||||
`[podcast.ws_connection] [fail] socketId=${client.id} errorClass=${error instanceof Error ? error.name : 'Error'} error="${sanitizeLogValue(error instanceof Error ? error.message : String(error))}" - websocket rejected`,
|
||||
);
|
||||
rejectSocketConnection(client, error);
|
||||
return null;
|
||||
}
|
||||
}
|
||||
|
||||
@@ -71,7 +77,7 @@ export class PodcastGateway implements OnGatewayInit, OnGatewayConnection, OnGat
|
||||
|
||||
@SubscribeMessage('subscribe:library')
|
||||
async handleSubscribeLibrary(client: Socket, libraryId: number): Promise<void> {
|
||||
const user = (client.data as { user?: RequestUser }).user;
|
||||
const user = await resolveSocketUser(client);
|
||||
if (!user || !Number.isInteger(libraryId) || libraryId <= 0) return;
|
||||
try {
|
||||
await this.libraries.verifyUserAccess(user.id, libraryId, user.isSuperuser);
|
||||
|
||||
@@ -0,0 +1,178 @@
|
||||
import { ValidationPipe } from '@nestjs/common';
|
||||
import { APP_GUARD, APP_INTERCEPTOR } from '@nestjs/core';
|
||||
import { JwtService } from '@nestjs/jwt';
|
||||
import { FastifyAdapter, type NestFastifyApplication } from '@nestjs/platform-fastify';
|
||||
import { IoAdapter } from '@nestjs/platform-socket.io';
|
||||
import { Test } from '@nestjs/testing';
|
||||
import { ThrottlerModule } from '@nestjs/throttler';
|
||||
import type { ScanProgressEvent } from '@bookorbit/types';
|
||||
import { io, type Socket as ClientSocket } from 'socket.io-client';
|
||||
import { afterAll, afterEach, beforeAll, beforeEach, describe, expect, it, vi } from 'vitest';
|
||||
|
||||
import { GlobalExceptionFilter } from '../../common/filters/http-exception.filter';
|
||||
import { GLOBAL_GUARDS } from '../../common/guards/global-guards';
|
||||
import { AuditInterceptor } from '../../common/interceptors/audit.interceptor';
|
||||
import { PermissionService } from '../../common/services/permission.service';
|
||||
import type { RequestUser } from '../../common/types/request-user';
|
||||
import { appConfig } from '../../config/config';
|
||||
import { DB } from '../../db';
|
||||
import { AchievementEventsService } from '../achievement/achievement-events.service';
|
||||
import { AuditEventsService } from '../audit/audit-events.service';
|
||||
import { AuthService } from '../auth/auth.service';
|
||||
import { ScanGateway } from './scan.gateway';
|
||||
import { ScanJobStore } from './scan-job-store.service';
|
||||
import { ScannerRepository } from './scanner.repository';
|
||||
|
||||
const READABLE_LIBRARY = 7;
|
||||
const UNREADABLE_LIBRARY = 9;
|
||||
const reader = { id: 42, username: 'reader', isSuperuser: false, permissions: [] } as unknown as RequestUser;
|
||||
|
||||
function progress(libraryId: number): ScanProgressEvent {
|
||||
return { jobId: 1, libraryId, status: 'running', processed: 1, total: 2, added: 0, updated: 0, missing: 0 };
|
||||
}
|
||||
|
||||
/**
|
||||
* The scan gateway over a real socket.io connection, inside a Nest app that installs the same global
|
||||
* guards, interceptor, pipe and filter as production.
|
||||
*
|
||||
* Nest 12 started running global enhancers on gateway messages, and the HTTP JWT guard crashed every
|
||||
* `subscribe:library` on a missing Authorization header: issue 1743. Mocked execution contexts
|
||||
* cannot see that kind of change in the framework, so this goes through Nest's own message pipeline.
|
||||
*/
|
||||
describe('scan gateway over socket.io behind the global guards', () => {
|
||||
let app: NestFastifyApplication;
|
||||
let baseUrl: string;
|
||||
let gateway: ScanGateway;
|
||||
let scanJobStore: ScanJobStore;
|
||||
let token: string;
|
||||
const authService = { validateSessionUser: vi.fn<(...args: unknown[]) => Promise<RequestUser | null>>() };
|
||||
const scannerRepository = {
|
||||
hasLibraryAccess: vi.fn((_userId: number, libraryId: number) => Promise.resolve(libraryId === READABLE_LIBRARY)),
|
||||
};
|
||||
const clients: ClientSocket[] = [];
|
||||
|
||||
beforeAll(async () => {
|
||||
const moduleRef = await Test.createTestingModule({
|
||||
imports: [ThrottlerModule.forRoot({ throttlers: [{ ttl: 60_000, limit: 1_000 }] })],
|
||||
providers: [
|
||||
ScanGateway,
|
||||
ScanJobStore,
|
||||
AchievementEventsService,
|
||||
AuditEventsService,
|
||||
{ provide: JwtService, useValue: new JwtService({ secret: 'scan-gateway-socket-test' }) },
|
||||
{ provide: AuthService, useValue: authService },
|
||||
{ provide: ScannerRepository, useValue: scannerRepository },
|
||||
{ provide: appConfig.KEY, useValue: { appUrl: 'http://localhost:6263' } },
|
||||
{ provide: PermissionService, useValue: {} },
|
||||
{ provide: DB, useValue: {} },
|
||||
{ provide: APP_INTERCEPTOR, useClass: AuditInterceptor },
|
||||
...GLOBAL_GUARDS.map((useClass) => ({ provide: APP_GUARD, useClass })),
|
||||
],
|
||||
}).compile();
|
||||
|
||||
app = moduleRef.createNestApplication<NestFastifyApplication>(new FastifyAdapter(), { logger: false });
|
||||
app.useWebSocketAdapter(new IoAdapter(app));
|
||||
app.useGlobalPipes(new ValidationPipe({ whitelist: true, forbidNonWhitelisted: true, transform: true }));
|
||||
app.useGlobalFilters(new GlobalExceptionFilter());
|
||||
await app.listen(0, '127.0.0.1');
|
||||
baseUrl = await app.getUrl();
|
||||
|
||||
gateway = app.get(ScanGateway);
|
||||
scanJobStore = app.get(ScanJobStore);
|
||||
token = app.get(JwtService).sign({ sub: reader.id, ver: 3, sid: 11, amr: 'password' });
|
||||
});
|
||||
|
||||
beforeEach(() => {
|
||||
authService.validateSessionUser.mockReset();
|
||||
scannerRepository.hasLibraryAccess.mockClear();
|
||||
});
|
||||
|
||||
afterEach(() => {
|
||||
for (const client of clients.splice(0)) client.close();
|
||||
scanJobStore.delete(READABLE_LIBRARY);
|
||||
});
|
||||
|
||||
afterAll(async () => {
|
||||
await app?.close();
|
||||
});
|
||||
|
||||
function connect(): ClientSocket {
|
||||
const client = io(`${baseUrl}/scan`, { transports: ['websocket'], auth: { token }, reconnection: false, forceNew: true });
|
||||
clients.push(client);
|
||||
return client;
|
||||
}
|
||||
|
||||
/** Rejects as soon as Nest answers with an `exception` event, which is how a crashing guard surfaces. */
|
||||
function nextEvent<T>(client: ClientSocket, event: string): Promise<T> {
|
||||
return new Promise((resolve, reject) => {
|
||||
client.once(event, resolve as never);
|
||||
client.once('exception', (error: unknown) => reject(new Error(`server answered with an exception: ${JSON.stringify(error)}`)));
|
||||
});
|
||||
}
|
||||
|
||||
function collect(client: ClientSocket, event: string): unknown[] {
|
||||
const seen: unknown[] = [];
|
||||
client.on(event, (payload: unknown) => seen.push(payload));
|
||||
return seen;
|
||||
}
|
||||
|
||||
/**
|
||||
* Keeps the handshake's session lookup open until the client's subscribe has reached the server
|
||||
* and had time to run, which is the order a fast client on a slow database produces.
|
||||
*/
|
||||
function holdSessionCheckUntilSubscribe(user: RequestUser | null): void {
|
||||
let finish!: (value: RequestUser | null) => void;
|
||||
authService.validateSessionUser.mockImplementationOnce(() => new Promise((resolve) => (finish = resolve)));
|
||||
gateway.server.once('connection', (socket) => {
|
||||
socket.prependAny((event: string) => {
|
||||
if (event === 'subscribe:library') setTimeout(() => finish(user), 50);
|
||||
});
|
||||
});
|
||||
}
|
||||
|
||||
it('subscribes a client that asks the moment it connects, while its session is still being checked', async () => {
|
||||
scanJobStore.create(31, READABLE_LIBRARY, 200, 'manual');
|
||||
holdSessionCheckUntilSubscribe(reader);
|
||||
const client = connect();
|
||||
const exceptions = collect(client, 'exception');
|
||||
client.on('connect', () => client.emit('subscribe:library', READABLE_LIBRARY));
|
||||
|
||||
const snapshot = await nextEvent<ScanProgressEvent>(client, 'scan:progress');
|
||||
|
||||
expect(snapshot).toMatchObject({ jobId: 31, libraryId: READABLE_LIBRARY, status: 'running', total: 200 });
|
||||
expect(scannerRepository.hasLibraryAccess).toHaveBeenCalledWith(reader.id, READABLE_LIBRARY);
|
||||
expect(exceptions).toEqual([]);
|
||||
});
|
||||
|
||||
it('broadcasts a library to subscribers who can read it and never to anyone else', async () => {
|
||||
authService.validateSessionUser.mockResolvedValue(reader);
|
||||
const client = connect();
|
||||
const exceptions = collect(client, 'exception');
|
||||
await nextEvent(client, 'connect');
|
||||
|
||||
client.emit('subscribe:library', UNREADABLE_LIBRARY);
|
||||
client.emit('subscribe:library', READABLE_LIBRARY);
|
||||
await vi.waitFor(() => expect(gateway.server.adapter.rooms.get(`library:${READABLE_LIBRARY}`)?.has(client.id!)).toBe(true));
|
||||
expect(gateway.server.adapter.rooms.get(`library:${UNREADABLE_LIBRARY}`)?.has(client.id!) ?? false).toBe(false);
|
||||
|
||||
// One connection delivers in order, so the readable event arriving first proves the other was never sent.
|
||||
const received = nextEvent<ScanProgressEvent>(client, 'scan:progress');
|
||||
gateway.emitProgress(progress(UNREADABLE_LIBRARY));
|
||||
gateway.emitProgress(progress(READABLE_LIBRARY));
|
||||
|
||||
await expect(received).resolves.toMatchObject({ libraryId: READABLE_LIBRARY });
|
||||
expect(exceptions).toEqual([]);
|
||||
});
|
||||
|
||||
it('drops a subscribe that arrives mid-handshake from a session that turns out to be revoked', async () => {
|
||||
holdSessionCheckUntilSubscribe(null);
|
||||
const client = connect();
|
||||
const unauthorized = collect(client, 'unauthorized');
|
||||
client.on('connect', () => client.emit('subscribe:library', READABLE_LIBRARY));
|
||||
|
||||
await expect(nextEvent(client, 'disconnect')).resolves.toBe('io server disconnect');
|
||||
|
||||
expect(unauthorized).toEqual([{ reason: 'invalid_token' }]);
|
||||
expect(scannerRepository.hasLibraryAccess).not.toHaveBeenCalled();
|
||||
});
|
||||
});
|
||||
@@ -7,12 +7,29 @@ function makeGateway() {
|
||||
const jwtService = { verify: vi.fn() };
|
||||
const authService = { validateSessionUser: vi.fn() };
|
||||
const scanJobStore = { get: vi.fn(), isRunning: vi.fn() };
|
||||
const scannerRepository = { hasLibraryAccess: vi.fn().mockResolvedValue(false) };
|
||||
const achievementEvents = new AchievementEventsService();
|
||||
const appConfiguration = { appUrl: 'http://localhost:6263' };
|
||||
const gateway = new ScanGateway(jwtService as any, authService as any, scanJobStore as any, achievementEvents, appConfiguration as any);
|
||||
return { gateway, jwtService, authService, scanJobStore, achievementEvents };
|
||||
const gateway = new ScanGateway(
|
||||
jwtService as any,
|
||||
authService as any,
|
||||
scanJobStore as any,
|
||||
scannerRepository as any,
|
||||
achievementEvents,
|
||||
appConfiguration as any,
|
||||
);
|
||||
return { gateway, jwtService, authService, scanJobStore, scannerRepository, achievementEvents };
|
||||
}
|
||||
|
||||
const reader = { id: 42, username: 'reader', isSuperuser: false };
|
||||
const admin = { id: 1, username: 'admin', isSuperuser: true };
|
||||
|
||||
function subscriber(user?: object) {
|
||||
return { id: 'sock-sub', data: user ? { user } : {}, join: vi.fn().mockResolvedValue(undefined), emit: vi.fn() } as any;
|
||||
}
|
||||
|
||||
const runningScan = { jobId: 9, processed: 50, total: 200, added: 10, updated: 5, missing: 2 };
|
||||
|
||||
function mockServer() {
|
||||
const emit = vi.fn();
|
||||
const to = vi.fn().mockReturnValue({ emit });
|
||||
@@ -190,25 +207,24 @@ describe('subscription lifecycle', () => {
|
||||
expect(() => gateway.afterInit({} as any)).not.toThrow();
|
||||
});
|
||||
|
||||
it('logs disconnects and emits an in-flight progress snapshot on subscribe', () => {
|
||||
const { gateway, scanJobStore } = makeGateway();
|
||||
scanJobStore.get.mockReturnValue({
|
||||
jobId: 9,
|
||||
processed: 50,
|
||||
total: 200,
|
||||
added: 10,
|
||||
updated: 5,
|
||||
missing: 2,
|
||||
});
|
||||
const join = vi.fn();
|
||||
const emit = vi.fn();
|
||||
const client = { join, emit } as any;
|
||||
it('logs disconnects without throwing', () => {
|
||||
const { gateway } = makeGateway();
|
||||
expect(() => gateway.handleDisconnect({ id: 'sock-3' } as any)).not.toThrow();
|
||||
});
|
||||
});
|
||||
|
||||
gateway.handleSubscribeLibrary(client, 88);
|
||||
gateway.handleDisconnect({ id: 'sock-3' } as any);
|
||||
describe('subscribe:library', () => {
|
||||
it('joins the room and sends the in-flight scan snapshot to a user with access to the library', async () => {
|
||||
const { gateway, scanJobStore, scannerRepository } = makeGateway();
|
||||
scannerRepository.hasLibraryAccess.mockResolvedValue(true);
|
||||
scanJobStore.get.mockReturnValue(runningScan);
|
||||
const client = subscriber(reader);
|
||||
|
||||
expect(join).toHaveBeenCalledWith('library:88');
|
||||
expect(emit).toHaveBeenCalledWith('scan:progress', {
|
||||
await gateway.handleSubscribeLibrary(client, 88);
|
||||
|
||||
expect(scannerRepository.hasLibraryAccess).toHaveBeenCalledWith(42, 88);
|
||||
expect(client.join).toHaveBeenCalledWith('library:88');
|
||||
expect(client.emit).toHaveBeenCalledWith('scan:progress', {
|
||||
jobId: 9,
|
||||
libraryId: 88,
|
||||
status: 'running',
|
||||
@@ -220,17 +236,99 @@ describe('subscription lifecycle', () => {
|
||||
});
|
||||
});
|
||||
|
||||
it('subscribes without emitting progress when no scan entry exists', () => {
|
||||
const { gateway, scanJobStore } = makeGateway();
|
||||
it('subscribes without emitting progress when no scan entry exists', async () => {
|
||||
const { gateway, scanJobStore, scannerRepository } = makeGateway();
|
||||
scannerRepository.hasLibraryAccess.mockResolvedValue(true);
|
||||
scanJobStore.get.mockReturnValue(undefined);
|
||||
const join = vi.fn();
|
||||
const emit = vi.fn();
|
||||
const client = { join, emit } as any;
|
||||
const client = subscriber(reader);
|
||||
|
||||
gateway.handleSubscribeLibrary(client, 7);
|
||||
await gateway.handleSubscribeLibrary(client, 7);
|
||||
|
||||
expect(join).toHaveBeenCalledWith('library:7');
|
||||
expect(emit).not.toHaveBeenCalled();
|
||||
expect(client.join).toHaveBeenCalledWith('library:7');
|
||||
expect(client.emit).not.toHaveBeenCalled();
|
||||
});
|
||||
|
||||
it('lets a superuser into any library without an access row', async () => {
|
||||
const { gateway, scannerRepository } = makeGateway();
|
||||
const client = subscriber(admin);
|
||||
|
||||
await gateway.handleSubscribeLibrary(client, 5);
|
||||
|
||||
expect(scannerRepository.hasLibraryAccess).not.toHaveBeenCalled();
|
||||
expect(client.join).toHaveBeenCalledWith('library:5');
|
||||
});
|
||||
|
||||
it('keeps a user out of a library they cannot read, so its book cards and scan counts never reach them', async () => {
|
||||
const { gateway, scanJobStore, scannerRepository } = makeGateway();
|
||||
scannerRepository.hasLibraryAccess.mockResolvedValue(false);
|
||||
scanJobStore.get.mockReturnValue(runningScan);
|
||||
const client = subscriber(reader);
|
||||
|
||||
await gateway.handleSubscribeLibrary(client, 3);
|
||||
|
||||
expect(scannerRepository.hasLibraryAccess).toHaveBeenCalledWith(42, 3);
|
||||
expect(client.join).not.toHaveBeenCalled();
|
||||
expect(client.emit).not.toHaveBeenCalled();
|
||||
});
|
||||
|
||||
it('ignores a socket that never authenticated', async () => {
|
||||
const { gateway, scanJobStore, scannerRepository } = makeGateway();
|
||||
scanJobStore.get.mockReturnValue(runningScan);
|
||||
const client = subscriber();
|
||||
|
||||
await gateway.handleSubscribeLibrary(client, 3);
|
||||
|
||||
expect(scannerRepository.hasLibraryAccess).not.toHaveBeenCalled();
|
||||
expect(client.join).not.toHaveBeenCalled();
|
||||
expect(client.emit).not.toHaveBeenCalled();
|
||||
});
|
||||
|
||||
it.each([0, -1, 1.5, Number.NaN, '7', null, undefined, { id: 7 }])('ignores the malformed library id %s without querying', async (libraryId) => {
|
||||
const { gateway, scannerRepository } = makeGateway();
|
||||
const client = subscriber(admin);
|
||||
|
||||
await gateway.handleSubscribeLibrary(client, libraryId as never);
|
||||
|
||||
expect(scannerRepository.hasLibraryAccess).not.toHaveBeenCalled();
|
||||
expect(client.join).not.toHaveBeenCalled();
|
||||
});
|
||||
|
||||
it('holds a subscription that arrives mid-handshake until the token check finishes, then honours it', async () => {
|
||||
const { gateway, jwtService, authService, scannerRepository } = makeGateway();
|
||||
jwtService.verify.mockReturnValue({ sub: 42, ver: 3, sid: 11 });
|
||||
let finishSessionCheck!: (user: object) => void;
|
||||
authService.validateSessionUser.mockReturnValue(new Promise((done) => (finishSessionCheck = done)));
|
||||
scannerRepository.hasLibraryAccess.mockResolvedValue(true);
|
||||
const client = { ...subscriber(), handshake: { auth: { token: 'valid' } }, disconnect: vi.fn() } as any;
|
||||
|
||||
const connected = gateway.handleConnection(client);
|
||||
const subscribed = gateway.handleSubscribeLibrary(client, 88);
|
||||
await Promise.resolve();
|
||||
expect(client.join).not.toHaveBeenCalledWith('library:88');
|
||||
|
||||
finishSessionCheck(reader);
|
||||
await Promise.all([connected, subscribed]);
|
||||
|
||||
expect(client.join).toHaveBeenCalledWith('user:42');
|
||||
expect(client.join).toHaveBeenCalledWith('library:88');
|
||||
expect(client.disconnect).not.toHaveBeenCalled();
|
||||
});
|
||||
|
||||
it('drops a subscription that arrives mid-handshake when the handshake is refused', async () => {
|
||||
const { gateway, jwtService, authService, scannerRepository } = makeGateway();
|
||||
jwtService.verify.mockReturnValue({ sub: 42, ver: 3, sid: 11 });
|
||||
let finishSessionCheck!: (user: object | null) => void;
|
||||
authService.validateSessionUser.mockReturnValue(new Promise((done) => (finishSessionCheck = done)));
|
||||
const client = { ...subscriber(), handshake: { auth: { token: 'revoked' } }, disconnect: vi.fn() } as any;
|
||||
|
||||
const connected = gateway.handleConnection(client);
|
||||
const subscribed = gateway.handleSubscribeLibrary(client, 88);
|
||||
finishSessionCheck(null);
|
||||
await Promise.all([connected, subscribed]);
|
||||
|
||||
expect(client.disconnect).toHaveBeenCalledTimes(1);
|
||||
expect(scannerRepository.hasLibraryAccess).not.toHaveBeenCalled();
|
||||
expect(client.join).not.toHaveBeenCalled();
|
||||
});
|
||||
});
|
||||
|
||||
|
||||
@@ -24,7 +24,9 @@ import {
|
||||
type BookProgressChangedPayload,
|
||||
} from '../achievement/achievement-events.service';
|
||||
import { ScanJobStore } from './scan-job-store.service';
|
||||
import { rejectSocketConnection } from '../../common/utils/ws-auth.utils';
|
||||
import { ScannerRepository } from './scanner.repository';
|
||||
import type { RequestUser } from '../../common/types/request-user';
|
||||
import { rejectSocketConnection, resolveSocketUser, trackSocketHandshake } from '../../common/utils/ws-auth.utils';
|
||||
import { appConfig } from '../../config/config';
|
||||
import { BOOK_COVER_CHANGED, BookCoverEventsService } from '../book-cover-store/book-cover-events.service';
|
||||
|
||||
@@ -44,6 +46,7 @@ export class ScanGateway implements OnGatewayInit, OnGatewayConnection, OnGatewa
|
||||
private readonly jwtService: JwtService,
|
||||
private readonly authService: AuthService,
|
||||
private readonly scanJobStore: ScanJobStore,
|
||||
private readonly scannerRepository: ScannerRepository,
|
||||
private readonly achievementEvents: AchievementEventsService,
|
||||
@Inject(appConfig.KEY) app: ConfigType<typeof appConfig>,
|
||||
@Optional() private readonly coverEvents?: BookCoverEventsService,
|
||||
@@ -71,6 +74,10 @@ export class ScanGateway implements OnGatewayInit, OnGatewayConnection, OnGatewa
|
||||
}
|
||||
|
||||
async handleConnection(client: Socket): Promise<void> {
|
||||
await trackSocketHandshake(client, this.authenticate(client));
|
||||
}
|
||||
|
||||
private async authenticate(client: Socket): Promise<RequestUser | null> {
|
||||
try {
|
||||
const token = client.handshake.auth?.token as string | undefined;
|
||||
if (!token) throw new UnauthorizedException('No token provided');
|
||||
@@ -82,11 +89,13 @@ export class ScanGateway implements OnGatewayInit, OnGatewayConnection, OnGatewa
|
||||
(client.data as Record<string, unknown>).user = user;
|
||||
await client.join(`user:${user.id}`);
|
||||
this.logger.debug(`[scanner.ws_connection] [start] userId=${user.id} socketId=${client.id} - websocket connected`);
|
||||
return user;
|
||||
} catch (err) {
|
||||
this.logger.warn(
|
||||
`[scanner.ws_connection] [fail] socketId=${client.id} errorClass=${err instanceof Error ? err.name : 'Error'} error="${sanitizeLogValue(err instanceof Error ? err.message : String(err))}" - websocket rejected`,
|
||||
);
|
||||
rejectSocketConnection(client, err);
|
||||
return null;
|
||||
}
|
||||
}
|
||||
|
||||
@@ -94,9 +103,19 @@ export class ScanGateway implements OnGatewayInit, OnGatewayConnection, OnGatewa
|
||||
this.logger.debug(`[scanner.ws_connection] [end] socketId=${client.id} - websocket disconnected`);
|
||||
}
|
||||
|
||||
/**
|
||||
* A library room carries the library's book cards and scan counts, so joining one needs the same
|
||||
* access as reading the library over HTTP.
|
||||
*/
|
||||
@SubscribeMessage('subscribe:library')
|
||||
handleSubscribeLibrary(client: Socket, libraryId: number): void {
|
||||
void client.join(`library:${libraryId}`);
|
||||
async handleSubscribeLibrary(client: Socket, libraryId: number): Promise<void> {
|
||||
const user = await resolveSocketUser(client);
|
||||
if (!user || !Number.isSafeInteger(libraryId) || libraryId <= 0) return;
|
||||
if (!user.isSuperuser && !(await this.scannerRepository.hasLibraryAccess(user.id, libraryId))) {
|
||||
this.logger.debug(`[scanner.ws_subscribe] [fail] userId=${user.id} libraryId=${libraryId} socketId=${client.id} - no library access`);
|
||||
return;
|
||||
}
|
||||
await client.join(`library:${libraryId}`);
|
||||
// Send current snapshot immediately so reconnecting clients catch up.
|
||||
const entry = this.scanJobStore.get(libraryId);
|
||||
if (entry) {
|
||||
|
||||
@@ -129,6 +129,15 @@ export class ScannerRepository {
|
||||
return row?.name ?? null;
|
||||
}
|
||||
|
||||
async hasLibraryAccess(userId: number, libraryId: number): Promise<boolean> {
|
||||
const [row] = await this.db
|
||||
.select({ userId: userLibraryAccess.userId })
|
||||
.from(userLibraryAccess)
|
||||
.where(and(eq(userLibraryAccess.userId, userId), eq(userLibraryAccess.libraryId, libraryId)))
|
||||
.limit(1);
|
||||
return row !== undefined;
|
||||
}
|
||||
|
||||
async findLibraryAccessibleUserIds(libraryId: number): Promise<number[]> {
|
||||
const [accessRows, superuserRows] = await Promise.all([
|
||||
this.db.select({ userId: userLibraryAccess.userId }).from(userLibraryAccess).where(eq(userLibraryAccess.libraryId, libraryId)),
|
||||
|
||||
Reference in New Issue
Block a user