lee/puter
1
0
Fork 0
mirror of https://github.com/HeyPuter/puter.git synced 2026-09-13 00:36:02 +00:00
Code Issues Actions 9 Packages Projects Releases Wiki Activity

feat: subscribe ACL and app scoping for event subscriptions (PUT-1672) (#3677)

Browse Source
This commit is contained in:
Daniel Salazar
2026-09-02 13:08:23 -07:00
committed by GitHub
parent fb8e2976dc
commit 66dc0a6396
11 changed files with 1314 additions and 251 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/backend/services/events/EventsService.test.ts
+120 -30
View File
@@ -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<string, 'hidden' | 'forbidden'>;
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<void> => {
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<string, unknown>;
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();
src/backend/services/events/EventsService.ts
+139 -66
View File
@@ -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<readonly string[]>;
ancestors?: () => Promise<ReadonlyArray<{ uid: string; path: string }>>;
/** 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<void> {
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<SubscriptionView[]> {
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<void> {
async reapSocket(holderUserId: number, socketId: string): Promise<void> {
// 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<void> {
// 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<SessionSubscription[]> {
if (rows.length === 0) return rows;
const node = this.#eventDescriptor(context);
const decisions = new Map<string, Promise<boolean>>();
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<void> {
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);
src/backend/services/events/authorization.integration.test.ts
+246
View File
@@ -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 <https://www.gnu.org/licenses/>.
*/
/**
* 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<void> => {
await env.server.services.acl.setUserUser(
owner.actor,
guest.actor,
descriptor(path),
mode,
);
};
const unshare = async (path: string, mode: AclMode): Promise<void> => {
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<string> => {
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<void> => {
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]);
});
});
src/backend/services/events/authorization.ts
+160 -29
View File
@@ -17,45 +17,176 @@
* along with this program. If not, see <https://www.gnu.org/licenses/>.
*/
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<boolean>;
getSafeAclError: (
actor: Actor,
resource: ResourceDescriptor,
mode: AclMode,
) => Promise<AclError>;
};
/** Existing ancestors of a path, deepest first. */
getAncestorChain: (
path: string,
) => Promise<ReadonlyArray<{ uid: string; path: string }>>;
getUser: (userId: number) => Promise<UserRow | null>;
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<EventAclDeps, 'getAncestorChain'>,
): 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<Actor | null> => {
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<AclMode> => {
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<boolean> => {
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;
};
src/backend/services/events/registry.test.ts
+4 -1
View File
@@ -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,
src/backend/services/events/registry.ts
+3 -3
View File
@@ -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;
src/backend/services/events/scoping.integration.test.ts
+315
View File
@@ -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 <https://www.gnu.org/licenses/>.
*/
/**
* 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<Actor> => {
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<void> => {
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<string[]> =>
(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<string> =>
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([]);
});
});
src/backend/services/events/subscriptionCache.ts
+4 -3
View File
@@ -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
src/backend/services/fs/FSService.ts
+1 -4
View File
@@ -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);
src/backend/stores/events/EventSubscriptionStore.test.ts
+97 -12
View File
@@ -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> = {},
): 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 () => {
src/backend/stores/events/EventSubscriptionStore.ts
+225 -103
View File
@@ -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 `{<userId>}` 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:{<userId>} SET which tokens anyone is watching
* ev:t:{<userId>}:<token> HASH subId -> row, for one watched token
* ev:s:{<userId>}:<socketId> SET what this socket holds, for reaping
* ev:g:{<userId>} STR subscription-set generation
* Keys carry a `{<userId>}` 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:{<ownerId>} SET which tokens anyone is watching
* ev:t:{<ownerId>}:<token> HASH subId -> row, for one watched token
* ev:s:{<holderId>}:<socketId> SET what this socket holds, for reaping
* ev:g:{<ownerId>} 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<number, SocketRef[]> => {
const grouped = new Map<number, SocketRef[]>();
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<number> {
const { userId, socketId, token, subId } = sub;
async add(sub: SessionSubscription): Promise<GenerationBump> {
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<number | null> {
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<GenerationBump> {
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<number | null> {
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<GenerationBump[]> {
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<void> {
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<void> {
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<void> {
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<boolean> {
return (await this.clients.redis.exists(watchedKey(userId))) === 1;
async userHasAny(ownerUserId: number): Promise<boolean> {
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<string[]> {
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<SessionSubscription[]> {
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<SessionSubscription[]> {
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<string, Set<string>>();
for (const ref of refs) {
const { token, subId } = parseSocketRef(ref);
const ids = byToken.get(token) ?? new Set<string>();
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<SessionSubscription | null> {
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 --------------------------------------------------