diff --git a/CLAUDE.md b/CLAUDE.md index cf8d9050c..38933c492 100644 --- a/CLAUDE.md +++ b/CLAUDE.md @@ -136,6 +136,7 @@ must meet, and how it is enforced), [`docs/feed.md`](docs/feed.md), [`docs/profile-fields.md`](docs/profile-fields.md), [`docs/testing.md`](docs/testing.md), [`docs/load-testing.md`](docs/load-testing.md) (JMeter load tests), [`docs/logging-monitoring.md`](docs/logging-monitoring.md), +[`docs/websocket-auth.md`](docs/websocket-auth.md) (who may subscribe to what, and turning enforcement on), [`docs/troubleshooting.md`](docs/troubleshooting.md). Mobile: [`android/README.md`](android/README.md), [`ios/README.md`](ios/README.md), [`docs/ios.md`](docs/ios.md) (iOS plan and remaining manual steps), diff --git a/backend/api/package.json b/backend/api/package.json index 29462148f..541d46fd6 100644 --- a/backend/api/package.json +++ b/backend/api/package.json @@ -1,6 +1,6 @@ { "name": "@compass/api", - "version": "1.93.1", + "version": "1.94.0", "private": true, "description": "Backend API endpoints", "main": "src/serve.ts", @@ -34,7 +34,7 @@ "compression": "^1.8.0", "cors": "2.8.5", "dayjs": "1.11.19", - "express": "5.0.0", + "express": "5.2.1", "firebase-admin": "13.5.0", "gcp-metadata": "6.1.0", "jsdom": "29.0.1", diff --git a/backend/api/src/app.ts b/backend/api/src/app.ts index 624ed82e0..f3c976df7 100644 --- a/backend/api/src/app.ts +++ b/backend/api/src/app.ts @@ -209,7 +209,7 @@ const requestMonitoring: RequestHandler = (req, _res, next) => { } const apiErrorHandler: ErrorRequestHandler = (error, _req, res, _next) => { - if (error instanceof APIError) { + if (error instanceof APIError && error.code !== 500) { log.info(error) if (!res.headersSent) { res.status(error.code).json(error.toJSON()) @@ -217,11 +217,9 @@ const apiErrorHandler: ErrorRequestHandler = (error, _req, res, _next) => { } else { log.error(error) if (!res.headersSent) { - const apiError = APIErrors.internalServerError(error.message || 'Internal server error', { - originalError: error.toString(), - context: 'Unhandled exception in request processing', - }) - res.status(500).json(apiError.toJSON()) + // Generic on purpose, including for a thrown APIError 500: many handlers build those from + // `error.message`, which can carry Postgres constraint and column names. The detail is in the log. + res.status(500).json(APIErrors.internalServerError('Internal server error').toJSON()) } } } diff --git a/backend/api/src/create-comment.ts b/backend/api/src/create-comment.ts index bd2ff1937..b50e82cc4 100644 --- a/backend/api/src/create-comment.ts +++ b/backend/api/src/create-comment.ts @@ -3,7 +3,6 @@ import {assertNotBlocked} from 'api/helpers/blocks' import {MAX_COMMENT_JSON_LENGTH, validateCommentAuthor} from 'api/helpers/comment' import {APIErrors, APIHandler} from 'api/helpers/endpoint' import {Notification} from 'common/notifications' -import {convertComment} from 'common/supabase/comment' import {type Row} from 'common/supabase/utils' import {User} from 'common/user' import {getNotificationDestinationsForUser} from 'common/user-notification-preferences' @@ -13,7 +12,7 @@ import {sendNewEndorsementEmail} from 'email/functions/helpers' import {createSupabaseDirectClient} from 'shared/supabase/init' import {insertNotificationToSupabase} from 'shared/supabase/notifications' import {getPrivateUser, getUser} from 'shared/utils' -import {broadcastUpdatedComment} from 'shared/websockets/helpers' +import {broadcastProfileCommentsChanged} from 'shared/websockets/helpers' export {MAX_COMMENT_JSON_LENGTH} @@ -50,7 +49,7 @@ export const createComment: APIHandler<'create-comment'> = async ( comment.id, ) - broadcastUpdatedComment(convertComment(comment)) + broadcastProfileCommentsChanged(onUser.id) return {status: 'success'} } diff --git a/backend/api/src/helpers/endpoint.ts b/backend/api/src/helpers/endpoint.ts index 4a01e5478..b25072e77 100644 --- a/backend/api/src/helpers/endpoint.ts +++ b/backend/api/src/helpers/endpoint.ts @@ -12,6 +12,7 @@ import {bannedFromWritingError, type BanReason} from 'common/moderation/ban' import {PrivateUser} from 'common/user' import {NextFunction, Request, Response} from 'express' import * as admin from 'firebase-admin' +import {getIp} from 'shared/analytics' import {getPrivateUserByKey, log} from 'shared/utils' import {z} from 'zod' @@ -90,7 +91,6 @@ export const parseCredentials = async (req: Request): Promise => { console.error('Error verifying Firebase JWT: ', err, scheme, { jwtHeader: _header, }) - debug('Failing Firebase JWT payload: ', payload) Sentry.captureException(err, { extra: {jwtHeader: _header}, }) @@ -115,7 +115,7 @@ export const lookupUser = async (creds: Credentials): Promise => { const key = creds.data const privateUser = await getPrivateUserByKey(key) if (!privateUser) { - debug('No private user exists with API key', key) + debug('No private user exists with the provided API key') throw APIErrors.unauthorized('No private user exists with the provided API key.') } return {uid: privateUser.id, creds: {privateUser, ...creds}} @@ -202,6 +202,7 @@ export type APIHandler = ( // API_RATE_LIMIT_PER_MIN_UNAUTHED // Endpoints can be exempted by adding their name to RATE_LIMIT_EXEMPT (comma-separated) const __rateLimitState: Map = new Map() +let __rateLimitWindow = 0 function getRateLimitConfig() { const authed = Number(process.env.API_RATE_LIMIT_PER_MIN_AUTHED ?? 120) @@ -211,8 +212,9 @@ function getRateLimitConfig() { function rateLimitKey(name: string, req: Request, auth?: AuthedUser) { if (auth) return `uid:${auth.uid}` - // fallback to IP for unauthenticated requests - return `ip:${req.ip}` + // fallback to IP for unauthenticated requests. Not `req.ip`: there is no `trust proxy`, so behind + // Cloud Run that is Google's front end, and every visitor would share a single bucket. + return `ip:${getIp(req)}` } function checkRateLimit(name: string, req: Request, res: Response, auth?: AuthedUser) { @@ -224,6 +226,13 @@ function checkRateLimit(name: string, req: Request, res: Response, auth?: Authed const windowMs = 60_000 const windowStart = Math.floor(now / windowMs) * windowMs + // Entries from a past window are never read again, and one key per client IP would otherwise grow + // for as long as the instance lives. + if (windowStart !== __rateLimitWindow) { + __rateLimitState.clear() + __rateLimitWindow = windowStart + } + let state = __rateLimitState.get(key) if (!state || state.windowStart !== windowStart) { state = {windowStart, count: 0} diff --git a/backend/api/src/hide-comment.ts b/backend/api/src/hide-comment.ts index 386a1b87d..2674d6b06 100644 --- a/backend/api/src/hide-comment.ts +++ b/backend/api/src/hide-comment.ts @@ -1,14 +1,23 @@ import {APIErrors, APIHandler} from 'api/helpers/endpoint' import {isAdminId} from 'common/envs/constants' -import {convertComment, convertVoteComment} from 'common/supabase/comment' +import {convertVoteComment} from 'common/supabase/comment' import {Row} from 'common/supabase/utils' import {createSupabaseDirectClient} from 'shared/supabase/init' -import {broadcastUpdatedComment, broadcastUpdatedVoteComment} from 'shared/websockets/helpers' +import { + broadcastProfileCommentsChanged, + broadcastUpdatedVoteComment, +} from 'shared/websockets/helpers' export const hideComment: APIHandler<'hide-comment'> = async ( {commentId, hide, commentType}, auth, ) => { + // Nothing records who hid a comment, so an author who could un-hide their own comment could also + // reverse a removal by a moderator or by the profile owner. + if (!hide && !isAdminId(auth.uid)) { + throw APIErrors.forbidden('Only admins can restore a hidden comment') + } + const pg = createSupabaseDirectClient() if (commentType === 'vote') { @@ -44,5 +53,5 @@ export const hideComment: APIHandler<'hide-comment'> = async ( await pg.none(`update profile_comments set hidden = $2 where id = $1`, [commentId, hide]) - broadcastUpdatedComment(convertComment(comment)) + broadcastProfileCommentsChanged(comment.on_user_id) } diff --git a/backend/api/src/react-to-message.ts b/backend/api/src/react-to-message.ts index 44950ae37..313309a43 100644 --- a/backend/api/src/react-to-message.ts +++ b/backend/api/src/react-to-message.ts @@ -3,6 +3,16 @@ import {createSupabaseDirectClient} from 'shared/supabase/init' import {APIErrors, APIHandler} from './helpers/endpoint' +// `reactions` maps each emoji to the ids of the members who used it. This takes one member ($2) off +// one emoji ($1), and drops the emoji once they were the last on it. The scalar comparison covers old +// rows holding a bare id rather than a list, which `MessageReactions` also tolerates. +const WITHOUT_REACTOR = ` + CASE + WHEN reactions -> $1 IN (to_jsonb($2::text), to_jsonb(ARRAY [$2::text])) + THEN reactions - $1 + ELSE jsonb_set(reactions, ARRAY [$1], (reactions -> $1) - $2::text) + END` + export const reactToMessage: APIHandler<'react-to-message'> = async ( {messageId, reaction, toDelete}, auth, @@ -28,10 +38,10 @@ export const reactToMessage: APIHandler<'react-to-message'> = async ( // Remove the reaction await pg.none( `UPDATE private_user_messages - SET reactions = reactions - $1 - WHERE id = $2 - AND reactions -> $1 ? $3`, - [reaction, messageId, auth.uid], + SET reactions = ${WITHOUT_REACTOR} + WHERE id = $3 + AND reactions -> $1 ? $2`, + [reaction, auth.uid, messageId], ) } else { // Toggle reaction @@ -39,8 +49,8 @@ export const reactToMessage: APIHandler<'react-to-message'> = async ( `UPDATE private_user_messages SET reactions = CASE - WHEN reactions -> $1 IS NOT NULL - THEN reactions - $1 + WHEN reactions -> $1 ? $2 + THEN ${WITHOUT_REACTOR} ELSE jsonb_set( COALESCE(reactions, '{}'::jsonb), ARRAY [$1], diff --git a/backend/api/tests/unit/create-comment.unit.test.ts b/backend/api/tests/unit/create-comment.unit.test.ts index b29d1ab42..80de3a935 100644 --- a/backend/api/tests/unit/create-comment.unit.test.ts +++ b/backend/api/tests/unit/create-comment.unit.test.ts @@ -1,6 +1,5 @@ import {createComment} from 'api/create-comment' import {AuthedUser} from 'api/helpers/endpoint' -import {convertComment} from 'common/supabase/comment' import {sqlMatch} from 'common/test-utils' import * as notificationPrefs from 'common/user-notification-preferences' import {richTextToString} from 'common/util/parse' @@ -78,7 +77,6 @@ describe('createComment', () => { content: mockContent.content, replyToCommentId: mockReplyToCommentId, } - const mockConvertCommentReturn = 'mockConverComment' ;(sharedUtils.getUser as jest.Mock) .mockResolvedValueOnce(mockCreator) @@ -91,7 +89,6 @@ describe('createComment', () => { ;(notificationPrefs.getNotificationDestinationsForUser as jest.Mock).mockReturnValue( mockNotificationDestination, ) - ;(convertComment as jest.Mock).mockReturnValue(mockConvertCommentReturn) const results = await createComment(mockProps, mockAuth, mockReq) @@ -128,8 +125,8 @@ describe('createComment', () => { mockOnUser, richTextToString(mockProps.content), ) - expect(websocketHelpers.broadcastUpdatedComment).toBeCalledTimes(1) - expect(websocketHelpers.broadcastUpdatedComment).toBeCalledWith(mockConvertCommentReturn) + expect(websocketHelpers.broadcastProfileCommentsChanged).toBeCalledTimes(1) + expect(websocketHelpers.broadcastProfileCommentsChanged).toBeCalledWith(mockOnUser.id) }) }) diff --git a/backend/api/tests/unit/hide-comment.unit.test.ts b/backend/api/tests/unit/hide-comment.unit.test.ts index e4fd096e1..4d03c294b 100644 --- a/backend/api/tests/unit/hide-comment.unit.test.ts +++ b/backend/api/tests/unit/hide-comment.unit.test.ts @@ -1,7 +1,6 @@ import {AuthedUser} from 'api/helpers/endpoint' import {hideComment} from 'api/hide-comment' import * as envConsts from 'common/envs/constants' -import {convertComment} from 'common/supabase/comment' import {sqlMatch} from 'common/test-utils' import * as supabaseInit from 'shared/supabase/init' import * as websocketHelpers from 'shared/websockets/helpers' @@ -44,11 +43,9 @@ describe('hideComment', () => { user_name: 'mockUserName', user_username: 'mockUserUsername', } - const mockConvertedComment = 'mockConvertedCommentValue' ;(mockPg.oneOrNone as jest.Mock).mockResolvedValue(mockComment) jest.spyOn(envConsts, 'isAdminId').mockReturnValue(true) - ;(convertComment as jest.Mock).mockReturnValue(mockConvertedComment) await hideComment(mockProps, mockAuth, mockReq) @@ -59,10 +56,10 @@ describe('hideComment', () => { ) expect(envConsts.isAdminId).toBeCalledTimes(1) expect(envConsts.isAdminId).toBeCalledWith(mockAuth.uid) - expect(convertComment).toBeCalledTimes(1) - expect(convertComment).toBeCalledWith(mockComment) - expect(websocketHelpers.broadcastUpdatedComment).toBeCalledTimes(1) - expect(websocketHelpers.broadcastUpdatedComment).toBeCalledWith(mockConvertedComment) + expect(websocketHelpers.broadcastProfileCommentsChanged).toBeCalledTimes(1) + expect(websocketHelpers.broadcastProfileCommentsChanged).toBeCalledWith( + mockComment.on_user_id, + ) }) it('should successfully hide the comment if the user is the one who made the comment', async () => { @@ -117,6 +114,38 @@ describe('hideComment', () => { await hideComment(mockProps, mockAuth, mockReq) }) }) + describe('when restoring a hidden comment', () => { + const mockAuth = {uid: '321'} as AuthedUser + const mockReq = {} as any + const mockComment = {hidden: true, id: 123, on_user_id: '321', user_id: '321'} + + it.each([undefined, 'vote'] as const)( + 'should refuse the author and the profile owner (commentType %s)', + async (commentType) => { + ;(mockPg.oneOrNone as jest.Mock).mockResolvedValue(mockComment) + jest.spyOn(envConsts, 'isAdminId').mockReturnValue(false) + + await expect( + hideComment({commentId: 'mockCommentId', hide: false, commentType}, mockAuth, mockReq), + ).rejects.toThrow('Only admins can restore a hidden comment') + + expect(mockPg.none).not.toBeCalled() + }, + ) + + it('should let an admin restore it', async () => { + ;(mockPg.oneOrNone as jest.Mock).mockResolvedValue(mockComment) + jest.spyOn(envConsts, 'isAdminId').mockReturnValue(true) + + await hideComment({commentId: 'mockCommentId', hide: false}, mockAuth, mockReq) + + expect(mockPg.none).toBeCalledWith(sqlMatch('update profile_comments set hidden = $2'), [ + 'mockCommentId', + false, + ]) + }) + }) + describe('when an error occurs', () => { it('should throw if the comment was not found', async () => { const mockProps = { diff --git a/backend/api/tests/unit/react-to-message.unit.test.ts b/backend/api/tests/unit/react-to-message.unit.test.ts index 9ed6ba8f5..59b6f8be4 100644 --- a/backend/api/tests/unit/react-to-message.unit.test.ts +++ b/backend/api/tests/unit/react-to-message.unit.test.ts @@ -78,9 +78,11 @@ describe('reactToMessage', () => { expect(result.success).toBeTruthy() expect(mockPg.oneOrNone).toBeCalledTimes(1) expect(mockPg.none).toBeCalledTimes(1) - expect(params1).toEqual([mockProps.reaction, mockProps.messageId, mockAuth.uid]) + expect(params1).toEqual([mockProps.reaction, mockAuth.uid, mockProps.messageId]) expect(sql1).toEqual(sqlMatch('UPDATE private_user_messages')) - expect(sql1).toEqual(sqlMatch('SET reactions = reactions - $1')) + // Takes this member off the emoji; the emoji key itself goes only with its last reactor. + expect(sql1).toEqual(sqlMatch('(reactions -> $1) - $2::text')) + expect(sql1).toEqual(sqlMatch('AND reactions -> $1 ? $2')) }) }) describe('when an error occurs', () => { diff --git a/backend/shared/CLAUDE.md b/backend/shared/CLAUDE.md index db8ff0230..2cf2b7020 100644 --- a/backend/shared/CLAUDE.md +++ b/backend/shared/CLAUDE.md @@ -23,6 +23,7 @@ backend/shared/src/ │ ├── users.ts, messages.ts, notifications.ts, options.ts Domain queries ├── websockets/ │ ├── server.ts, switchboard.ts +│ ├── topics.ts who may subscribe to what (closed by default) │ └── helpers.ts broadcastUpdatedUser, broadcastUpdatedPrivateUser, ... │ (topics here are what the web `useApiSubscription` listens to) ├── monitoring/ @@ -40,7 +41,8 @@ backend/shared/src/ - **SQL**: never string-concatenate. Use the helpers in `supabase/utils` for inserts/updates, or compose with `sql-builder` for reads. Lowercase SQL by convention. - **New websocket topic**: add a `broadcastX` function in `websockets/helpers.ts` so the frontend has a typed - topic to subscribe to. + topic to subscribe to, and list the topic in `websockets/topics.ts`, or subscribers are refused. See + [`docs/websocket-auth.md`](../../docs/websocket-auth.md). - **Logging**: `log` from `monitoring/log`. `debug()` from `common/logger` is fine for verbose dev logs but `log` carries structured context to Cloud Logging. - **Adding a domain helper**: if a query is reused by more than one API endpoint, lift it into diff --git a/backend/shared/src/monitoring/metrics.ts b/backend/shared/src/monitoring/metrics.ts index fbc61ac20..326e375d9 100644 --- a/backend/shared/src/monitoring/metrics.ts +++ b/backend/shared/src/monitoring/metrics.ts @@ -35,6 +35,10 @@ export const CUSTOM_METRICS = { metricKind: 'CUMULATIVE', valueKind: 'int64Value', }, + 'ws/unauthed_private_sessions': { + metricKind: 'CUMULATIVE', + valueKind: 'int64Value', + }, 'http/request_count': { metricKind: 'CUMULATIVE', valueKind: 'int64Value', diff --git a/backend/shared/src/websockets/helpers.ts b/backend/shared/src/websockets/helpers.ts index 8e93e6a21..b317e783d 100644 --- a/backend/shared/src/websockets/helpers.ts +++ b/backend/shared/src/websockets/helpers.ts @@ -1,4 +1,4 @@ -import {type ProfileComment, type VoteComment} from 'common/comment' +import {type VoteComment} from 'common/comment' import {type User} from 'common/user' import {broadcast, broadcastMulti} from './server' @@ -12,8 +12,10 @@ export function broadcastUpdatedUser(user: Partial & {id: string}) { broadcast(`user/${user.id}`, {user}) } -export function broadcastUpdatedComment(comment: ProfileComment) { - broadcast(`user/${comment.onUserId}/comment`, {comment}) +// Carries no comment on purpose: this topic is open to anyone, while who may read a profile's comments +// depends on the viewer. Subscribers refetch, and `get-profile-comments` applies those rules. +export function broadcastProfileCommentsChanged(onUserId: string) { + broadcast(`user/${onUserId}/comment`, {}) } export function broadcastUpdatedVoteComment(comment: VoteComment) { diff --git a/backend/shared/src/websockets/server.ts b/backend/shared/src/websockets/server.ts index 6c9e74075..c5e13f74b 100644 --- a/backend/shared/src/websockets/server.ts +++ b/backend/shared/src/websockets/server.ts @@ -6,14 +6,17 @@ import { CLIENT_MESSAGE_SCHEMA, ClientMessage, ServerMessage, + WS_MAX_SESSION_MS, } from 'common/api/websockets' import {IS_LOCAL} from 'common/hosting/constants' import {debug} from 'common/logger' +import * as admin from 'firebase-admin' import {isError} from 'lodash' import {log, metrics} from 'shared/utils' import {RawData, Server as WebSocketServer, WebSocket} from 'ws' import {Switchboard} from './switchboard' +import {canSubscribe, holdsForeignTopic} from './topics' // Extend the type definition locally interface HeartbeatWebSocket extends WebSocket { @@ -60,18 +63,60 @@ function parseMessage(data: RawData): ClientMessage { } } -function processMessage(ws: HeartbeatWebSocket, data: RawData): ServerMessage<'ack'> { +// Off until the store apps that predate `identify` have aged out: they can't authenticate, so turning +// this on stops their live updates. See docs/websocket-auth.md. +const enforceTopicAuth = () => process.env.WS_ENFORCE_TOPIC_AUTH === 'true' + +async function identify(ws: HeartbeatWebSocket, token: string | undefined) { + let uid: string | undefined + let invalid = false + if (token) { + try { + uid = (await admin.auth().verifyIdToken(token)).uid + } catch (err) { + debug('Websocket identify failed', err) + invalid = true + } + } + // The socket can close while the token is being verified. + if (!SWITCHBOARD.clients.has(ws)) return + + if (uid) SWITCHBOARD.identify(ws, uid) + else SWITCHBOARD.deidentify(ws) + + // The connection may have just changed hands, so what it held under the last identity goes. + if (enforceTopicAuth()) { + const client = SWITCHBOARD.getClient(ws) + const lost = Array.from(client.subscriptions).filter((topic) => !canSubscribe(topic, uid)) + SWITCHBOARD.unsubscribe(ws, ...lost) + } + + if (invalid) throw new Error('Invalid token.') +} + +function subscribe(ws: HeartbeatWebSocket, topics: string[]) { + const {uid} = SWITCHBOARD.getClient(ws) + // A refused topic is dropped, not failed: nothing in the older clients handles a failed ack, and + // the current one subscribes before its token is verified, then again after. + const allowed = enforceTopicAuth() ? topics.filter((topic) => canSubscribe(topic, uid)) : topics + SWITCHBOARD.subscribe(ws, ...allowed) +} + +async function processMessage( + ws: HeartbeatWebSocket, + data: RawData, +): Promise> { try { const msg = parseMessage(data) const {type, txid} = msg try { switch (type) { case 'identify': { - SWITCHBOARD.identify(ws, msg.uid) + await identify(ws, msg.token) break } case 'subscribe': { - SWITCHBOARD.subscribe(ws, ...msg.topics) + subscribe(ws, msg.topics) break } case 'unsubscribe': { @@ -137,7 +182,6 @@ export function listen(server: HttpServer, path: string) { let deadConnectionCleaner: NodeJS.Timeout | undefined const HEARTBEAT_INTERVAL = 25000 // 25s - const MAX_SESSION_DURATION = 4 * 60 * 1000 // 4 minutes wss.on('listening', () => { log.info(`Web socket server listening on ${path}. ${getWebsocketUrl()}`) @@ -161,14 +205,14 @@ export function listen(server: HttpServer, path: string) { ws.isAlive = true // debug('Received pong from client'); - // 2. Hard Session Limit: Sever connection after 5 mins to allow Scale-to-Zero + // 2. Hard Session Limit: Sever connection after 4 mins to allow Scale-to-Zero // This stops bots/idle users from resetting the Cloud Run 15-min idle timer indefinitely. const sessionTimeout = setTimeout(() => { if (ws.readyState === WebSocket.OPEN) { - debug('Closing WS: Max session duration reached (5m)') + debug('Closing WS: Max session duration reached') ws.close(1000, 'Session limit reached') // 1000 is a normal closure } - }, MAX_SESSION_DURATION) + }, WS_MAX_SESSION_MS) ws.on('pong', () => (ws.isAlive = true)) @@ -177,8 +221,8 @@ export function listen(server: HttpServer, path: string) { debug('WS client connected.') SWITCHBOARD.connect(ws) - ws.on('message', (data) => { - const result = processMessage(ws, data) + ws.on('message', async (data) => { + const result = await processMessage(ws, data) // mqp: check ws.readyState before sending? if (ws.readyState === WebSocket.OPEN) { ws.send(JSON.stringify(result)) @@ -190,6 +234,8 @@ export function listen(server: HttpServer, path: string) { metrics.inc('ws/connections_terminated') metrics.set('ws/open_connections', wss.clients.size) debug(`WS client disconnected.`, {code, reason: reason.toString()}) + const {uid, subscriptions} = SWITCHBOARD.getClient(ws) + if (holdsForeignTopic(subscriptions, uid)) metrics.inc('ws/unauthed_private_sessions') SWITCHBOARD.disconnect(ws) }) diff --git a/backend/shared/src/websockets/topics.ts b/backend/shared/src/websockets/topics.ts new file mode 100644 index 000000000..ba58a46ee --- /dev/null +++ b/backend/shared/src/websockets/topics.ts @@ -0,0 +1,40 @@ +import {IS_LOCAL} from 'common/hosting/constants' + +const PUBLIC_TOPICS = [/^user\/[^/]+\/comment$/, /^vote\/[^/]+\/comment$/] + +// One capture group: the member the topic belongs to. +const OWNED_TOPICS = [ + /^user\/([^/]+)$/, + /^private-user\/([^/]+)$/, + /^user-notifications\/([^/]+)$/, + /^private-user-messages\/([^/]+)(?:\/typing)?$/, +] + +// What the web hooks subscribe to while nobody is signed in. Nothing is ever broadcast on these. +const PLACEHOLDER_OWNERS = ['_', 'undefined'] + +const topicOwner = (topic: string) => { + for (const pattern of OWNED_TOPICS) { + const owner = topic.match(pattern)?.[1] + if (owner) return owner + } + return undefined +} + +/** + * Whether a connection identified as `uid` may listen to `topic`. Closed by default: a new topic is + * refused until it is listed above, so forgetting to classify one cannot make it public. + */ +export const canSubscribe = (topic: string, uid: string | undefined) => { + if (topic === '*') return IS_LOCAL + if (PUBLIC_TOPICS.some((pattern) => pattern.test(topic))) return true + const owner = topicOwner(topic) + return owner !== undefined && owner === uid +} + +/** Whether a session is listening to a real member's private topic without being that member. */ +export const holdsForeignTopic = (topics: Iterable, uid: string | undefined) => + Array.from(topics).some((topic) => { + const owner = topicOwner(topic) + return owner !== undefined && owner !== uid && !PLACEHOLDER_OWNERS.includes(owner) + }) diff --git a/backend/shared/tests/unit/websockets.unit.test.ts b/backend/shared/tests/unit/websockets.unit.test.ts new file mode 100644 index 000000000..775b7601c --- /dev/null +++ b/backend/shared/tests/unit/websockets.unit.test.ts @@ -0,0 +1,176 @@ +import {createServer, Server} from 'node:http' +import {AddressInfo} from 'node:net' + +import {broadcast, listen} from 'shared/websockets/server' +import {canSubscribe, holdsForeignTopic} from 'shared/websockets/topics' +import {WebSocket} from 'ws' + +jest.mock('firebase-admin', () => ({ + auth: () => ({ + verifyIdToken: async (token: string) => { + if (!token.startsWith('valid-')) throw new Error('bad token') + return {uid: token.replace('valid-', '')} + }, + }), +})) + +describe('canSubscribe', () => { + it.each([ + 'user/alice', + 'private-user/alice', + 'user-notifications/alice', + 'private-user-messages/alice', + 'private-user-messages/alice/typing', + ])('limits %s to its owner', (topic) => { + expect(canSubscribe(topic, 'alice')).toBe(true) + expect(canSubscribe(topic, 'mallory')).toBe(false) + expect(canSubscribe(topic, undefined)).toBe(false) + }) + + it.each(['user/alice/comment', 'vote/12/comment'])('leaves %s public', (topic) => { + expect(canSubscribe(topic, undefined)).toBe(true) + }) + + it('refuses a topic it does not know', () => { + expect(canSubscribe('some-new-topic/alice', 'alice')).toBe(false) + }) +}) + +describe('holdsForeignTopic', () => { + it('ignores own, public and signed-out placeholder topics', () => { + const topics = ['user/alice', 'vote/1/comment', 'user/_', 'user-notifications/undefined'] + expect(holdsForeignTopic(topics, 'alice')).toBe(false) + }) + + it('flags another member’s private topic', () => { + expect(holdsForeignTopic(['user-notifications/alice'], undefined)).toBe(true) + expect(holdsForeignTopic(['user-notifications/alice'], 'mallory')).toBe(true) + }) +}) + +describe('websocket server', () => { + let server: Server + let wss: ReturnType + let sockets: WebSocket[] + let txid = 0 + + const connect = async () => { + const port = (server.address() as AddressInfo).port + const ws = new WebSocket(`ws://127.0.0.1:${port}/ws`) + sockets.push(ws) + const received: any[] = [] + ws.on('message', (raw) => { + const msg = JSON.parse(raw.toString()) + if (msg.type === 'broadcast') received.push(msg) + }) + await new Promise((resolve) => ws.once('open', resolve)) + const send = (msg: object) => + new Promise((resolve) => { + const id = txid++ + const onMessage = (raw: Buffer) => { + const reply = JSON.parse(raw.toString()) + if (reply.type === 'ack' && reply.txid === id) { + ws.off('message', onMessage) + resolve(reply) + } + } + ws.on('message', onMessage) + ws.send(JSON.stringify({...msg, txid: id})) + }) + return {send, received} + } + + // Sent last: its arrival shows every earlier broadcast on the same socket has been delivered. + const flush = async (client: Awaited>) => { + await client.send({type: 'subscribe', topics: ['vote/flush/comment']}) + broadcast('vote/flush/comment', {}) + await new Promise((resolve) => { + const check = () => + client.received.some((m) => m.topic === 'vote/flush/comment') + ? resolve() + : setTimeout(check, 5) + check() + }) + return client.received.filter((m) => m.topic !== 'vote/flush/comment') + } + + const secret = 'user-notifications/alice' + + beforeEach(async () => { + sockets = [] + server = createServer() + wss = listen(server, '/ws') + await new Promise((resolve) => server.listen(0, '127.0.0.1', resolve)) + }) + + afterEach(async () => { + delete process.env.WS_ENFORCE_TOPIC_AUTH + sockets.forEach((ws) => ws.terminate()) + wss.close() + await new Promise((resolve) => server.close(resolve)) + }) + + describe('with enforcement on', () => { + beforeEach(() => { + process.env.WS_ENFORCE_TOPIC_AUTH = 'true' + }) + + it('keeps an anonymous connection off a private topic, and still acks', async () => { + const client = await connect() + const ack = await client.send({type: 'subscribe', topics: [secret]}) + expect(ack.success).toBe(true) + broadcast(secret, {notification: 'private'}) + expect(await flush(client)).toEqual([]) + }) + + it('ignores the claimed uid and goes by the token', async () => { + const client = await connect() + await client.send({type: 'identify', uid: 'alice', token: 'valid-mallory'}) + await client.send({type: 'subscribe', topics: [secret]}) + broadcast(secret, {notification: 'private'}) + expect(await flush(client)).toEqual([]) + }) + + it('refuses an identify with a bad token', async () => { + const client = await connect() + const ack = await client.send({type: 'identify', uid: 'alice', token: 'forged'}) + expect(ack.success).toBe(false) + await client.send({type: 'subscribe', topics: [secret]}) + broadcast(secret, {notification: 'private'}) + expect(await flush(client)).toEqual([]) + }) + + it('delivers to the owner', async () => { + const client = await connect() + await client.send({type: 'identify', uid: 'alice', token: 'valid-alice'}) + await client.send({type: 'subscribe', topics: [secret]}) + broadcast(secret, {notification: 'private'}) + expect(await flush(client)).toHaveLength(1) + }) + + it('drops private topics on sign-out', async () => { + const client = await connect() + await client.send({type: 'identify', uid: 'alice', token: 'valid-alice'}) + await client.send({type: 'subscribe', topics: [secret]}) + await client.send({type: 'identify', uid: ''}) + broadcast(secret, {notification: 'private'}) + expect(await flush(client)).toEqual([]) + }) + + it('drops the old member’s topics when the connection changes hands', async () => { + const client = await connect() + await client.send({type: 'identify', uid: 'alice', token: 'valid-alice'}) + await client.send({type: 'subscribe', topics: [secret]}) + await client.send({type: 'identify', uid: 'bob', token: 'valid-bob'}) + broadcast(secret, {notification: 'private'}) + expect(await flush(client)).toEqual([]) + }) + }) + + it('keeps today’s behaviour for old clients until enforcement is on', async () => { + const client = await connect() + await client.send({type: 'subscribe', topics: [secret]}) + broadcast(secret, {notification: 'private'}) + expect(await flush(client)).toHaveLength(1) + }) +}) diff --git a/backend/supabase/migration.sql b/backend/supabase/migration.sql index 05491d3ab..f3403ea61 100644 --- a/backend/supabase/migration.sql +++ b/backend/supabase/migration.sql @@ -98,4 +98,6 @@ BEGIN; \i backend/supabase/migrations/20261029_private_votes_and_stars.sql \i backend/supabase/migrations/20261031_compatibility_scores_prompt.sql \i backend/supabase/migrations/20261102_add_vote_edits.sql +\i backend/supabase/migrations/20261105_lock_user_notifications.sql +\i backend/supabase/migrations/20261105_lock_deleted_users_insert.sql COMMIT; diff --git a/backend/supabase/migrations/20261105_lock_deleted_users_insert.sql b/backend/supabase/migrations/20261105_lock_deleted_users_insert.sql new file mode 100644 index 000000000..9c5c6bc06 --- /dev/null +++ b/backend/supabase/migrations/20261105_lock_deleted_users_insert.sql @@ -0,0 +1,9 @@ +-- deleted_users took inserts from anyone holding the anon key, which let a visitor write fake rows +-- into the account-deletion reasons. The only writer is `delete-me`, through the service-role +-- backend, which RLS does not apply to. + +drop policy if exists "anyone can insert" on deleted_users; + +revoke insert on table deleted_users from anon, authenticated; + +notify pgrst, 'reload schema'; diff --git a/backend/supabase/migrations/20261105_lock_user_notifications.sql b/backend/supabase/migrations/20261105_lock_user_notifications.sql new file mode 100644 index 000000000..47453a2ee --- /dev/null +++ b/backend/supabase/migrations/20261105_lock_user_notifications.sql @@ -0,0 +1,13 @@ +-- user_notifications was still readable with the anon key, so anyone could pull every member's +-- notifications, the same data the websocket topic `user-notifications/` is now restricted to +-- its owner for. The web reads them through the `get-notifications` endpoint and every write comes +-- from the service-role backend, so nothing depends on the client grant. +-- +-- The revoke is what closes this. The policy name is a best guess, in case prod has one: it does not +-- matter if it is named differently, as a policy cannot grant what the role has no privilege for. + +drop policy if exists "public read" on user_notifications; + +revoke select on table user_notifications from anon, authenticated; + +notify pgrst, 'reload schema'; diff --git a/common/src/api/schema.ts b/common/src/api/schema.ts index 887abec5a..764b82762 100644 --- a/common/src/api/schema.ts +++ b/common/src/api/schema.ts @@ -89,6 +89,12 @@ import {FullUser, HiddenProfile} from './user-types' // mqp: very unscientific, // mqp: very unscientific, just balancing our willingness to accept load // with user willingness to put up with stale data +// `z.string().url()` alone accepts any scheme, `javascript:` included. +const httpUrl = z + .string() + .url() + .regex(/^https?:\/\//i, 'Must be an http(s) URL') + export const DEFAULT_CACHE_STRATEGY = 'public, max-age=5, stale-while-revalidate=10' /** @@ -491,7 +497,9 @@ export const API = (_apiTypeCheck = { method: 'POST', authed: true, rateLimited: true, - props: profileSchema.partial(), + // The referrer is set once, at signup. Being editable afterwards would let a member credit + // themselves or anyone else in the referral stats. + props: profileSchema.omit({referred_by_username: true}).partial(), returns: {} as ProfileRow, summary: 'Update profile fields for the authenticated user. `interests`, `causes` and `work` (option ids or new names) are written in the same transaction, so a form save is one request and one compatibility recompute.', @@ -1743,7 +1751,7 @@ export const API = (_apiTypeCheck = { description: z.string().max(2000).optional(), locationType: z.enum(['in_person', 'online']), locationAddress: z.string().max(500).optional(), - locationUrl: z.string().url().max(500).optional(), + locationUrl: httpUrl.max(500).optional(), eventStartTime: z.string().datetime(), eventEndTime: z.string().datetime().optional(), maxParticipants: z.number().int().min(1).optional(), @@ -1776,7 +1784,7 @@ export const API = (_apiTypeCheck = { description: z.string().max(2000).optional(), locationType: z.enum(['in_person', 'online']), locationAddress: z.string().max(500).optional(), - locationUrl: z.string().url().max(500).optional(), + locationUrl: httpUrl.max(500).optional(), eventStartTime: z.string(), eventEndTime: z.string().optional(), maxParticipants: z.number().min(1).max(1000).optional(), diff --git a/common/src/api/websocket-client.ts b/common/src/api/websocket-client.ts index 660393172..faf46d66a 100644 --- a/common/src/api/websocket-client.ts +++ b/common/src/api/websocket-client.ts @@ -1,4 +1,4 @@ -import {ClientMessage, ClientMessageType, ServerMessage} from './websockets' +import {ClientMessage, ClientMessageType, ServerMessage, WS_MAX_SESSION_MS} from './websockets' // mqp: useful for debugging const VERBOSE_LOGGING = false @@ -8,6 +8,15 @@ const TIMEOUT_MS = 5000 const RECONNECT_WAIT_MS = 5000 +// Spreads out the tabs that all lost their connection at once, after a deploy for instance. +const RECONNECT_JITTER_MS = 3000 + +const ACTIVITY_EVENTS = ['pointerdown', 'keydown', 'touchstart', 'scroll', 'focus'] + +const isTabVisible = () => typeof document === 'undefined' || document.visibilityState === 'visible' + +const jitter = () => Math.random() * RECONNECT_JITTER_MS + type ConnectingState = typeof WebSocket.CONNECTING type OpenState = typeof WebSocket.OPEN type ClosingState = typeof WebSocket.CLOSING @@ -30,6 +39,8 @@ export function formatState(state: ReadyState) { } } +export type RealtimeIdentity = {uid: string; token: string} + export type BroadcastHandler = (msg: ServerMessage<'broadcast'>) => void type OutstandingTxn = { @@ -40,7 +51,12 @@ type OutstandingTxn = { /** Client for the API websocket realtime server. Automatically manages reconnection * and resubscription on disconnect, and allows subscribers to get a callback - * when something is broadcasted. */ + * when something is broadcasted. + * + * It only reconnects for a tab someone is using: visible, with activity during the last session. + * An open socket keeps a Cloud Run instance awake, so a forgotten tab has to let go of it, and comes + * back on the next sign of life. Broadcasts sent in between are lost, which is what `onReconnect` is + * for. */ export class APIRealtimeClient { ws!: WebSocket url: string @@ -51,13 +67,60 @@ export class APIRealtimeClient { subscriptions: Map connectTimeout?: NodeJS.Timeout heartbeat?: number | undefined + getIdentity?: () => Promise + identified: boolean + // identify calls run one at a time, so a slow sign-in cannot land after the sign-out that followed + identityQueue: Promise + lastActivity: number + closedManually: boolean + hasOpened: boolean + reconnectHandlers: Set<() => void> - constructor(url: string) { + constructor(url: string, getIdentity?: () => Promise) { this.url = url this.txid = 0 this.txns = new Map() this.subscriptions = new Map() + this.getIdentity = getIdentity + this.identified = false + this.identityQueue = Promise.resolve() + this.lastActivity = Date.now() + this.closedManually = false + this.hasOpened = false + this.reconnectHandlers = new Set() this.connect() + this.listenForActivity() + } + + listenForActivity() { + if (typeof window === 'undefined' || !window.addEventListener) return + const onActivity = () => this.noteActivity() + // Capture, because a scroll inside an element does not bubble up to the window. + for (const event of ACTIVITY_EVENTS) { + window.addEventListener(event, onActivity, {capture: true, passive: true}) + } + document.addEventListener('visibilitychange', () => { + if (isTabVisible()) this.noteActivity() + }) + } + + noteActivity() { + this.lastActivity = Date.now() + if (this.state === WebSocket.CLOSED && !this.closedManually) { + this.waitAndReconnect(jitter()) + } + } + + get inUse() { + return isTabVisible() && Date.now() - this.lastActivity < WS_MAX_SESSION_MS + } + + /** Runs after a reconnect, not the first connect, to fetch what was broadcast in between. */ + onReconnect(handler: () => void) { + this.reconnectHandlers.add(handler) + return () => { + this.reconnectHandlers.delete(handler) + } } get state() { @@ -65,6 +128,7 @@ export class APIRealtimeClient { } close() { + this.closedManually = true this.ws.close(1000, 'Closed manually.') clearTimeout(this.connectTimeout) } @@ -80,7 +144,7 @@ export class APIRealtimeClient { console.error('API websocket error: ', ev) // this can fire without an onclose if this is the first time we ever try // to connect, so we need to turn on our reconnect in that case - this.waitAndReconnect() + if (this.inUse) this.waitAndReconnect() } this.ws.onopen = (_ev) => { if (VERBOSE_LOGGING) { @@ -92,11 +156,14 @@ export class APIRealtimeClient { this.sendMessage('ping', {}).catch(console.error) } }, 25000) - if (this.subscriptions.size > 0) { - this.sendMessage('subscribe', { - topics: Array.from(this.subscriptions.keys()), - }).catch(console.error) - } + this.identified = false + this.resubscribe().catch(console.error) + const isReconnect = this.hasOpened + this.hasOpened = true + // After the identity, so the private topics are live again before anyone refetches. + void this.refreshIdentity().then(() => { + if (isReconnect) this.reconnectHandlers.forEach((handler) => handler()) + }) } this.ws.onclose = (ev) => { // note that if the connection closes due to an error, onerror fires and then this @@ -114,19 +181,19 @@ export class APIRealtimeClient { } this.txns.clear() - // 1000 is RFC code for normal on-purpose closure - if (ev.code !== 1000) { - this.waitAndReconnect() + // 1000 is also what the server sends at the session limit, so it does not mean we are done. + if (!this.closedManually && this.inUse) { + this.waitAndReconnect(ev.code === 1000 ? jitter() : RECONNECT_WAIT_MS) } } } - waitAndReconnect() { + waitAndReconnect(delayMs = RECONNECT_WAIT_MS) { if (this.connectTimeout == null) { this.connectTimeout = setTimeout(() => { this.connectTimeout = undefined this.connect() - }, RECONNECT_WAIT_MS) + }, delayMs) } } @@ -194,8 +261,27 @@ export class APIRealtimeClient { } } - async identify(uid: string) { - return await this.sendMessage('identify', {uid}) + async resubscribe() { + if (this.subscriptions.size > 0) { + await this.sendMessage('subscribe', {topics: Array.from(this.subscriptions.keys())}) + } + } + + /** Tell the server who is signed in. Call on connect and whenever that changes. */ + refreshIdentity() { + this.identityQueue = this.identityQueue.then(() => this.sendIdentity()).catch(console.error) + return this.identityQueue + } + + private async sendIdentity() { + if (!this.getIdentity) return + const identity = await this.getIdentity() + if (!identity && !this.identified) return + await this.sendMessage('identify', identity ?? {uid: ''}) + this.identified = !!identity + // The server drops private topics it could not yet match to this connection, which includes + // anything subscribed while the token was still being verified, so send them again. + if (identity) await this.resubscribe() } async subscribe(topics: string[], handler: BroadcastHandler) { diff --git a/common/src/api/websockets.ts b/common/src/api/websockets.ts index 04e09dff0..f01e596fb 100644 --- a/common/src/api/websockets.ts +++ b/common/src/api/websockets.ts @@ -1,10 +1,18 @@ import {z} from 'zod' +// The server closes every connection after this long, so idle clients cannot hold a Cloud Run instance +// awake. The client uses the same figure to decide whether anyone is still there to reconnect for. +export const WS_MAX_SESSION_MS = 4 * 60 * 1000 + export const CLIENT_MESSAGE_SCHEMAS = { identify: z.object({ type: z.literal('identify'), txid: z.number(), + // Ignored by the server, which takes the identity from `token` alone. Still sent because an API + // revision without token support requires the field. uid: z.string(), + // A Firebase ID token. Leaving it out signs the connection out. + token: z.string().optional(), }), subscribe: z.object({ type: z.literal('subscribe'), diff --git a/common/src/discord/core.ts b/common/src/discord/core.ts index 3e6b9cb14..bbd9241ac 100644 --- a/common/src/discord/core.ts +++ b/common/src/discord/core.ts @@ -19,7 +19,9 @@ export const sendDiscordMessage = async (content: string, channel: string) => { const response = await fetch(webhookUrl!, { method: 'POST', headers: {'Content-Type': 'application/json'}, - body: JSON.stringify({content}), + // Messages quote text from visitors (the contact form, reports), which must not be able to ping + // @everyone or a role in the staff channels. + body: JSON.stringify({content, allowed_mentions: {parse: []}}), }) if (!response.ok) { diff --git a/common/src/hosting/constants.ts b/common/src/hosting/constants.ts index 05ac3b159..7888ac4e5 100644 --- a/common/src/hosting/constants.ts +++ b/common/src/hosting/constants.ts @@ -2,7 +2,9 @@ export const IS_WEBVIEW_DEV_PHONE = process.env.NEXT_PUBLIC_WEBVIEW_DEV_PHONE == export const IS_LOCAL_ANDROID = IS_WEBVIEW_DEV_PHONE || process.env.NEXT_PUBLIC_LOCAL_ANDROID === '1' export const IS_WEBVIEW = !!process.env.NEXT_PUBLIC_WEBVIEW -export const IS_GOOGLE_CLOUD = !!process.env.GOOGLE_CLOUD_PROJECT +// `K_SERVICE` is set by Cloud Run itself. GOOGLE_CLOUD_PROJECT comes from our own config, and losing it +// alone must not turn a deployment into "local", which skips the internal-route key check. +export const IS_GOOGLE_CLOUD = !!process.env.GOOGLE_CLOUD_PROJECT || !!process.env.K_SERVICE export const IS_VERCEL = !!process.env.NEXT_PUBLIC_VERCEL export const IS_DEPLOYED = IS_GOOGLE_CLOUD || IS_VERCEL || IS_WEBVIEW export const IS_LOCAL = !IS_DEPLOYED diff --git a/common/tests/unit/websocket-client.test.ts b/common/tests/unit/websocket-client.test.ts new file mode 100644 index 000000000..67dc8b5d4 --- /dev/null +++ b/common/tests/unit/websocket-client.test.ts @@ -0,0 +1,192 @@ +import {APIRealtimeClient, RealtimeIdentity} from 'common/api/websocket-client' + +class FakeSocket { + static OPEN = 1 + static CONNECTING = 0 + static CLOSING = 2 + static CLOSED = 3 + static last: FakeSocket + static count = 0 + readyState = 1 + sent: any[] = [] + onopen?: (ev: unknown) => void + onmessage?: (ev: {data: string}) => void + onclose?: (ev: unknown) => void + onerror?: (ev: unknown) => void + constructor() { + FakeSocket.last = this + FakeSocket.count++ + } + close() {} + send(raw: string) { + const msg = JSON.parse(raw) + this.sent.push(msg) + // Acked on a later tick, like a real server. + setTimeout(() => + this.onmessage?.({data: JSON.stringify({type: 'ack', txid: msg.txid, success: true})}), + ) + } +} + +const sentOfType = (type: string) => FakeSocket.last.sent.filter((m) => m.type === type) + +describe('APIRealtimeClient identity', () => { + let identity: RealtimeIdentity | undefined + + const open = async () => { + const client = new APIRealtimeClient('ws://test', async () => identity) + FakeSocket.last.onopen?.({}) + await client.identityQueue + return client + } + + beforeEach(() => { + ;(global as any).WebSocket = FakeSocket + ;(global as any).window = {setInterval: () => 0} + identity = {uid: 'alice', token: 'token-1'} + }) + + afterEach(() => { + delete (global as any).WebSocket + delete (global as any).window + }) + + it('identifies on connect, then subscribes again so private topics are not lost', async () => { + const client = await open() + await client.subscribe(['user/alice'], () => {}) + await client.refreshIdentity() + + expect(sentOfType('identify').at(-1)).toMatchObject({uid: 'alice', token: 'token-1'}) + const last = FakeSocket.last.sent.at(-1) + expect(last).toMatchObject({type: 'subscribe', topics: ['user/alice']}) + }) + + it('sends a tokenless identify on sign-out', async () => { + const client = await open() + identity = undefined + await client.refreshIdentity() + + const last = sentOfType('identify').at(-1) + expect(last.token).toBeUndefined() + }) + + it('stays quiet for a visitor who never signed in', async () => { + identity = undefined + await open() + expect(sentOfType('identify')).toEqual([]) + }) +}) + +describe('APIRealtimeClient reconnection', () => { + const SESSION_MS = 4 * 60 * 1000 + const ONE_MINUTE = 60 * 1000 + let listeners: Record void> + let client: APIRealtimeClient + + const serverCloses = (code = 1000) => { + FakeSocket.last.readyState = 3 + FakeSocket.last.onclose?.({code}) + } + const flushMicrotasks = async () => { + for (let i = 0; i < 10; i++) await Promise.resolve() + } + + beforeEach(() => { + jest.useFakeTimers() + listeners = {} + const addEventListener = (event: string, handler: () => void) => { + listeners[event] = handler + } + ;(global as any).WebSocket = FakeSocket + ;(global as any).window = {setInterval: () => 0, addEventListener} + ;(global as any).document = {visibilityState: 'visible', addEventListener} + FakeSocket.count = 0 + client = new APIRealtimeClient('ws://test') + FakeSocket.last.onopen?.({}) + }) + + afterEach(() => { + jest.useRealTimers() + delete (global as any).WebSocket + delete (global as any).window + delete (global as any).document + }) + + it('reconnects after the session limit for a tab in use', () => { + jest.advanceTimersByTime(ONE_MINUTE) + listeners.pointerdown() + jest.advanceTimersByTime(SESSION_MS - ONE_MINUTE) + serverCloses() + jest.advanceTimersByTime(5000) + expect(FakeSocket.count).toBe(2) + }) + + it('lets go of a visible tab nobody has touched', () => { + jest.advanceTimersByTime(SESSION_MS) + serverCloses() + jest.advanceTimersByTime(60 * ONE_MINUTE) + expect(FakeSocket.count).toBe(1) + }) + + it('stops within two sessions of the last activity', () => { + jest.advanceTimersByTime(SESSION_MS - 1000) + listeners.keydown() + jest.advanceTimersByTime(1000) + serverCloses() + jest.advanceTimersByTime(5000) + expect(FakeSocket.count).toBe(2) + + FakeSocket.last.onopen?.({}) + jest.advanceTimersByTime(SESSION_MS) + serverCloses() + jest.advanceTimersByTime(60 * ONE_MINUTE) + expect(FakeSocket.count).toBe(2) + }) + + it('comes back on the next activity', () => { + jest.advanceTimersByTime(SESSION_MS) + serverCloses() + listeners.pointerdown() + jest.advanceTimersByTime(5000) + expect(FakeSocket.count).toBe(2) + }) + + it('lets go of a hidden tab, and comes back when it is shown', () => { + ;(global as any).document.visibilityState = 'hidden' + serverCloses() + jest.advanceTimersByTime(60 * ONE_MINUTE) + expect(FakeSocket.count).toBe(1) + ;(global as any).document.visibilityState = 'visible' + listeners.visibilitychange() + jest.advanceTimersByTime(5000) + expect(FakeSocket.count).toBe(2) + }) + + it('does not retry an abnormal close for a tab nobody is using', () => { + jest.advanceTimersByTime(SESSION_MS) + serverCloses(1006) + jest.advanceTimersByTime(60 * ONE_MINUTE) + expect(FakeSocket.count).toBe(1) + }) + + it('stays closed after a manual close, whatever the activity', () => { + client.close() + serverCloses() + listeners.pointerdown() + jest.advanceTimersByTime(5000) + expect(FakeSocket.count).toBe(1) + }) + + it('calls onReconnect after a reconnect but not the first connect', async () => { + const handler = jest.fn() + client.onReconnect(handler) + await flushMicrotasks() + expect(handler).not.toBeCalled() + + serverCloses() + jest.advanceTimersByTime(5000) + FakeSocket.last.onopen?.({}) + await flushMicrotasks() + expect(handler).toBeCalledTimes(1) + }) +}) diff --git a/docs/database-schema.md b/docs/database-schema.md index cb6c84355..faf5b579a 100644 --- a/docs/database-schema.md +++ b/docs/database-schema.md @@ -594,10 +594,11 @@ anyone", in bulk, up to PostgREST's row cap. A `public read ... using (true)` po only right for data that is public by design (prompts, options, votes and their results, the canonical taxonomies). Every table that says something about a member — `profiles`, `users`, `compatibility_scores`, `user_activity`, `profile_stars`, `profile_likes`, `profile_ships`, -`bookmarked_searches`, `compatibility_answers`, `profile_comments`, every `private_*` table — has -had its `anon`/`authenticated` SELECT grant revoked (migrations +`bookmarked_searches`, `compatibility_answers`, `profile_comments`, `user_notifications`, every +`private_*` table — has had its `anon`/`authenticated` SELECT grant revoked (migrations `20260730_cap_profiles_users_reads.sql`, `20260731_lock_activity_stars_compat.sql`, -`20261028_lock_likes_ships.sql`, `20261028_lock_client_read_tables.sql`). The client reads them one +`20261028_lock_likes_ships.sql`, `20261028_lock_client_read_tables.sql`, +`20261105_lock_user_notifications.sql`). The client reads them one of two ways: - **A capped `security definer` RPC** (`get_display_users`, `get_profile_by_user_id`, diff --git a/docs/websocket-auth.md b/docs/websocket-auth.md new file mode 100644 index 000000000..3b63865b1 --- /dev/null +++ b/docs/websocket-auth.md @@ -0,0 +1,41 @@ +# WebSocket auth + +The API websocket carries a mix of public and private topics. `backend/shared/src/websockets/topics.ts` +decides who may listen to what, and refuses anything it does not list. + +| Topic | Who may subscribe | +| ------------------------------------ | ----------------- | +| `user//comment` | Anyone | +| `vote//comment` | Anyone | +| `user/` | `` only | +| `private-user/` | `` only | +| `user-notifications/` | `` only | +| `private-user-messages/` | `` only | +| `private-user-messages//typing` | `` only | + +A public topic must not carry anything that depends on the viewer. `user//comment` is an empty +"something changed" signal for that reason: subscribers refetch from `get-profile-comments`, which applies +the members-only, block and hidden rules. `vote//comment` still carries the comment, because +`vote_comments` can be read with the anon key anyway. + +## How a connection is identified + +The client sends `identify` with a Firebase ID token on every connect and on every sign-in or sign-out. +The server verifies the token and takes the uid from it; the `uid` field of the message is ignored. +An `identify` without a token signs the connection out. The token is only checked at `identify`, which is +safe because the server closes every connection after 4 minutes, so no identity outlives that. + +## Reconnecting + +An open socket keeps a Cloud Run instance awake, so the client only reconnects for a tab someone is using: +visible, with a click, key press, scroll or touch during the last session. Otherwise it stays disconnected +until the next activity or the tab becoming visible. A forgotten tab lets go within two sessions (8 +minutes) of the last activity, which is what lets the API scale to zero. Abnormal closes follow the same +rule. + +Nothing is replayed, so whatever was broadcast while disconnected is gone. A subscription that shows server +state should pass `onReconnect` to `useApiSubscription` and refetch there. Background tabs miss live updates +by design, including the unread count in the tab title, until they are looked at again. + +A refused topic is dropped and still acked as a success. Older clients have nothing handling a failed ack, +and the current client subscribes before its token is verified, then again once it is. diff --git a/web/components/comments/dropdown-menu.tsx b/web/components/comments/dropdown-menu.tsx index 8fc4defaa..bd4e1cbf8 100644 --- a/web/components/comments/dropdown-menu.tsx +++ b/web/components/comments/dropdown-menu.tsx @@ -49,6 +49,8 @@ export default function DropdownMenu(props: { const icon = props.icon ??