diff --git a/src/backend/services/events/appSockets.integration.test.ts b/src/backend/services/events/appSockets.integration.test.ts new file mode 100644 index 000000000..16c282fae --- /dev/null +++ b/src/backend/services/events/appSockets.integration.test.ts @@ -0,0 +1,307 @@ +/* + * Copyright (C) 2024-present Puter Technologies Inc. + * + * This file is part of Puter. + * + * Puter is free software: you can redistribute it and/or modify + * it under the terms of the GNU Affero General Public License as published + * by the Free Software Foundation, either version 3 of the License, or + * (at your option) any later version. + * + * This program is distributed in the hope that it will be useful, + * but WITHOUT ANY WARRANTY; without even the implied warranty of + * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the + * GNU Affero General Public License for more details. + * + * You should have received a copy of the GNU Affero General Public License + * along with this program. If not, see . + */ + +/** + * What an app connection can and cannot hear, over a real socket. + * + * The isolation claim is the whole point of letting an app hold one: it sees + * its own subscriptions' deliveries and nothing else — not the desktop's + * filesystem fan, not another app's deliveries. Only a live server can show + * that, so this drives real socket.io clients against real HTTP writes. + */ + +import { io as ioClient, type Socket as ClientSocket } from 'socket.io-client'; +import { v4 as uuidv4 } from 'uuid'; +import { afterAll, afterEach, beforeAll, describe, expect, it, vi } from 'vitest'; +import { setupPuterTestEnv, type PuterTestEnv } from '../../testUtil.js'; +import type { IConfig } from '../../types.js'; +import { + appSocketRoom, + SocketService, +} from '../socket/SocketService.js'; +import { + EVENTS_DELIVERY_CHANNEL, + EVENTS_SUBSCRIBE_VERB, + type DeliveryEnvelope, +} from './EventsService.js'; + +const BOOT_TIMEOUT_MS = 120_000; + +let env: PuterTestEnv; +let userId: number; +let username: string; +let sessionToken: string; + +/** Two apps, so "only its own" has something to be measured against. */ +let appOne: { uid: string; token: string }; +let appTwo: { uid: string; token: string }; + +const openSockets: ClientSocket[] = []; + +const socketService = () => + env.server.services.socket as unknown as SocketService; + +const makeApp = async ( + forUserId = userId, + forUsername = username, +): Promise<{ uid: string; token: string }> => { + const uid = `app-${uuidv4()}`; + await env.server.clients.db.write( + 'INSERT INTO `apps` (`uid`, `name`, `title`, `index_url`, `owner_user_id`) VALUES (?, ?, ?, ?, ?)', + [uid, uid, uid, `https://${uid}.example/`, forUserId], + ); + const user = await env.server.stores.user.getByUsername(forUsername); + const token = await env.server.services.auth.getUserAppToken( + { user: user as never, effectiveApp: null }, + uid, + ); + // An app's own data directory is the one place it can always list, which + // is what makes it the honest anchor for these. + await env.server.services.fs.mkdir(forUserId, { + path: `/${forUsername}/AppData/${uid}`, + createMissingParents: true, + }); + return { uid, token }; +}; + +const connect = ( + token: string, + opts: Record = {}, +): Promise => + new Promise((resolve, reject) => { + const socket = ioClient(env.origin, { + auth: { auth_token: token }, + transports: ['websocket'], + reconnection: false, + ...opts, + }); + openSockets.push(socket); + socket.on('connect', () => resolve(socket)); + socket.on('connect_error', (err: Error) => reject(err)); + }); + +const subscribe = ( + socket: ClientSocket, + subject: string, +): Promise<{ ok: boolean; sub?: { subId: string } }> => + new Promise((resolve) => + socket.emit(EVENTS_SUBSCRIBE_VERB, { subject }, resolve), + ); + +/** Every message a socket saw on one channel, in arrival order. */ +const collect = (socket: ClientSocket, channel: string): unknown[] => { + const seen: unknown[] = []; + socket.on(channel, (payload: unknown) => seen.push(payload)); + return seen; +}; + +const deliveries = (socket: ClientSocket): DeliveryEnvelope[] => + collect(socket, EVENTS_DELIVERY_CHANNEL) as DeliveryEnvelope[]; + +/** Create a directory the way a client does — the write that fans out. */ +const mkdirOverHttp = async (path: string): Promise => { + const response = await fetch(new URL('/mkdir', env.apiOrigin), { + method: 'POST', + headers: { + 'content-type': 'application/json', + authorization: `Bearer ${sessionToken}`, + }, + body: JSON.stringify({ path, create_missing_parents: true }), + }); + expect(response.status).toBe(200); +}; + +const settle = (seen: unknown[], count = 1) => + vi.waitFor(() => expect(seen.length).toBeGreaterThanOrEqual(count), { + timeout: 5_000, + interval: 25, + }); + +beforeAll(async () => { + env = await setupPuterTestEnv({ events: { enabled: true } } as IConfig); + username = env.users.user.username; + sessionToken = env.users.user.token; + const user = await env.server.stores.user.getByUsername(username); + userId = user!.id; + + appOne = await makeApp(); + appTwo = await makeApp(); +}, BOOT_TIMEOUT_MS); + +afterEach(async () => { + while (openSockets.length) openSockets.pop()?.disconnect(); + await vi.waitFor(() => { + for (const room of [ + String(userId), + appSocketRoom(userId, appOne.uid), + appSocketRoom(userId, appTwo.uid), + ]) + expect(socketService().has({ room })).toBe(false); + }); +}); + +afterAll(async () => { + await env?.shutdown(); +}); + +describe('an app connection', () => { + it('is admitted, and only into its own room', async () => { + const socket = await connect(appOne.token); + + expect(socket.connected).toBe(true); + expect( + socketService().has({ room: appSocketRoom(userId, appOne.uid) }), + ).toBe(true); + // The user room is what carries the desktop's filesystem fan. + expect(socketService().has({ room: userId })).toBe(false); + }); + + it('hears its own subscription and nothing the desktop hears', async () => { + const app = await connect(appOne.token); + const desktop = await connect(sessionToken); + const appDelivered = deliveries(app); + const appItems = collect(app, 'item.added'); + const appCache = collect(app, 'cache.updated'); + const desktopItems = collect(desktop, 'item.added'); + const desktopDelivered = deliveries(desktop); + + const anchor = `/${username}/AppData/${appOne.uid}`; + const ack = await subscribe(app, `fs:${anchor}`); + expect(ack.ok).toBe(true); + + const created = `${anchor}/reports-${uuidv4().slice(0, 8)}`; + await mkdirOverHttp(created); + await settle(appDelivered); + await settle(desktopItems); + + expect(appDelivered).toHaveLength(1); + expect(appDelivered[0].subId).toBe(ack.sub!.subId); + expect(appDelivered[0].event).toMatchObject({ + op: 'add', + path: created, + self: true, + }); + // The legacy channel still reaches the desktop, and only the desktop; + // the desktop, subscribed to nothing, gains nothing new. + expect(desktopItems.length).toBeGreaterThan(0); + expect(appItems).toEqual([]); + expect(appCache).toEqual([]); + expect(desktopDelivered).toEqual([]); + }); + + it('delivers only the projected shape', async () => { + const app = await connect(appOne.token); + const delivered = deliveries(app); + const anchor = `/${username}/AppData/${appOne.uid}`; + await subscribe(app, `fs:${anchor}`); + + await mkdirOverHttp(`${anchor}/shape-${uuidv4().slice(0, 8)}`); + await settle(delivered); + + expect(Object.keys(delivered[0]).sort()).toEqual(['event', 'subId']); + expect(Object.keys(delivered[0].event).sort()).toEqual([ + 'id', + 'op', + 'path', + 'self', + 'seq', + 'subject', + 'ts', + 'uid', + ]); + }); + + it('never sees another app`s deliveries', async () => { + const one = await connect(appOne.token); + const two = await connect(appTwo.token); + const oneDelivered = deliveries(one); + const twoDelivered = deliveries(two); + + const oneAnchor = `/${username}/AppData/${appOne.uid}`; + await subscribe(one, `fs:${oneAnchor}`); + await subscribe(two, `fs:/${username}/AppData/${appTwo.uid}`); + + await mkdirOverHttp(`${oneAnchor}/private-${uuidv4().slice(0, 8)}`); + await settle(oneDelivered); + + expect(oneDelivered).toHaveLength(1); + expect(twoDelivered).toEqual([]); + }); + + it('works from a standalone origin', async () => { + const app = await connect(appOne.token, { + extraHeaders: { Origin: 'https://standalone.example' }, + }); + const delivered = deliveries(app); + const anchor = `/${username}/AppData/${appOne.uid}`; + await subscribe(app, `fs:${anchor}`); + + await mkdirOverHttp(`${anchor}/standalone-${uuidv4().slice(0, 8)}`); + await settle(delivered); + + expect(delivered).toHaveLength(1); + }); + + it('spends the account`s per-origin allowance like any other socket', async () => { + const previous = SocketService.MAX_SOCKETS_PER_ORIGIN; + SocketService.MAX_SOCKETS_PER_ORIGIN = 1; + try { + const origin = { extraHeaders: { Origin: 'https://capped.example' } }; + const first = await connect(appOne.token, origin); + const second = await connect(appTwo.token, origin); + + await new Promise((resolve) => setTimeout(resolve, 500)); + expect(first.connected).toBe(true); + expect(second.connected).toBe(false); + } finally { + SocketService.MAX_SOCKETS_PER_ORIGIN = previous; + } + }); + + it('is dropped together with the desktop when the account`s sessions are revoked', async () => { + // A separate account: revoking sessions must not disturb the shared + // fixtures the rest of this file connects with. + const other = env.users.other; + const otherRow = await env.server.stores.user.getByUsername( + other.username, + ); + const otherApp = await makeApp(otherRow!.id, other.username); + + const desktop = await connect(other.token); + const app = await connect(otherApp.token); + expect(desktop.connected).toBe(true); + expect(app.connected).toBe(true); + + const authService = env.server.services.auth as unknown as { + revokeAllSessionsForUserId: (id: number) => Promise; + }; + await authService.revokeAllSessionsForUserId(otherRow!.id); + + // Eviction goes by account, not by which session minted the socket — + // the app connection is not in the desktop's room, so it has to be + // reached the same way the desktop is. + await vi.waitFor( + () => { + expect(desktop.connected).toBe(false); + expect(app.connected).toBe(false); + }, + { timeout: 5_000 }, + ); + }); +}); diff --git a/src/backend/services/socket/SocketService.test.ts b/src/backend/services/socket/SocketService.test.ts index 5bdbc492c..50695233e 100644 --- a/src/backend/services/socket/SocketService.test.ts +++ b/src/backend/services/socket/SocketService.test.ts @@ -18,6 +18,7 @@ */ import { io as ioClient, type Socket as ClientSocket } from 'socket.io-client'; +import { v4 as uuidv4 } from 'uuid'; import { afterAll, beforeAll, describe, expect, it, vi } from 'vitest'; import type { Actor } from '../../core/actor.js'; import type { PuterServer } from '../../server.js'; @@ -29,8 +30,10 @@ import { } from '../../testUtil.js'; import type { AuthResult } from '../auth/AuthService.js'; import { + accountSocketRoom, buildSocketReauthError, decideSocketAuth, + socketRoomsFor, SocketService, type SocketReauthError, } from './SocketService.js'; @@ -106,6 +109,13 @@ describe('decideSocketAuth', () => { expect((decision.reject as { data?: unknown }).data).toBeUndefined(); }); + it('admits an app-under-user actor where app sockets are allowed', () => { + const decision = decideSocketAuth({ actor: appActor } as AuthResult, { + allowAppActors: true, + }); + expect(decision).toEqual({ accept: appActor }); + }); + it('rejects an access-token actor with a specific message', () => { const decision = decideSocketAuth({ actor: accessTokenActor, @@ -114,6 +124,38 @@ describe('decideSocketAuth', () => { expect(decision.reject.message).toMatch(/only user tokens/); }); + it('keeps rejecting an access-token actor where app sockets are allowed', () => { + // Including one an app issued: the allowance is for app-under-user + // credentials, not for anything that resolves to an app. + const issuedByApp: Actor = { + ...accessTokenActor, + accessToken: { + uid: 'tok-2', + issuer: appActor, + authorized: null, + }, + effectiveApp: appActor.app, + }; + for (const actor of [accessTokenActor, issuedByApp]) { + const decision = decideSocketAuth({ actor } as AuthResult, { + allowAppActors: true, + }); + if (!('reject' in decision)) throw new Error('expected reject'); + expect(decision.reject.message).toMatch(/only user tokens/); + } + }); + + it('applies the account gates to an admitted app actor', () => { + const decision = decideSocketAuth( + { + actor: { ...appActor, user: { ...appActor.user, suspended: 1 } }, + } as unknown as AuthResult, + { allowAppActors: true }, + ); + if (!('reject' in decision)) throw new Error('expected reject'); + expect(decision.reject.message).toMatch(/suspended/i); + }); + it('rejects a suspended user — the HTTP gate the handshake never runs', () => { const decision = decideSocketAuth({ actor: { @@ -161,6 +203,25 @@ describe('decideSocketAuth', () => { }); }); +// -- socketRoomsFor ---------------------------------------------------- + +describe('socketRoomsFor', () => { + it('puts a session in the user room', () => { + expect( + socketRoomsFor({ user: { id: 7, uuid: 'u-7', username: 'u' } }), + ).toEqual(['7', accountSocketRoom(7)]); + }); + + it('keeps an app out of the user room and in its own', () => { + const rooms = socketRoomsFor({ + user: { id: 7, uuid: 'u-7', username: 'u' }, + app: { uid: 'app-1' }, + }); + expect(rooms).toEqual(['u7:aapp-1', accountSocketRoom(7)]); + expect(rooms).not.toContain('7'); + }); +}); + // -- Live socket.io integration -------------------------------------- describe('SocketService (live socket.io)', () => { @@ -234,6 +295,23 @@ describe('SocketService (live socket.io)', () => { ); }); + it('rejects an app-under-user token while events are off', async () => { + const row = await server.stores.user.getByUsername(user.username); + const appUid = `app-${uuidv4()}`; + await server.clients.db.write( + 'INSERT INTO `apps` (`uid`, `name`, `title`, `index_url`, `owner_user_id`) VALUES (?, ?, ?, ?, ?)', + [appUid, appUid, appUid, `https://${appUid}.example/`, row!.id], + ); + const appToken = await server.services.auth.getUserAppToken( + { user: row as never, effectiveApp: null }, + appUid, + ); + + await expect(connect({ auth_token: appToken })).rejects.toThrow( + /only user tokens/, + ); + }); + it('accepts a user session token, joins the per-user room, and announces the connect', async () => { const row = await server.stores.user.getByUsername(user.username); const userId = row!.id; diff --git a/src/backend/services/socket/SocketService.ts b/src/backend/services/socket/SocketService.ts index 0ebcb7eb8..b61ade1f2 100644 --- a/src/backend/services/socket/SocketService.ts +++ b/src/backend/services/socket/SocketService.ts @@ -60,6 +60,15 @@ export const buildSocketReauthError = (reauth: { /** Pure decision from an `AuthResult` to a socket-side accept/reject. */ export type SocketAuthDecision = { accept: Actor } | { reject: Error }; +export interface SocketAuthOptions { + /** + * Whether an app-under-user actor may hold a connection. Off, the handshake + * is exactly what it has always been. On, an app socket is admitted — into + * its own room and nothing else (see `socketRoomsFor`). + */ + allowAppActors?: boolean; +} + /** * Map an `AuthService.authenticate()` result onto the socket-handshake verdict. * Order matters: @@ -67,8 +76,9 @@ export type SocketAuthDecision = { accept: Actor } | { reject: Error }; * 1. `reauth` → structured `reauth_required` error so the client can drive the * same migration / re-login flow it does for HTTP. * 2. Missing actor → generic `socket auth failed`. - * 3. App-under-user / access-token actor → rejected with a specific message; - * sockets only accept plain user actors. + * 3. Access-token actor → rejected, always. App-under-user actor → rejected unless + * `allowAppActors`, since a subscription feed is the only thing an app + * connection is for. * 4. Suspended, or pending a verification → rejected. A socket carries the same * filesystem entries, upload paths and notification bodies as the HTTP * routes, which get these two from `requireAuthGate` / @@ -78,7 +88,10 @@ export type SocketAuthDecision = { accept: Actor } | { reject: Error }; * * Pure / no side effects — the middleware logs the reauth event. */ -export const decideSocketAuth = (result: AuthResult): SocketAuthDecision => { +export const decideSocketAuth = ( + result: AuthResult, + options: SocketAuthOptions = {}, +): SocketAuthDecision => { if (result.reauth) { return { reject: buildSocketReauthError(result.reauth) }; } @@ -86,7 +99,10 @@ export const decideSocketAuth = (result: AuthResult): SocketAuthDecision => { if (!actor || !actor.user) { return { reject: new Error('socket auth failed') }; } - if (isAppActor(actor) || isAccessTokenActor(actor)) { + if ( + isAccessTokenActor(actor) || + (isAppActor(actor) && !options.allowAppActors) + ) { return { reject: new Error('socket auth: only user tokens accepted') }; } try { @@ -111,6 +127,34 @@ export interface SocketSpecifier { socket?: string; } +/** The room an app-under-user socket receives its own deliveries in. */ +export const appSocketRoom = ( + userId: number | string, + appUid: string, +): string => `u${userId}:a${appUid}`; + +/** + * A room every one of an account's sockets joins and nothing is ever emitted + * to. The user room is what a session revoke drops, and an app socket is + * deliberately not in it — this is the handle that reaches those. + */ +export const accountSocketRoom = (userId: number | string): string => + `u${userId}:all`; + +/** + * Which rooms a socket joins. An app socket gets its own per-(user, app) room + * and never the user room, which carries the whole `outer.gui.*` fan and is the + * reason app actors were refused outright. + */ +export const socketRoomsFor = (actor: Actor): string[] => { + const userId = String(actor.user!.id); + const appUid = isAppActor(actor) ? actor.app?.uid : undefined; + return [ + appUid ? appSocketRoom(userId, appUid) : userId, + accountSocketRoom(userId), + ]; +}; + // -- Redis key format for cross-node FS-cache invalidation ---------- // // puter-js (browser) polls `GET /cache/last-change-timestamp` and purges @@ -161,9 +205,9 @@ interface AuthenticatedSocket extends Socket { * Socket.io wrapper with: * * 1. Auth middleware — reads `handshake.auth.auth_token`, validates it via - * `AuthService`, rejects anything other than plain user actors (no - * app-under-user, no access-token), and joins the socket to a per-user room - * keyed by `user.id`. + * `AuthService`, rejects access-token actors (and app-under-user actors + * unless events are enabled), and joins the socket to its room: the per-user + * room keyed by `user.id` for a session, a per-(user, app) room for an app. * 2. Event bus → socket fan-out — subscribes to the known set of `outer.gui.*` * mutation events and pushes each to the affected users' rooms. Strips the * `outer.gui.` prefix before emitting. @@ -314,6 +358,14 @@ export class SocketService extends PuterService { // -- Auth + connection wiring ----------------------------------- + /** + * An app connection exists to carry event subscriptions, so it is admitted + * only where those are switched on. + */ + #authOptions(): SocketAuthOptions { + return { allowAppActors: this.config.events?.enabled === true }; + } + #installAuthMiddleware(): void { if (!this.#io) return; const authService = this.services.auth as AuthService | undefined; @@ -369,7 +421,7 @@ export class SocketService extends PuterService { ); } - const decision = decideSocketAuth(result); + const decision = decideSocketAuth(result, this.#authOptions()); if ('reject' in decision) { next(decision.reject); return; @@ -379,7 +431,7 @@ export class SocketService extends PuterService { socket.authToken = token; // user.id is numeric in the DB; stringify for room name // so adapter lookups key on a stable type. - socket.join(String(decision.accept.user!.id)); + socket.join(socketRoomsFor(decision.accept)); next(); } catch (err) { console.warn('[socket] auth error', err); @@ -424,13 +476,12 @@ export class SocketService extends PuterService { /** * Simultaneous connections per (user, origin). * - * The natural split would be per app, but there isn't one to key on: - * `decideSocketAuth` accepts only plain user actors, so an app-token actor - * never reaches this code and every socket here belongs to a session. The - * requesting origin is the next-best proxy — it separates our own pages - * from a third-party site embedding the SDK against the same session, which - * is the split that matters. Without it a single looping page consumes the - * account's whole allowance and takes every other window offline with it. + * The natural split would be per app, but most sockets have no app to key + * on — a session carries none, and an app connection is the minority case. + * The requesting origin covers both: it separates our own pages from a + * third-party site embedding the SDK against the same session, and an app + * connects from its own origin. Without it a single looping page consumes + * the account's whole allowance and takes every other window offline. * * A browser sets `Origin` itself, so a page can't lie about its own; a * non-browser client can put anything there, which is exactly why the @@ -563,6 +614,7 @@ export class SocketService extends PuterService { try { decision = decideSocketAuth( await authService.authenticate(token, {}), + this.#authOptions(), ); } catch { decision = { reject: new Error('socket reauth failed') }; @@ -594,8 +646,9 @@ export class SocketService extends PuterService { * Cluster-wide: `disconnectSockets` publishes through the adapter, so a * revoke handled on one node reaches sockets terminated on another. * - * Every connection for the account goes, not just the revoked session's. - * Narrowing would mean matching each socket to its session via + * Every connection for the account goes, not just the revoked session's — + * which is what the account room is for, since an app socket is not in the + * user room. Narrowing would mean matching each socket to its session via * `fetchSockets`, which the adapter implements on top of `serverCount()` — * and that path is unavailable with our Redis client. Dropping the room is * the safe direction: a connection whose session survived reconnects on its @@ -604,7 +657,7 @@ export class SocketService extends PuterService { async #evictUserSockets(userId: number): Promise { const io = this.#io; if (!io || !userId) return; - await io.in(String(userId)).disconnectSockets(true); + await io.in(accountSocketRoom(userId)).disconnectSockets(true); } async #allowSocketEvent(userId: number, event: string): Promise { @@ -637,6 +690,12 @@ export class SocketService extends PuterService { // answer with `events_disabled` rather than going unanswered. this.services.events.attachSocket(socket, actor); + // Everything below is the desktop session's own traffic: two verbs + // that reach the user's other tabs, and a connect announcement + // whose listeners read it as "the UI is up". An app connection is + // none of those things. + if (isAppActor(actor)) return; + // Peer-echo: one tab notifies others that trash is empty. socket.on('trash.is_empty', (msg: unknown) => { void this.#allowSocketEvent(userId, 'trash.is_empty').then( diff --git a/src/puter-js/src/modules/FileSystem/index.js b/src/puter-js/src/modules/FileSystem/index.js index d2b613473..8c17e11a4 100644 --- a/src/puter-js/src/modules/FileSystem/index.js +++ b/src/puter-js/src/modules/FileSystem/index.js @@ -130,13 +130,6 @@ export class PuterJSFileSystemModule extends PuterModule { } bindSocketEvents () { - // this.socket.on('cache.updated', (msg) => { - // // check original_client_socket_id and if it matches this.socket.id, don't post update - // if (msg.original_client_socket_id !== this.socket.id) { - // this.invalidateCache(); - // } - // }); - this.socket.on('item.renamed', (item) => { puter._cache.flushall(); });