From 66dc0a63969463a2cb263dc6eb52d57058a9080b Mon Sep 17 00:00:00 2001 From: Daniel Salazar Date: Wed, 2 Sep 2026 13:08:23 -0700 Subject: [PATCH] feat: subscribe ACL and app scoping for event subscriptions (PUT-1672) (#3677) --- .../services/events/EventsService.test.ts | 150 ++++++-- src/backend/services/events/EventsService.ts | 205 +++++++---- .../events/authorization.integration.test.ts | 246 +++++++++++++ src/backend/services/events/authorization.ts | 189 ++++++++-- src/backend/services/events/registry.test.ts | 5 +- src/backend/services/events/registry.ts | 6 +- .../events/scoping.integration.test.ts | 315 +++++++++++++++++ .../services/events/subscriptionCache.ts | 7 +- src/backend/services/fs/FSService.ts | 5 +- .../events/EventSubscriptionStore.test.ts | 109 +++++- .../stores/events/EventSubscriptionStore.ts | 328 ++++++++++++------ 11 files changed, 1314 insertions(+), 251 deletions(-) create mode 100644 src/backend/services/events/authorization.integration.test.ts create mode 100644 src/backend/services/events/scoping.integration.test.ts diff --git a/src/backend/services/events/EventsService.test.ts b/src/backend/services/events/EventsService.test.ts index 3a217772c..62ddcb2e8 100644 --- a/src/backend/services/events/EventsService.test.ts +++ b/src/backend/services/events/EventsService.test.ts @@ -140,6 +140,38 @@ const actorFor = (id = userId): Actor => effectiveApp: null, }) as unknown as Actor; +/** + * Paths the ACL says no to, and how loudly. The real check is exercised against + * real grants in the integration suite; here access is data, so a test can say + * "this went away" without staging a share. + */ +let denied: Map; + +const aclService = () => ({ + check: async (_actor: Actor, resource: { path: string }) => + !denied.has(resource.path), + getSafeAclError: async (_actor: Actor, resource: { path: string }) => + denied.get(resource.path) === 'forbidden' + ? { status: 403, message: 'Forbidden', fields: { code: 'forbidden' } } + : { + status: 404, + message: 'Subject does not exist', + fields: { code: 'subject_does_not_exist' }, + }, +}); + +const userStore = { + getById: async (id: number) => ({ + id, + uuid: `user-${id}`, + username: `u${id}`, + }), +}; + +const appStore = { + getByUid: async (uid: string) => ({ uid, id: 1 }), +}; + /** * Each service gets its own outbox. A delivery still in flight when a test * ends must land in that test's record, not in the next one's. @@ -161,7 +193,12 @@ const buildService = ( redis, event: bus, } as never, - { eventSubscription: store, fsEntry: fsEntryStore } as never, + { + eventSubscription: store, + fsEntry: fsEntryStore, + user: userStore, + app: appStore, + } as never, { socket: { send: vi.fn(async (spec: { socket?: string }, _key, data) => { @@ -176,6 +213,7 @@ const buildService = ( ancestorChain(path), ), }, + acl: aclService(), } as never, ); built.onDelivered = (envelope) => counted.push(envelope); @@ -233,16 +271,23 @@ const subscribe = async (subject: string, socket = socketId) => */ const seedSubscriptions = async ( count: number, - row: { token: string; anchorUid: string; anchorPath: string; match: string | null }, + row: { + token: string; + anchorUid: string; + anchorPath: string; + match: string | null; + }, ): Promise => { for (let i = 0; i < count; i++) await store.add({ subId: `seed-${seq}-${i}`, socketId: `socket-${seq}-${i}`, - userId, + holderUserId: userId, + ownerUserId: userId, subject: 'fs:seeded', op: null, appUid: null, + permission: 'list', ...row, }); }; @@ -251,7 +296,7 @@ const seedSubscriptions = async ( const dispatch = async (node: FSEntry, key = 'fs.write.file' as const) => service.dispatchFs(key, node, { actingUserId: userId, - ancestorUids: async () => ancestorChain(node.path).map((a) => a.uid), + ancestors: async () => ancestorChain(node.path), }); beforeEach(() => { @@ -260,6 +305,7 @@ beforeEach(() => { socketId = `socket-${seq}`; commands = []; entries = new Map(); + denied = new Map(); redis = countingRedis(new MockRedis.Cluster(['redis://localhost:7001'])); store = new EventSubscriptionStore( {} as IConfig, @@ -302,7 +348,7 @@ describe('the feature switch', () => { const { file } = seedTree(); await off.dispatchFs('fs.write.file', file, { - ancestorUids: async () => { + ancestors: async () => { throw new Error('the tree must not be walked'); }, }); @@ -335,13 +381,11 @@ describe('subscribing', () => { ); }); - it('answers a node someone else owns as absent, not as refused', async () => { + it('answers a node the caller cannot see as absent, not as refused', async () => { const { documents } = seedTree(); - register({ ...documents, userId: userId + 500 } as FSEntry); + denied.set(documents.path, 'hidden'); - await expect( - subscribe(`fs:${documents.uid}`), - ).rejects.toSatisfy( + await expect(subscribe(`fs:${documents.uid}`)).rejects.toSatisfy( (err: unknown) => isHttpError(err) && err.statusCode === 404 && @@ -349,6 +393,29 @@ describe('subscribing', () => { ); }); + it('refuses a node the caller can see but not list', async () => { + const { documents } = seedTree(); + denied.set(documents.path, 'forbidden'); + + await expect(subscribe(`fs:${documents.uid}`)).rejects.toSatisfy( + (err: unknown) => + isHttpError(err) && + err.statusCode === 403 && + err.legacyCode === 'forbidden', + ); + }); + + it('stores the anchor`s owner, not the subscriber, as the keyspace', async () => { + const { documents } = seedTree(); + const owner = userId + 500; + register({ ...documents, userId: owner } as FSEntry); + + await subscribe(`fs:${documents.uid}`); + + await expect(store.userHasAny(owner)).resolves.toBe(true); + await expect(store.userHasAny(userId)).resolves.toBe(false); + }); + it('files the missing remainder as the filter', async () => { const { documents } = seedTree(); @@ -431,13 +498,13 @@ describe('what a dispatch costs', () => { it('walks the tree only for a user who has subscriptions', async () => { const { file } = seedTree(); - const walk = vi.fn(async () => ['docs']); + const walk = vi.fn(async () => [{ uid: 'docs', path: '/docs' }]); await service.dispatchFs('fs.write.file', file, { - ancestorUids: walk, + ancestors: walk, }); await service.dispatchFs('fs.write.file', file, { - ancestorUids: walk, + ancestors: walk, }); expect(walk).not.toHaveBeenCalled(); @@ -468,7 +535,8 @@ describe('cross-process invalidation', () => { await store.add({ subId: 'remote-sub', socketId: 'remote-socket', - userId, + holderUserId: userId, + ownerUserId: userId, subject: `fs:${documents.uid}`, token: `f#${documents.uid}`, anchorUid: documents.uid, @@ -476,6 +544,7 @@ describe('cross-process invalidation', () => { match: null, op: null, appUid: null, + permission: 'list', }); const generation = await store.getGeneration(userId); @@ -587,42 +656,63 @@ describe('matching', () => { await service.dispatchFs('fs.write.file', file, { actingUserId: userId + 900, - ancestorUids: async () => [documents.uid], + ancestors: async () => [ + { uid: documents.uid, path: documents.path }, + ], }); await flush(); expect(sent[0].envelope.event).toMatchObject({ self: false }); }); - it('will not deliver a row whose holder does not own the node', async () => { + it('stops delivering the moment the holder`s access goes', async () => { const { documents, file } = seedTree(); - const sub = await subscribe(`fs:${documents.uid}`); - // Holder and owner cannot disagree through the API yet, so the stored - // row is where they are made to — which is what the delivery-side - // check is for once a shared anchor can produce that pair. - const rowKey = `ev:t:{${userId}}:f#${documents.uid}`; - const stored = JSON.parse( - (await redis.hget(rowKey, sub.subId)) as string, - ) as Record; - await redis.hset( - rowKey, - sub.subId, - JSON.stringify({ ...stored, userId: userId + 700 }), - ); + await subscribe(`fs:${documents.uid}`); + // The row is still there and still matches; only the answer to "may + // this holder list the node" changed. + denied.set(file.path, 'hidden'); await dispatch(file); await flush(); expect(sent).toEqual([]); }); + it('re-checks the node the event is about, not the anchor', async () => { + const { documents } = seedTree(); + await subscribe(`fs:/u${userId}/Documents/**`); + const reachable = register( + entry({ + uid: `open-${seq}`, + path: `${documents.path}/open/notes.txt`, + }), + ); + const closed = register( + entry({ + uid: `closed-${seq}`, + path: `${documents.path}/closed/secret.txt`, + }), + ); + denied.set(closed.path, 'hidden'); + + await dispatch(reachable); + await dispatch(closed); + await flush(); + + expect(sent.map((s) => (s.envelope.event as { uid: string }).uid)).toEqual([ + reachable.uid, + ]); + }); + it('publishes nothing for an event with no registry entry', async () => { const { documents, file } = seedTree(); await subscribe(`fs:${documents.uid}`); commands = []; await service.dispatchFs('fs.copy.node', file, { - ancestorUids: async () => [documents.uid], + ancestors: async () => [ + { uid: documents.uid, path: documents.path }, + ], }); await flush(); diff --git a/src/backend/services/events/EventsService.ts b/src/backend/services/events/EventsService.ts index 473fba5d9..ec1d474de 100644 --- a/src/backend/services/events/EventsService.ts +++ b/src/backend/services/events/EventsService.ts @@ -30,15 +30,20 @@ import { HttpError } from '../../core/http/HttpError.js'; import { checkRateLimit } from '../../core/http/middleware/rateLimit.js'; import { SESSION_SUBSCRIPTION_TTL_SECONDS, + type GenerationBump, type SessionSubscription, } from '../../stores/events/EventSubscriptionStore.js'; import type { FSEntry } from '../../stores/fs/FSEntry.js'; +import type { ResourceDescriptor } from '../acl/ACLService.js'; import { resolveNode } from '../fs/resolveNode.js'; import { PuterService } from '../types.js'; import { resolveFsAnchor, type FsAnchorDeps } from './anchors.js'; import { assertSubscribeAuthorized, - checkSubscribeAuthorization, + checkDeliveryAuthorized, + nodeDescriptor, + rowInActorScope, + type EventAclDeps, } from './authorization.js'; import { DeliveryCoalescer } from './coalescer.js'; import { @@ -70,7 +75,11 @@ import { parseSubject, type FsOp } from './subjects.js'; * rather than a timer, so it costs nothing once warm. * 3. One Redis command against the user's watched tokens. * - * Only past all three does anything read a subscription or walk the tree. + * Only past all three does anything read a subscription or walk the tree, and + * only a row that survived op and filter matching pays for the ACL re-check + * that decides whether its holder may still be told. Everything is keyed by the + * owner of the resource, because that is all a write knows about itself. + * * Nothing here reaches the caller: an event that could not be dispatched is an * event nobody hears, and the write that produced it still succeeded. */ @@ -134,8 +143,10 @@ export interface FsDispatchOptions { /** * Resolved lazily and only past the second gate — walking the tree for a * user nobody subscribed on behalf of is the cost this exists to avoid. + * Paths as well as uids: the same walk answers the token lookup and the + * per-delivery ACL re-check. */ - ancestorUids?: () => Promise; + ancestors?: () => Promise>; /** Who performed the write, for the `self` flag. */ actingUserId?: number; } @@ -295,10 +306,10 @@ export class EventsService extends PuterService { request: SubscribeRequest, ): Promise<{ sub: SubscriptionView }> { if (!this.enabled) throw disabled(); - const userId = actor.user?.id; - if (userId === undefined) throw disabled(); + const holderUserId = actor.user?.id; + if (holderUserId === undefined) throw disabled(); - await this.#spendCallBudget(userId); + await this.#spendCallBudget(holderUserId); const rawSubject = String(request?.subject ?? ''); const parsed = parseSubject(rawSubject); @@ -314,7 +325,9 @@ export class EventsService extends PuterService { }); // The resolver answers where a subscription keys, not whose it is, so - // the owner comes from the anchor node itself. + // the owner comes from the anchor node itself — and that is the + // keyspace the row is indexed in, because dispatch only ever knows + // whose resource changed. const entry = await resolveNode(this.stores.fsEntry, { uid: anchor.uid, }); @@ -322,10 +335,11 @@ export class EventsService extends PuterService { throw new HttpError(404, `No such entry: ${anchor.path}`, { legacyCode: 'subject_does_not_exist', }); - assertSubscribeAuthorized( - { userId }, - { ownerUserId: entry.userId }, + const permission = await assertSubscribeAuthorized( + actor, + { uid: anchor.uid, path: anchor.path }, rawSubject, + this.#aclDeps(), ); // Compile now so an unusable pattern fails this call rather than every @@ -335,7 +349,8 @@ export class EventsService extends PuterService { const sub: SessionSubscription = { subId: randomUUID(), socketId, - userId, + holderUserId, + ownerUserId: entry.userId, subject: rawSubject, token: anchor.token, anchorUid: anchor.uid, @@ -343,11 +358,12 @@ export class EventsService extends PuterService { match: anchor.match, op: anchor.op, appUid: actor.effectiveApp?.uid ?? null, + permission, }; - const generation = await this.stores.eventSubscription.add(sub); - this.#publishGeneration(userId, generation); - this.#startRefresh(userId, socketId); + const bump = await this.stores.eventSubscription.add(sub); + this.#publishGeneration(bump); + this.#startRefresh(holderUserId, socketId); return { sub: toView(sub) }; } @@ -358,46 +374,62 @@ export class EventsService extends PuterService { request: UnsubscribeRequest, ): Promise { if (!this.enabled) throw disabled(); - const userId = actor.user?.id; - if (userId === undefined) throw disabled(); + const holderUserId = actor.user?.id; + if (holderUserId === undefined) throw disabled(); - await this.#spendCallBudget(userId); + await this.#spendCallBudget(holderUserId); const subId = String(request?.subId ?? ''); if (!subId) throw unknownSubscription(); - const generation = await this.stores.eventSubscription.remove( - userId, + const sub = await this.stores.eventSubscription.getForSocket( + holderUserId, socketId, subId, ); - // An id this socket never held reads as absent rather than refused — - // otherwise unsubscribe reports whether someone else's id exists. - if (generation === null) throw unknownSubscription(); + // An id this socket never held — or one another app created — reads as + // absent rather than refused: a 403 here is an oracle for subIds. + if (!sub || !rowInActorScope(actor, sub)) throw unknownSubscription(); + const bump = await this.stores.eventSubscription.remove(sub); this.#forget(subId); - this.#publishGeneration(userId, generation); + this.#publishGeneration(bump); + } + + /** What this actor holds on one connection, scoped to what it may see. */ + async listSubscriptions( + actor: Actor, + socketId: string, + ): Promise { + if (!this.enabled) throw disabled(); + const holderUserId = actor.user?.id; + if (holderUserId === undefined) throw disabled(); + + const held = await this.stores.eventSubscription.listForSocket( + holderUserId, + socketId, + ); + return held.filter((sub) => rowInActorScope(actor, sub)).map(toView); } /** Disconnect handler; also covers a socket the server dropped. */ - async reapSocket(userId: number, socketId: string): Promise { + async reapSocket(holderUserId: number, socketId: string): Promise { // Nothing could have been registered, so nothing has to be looked up. // A switch flipped off under live subscriptions leaves them to the TTL. if (!this.enabled) return; - this.#stopRefresh(userId, socketId); + this.#stopRefresh(holderUserId, socketId); try { const held = await this.stores.eventSubscription.listForSocket( - userId, + holderUserId, socketId, ); for (const sub of held) this.#forget(sub.subId); - const generation = await this.stores.eventSubscription.reapSocket( - userId, + const bumps = await this.stores.eventSubscription.reapSocket( + holderUserId, socketId, ); - if (generation !== null) - this.#publishGeneration(userId, generation); + for (const bump of bumps) this.#publishGeneration(bump); } catch (err) { // The TTL backstop exists for exactly this. console.warn('[events] failed to reap socket subscriptions', err); @@ -420,43 +452,41 @@ export class EventsService extends PuterService { const subject = lookupPublicSubject(key); if (!subject) return; - const userId = entry?.userId; - if (typeof userId !== 'number') return; + const ownerUserId = entry?.userId; + if (typeof ownerUserId !== 'number') return; - if (!(await this.#userHasAny(userId))) return; + if (!(await this.#userHasAny(ownerUserId))) return; - const ancestorUids = options.ancestorUids - ? await options.ancestorUids() - : []; + const ancestors = options.ancestors ? await options.ancestors() : []; const context: EventContext = { key, entry, - ancestorUids, + ancestors, id: randomUUID(), ts: Date.now(), }; const watched = await this.stores.eventSubscription.watchedTokens( - userId, + ownerUserId, subject.tokens(context), ); if (watched.length === 0) return; const rows = await this.stores.eventSubscription.getForTokens( - userId, + ownerUserId, watched, ); if (rows.length === 0) return; - this.#route(subject, context, rows, options.actingUserId); + await this.#route(subject, context, rows, options.actingUserId); } - #route( + async #route( subject: PublicSubject, context: EventContext, rows: SessionSubscription[], actingUserId: number | undefined, - ): void { + ): Promise { // One throwaway projection reads the op off the registry entry rather // than re-deriving it from the subject string. const { op } = subject.project({ ...context, self: false, seq: 0 }); @@ -464,20 +494,22 @@ export class EventsService extends PuterService { const evaluated = evaluateWithCap( rows, - (row) => this.#passes(row, op, matchOn, context), + (row) => this.#passes(row, op, matchOn), FILTER_EVALUATIONS_PER_EVENT, ); - const matched = evaluated.matched.slice( - 0, - EVENTS_MATCHED_SUBSCRIPTIONS_PER_EVENT, + const matched = await this.#stillAuthorized( + evaluated.matched.slice(0, EVENTS_MATCHED_SUBSCRIPTIONS_PER_EVENT), + context, ); let seq = 0; for (const row of matched) { const event = subject.project({ ...context, - self: actingUserId === undefined || actingUserId === row.userId, + self: + actingUserId === undefined || + actingUserId === row.holderUserId, seq: seq++, }); this.#coalesce().push(coalesceKey(row.subId, event.subject), { @@ -508,20 +540,8 @@ export class EventsService extends PuterService { } /** Op filter first — a comparison, where the glob is not. */ - #passes( - row: SessionSubscription, - op: FsOp, - matchOn: string, - context: EventContext, - ): boolean { + #passes(row: SessionSubscription, op: FsOp, matchOn: string): boolean { if (row.op !== null && row.op !== op) return false; - if ( - !checkSubscribeAuthorization( - { userId: row.userId }, - { ownerUserId: context.entry.userId }, - ) - ) - return false; if (!row.match) return true; const relative = relativeTo(row.anchorPath, matchOn); @@ -529,6 +549,50 @@ export class EventsService extends PuterService { return this.#matcherFor(row).test(relative); } + /** + * Re-run each surviving row's access against the node the event is about, + * not against its anchor: a filter that reaches into something the holder + * cannot list must not deliver from it, and a share revoked after the + * subscription was made must stop delivering at once rather than when the + * row is next touched. + * + * Last of the filters, because it is the only one that can cost a lookup — + * and rows that share an identity and a grant share one decision. + */ + async #stillAuthorized( + rows: SessionSubscription[], + context: EventContext, + ): Promise { + if (rows.length === 0) return rows; + + const node = this.#eventDescriptor(context); + const decisions = new Map>(); + const allowed = await Promise.all( + rows.map((row) => { + const key = `${row.holderUserId}|${row.appUid ?? ''}|${row.permission}`; + let decision = decisions.get(key); + if (!decision) { + decision = checkDeliveryAuthorized( + row, + node, + this.#aclDeps(), + ); + decisions.set(key, decision); + } + return decision; + }), + ); + return rows.filter((_row, i) => allowed[i]); + } + + /** The event's node as ACL wants it, reusing the walk dispatch already did. */ + #eventDescriptor(context: EventContext): ResourceDescriptor { + return nodeDescriptor( + { uid: context.entry.uid, path: context.entry.path }, + { getAncestorChain: async () => context.ancestors }, + ); + } + #matcherFor(row: SessionSubscription): CompiledMatch { const cached = this.#compiled.get(row.subId); if (cached && cached.pattern === row.match) return cached; @@ -654,7 +718,7 @@ export class EventsService extends PuterService { } } - #publishGeneration(userId: number, generation: number): void { + #publishGeneration({ userId, generation }: GenerationBump): void { this.#cache.bump(userId, generation); try { this.clients.event.emit( @@ -681,6 +745,15 @@ export class EventsService extends PuterService { }; } + #aclDeps(): EventAclDeps { + return { + acl: this.services.acl, + getAncestorChain: (path) => this.services.fs.getAncestorChain(path), + getUser: (userId) => this.stores.user.getById(userId), + getApp: (uid) => this.stores.app.getByUid(uid), + }; + } + async #spendCallBudget(userId: number): Promise { const ok = await checkRateLimit( `${EVENTS_SUBSCRIBE_LIMIT.scope}:${userId}`, @@ -695,13 +768,13 @@ export class EventsService extends PuterService { * and renewing is what separates one of those from a node that died without * reaping. */ - #startRefresh(userId: number, socketId: string): void { - const key = `${userId}|${socketId}`; + #startRefresh(holderUserId: number, socketId: string): void { + const key = `${holderUserId}|${socketId}`; if (this.#refreshTimers.has(key)) return; const timer = setInterval( () => { void this.stores.eventSubscription - .refresh(userId, socketId) + .refresh(holderUserId, socketId) .catch(() => {}); }, Math.floor((SESSION_SUBSCRIPTION_TTL_SECONDS * 1000) / 3), @@ -710,8 +783,8 @@ export class EventsService extends PuterService { this.#refreshTimers.set(key, timer); } - #stopRefresh(userId: number, socketId: string): void { - const key = `${userId}|${socketId}`; + #stopRefresh(holderUserId: number, socketId: string): void { + const key = `${holderUserId}|${socketId}`; const timer = this.#refreshTimers.get(key); if (!timer) return; clearInterval(timer); diff --git a/src/backend/services/events/authorization.integration.test.ts b/src/backend/services/events/authorization.integration.test.ts new file mode 100644 index 000000000..1f6083168 --- /dev/null +++ b/src/backend/services/events/authorization.integration.test.ts @@ -0,0 +1,246 @@ +/* + * 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 . + */ + +/** + * Subscribing, and staying subscribed, against real grants. + * + * The unit tests treat access as data; these stage actual shares, so they are + * what shows that a subscription on someone else's folder can be made at all, + * that the row is found from the owner's side when that owner writes, and that + * taking the share away stops the delivery rather than the write. + */ + +import { afterAll, beforeAll, describe, expect, it, vi } from 'vitest'; +import { EVENTS_COALESCE_WINDOW_MS } from '../../controllers/events/limits.js'; +import { makeActor, type Actor } from '../../core/actor.js'; +import { isHttpError } from '../../core/http/HttpError.js'; +import { setupPuterTestEnv, type PuterTestEnv } from '../../testUtil.js'; +import type { IConfig } from '../../types.js'; +import type { AclMode } from '../acl/ACLService.js'; +import type { DeliveryEnvelope } from './EventsService.js'; + +const BOOT_TIMEOUT_MS = 120_000; + +let env: PuterTestEnv; +let owner: { actor: Actor; username: string; token: string; id: number }; +let guest: { actor: Actor; username: string; token: string; id: number }; +let delivered: DeliveryEnvelope[]; + +const events = () => env.server.services.events; +const fs = () => env.server.services.fs; + +const descriptor = (path: string) => ({ + path, + resolveAncestors: () => fs().getAncestorChain(path), +}); + +const share = async (path: string, mode: AclMode): Promise => { + await env.server.services.acl.setUserUser( + owner.actor, + guest.actor, + descriptor(path), + mode, + ); +}; + +const unshare = async (path: string, mode: AclMode): Promise => { + const entry = await env.server.stores.fsEntry.getEntryByPath(path); + await env.server.services.permission.revokeUserUserPermission( + owner.actor, + guest.username, + `fs:${entry!.uid}:${mode}`, + ); +}; + +const folder = async (path: string): Promise => { + await fs().mkdir(owner.id, { path, createMissingParents: true }); + return path; +}; + +/** A write by the owner, through the route a client actually calls. */ +const mkdirAsOwner = async (path: string): Promise => { + const response = await fetch(new URL('/mkdir', env.apiOrigin), { + method: 'POST', + headers: { + 'content-type': 'application/json', + authorization: `Bearer ${owner.token}`, + }, + body: JSON.stringify({ path, create_missing_parents: true }), + }); + expect(response.status).toBe(200); +}; + +const subscribeAs = async ( + who: { actor: Actor }, + socketId: string, + subject: string, +) => (await events().subscribe(who.actor, socketId, { subject })).sub; + +const settle = (count = 1) => + vi.waitFor(() => expect(delivered.length).toBeGreaterThanOrEqual(count), { + timeout: EVENTS_COALESCE_WINDOW_MS * 12, + interval: 25, + }); + +const quiet = () => + new Promise((resolve) => + setTimeout(resolve, EVENTS_COALESCE_WINDOW_MS * 3), + ); + +const uniquePath = (base: string) => + `${base}/n-${Math.random().toString(36).slice(2, 8)}`; + +beforeAll(async () => { + env = await setupPuterTestEnv({ events: { enabled: true } } as IConfig); + + const ownerRow = await env.server.stores.user.getByUsername( + env.users.user.username, + ); + const guestRow = await env.server.stores.user.getByUsername( + env.users.other.username, + ); + owner = { + actor: makeActor({ user: ownerRow as never }), + username: env.users.user.username, + token: env.users.user.token, + id: ownerRow!.id, + }; + guest = { + actor: makeActor({ user: guestRow as never }), + username: env.users.other.username, + token: env.users.other.token, + id: guestRow!.id, + }; + + delivered = []; + events().onDelivered = (envelope) => delivered.push(envelope); +}, BOOT_TIMEOUT_MS); + +afterAll(async () => { + await env?.shutdown(); +}); + +describe('who may subscribe', () => { + it('lets a guest subscribe to a folder shared with them', async () => { + const path = await folder(`/${owner.username}/shared-listable`); + await share(path, 'list'); + + const sub = await subscribeAs(guest, 'guest-a', `fs:${path}`); + + expect(sub.anchor.path).toBe(path); + }); + + it('answers a folder that was never shared as absent', async () => { + const path = await folder(`/${owner.username}/never-shared`); + + await expect( + subscribeAs(guest, 'guest-b', `fs:${path}`), + ).rejects.toSatisfy( + (err: unknown) => + isHttpError(err) && + err.statusCode === 404 && + err.legacyCode === 'subject_does_not_exist', + ); + }); + + it('refuses a folder the guest can see but not list', async () => { + const path = await folder(`/${owner.username}/shared-visible`); + await share(path, 'see'); + + await expect( + subscribeAs(guest, 'guest-c', `fs:${path}`), + ).rejects.toSatisfy( + (err: unknown) => + isHttpError(err) && + err.statusCode === 403 && + err.legacyCode === 'forbidden', + ); + }); + + it('will not let a filter buy reach the anchor check refuses', async () => { + const shared = await folder(`/${owner.username}/reach-shared`); + await folder(`/${owner.username}/reach-private`); + await share(shared, 'list'); + + await expect( + subscribeAs( + guest, + 'guest-d', + `fs:/${owner.username}/reach-private/**`, + ), + ).rejects.toSatisfy( + (err: unknown) => isHttpError(err) && err.statusCode === 404, + ); + }); +}); + +describe('delivering across an account boundary', () => { + it('finds the guest`s row from the owner`s write', async () => { + const path = await folder(`/${owner.username}/shared-writes`); + await share(path, 'list'); + const theirs = await subscribeAs(guest, 'guest-e', `fs:${path}`); + const mine = await subscribeAs(owner, 'owner-e', `fs:${path}`); + delivered.length = 0; + + await mkdirAsOwner(uniquePath(path)); + await settle(2); + + const byId = new Map(delivered.map((d) => [d.subId, d.event])); + // The guest hears an event about someone else's write; the owner + // hears their own. + expect(byId.get(theirs.subId)).toMatchObject({ self: false }); + expect(byId.get(mine.subId)).toMatchObject({ self: true }); + }); + + it('stops delivering the moment the share is revoked', async () => { + const path = await folder(`/${owner.username}/shared-revoked`); + await share(path, 'list'); + await subscribeAs(guest, 'guest-f', `fs:${path}`); + + await unshare(path, 'list'); + delivered.length = 0; + await mkdirAsOwner(uniquePath(path)); + await quiet(); + + // The row is still registered; it just no longer authorizes anything. + expect(delivered).toEqual([]); + }); + + it('delivers only from the part of the anchor the guest can still list', async () => { + const shared = await folder(`/${owner.username}/narrowed`); + const allowed = await folder(`${shared}/allowed`); + const closed = await folder(`${shared}/closed`); + await share(shared, 'list'); + const sub = await subscribeAs(guest, 'guest-g', `fs:${shared}/**`); + + // The share narrows to one subfolder: the subscription's anchor is no + // longer listable, and the filter still spans both. + await share(allowed, 'list'); + await unshare(shared, 'list'); + delivered.length = 0; + + await mkdirAsOwner(uniquePath(closed)); + await quiet(); + expect(delivered).toEqual([]); + + await mkdirAsOwner(uniquePath(allowed)); + await settle(); + expect(delivered.map((d) => d.subId)).toEqual([sub.subId]); + }); +}); diff --git a/src/backend/services/events/authorization.ts b/src/backend/services/events/authorization.ts index f2fb5024b..492bff9c2 100644 --- a/src/backend/services/events/authorization.ts +++ b/src/backend/services/events/authorization.ts @@ -17,45 +17,176 @@ * along with this program. If not, see . */ +import { makeActor, type Actor } from '../../core/actor.js'; import { HttpError } from '../../core/http/HttpError.js'; +import type { UserRow } from '../../stores/user/UserStore.js'; +import type { + AclError, + AclMode, + ResourceDescriptor, +} from '../acl/ACLService.js'; /** - * Who may subscribe to an anchor, and who may still be delivered from it. + * Who may subscribe to an anchor, who may still be delivered from it, and which + * of their rows an actor is allowed to see. * - * Owner-only today. The real rule is an ACL check for `list` on the anchor, - * with the grant it succeeded under stored on the subscription and re-checked - * at delivery; both call sites below are shaped for that and neither has to - * move when it lands. What must not change is the failure: a node the caller - * cannot reach is answered as absent, because a distinguishable "forbidden" - * turns subscribe into a way to ask whether a path exists. + * Subscribing is the same check as reading the resource, at mode `list` — `see` + * is not enough, because a subscription reports the names of things appearing + * under the anchor. The mode the check passed under is stored on the row and + * re-run per event, against the node the event is about rather than the anchor: + * that is what keeps a match filter from reaching anything the holder could not + * have subscribed to directly, and what makes a revoked share stop delivering + * without anyone having to find and delete the row. + * + * The failure is a `getSafeAclError` failure, not a bare 403: a node the caller + * cannot even `see` is answered as absent, because a distinguishable + * "forbidden" turns subscribe into a way to ask whether a path exists. */ -export interface AnchorOwnership { - /** The user the anchor node belongs to. */ - ownerUserId: number; +/** Minimum ACL mode a subscription needs on its anchor. */ +export const SUBSCRIBE_MODE: AclMode = 'list'; + +/** A node an authorization decision is about. */ +export interface AuthorizedNode { + uid: string; + path: string; } -export interface SubscribingActor { - userId: number; +/** What a stored subscription carries about the grant it was made under. */ +export interface SubscriptionGrant { + holderUserId: number; + appUid: string | null; + permission: AclMode; } -const subjectDoesNotExist = (subject: string): HttpError => - new HttpError(404, `No such entry: ${subject}`, { - legacyCode: 'subject_does_not_exist', - }); +export interface EventAclDeps { + acl: { + check: ( + actor: Actor, + resource: ResourceDescriptor, + mode: AclMode, + ) => Promise; + getSafeAclError: ( + actor: Actor, + resource: ResourceDescriptor, + mode: AclMode, + ) => Promise; + }; + /** Existing ancestors of a path, deepest first. */ + getAncestorChain: ( + path: string, + ) => Promise>; + getUser: (userId: number) => Promise; + getApp: (uid: string) => Promise<{ id?: number } | null>; +} -/** Whether this actor may hold a subscription on this anchor. */ -export const checkSubscribeAuthorization = ( - actor: SubscribingActor, - anchor: AnchorOwnership, -): boolean => actor.userId === anchor.ownerUserId; - -/** The same decision, as the subscribe path needs it: pass, or 404. */ -export const assertSubscribeAuthorized = ( - actor: SubscribingActor, - anchor: AnchorOwnership, +/** + * Descriptor for one node, walking the tree at most once however many + * subscriptions ask about it. The chain must start with the node itself, which + * the store no longer returns once the node is gone — a removal is exactly the + * event whose own grant may live on the node being removed. + */ +export const nodeDescriptor = ( + node: AuthorizedNode, + deps: Pick, +): ResourceDescriptor => { + let ancestors: Promise< + ReadonlyArray<{ uid: string; path: string }> + > | null = null; + return { + path: node.path, + resolveAncestors: () => { + ancestors ??= deps + .getAncestorChain(node.path) + .then((chain) => + chain[0]?.uid === node.uid + ? chain + : [{ uid: node.uid, path: node.path }, ...chain], + ); + return ancestors; + }, + }; +}; + +/** + * The identity a stored subscription acts as when its access is re-checked. An + * app's row is re-checked as that app, not as the user who launched it — and + * the app is resolved rather than taken from the row, because a grant to an app + * is stored against its numeric id and is invisible to an actor without one. + */ +const grantActor = async ( + grant: SubscriptionGrant, + user: UserRow, + deps: EventAclDeps, +): Promise => { + if (!grant.appUid) return makeActor({ user, app: null }); + const app = await deps.getApp(grant.appUid); + if (!app) return null; + return makeActor({ user, app: { uid: grant.appUid, id: app.id } }); +}; + +/** + * Authorize a subscribe. Returns the mode the check succeeded under, which the + * row stores; throws the safe error otherwise. + */ +export const assertSubscribeAuthorized = async ( + actor: Actor, + anchor: AuthorizedNode, subject: string, -): void => { - if (!checkSubscribeAuthorization(actor, anchor)) - throw subjectDoesNotExist(subject); + deps: EventAclDeps, +): Promise => { + const resource = nodeDescriptor(anchor, deps); + if (await deps.acl.check(actor, resource, SUBSCRIBE_MODE)) + return SUBSCRIBE_MODE; + + const safe = await deps.acl.getSafeAclError( + actor, + resource, + SUBSCRIBE_MODE, + ); + if (safe.status === 404) + throw new HttpError(404, `No such entry: ${subject}`, { + legacyCode: 'subject_does_not_exist', + }); + throw new HttpError(403, safe.message, { + legacyCode: safe.fields.code, + }); +}; + +/** + * Whether a stored subscription may still be delivered an event about `node`. + * Anything that cannot be decided is a no: a delivery is not worth failing a + * write over, and silence is the safe direction. + */ +export const checkDeliveryAuthorized = async ( + grant: SubscriptionGrant, + node: ResourceDescriptor, + deps: EventAclDeps, +): Promise => { + try { + const user = await deps.getUser(grant.holderUserId); + if (!user) return false; + const actor = await grantActor(grant, user, deps); + if (!actor) return false; + return await deps.acl.check(actor, node, grant.permission); + } catch { + return false; + } +}; + +/** + * Whether a row is inside what this actor may see and remove. An app-context + * actor — an app token, or an access token an app issued — is confined to the + * rows its own app created; a user-context one sees everything of theirs across + * apps, which is what makes the account the revoke surface. + */ +export const rowInActorScope = ( + actor: Actor, + row: { appUid: string | null }, +): boolean => { + const app = actor.effectiveApp; + // Unresolved is not "no app": reading it that way is what would hand an + // app the account-wide view. + if (app === undefined) return false; + return app === null || row.appUid === app.uid; }; diff --git a/src/backend/services/events/registry.test.ts b/src/backend/services/events/registry.test.ts index a1fd840bb..d0847384b 100644 --- a/src/backend/services/events/registry.test.ts +++ b/src/backend/services/events/registry.test.ts @@ -80,7 +80,10 @@ const entry = { const delivery: DeliveryContext = { key: 'fs.write.file', entry, - ancestorUids: ['uid-parent', 'uid-home'], + ancestors: [ + { uid: 'uid-parent', path: '/u/Documents' }, + { uid: 'uid-home', path: '/u' }, + ], id: 'ev-1', ts: 1_700_000_001, self: true, diff --git a/src/backend/services/events/registry.ts b/src/backend/services/events/registry.ts index 3cb7e5dbe..9b7cbd9b3 100644 --- a/src/backend/services/events/registry.ts +++ b/src/backend/services/events/registry.ts @@ -50,8 +50,8 @@ export interface ProjectedEvent { export interface EventContext { key: EventKey; entry: FSEntry; - /** Ancestor uids of `entry`, deepest first. */ - ancestorUids: readonly string[]; + /** Existing ancestors of `entry`, deepest first. */ + ancestors: ReadonlyArray<{ uid: string; path: string }>; id: string; ts: number; } @@ -99,7 +99,7 @@ export interface UnpublishedInternalEvent { // uid alone and a deep write still matches. const fsTokens = (event: EventContext): string[] => [ fsAnchorToken(event.entry.uid), - ...event.ancestorUids.map(fsAnchorToken), + ...event.ancestors.map((ancestor) => fsAnchorToken(ancestor.uid)), ]; const fsMatchOn = (event: EventContext): string => event.entry.path; diff --git a/src/backend/services/events/scoping.integration.test.ts b/src/backend/services/events/scoping.integration.test.ts new file mode 100644 index 000000000..48ee2912c --- /dev/null +++ b/src/backend/services/events/scoping.integration.test.ts @@ -0,0 +1,315 @@ +/* + * 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 each credential shape may see and remove, and whether it can hold a + * connection at all. + * + * Every actor here is minted and then authenticated for real, because the whole + * question is what `effectiveApp` resolves to at the end of a token chain: an + * app sees only what it created, a credential acting for the account sees + * across apps, and an id belonging to another app is answered as absent rather + * than refused. + */ + +import { io as ioClient, type Socket as ClientSocket } from 'socket.io-client'; +import { afterAll, beforeAll, describe, expect, it, vi } from 'vitest'; +import { v4 as uuidv4 } from 'uuid'; +import { EVENTS_COALESCE_WINDOW_MS } from '../../controllers/events/limits.js'; +import type { Actor } from '../../core/actor.js'; +import { isHttpError } from '../../core/http/HttpError.js'; +import { setupPuterTestEnv, type PuterTestEnv } from '../../testUtil.js'; +import type { IConfig } from '../../types.js'; +import type { AuthResult } from '../auth/AuthService.js'; +import { decideSocketAuth } from '../socket/SocketService.js'; +import type { DeliveryEnvelope } from './EventsService.js'; + +const BOOT_TIMEOUT_MS = 120_000; +const SOCKET_ID = 'matrix-socket'; + +let env: PuterTestEnv; +let username: string; +let userId: number; +let anchor: string; + +/** One actor per row of the scoping matrix, each from a real credential. */ +let session: Actor; +let appOne: Actor; +let appTwo: Actor; +let appAccessToken: Actor; +let personalAccessToken: Actor; +let worker: Actor; + +/** Raw token strings, for the shapes a live handshake is attempted with. */ +let appAccessTokenStr: string; + +let appOneUid: string; +let appTwoUid: string; + +const events = () => env.server.services.events; +const auth = () => env.server.services.auth; + +const actorFor = async (token: string): Promise => { + const result = await auth().authenticate(token); + expect(result.actor, 'credential did not authenticate').toBeDefined(); + return result.actor!; +}; + +/** An app the user has granted `list` on the shared anchor. */ +const makeAppActor = async (): Promise<{ uid: string; actor: Actor }> => { + 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/`, userId], + ); + const entry = await env.server.stores.fsEntry.getEntryByPath(anchor); + await env.server.services.permission.grantUserAppPermission( + session, + uid, + `fs:${entry!.uid}:list`, + ); + const token = await auth().getUserAppToken(session, uid); + return { uid, actor: await actorFor(token) }; +}; + +const subscribeAs = async (actor: Actor) => + (await events().subscribe(actor, SOCKET_ID, { subject: `fs:${anchor}` })) + .sub; + +/** A write by the user, through the route a client actually calls. */ +const writeAsUser = async (path: string): Promise => { + const response = await fetch(new URL('/mkdir', env.apiOrigin), { + method: 'POST', + headers: { + 'content-type': 'application/json', + authorization: `Bearer ${env.users.user.token}`, + }, + body: JSON.stringify({ path, create_missing_parents: true }), + }); + expect(response.status).toBe(200); +}; + +const heldBy = async (actor: Actor): Promise => + (await events().listSubscriptions(actor, SOCKET_ID)).map((sub) => sub.subId); + +const absent = (err: unknown) => + isHttpError(err) && + err.statusCode === 404 && + err.legacyCode === 'subscription_does_not_exist'; + +/** Attempt a live handshake; resolves with the rejection message, or throws if it connects. */ +const attemptConnect = (token: string): Promise => + new Promise((resolve, reject) => { + const socket: ClientSocket = ioClient(env.origin, { + auth: { auth_token: token }, + transports: ['websocket'], + reconnection: false, + }); + socket.on('connect', () => { + socket.disconnect(); + reject(new Error('socket connected; expected a handshake rejection')); + }); + socket.on('connect_error', (err: Error) => { + socket.disconnect(); + resolve(err.message); + }); + }); + +beforeAll(async () => { + env = await setupPuterTestEnv({ events: { enabled: true } } as IConfig); + username = env.users.user.username; + const row = await env.server.stores.user.getByUsername(username); + userId = row!.id; + + anchor = `/${username}/matrix`; + await env.server.services.fs.mkdir(userId, { + path: anchor, + createMissingParents: true, + }); + + session = await actorFor(env.users.user.token); + ({ uid: appOneUid, actor: appOne } = await makeAppActor()); + ({ uid: appTwoUid, actor: appTwo } = await makeAppActor()); + + const entry = await env.server.stores.fsEntry.getEntryByPath(anchor); + appAccessTokenStr = await auth().createAccessToken( + appOne, + [[`fs:${entry!.uid}:list`]], + { label: 'matrix' }, + ); + appAccessToken = await actorFor(appAccessTokenStr); + personalAccessToken = await actorFor(env.users.user.apiToken); + worker = await actorFor(env.users.user.workerToken); +}, BOOT_TIMEOUT_MS); + +afterAll(async () => { + await env?.shutdown(); +}); + +describe('which credentials reach the session verbs at all', () => { + const accepted = (actor: Actor) => { + const decision = decideSocketAuth({ actor } as AuthResult, { + allowAppActors: true, + }); + return 'accept' in decision; + }; + + it('admits a session, an app and a worker, and refuses access tokens', () => { + expect(accepted(session)).toBe(true); + expect(accepted(appOne)).toBe(true); + // A worker session is user-shaped — no app, no access token — so it + // connects like one and is scoped by what it was minted with. + expect(worker.session?.kind).toBe('worker'); + expect(accepted(worker)).toBe(true); + + // Both access-token shapes are refused at the handshake, so neither + // ever holds a socket to call a session verb on. + expect(accepted(appAccessToken)).toBe(false); + expect(accepted(personalAccessToken)).toBe(false); + }); + + it('refuses both access-token shapes at a live handshake, even with events on', async () => { + // `accepted` above is the pure decision function; this drives an + // actual connection against a server booted with events enabled, so + // the wiring — not just the predicate — is what is on the hook. + await expect( + attemptConnect(env.users.user.apiToken), + ).resolves.toMatch(/only user tokens/); + await expect(attemptConnect(appAccessTokenStr)).resolves.toMatch( + /only user tokens/, + ); + }); + + it('refuses an app while events are off, whatever the token says', () => { + expect( + 'reject' in decideSocketAuth({ actor: appOne } as AuthResult), + ).toBe(true); + }); +}); + +describe('what an actor sees and removes', () => { + it('stamps the creating app on the row, and nothing on a session`s', async () => { + const own = await subscribeAs(session); + const theirs = await subscribeAs(appOne); + + const rows = await env.server.stores.eventSubscription.listForSocket( + userId, + SOCKET_ID, + ); + const byId = new Map(rows.map((row) => [row.subId, row])); + expect(byId.get(own.subId)?.appUid).toBeNull(); + expect(byId.get(theirs.subId)?.appUid).toBe(appOneUid); + + // An access token an app issued acts as that app, one hop through its + // issuer — which is what the whole scope keys on. + expect(appAccessToken.effectiveApp?.uid).toBe(appOneUid); + expect(personalAccessToken.effectiveApp).toBeNull(); + expect(worker.effectiveApp).toBeNull(); + + await events().unsubscribe(session, SOCKET_ID, { subId: own.subId }); + await events().unsubscribe(session, SOCKET_ID, { subId: theirs.subId }); + }); + + it('shows an app its own rows and no others', async () => { + const mine = await subscribeAs(appOne); + const theirs = await subscribeAs(appTwo); + const account = await subscribeAs(session); + + await expect(heldBy(appOne)).resolves.toEqual([mine.subId]); + await expect(heldBy(appTwo)).resolves.toEqual([theirs.subId]); + // The token the app issued inherits exactly the app's view. + await expect(heldBy(appAccessToken)).resolves.toEqual([mine.subId]); + + for (const wide of [session, personalAccessToken, worker]) + expect((await heldBy(wide)).sort()).toEqual( + [mine.subId, theirs.subId, account.subId].sort(), + ); + + await events().unsubscribe(session, SOCKET_ID, { subId: mine.subId }); + await events().unsubscribe(session, SOCKET_ID, { subId: theirs.subId }); + await events().unsubscribe(session, SOCKET_ID, { + subId: account.subId, + }); + }); + + it('answers another app`s subscription id as absent', async () => { + const mine = await subscribeAs(appOne); + + await expect( + events().unsubscribe(appTwo, SOCKET_ID, { subId: mine.subId }), + ).rejects.toSatisfy(absent); + // Still there: the refusal removed nothing. + await expect(heldBy(appOne)).resolves.toEqual([mine.subId]); + + await events().unsubscribe(appOne, SOCKET_ID, { subId: mine.subId }); + await expect(heldBy(appOne)).resolves.toEqual([]); + }); + + it('answers a session`s subscription id as absent to an app', async () => { + const account = await subscribeAs(session); + + await expect( + events().unsubscribe(appOne, SOCKET_ID, { subId: account.subId }), + ).rejects.toSatisfy(absent); + + await events().unsubscribe(session, SOCKET_ID, { + subId: account.subId, + }); + }); + + it('lets the account remove what an app left behind', async () => { + const theirs = await subscribeAs(appTwo); + expect(appTwoUid).not.toBe(appOneUid); + + await events().unsubscribe(session, SOCKET_ID, { subId: theirs.subId }); + + await expect(heldBy(session)).resolves.toEqual([]); + }); +}); + +describe('an app`s row is delivered as the app', () => { + it('follows the grant the app holds, and stops when it is taken back', async () => { + const delivered: DeliveryEnvelope[] = []; + events().onDelivered = (envelope) => delivered.push(envelope); + const sub = await subscribeAs(appOne); + const entry = await env.server.stores.fsEntry.getEntryByPath(anchor); + + await writeAsUser(`${anchor}/granted-${uuidv4().slice(0, 8)}`); + await vi.waitFor(() => expect(delivered).toHaveLength(1), { + timeout: EVENTS_COALESCE_WINDOW_MS * 12, + interval: 25, + }); + expect(delivered[0].subId).toBe(sub.subId); + + // The user takes the app's access away; the row outlives the grant, + // and the re-check is what makes it stop. + await env.server.services.permission.revokeUserAppPermission( + session, + appOneUid, + `fs:${entry!.uid}:list`, + ); + delivered.length = 0; + + await writeAsUser(`${anchor}/revoked-${uuidv4().slice(0, 8)}`); + await new Promise((resolve) => + setTimeout(resolve, EVENTS_COALESCE_WINDOW_MS * 3), + ); + expect(delivered).toEqual([]); + }); +}); diff --git a/src/backend/services/events/subscriptionCache.ts b/src/backend/services/events/subscriptionCache.ts index e3e577120..83be4bcda 100644 --- a/src/backend/services/events/subscriptionCache.ts +++ b/src/backend/services/events/subscriptionCache.ts @@ -18,10 +18,11 @@ */ /** - * The "does this user have any subscriptions at all" answer, per process. + * The "does anyone watch anything of this user's" answer, per process — asked + * of whoever owns the resource being written, which is all a write knows. * - * Nearly every user has none, and this is what lets a write on their behalf - * cost nothing: after the first miss the answer is in memory and dispatch never + * Nearly every user has nothing watched, and this is what lets a write cost + * nothing: after the first miss the answer is in memory and dispatch never * touches Redis again. That only holds if invalidation is pushed rather than * polled, so entries are keyed by a per-user generation the subscribe and * unsubscribe paths bump and broadcast — never by a timer, which would put the diff --git a/src/backend/services/fs/FSService.ts b/src/backend/services/fs/FSService.ts index 4a8ec19d8..97cfee8b9 100644 --- a/src/backend/services/fs/FSService.ts +++ b/src/backend/services/fs/FSService.ts @@ -3819,10 +3819,7 @@ export class FSService extends PuterService { .dispatchFs(key, entry, { actingUserId: (Context.get('actor') as Actor | undefined) ?.user?.id, - ancestorUids: async () => - (await this.getAncestorChain(entry.path)).map( - (ancestor) => ancestor.uid, - ), + ancestors: () => this.getAncestorChain(entry.path), }) .catch((err: unknown) => { console.warn('[fs] event dispatch failed', err); diff --git a/src/backend/stores/events/EventSubscriptionStore.test.ts b/src/backend/stores/events/EventSubscriptionStore.test.ts index 323acf39a..e9b778e19 100644 --- a/src/backend/stores/events/EventSubscriptionStore.test.ts +++ b/src/backend/stores/events/EventSubscriptionStore.test.ts @@ -41,7 +41,8 @@ const makeSub = ( ): SessionSubscription => ({ subId: over.subId ?? `sub-${Math.random().toString(36).slice(2)}`, socketId: 'socket-a', - userId: USER, + holderUserId: USER, + ownerUserId: USER, subject: 'fs:/testuser/Documents', token: 'f#anchor', anchorUid: 'anchor', @@ -49,9 +50,17 @@ const makeSub = ( match: null, op: null, appUid: null, + permission: 'list', ...over, }); +/** A row on someone else's node: the shared-folder case, in one place. */ +const sharedWith = ( + holderUserId: number, + over: Partial = {}, +): SessionSubscription => + makeSub({ holderUserId, ownerUserId: USER, ...over }); + beforeEach(() => { USER = ++userSeq; redis = new MockRedis.Cluster(['redis://localhost:7001']); @@ -76,6 +85,26 @@ describe('registration', () => { ]); }); + it('indexes a shared anchor under its owner, not its subscriber', async () => { + const holder = USER + 5000; + const sub = sharedWith(holder); + + const bump = await store.add(sub); + + // Dispatch only knows whose resource changed, so that is the side the + // row has to be findable from. + expect(bump.userId).toBe(USER); + await expect(store.userHasAny(USER)).resolves.toBe(true); + await expect(store.userHasAny(holder)).resolves.toBe(false); + await expect(store.getForTokens(USER, ['f#anchor'])).resolves.toEqual([ + sub, + ]); + // The socket set is the one side that is the holder's. + expect(await redis.smembers(`ev:s:{${holder}}:socket-a`)).toEqual([ + `${USER}|f#anchor|${sub.subId}`, + ]); + }); + it('keys every one of a user`s keys into one cluster slot', async () => { await store.add(makeSub()); const keys = await redis.keys(`ev:*{${USER}}*`); @@ -100,6 +129,20 @@ describe('registration', () => { SESSION_SUBSCRIPTION_TTL_SECONDS - 10, ); }); + + it('refreshes the owner`s keys, not only the holder`s', async () => { + const holder = USER + 5000; + await store.add(sharedWith(holder)); + await redis.expire(`ev:w:{${USER}}`, 5); + await redis.expire(`ev:t:{${USER}}:f#anchor`, 5); + + await store.refresh(holder, 'socket-a'); + + for (const key of [`ev:w:{${USER}}`, `ev:t:{${USER}}:f#anchor`]) + expect(await redis.ttl(key)).toBeGreaterThan( + SESSION_SUBSCRIPTION_TTL_SECONDS - 10, + ); + }); }); describe('the per-socket cap', () => { @@ -123,7 +166,7 @@ describe('the per-socket cap', () => { await expect( store.add(makeSub({ subId: 'b-0', socketId: 'socket-b' })), - ).resolves.toBeGreaterThan(0); + ).resolves.toMatchObject({ userId: USER }); }); }); @@ -132,7 +175,7 @@ describe('removal', () => { const sub = makeSub(); await store.add(sub); - await store.remove(USER, sub.socketId, sub.subId); + await store.remove(sub); await expect(store.watchedTokens(USER, ['f#anchor'])).resolves.toEqual( [], @@ -146,7 +189,7 @@ describe('removal', () => { await store.add(mine); await store.add(theirs); - await store.remove(USER, 'socket-a', 'mine'); + await store.remove(mine); await expect(store.watchedTokens(USER, ['f#anchor'])).resolves.toEqual([ 'f#anchor', @@ -156,12 +199,32 @@ describe('removal', () => { ]); }); - it('reports an id this socket never held as absent', async () => { - await store.add(makeSub({ subId: 'mine', socketId: 'socket-a' })); + it('reads back only what the asking socket holds', async () => { + const mine = makeSub({ subId: 'mine', socketId: 'socket-a' }); + await store.add(mine); await expect( - store.remove(USER, 'socket-b', 'mine'), + store.getForSocket(USER, 'socket-a', 'mine'), + ).resolves.toEqual(mine); + await expect( + store.getForSocket(USER, 'socket-b', 'mine'), ).resolves.toBeNull(); + await expect( + store.getForSocket(USER, 'socket-a', 'not-a-sub'), + ).resolves.toBeNull(); + }); + + it('finds a shared-anchor row from the holder`s socket', async () => { + const holder = USER + 5000; + const sub = sharedWith(holder, { subId: 'shared' }); + await store.add(sub); + + await expect( + store.getForSocket(holder, 'socket-a', 'shared'), + ).resolves.toEqual(sub); + + await store.remove(sub); + await expect(store.userHasAny(USER)).resolves.toBe(false); }); }); @@ -196,7 +259,29 @@ describe('disconnect', () => { }); it('says nothing changed when the socket held nothing', async () => { - await expect(store.reapSocket(USER, 'socket-z')).resolves.toBeNull(); + await expect(store.reapSocket(USER, 'socket-z')).resolves.toEqual([]); + }); + + it('moves the generation of every owner the socket watched', async () => { + const holder = USER + 5000; + const otherOwner = USER + 6000; + await store.add(sharedWith(holder)); + await store.add( + makeSub({ + subId: 'elsewhere', + holderUserId: holder, + ownerUserId: otherOwner, + token: 'f#other', + }), + ); + + const bumps = await store.reapSocket(holder, 'socket-a'); + + expect(bumps.map((bump) => bump.userId).sort()).toEqual( + [USER, otherOwner].sort(), + ); + await expect(store.userHasAny(USER)).resolves.toBe(false); + await expect(store.userHasAny(otherOwner)).resolves.toBe(false); }); }); @@ -204,11 +289,11 @@ describe('the generation counter', () => { it('advances on every registration and removal', async () => { const first = await store.add(makeSub({ subId: 'a' })); const second = await store.add(makeSub({ subId: 'b', token: 'f#two' })); - expect(second).toBeGreaterThan(first); + expect(second.generation).toBeGreaterThan(first.generation); - const third = await store.remove(USER, 'socket-a', 'a'); - expect(third).toBeGreaterThan(second); - await expect(store.getGeneration(USER)).resolves.toBe(third); + const third = await store.remove(makeSub({ subId: 'a' })); + expect(third.generation).toBeGreaterThan(second.generation); + await expect(store.getGeneration(USER)).resolves.toBe(third.generation); }); it('outlives the subscriptions it orders', async () => { diff --git a/src/backend/stores/events/EventSubscriptionStore.ts b/src/backend/stores/events/EventSubscriptionStore.ts index 09d0d80c7..2247ed83f 100644 --- a/src/backend/stores/events/EventSubscriptionStore.ts +++ b/src/backend/stores/events/EventSubscriptionStore.ts @@ -20,6 +20,7 @@ import { EVENTS_SESSION_SUBSCRIPTIONS_PER_SOCKET } from '../../controllers/events/limits.js'; import { HttpError } from '../../core/http/HttpError.js'; import type { FsOp } from '../../services/events/subjects.js'; +import type { AclMode } from '../../services/acl/ACLService.js'; import { PuterStore } from '../types.js'; /** @@ -27,14 +28,25 @@ import { PuterStore } from '../types.js'; * when it disconnects. Nothing here outlives a connection, so none of it * belongs in a table. * - * Every key a user owns carries the same `{}` hash tag, so one user's - * whole set lives in one cluster slot and a pipeline over it never crosses - * slots. Four keys, each answering one question: + * Rows are indexed by the **owner of the anchor node**, not by the subscriber: + * dispatch knows only whose resource changed, and a subscription on a folder + * shared with someone else has to be found from that side. The subscriber is + * still on the row — it is who the delivery is for and whose access is + * re-checked — but it is not what anything is keyed by. * - * ev:w:{} SET which tokens anyone is watching - * ev:t:{}: HASH subId -> row, for one watched token - * ev:s:{}: SET what this socket holds, for reaping - * ev:g:{} STR subscription-set generation + * Keys carry a `{}` hash tag so one user's set lives in one cluster + * slot and a pipeline over it never crosses slots. Four keys, each answering + * one question: + * + * ev:w:{} SET which tokens anyone is watching + * ev:t:{}: HASH subId -> row, for one watched token + * ev:s:{}: SET what this socket holds, for reaping + * ev:g:{} STR subscription-set generation + * + * The socket set is the one keyed by the holder — it is read on disconnect, + * when all that is known is whose connection went — so its members name the + * owner whose keyspace each row lives in, and a write touching both sides + * splits into one pipeline per slot. * * `ev:w` is what dispatch asks first and is the only one on the hot path. * Membership in it is exact rather than approximate: a token's row hash going @@ -51,7 +63,10 @@ import { PuterStore } from '../types.js'; export interface SessionSubscription { subId: string; socketId: string; - userId: number; + /** Who subscribed: the delivery target, and whose access is re-checked. */ + holderUserId: number; + /** Owner of the anchor node: the keyspace this row is indexed in. */ + ownerUserId: number; /** The subject as the client asked for it. */ subject: string; /** Anchor token the row is indexed under. */ @@ -61,7 +76,16 @@ export interface SessionSubscription { /** Glob relative to the anchor, or `null` for a node-form subscription. */ match: string | null; op: FsOp | null; + /** The app that created the row, and the scope of the three verbs. */ appUid: string | null; + /** ACL mode the subscribe check passed under; re-checked per delivery. */ + permission: AclMode; +} + +/** One owner's generation after a change to the set of rows keyed under them. */ +export interface GenerationBump { + userId: number; + generation: number; } // -- Keys ------------------------------------------------------------- @@ -73,11 +97,35 @@ const socketKey = (userId: number | string, socketId: string): string => `ev:s:{${userId}}:${socketId}`; const generationKey = (userId: number | string): string => `ev:g:{${userId}}`; -/** `ev:s` members name the row they point at. */ -const socketRef = (token: string, subId: string): string => `${token}|${subId}`; -const parseSocketRef = (ref: string): { token: string; subId: string } => { - const at = ref.indexOf('|'); - return { token: ref.slice(0, at), subId: ref.slice(at + 1) }; +/** `ev:s` members name the row they point at, and the keyspace it is in. */ +interface SocketRef { + ownerUserId: number; + token: string; + subId: string; +} + +const socketRef = (ref: SocketRef): string => + `${ref.ownerUserId}|${ref.token}|${ref.subId}`; + +const parseSocketRef = (ref: string): SocketRef => { + const owner = ref.indexOf('|'); + const token = ref.indexOf('|', owner + 1); + return { + ownerUserId: Number(ref.slice(0, owner)), + token: ref.slice(owner + 1, token), + subId: ref.slice(token + 1), + }; +}; + +/** Group refs by the keyspace they live in, so no pipeline crosses slots. */ +const byOwner = (refs: readonly SocketRef[]): Map => { + const grouped = new Map(); + for (const ref of refs) { + const held = grouped.get(ref.ownerUserId) ?? []; + held.push(ref); + grouped.set(ref.ownerUserId, held); + } + return grouped; }; // -- Lifetimes -------------------------------------------------------- @@ -107,70 +155,89 @@ export class EventSubscriptionStore extends PuterStore { // -- Writes ------------------------------------------------------ /** - * Register one subscription. Returns the generation the write produced, so - * the caller can broadcast it. + * Register one subscription. Returns the owner's new generation, so the + * caller can broadcast it — that is the keyspace dispatch reads. * * Ordering is deliberate: the row lands before the token joins the watched * set, so dispatch never sees a token whose rows it cannot read yet. */ - async add(sub: SessionSubscription): Promise { - const { userId, socketId, token, subId } = sub; + async add(sub: SessionSubscription): Promise { + const { holderUserId, ownerUserId, socketId, token, subId } = sub; const held = await this.clients.redis.scard( - socketKey(userId, socketId), + socketKey(holderUserId, socketId), ); if (held >= EVENTS_SESSION_SUBSCRIPTIONS_PER_SOCKET) throw subscriptionLimitReached(); - const pipeline = this.clients.redis.pipeline(); - pipeline.hset(tokenKey(userId, token), subId, JSON.stringify(sub)); - pipeline.expire( - tokenKey(userId, token), + const rows = this.clients.redis.pipeline(); + rows.hset(tokenKey(ownerUserId, token), subId, JSON.stringify(sub)); + rows.expire( + tokenKey(ownerUserId, token), SESSION_SUBSCRIPTION_TTL_SECONDS, ); - pipeline.sadd(socketKey(userId, socketId), socketRef(token, subId)); - pipeline.expire( - socketKey(userId, socketId), - SESSION_SUBSCRIPTION_TTL_SECONDS, - ); - pipeline.sadd(watchedKey(userId), token); - pipeline.expire(watchedKey(userId), SESSION_SUBSCRIPTION_TTL_SECONDS); - await pipeline.exec(); + rows.sadd(watchedKey(ownerUserId), token); + rows.expire(watchedKey(ownerUserId), SESSION_SUBSCRIPTION_TTL_SECONDS); + await rows.exec(); - return this.bumpGeneration(userId); + const holder = this.clients.redis.pipeline(); + holder.sadd( + socketKey(holderUserId, socketId), + socketRef({ ownerUserId, token, subId }), + ); + holder.expire( + socketKey(holderUserId, socketId), + SESSION_SUBSCRIPTION_TTL_SECONDS, + ); + await holder.exec(); + + return { + userId: ownerUserId, + generation: await this.bumpGeneration(ownerUserId), + }; } /** - * Drop one subscription. Returns the new generation, or `null` when the - * subscription was not this socket's to remove — the caller answers that as - * a 404 rather than telling a client which ids exist. + * Drop one subscription the caller has already read back. Taking the row + * rather than an id keeps the scope decision — whose row this is, and which + * app's — with the actor, where it belongs. */ - async remove( - userId: number, - socketId: string, - subId: string, - ): Promise { - const refs = await this.clients.redis.smembers( - socketKey(userId, socketId), - ); - const ref = refs.find((r) => parseSocketRef(r).subId === subId); - if (!ref) return null; - - await this.#dropRefs(userId, socketId, [ref]); - return this.bumpGeneration(userId); + async remove(sub: SessionSubscription): Promise { + await this.#dropRefs(sub.holderUserId, sub.socketId, [ + { + ownerUserId: sub.ownerUserId, + token: sub.token, + subId: sub.subId, + }, + ]); + return { + userId: sub.ownerUserId, + generation: await this.bumpGeneration(sub.ownerUserId), + }; } /** * Drop everything a socket held. Runs on disconnect; the TTL is what covers - * the disconnect that never runs. + * the disconnect that never runs. One socket can hold rows in several + * owners' keyspaces, so several generations may move. */ - async reapSocket(userId: number, socketId: string): Promise { - const refs = await this.clients.redis.smembers( - socketKey(userId, socketId), - ); - if (refs.length === 0) return null; - await this.#dropRefs(userId, socketId, refs); - return this.bumpGeneration(userId); + async reapSocket( + holderUserId: number, + socketId: string, + ): Promise { + const refs = ( + await this.clients.redis.smembers(socketKey(holderUserId, socketId)) + ).map(parseSocketRef); + if (refs.length === 0) return []; + + await this.#dropRefs(holderUserId, socketId, refs); + const bumps: GenerationBump[] = []; + for (const ownerUserId of byOwner(refs).keys()) + bumps.push({ + userId: ownerUserId, + generation: await this.bumpGeneration(ownerUserId), + }); + return bumps; } /** @@ -180,49 +247,75 @@ export class EventSubscriptionStore extends PuterStore { * anchor. */ async #dropRefs( - userId: number, + holderUserId: number, socketId: string, - refs: string[], + refs: readonly SocketRef[], ): Promise { - const parsed = refs.map(parseSocketRef); - - const drop = this.clients.redis.pipeline(); - for (const { token, subId } of parsed) - drop.hdel(tokenKey(userId, token), subId); - drop.srem(socketKey(userId, socketId), ...refs); - await drop.exec(); - - const tokens = [...new Set(parsed.map((p) => p.token))]; - const counts = this.clients.redis.pipeline(); - for (const token of tokens) counts.hlen(tokenKey(userId, token)); - const results = (await counts.exec()) ?? []; - - const orphaned = tokens.filter( - (_token, i) => Number(results[i]?.[1] ?? 0) === 0, + await this.clients.redis.srem( + socketKey(holderUserId, socketId), + ...refs.map(socketRef), ); - if (orphaned.length > 0) - await this.clients.redis.srem(watchedKey(userId), ...orphaned); + + for (const [ownerUserId, owned] of byOwner(refs)) { + const drop = this.clients.redis.pipeline(); + for (const { token, subId } of owned) + drop.hdel(tokenKey(ownerUserId, token), subId); + await drop.exec(); + + const tokens = [...new Set(owned.map((ref) => ref.token))]; + const counts = this.clients.redis.pipeline(); + for (const token of tokens) + counts.hlen(tokenKey(ownerUserId, token)); + const results = (await counts.exec()) ?? []; + + const orphaned = tokens.filter( + (_token, i) => Number(results[i]?.[1] ?? 0) === 0, + ); + if (orphaned.length > 0) + await this.clients.redis.srem( + watchedKey(ownerUserId), + ...orphaned, + ); + } } - /** Keep a live socket's keys ahead of the TTL backstop. */ - async refresh(userId: number, socketId: string): Promise { - const pipeline = this.clients.redis.pipeline(); - pipeline.expire( - socketKey(userId, socketId), + /** + * Keep a live socket's keys ahead of the TTL backstop — its own, and the + * watched sets its rows live in, which may belong to other users. + */ + async refresh(holderUserId: number, socketId: string): Promise { + const refs = ( + await this.clients.redis.smembers(socketKey(holderUserId, socketId)) + ).map(parseSocketRef); + + await this.clients.redis.expire( + socketKey(holderUserId, socketId), SESSION_SUBSCRIPTION_TTL_SECONDS, ); - pipeline.expire(watchedKey(userId), SESSION_SUBSCRIPTION_TTL_SECONDS); - await pipeline.exec(); + + for (const [ownerUserId, owned] of byOwner(refs)) { + const pipeline = this.clients.redis.pipeline(); + pipeline.expire( + watchedKey(ownerUserId), + SESSION_SUBSCRIPTION_TTL_SECONDS, + ); + for (const token of new Set(owned.map((ref) => ref.token))) + pipeline.expire( + tokenKey(ownerUserId, token), + SESSION_SUBSCRIPTION_TTL_SECONDS, + ); + await pipeline.exec(); + } } // -- Reads ------------------------------------------------------- /** - * Whether this user has any subscriptions at all. One command, and the only - * thing a cold process needs before it can answer from memory. + * Whether anyone watches anything of this owner's at all. One command, and + * the only thing a cold process needs before it can answer from memory. */ - async userHasAny(userId: number): Promise { - return (await this.clients.redis.exists(watchedKey(userId))) === 1; + async userHasAny(ownerUserId: number): Promise { + return (await this.clients.redis.exists(watchedKey(ownerUserId))) === 1; } /** @@ -230,12 +323,12 @@ export class EventSubscriptionStore extends PuterStore { * and one command whatever the depth of the tree. */ async watchedTokens( - userId: number, + ownerUserId: number, tokens: readonly string[], ): Promise { if (tokens.length === 0) return []; const flags = await this.clients.redis.smismember( - watchedKey(userId), + watchedKey(ownerUserId), ...tokens, ); return tokens.filter((_token, i) => Number(flags[i]) === 1); @@ -243,12 +336,13 @@ export class EventSubscriptionStore extends PuterStore { /** The rows behind a set of watched tokens. */ async getForTokens( - userId: number, + ownerUserId: number, tokens: readonly string[], ): Promise { if (tokens.length === 0) return []; const pipeline = this.clients.redis.pipeline(); - for (const token of tokens) pipeline.hvals(tokenKey(userId, token)); + for (const token of tokens) + pipeline.hvals(tokenKey(ownerUserId, token)); const results = (await pipeline.exec()) ?? []; const subs: SessionSubscription[] = []; @@ -264,27 +358,55 @@ export class EventSubscriptionStore extends PuterStore { return subs; } - /** Everything one socket holds, newest first is not meaningful here. */ + /** Everything one socket holds, across every keyspace its rows live in. */ async listForSocket( - userId: number, + holderUserId: number, socketId: string, ): Promise { - const refs = await this.clients.redis.smembers( - socketKey(userId, socketId), - ); + const refs = ( + await this.clients.redis.smembers(socketKey(holderUserId, socketId)) + ).map(parseSocketRef); if (refs.length === 0) return []; - const byToken = new Map>(); - for (const ref of refs) { - const { token, subId } = parseSocketRef(ref); - const ids = byToken.get(token) ?? new Set(); - ids.add(subId); - byToken.set(token, ids); + const held: SessionSubscription[] = []; + for (const [ownerUserId, owned] of byOwner(refs)) { + const wanted = new Set(owned.map((ref) => ref.subId)); + const rows = await this.getForTokens(ownerUserId, [ + ...new Set(owned.map((ref) => ref.token)), + ]); + held.push(...rows.filter((row) => wanted.has(row.subId))); } + return held; + } - const tokens = [...byToken.keys()]; - const rows = await this.getForTokens(userId, tokens); - return rows.filter((row) => byToken.get(row.token)?.has(row.subId)); + /** + * One row this socket holds, by id. An id the socket never held reads as + * absent, which is what keeps unsubscribe from answering whether someone + * else's id exists. + */ + async getForSocket( + holderUserId: number, + socketId: string, + subId: string, + ): Promise { + const refs = await this.clients.redis.smembers( + socketKey(holderUserId, socketId), + ); + const ref = refs + .map(parseSocketRef) + .find((candidate) => candidate.subId === subId); + if (!ref) return null; + + const raw = await this.clients.redis.hget( + tokenKey(ref.ownerUserId, ref.token), + subId, + ); + if (!raw) return null; + try { + return JSON.parse(raw) as SessionSubscription; + } catch { + return null; + } } // -- Generation --------------------------------------------------