From 311a31da58ebfc079653f860ea4cf4ed9a051d42 Mon Sep 17 00:00:00 2001 From: dakkar Date: Thu, 15 Aug 2024 11:35:51 +0100 Subject: rough rate limiting for websockets --- packages/backend/src/server/api/stream/Connection.ts | 7 +++++++ 1 file changed, 7 insertions(+) (limited to 'packages/backend/src/server/api/stream') diff --git a/packages/backend/src/server/api/stream/Connection.ts b/packages/backend/src/server/api/stream/Connection.ts index 7dd7db24e5..dfc6f0d298 100644 --- a/packages/backend/src/server/api/stream/Connection.ts +++ b/packages/backend/src/server/api/stream/Connection.ts @@ -25,6 +25,7 @@ import type Channel from './channel.js'; export default class Connection { public user?: MiUser; public token?: MiAccessToken; + private rateLimiter?: () => Promise; private wsConnection: WebSocket.WebSocket; public subscriber: StreamEventEmitter; private channels: Channel[] = []; @@ -48,9 +49,11 @@ export default class Connection { user: MiUser | null | undefined, token: MiAccessToken | null | undefined, + rateLimiter: () => Promise, ) { if (user) this.user = user; if (token) this.token = token; + if (rateLimiter) this.rateLimiter = rateLimiter; } @bindThis @@ -103,6 +106,10 @@ export default class Connection { private async onWsConnectionMessage(data: WebSocket.RawData) { let obj: Record; + if (this.rateLimiter && await this.rateLimiter()) { + return; + } + try { obj = JSON.parse(data.toString()); } catch (e) { -- cgit v1.2.3-freya From 6d3f9503ed1fd04718396b248cc5a753245c0f67 Mon Sep 17 00:00:00 2001 From: Julia Johannesen Date: Fri, 16 Aug 2024 17:13:20 -0400 Subject: Limit number of rate limit requests --- .../src/server/api/StreamingApiServerService.ts | 5 +++++ .../backend/src/server/api/stream/Connection.ts | 26 ++++++++++++++++++++-- 2 files changed, 29 insertions(+), 2 deletions(-) (limited to 'packages/backend/src/server/api/stream') diff --git a/packages/backend/src/server/api/StreamingApiServerService.ts b/packages/backend/src/server/api/StreamingApiServerService.ts index 1435169812..f48af45fb1 100644 --- a/packages/backend/src/server/api/StreamingApiServerService.ts +++ b/packages/backend/src/server/api/StreamingApiServerService.ts @@ -26,12 +26,15 @@ import proxyAddr from 'proxy-addr'; import ms from 'ms'; import type * as http from 'node:http'; import type { IEndpointMeta } from './endpoints.js'; +import { LoggerService } from '@/core/LoggerService.js'; +import type Logger from '@/logger.js'; @Injectable() export class StreamingApiServerService { #wss: WebSocket.WebSocketServer; #connections = new Map(); #cleanConnectionsIntervalId: NodeJS.Timeout | null = null; + #logger: Logger; constructor( @Inject(DI.redisForSub) @@ -49,6 +52,7 @@ export class StreamingApiServerService { private channelFollowingService: ChannelFollowingService, private rateLimiterService: RateLimiterService, private roleService: RoleService, + private loggerService: LoggerService, ) { } @@ -155,6 +159,7 @@ export class StreamingApiServerService { this.notificationService, this.cacheService, this.channelFollowingService, + this.loggerService, user, app, rateLimiter, ); diff --git a/packages/backend/src/server/api/stream/Connection.ts b/packages/backend/src/server/api/stream/Connection.ts index dfc6f0d298..0a7828d163 100644 --- a/packages/backend/src/server/api/stream/Connection.ts +++ b/packages/backend/src/server/api/stream/Connection.ts @@ -17,6 +17,8 @@ import { ChannelFollowingService } from '@/core/ChannelFollowingService.js'; import type { ChannelsService } from './ChannelsService.js'; import type { EventEmitter } from 'events'; import type Channel from './channel.js'; +import { LoggerService } from '@/core/LoggerService.js'; +import type Logger from '@/logger.js'; /** * Main stream connection @@ -39,6 +41,9 @@ export default class Connection { public userIdsWhoMeMutingRenotes: Set = new Set(); public userMutedInstances: Set = new Set(); private fetchIntervalId: NodeJS.Timeout | null = null; + private activeRateLimitRequests: number = 0; + private closingConnection: boolean = false; + private logger: Logger; constructor( private channelsService: ChannelsService, @@ -46,6 +51,7 @@ export default class Connection { private notificationService: NotificationService, private cacheService: CacheService, private channelFollowingService: ChannelFollowingService, + private loggerService: LoggerService, user: MiUser | null | undefined, token: MiAccessToken | null | undefined, @@ -54,6 +60,8 @@ export default class Connection { if (user) this.user = user; if (token) this.token = token; if (rateLimiter) this.rateLimiter = rateLimiter; + + this.logger = loggerService.getLogger('streaming', 'coral', false); } @bindThis @@ -106,8 +114,22 @@ export default class Connection { private async onWsConnectionMessage(data: WebSocket.RawData) { let obj: Record; - if (this.rateLimiter && await this.rateLimiter()) { - return; + if (this.closingConnection) return; + + if (this.rateLimiter) { + if (this.activeRateLimitRequests <= 128) { + this.activeRateLimitRequests++; + const shouldRateLimit = await this.rateLimiter(); + this.activeRateLimitRequests--; + + if (shouldRateLimit) return; + if (this.closingConnection) return; + } else { + this.logger.warn('Closing a connection due to an excessive influx of messages.'); + this.closingConnection = true; + this.wsConnection.close(1008, 'Please stop spamming the streaming API.'); + return; + } } try { -- cgit v1.2.3-freya From 76bbc10aa1305ce88c61783a6ac8aec4371eaa2b Mon Sep 17 00:00:00 2001 From: syuilo <4439005+syuilo@users.noreply.github.com> Date: Sat, 17 Aug 2024 09:30:03 +0900 Subject: fix(backend): 無制限にストリーミングのチャンネルに接続できる問題を修正 MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- packages/backend/src/server/api/stream/Connection.ts | 6 ++++++ 1 file changed, 6 insertions(+) (limited to 'packages/backend/src/server/api/stream') diff --git a/packages/backend/src/server/api/stream/Connection.ts b/packages/backend/src/server/api/stream/Connection.ts index 0a7828d163..e8cd557c1c 100644 --- a/packages/backend/src/server/api/stream/Connection.ts +++ b/packages/backend/src/server/api/stream/Connection.ts @@ -20,6 +20,8 @@ import type Channel from './channel.js'; import { LoggerService } from '@/core/LoggerService.js'; import type Logger from '@/logger.js'; +const MAX_CHANNELS_PER_CONNECTION = 32; + /** * Main stream connection */ @@ -283,6 +285,10 @@ export default class Connection { */ @bindThis public connectChannel(id: string, params: any, channel: string, pong = false) { + if (this.channels.length >= MAX_CHANNELS_PER_CONNECTION) { + return; + } + const channelService = this.channelsService.getChannelService(channel); if (channelService.requireCredential && this.user == null) { -- cgit v1.2.3-freya From 9c1c1e9f099c25cf33f5804fdf3fba547ba73c92 Mon Sep 17 00:00:00 2001 From: Julia Johannesen Date: Sat, 17 Aug 2024 13:08:46 -0400 Subject: Fix logging stuff --- packages/backend/src/server/api/StreamingApiServerService.ts | 1 - packages/backend/src/server/api/stream/Connection.ts | 2 +- 2 files changed, 1 insertion(+), 2 deletions(-) (limited to 'packages/backend/src/server/api/stream') diff --git a/packages/backend/src/server/api/StreamingApiServerService.ts b/packages/backend/src/server/api/StreamingApiServerService.ts index f48af45fb1..db948122bf 100644 --- a/packages/backend/src/server/api/StreamingApiServerService.ts +++ b/packages/backend/src/server/api/StreamingApiServerService.ts @@ -34,7 +34,6 @@ export class StreamingApiServerService { #wss: WebSocket.WebSocketServer; #connections = new Map(); #cleanConnectionsIntervalId: NodeJS.Timeout | null = null; - #logger: Logger; constructor( @Inject(DI.redisForSub) diff --git a/packages/backend/src/server/api/stream/Connection.ts b/packages/backend/src/server/api/stream/Connection.ts index e8cd557c1c..0914cdbb22 100644 --- a/packages/backend/src/server/api/stream/Connection.ts +++ b/packages/backend/src/server/api/stream/Connection.ts @@ -53,7 +53,7 @@ export default class Connection { private notificationService: NotificationService, private cacheService: CacheService, private channelFollowingService: ChannelFollowingService, - private loggerService: LoggerService, + loggerService: LoggerService, user: MiUser | null | undefined, token: MiAccessToken | null | undefined, -- cgit v1.2.3-freya From 3dd993a76a5e5d87a0b31e1eff5093958f239021 Mon Sep 17 00:00:00 2001 From: Julia Johannesen Date: Sat, 17 Aug 2024 14:27:43 -0400 Subject: Add IP and user ID to connection close message --- packages/backend/src/server/api/StreamingApiServerService.ts | 2 +- packages/backend/src/server/api/stream/Connection.ts | 6 +++++- 2 files changed, 6 insertions(+), 2 deletions(-) (limited to 'packages/backend/src/server/api/stream') diff --git a/packages/backend/src/server/api/StreamingApiServerService.ts b/packages/backend/src/server/api/StreamingApiServerService.ts index 19c78fd4d1..2070ab6106 100644 --- a/packages/backend/src/server/api/StreamingApiServerService.ts +++ b/packages/backend/src/server/api/StreamingApiServerService.ts @@ -159,7 +159,7 @@ export class StreamingApiServerService { this.cacheService, this.channelFollowingService, this.loggerService, - user, app, + user, app, requestIp, rateLimiter, ); diff --git a/packages/backend/src/server/api/stream/Connection.ts b/packages/backend/src/server/api/stream/Connection.ts index 0914cdbb22..b71a99b89e 100644 --- a/packages/backend/src/server/api/stream/Connection.ts +++ b/packages/backend/src/server/api/stream/Connection.ts @@ -57,6 +57,7 @@ export default class Connection { user: MiUser | null | undefined, token: MiAccessToken | null | undefined, + private ip: string, rateLimiter: () => Promise, ) { if (user) this.user = user; @@ -127,7 +128,10 @@ export default class Connection { if (shouldRateLimit) return; if (this.closingConnection) return; } else { - this.logger.warn('Closing a connection due to an excessive influx of messages.'); + let connectionInfo = `IP ${this.ip}`; + if (this.user) connectionInfo += `, user ID ${this.user.id}`; + + this.logger.warn(`Closing a connection (${connectionInfo}) due to an excessive influx of messages.`); this.closingConnection = true; this.wsConnection.close(1008, 'Please stop spamming the streaming API.'); return; -- cgit v1.2.3-freya