feat: re-authorize live subscriptions; revoke on permission loss or token expiry (#1414)

[kriszyp]

Draft — security-sensitive, hot-path adjacent. Posting for a review pass before merge.

Summary

Subscribe-time authorization is point-in-time: once an SSE/WebSocket/MQTT stream is open it keeps delivering even after the principal loses access (drop_user, role/permission change) or the bearer token it was opened with expires (#1414). This adds continuous re-authorization that terminates such subscriptions.

Re-authorization is table/RBAC-level — it re-runs the same allowRead the subscription was granted with, against a freshly-fetched user. There is no per-record evaluation (consistent with the allowRead grant model and default RBAC's table-level granularity).

How it works

• server/liveSubscriptionAuth.ts — a registry of live subscriptions, each with a recheck() and a terminate(). Swept:
  1. immediately on the ITC user-change broadcast — serverHandlers.userHandler rebuilds the user/role cache before firing listeners, so the recheck observes current permissions; and
  2. on a 30s interval (your choice (a)) as a backstop and to catch token expiry, which is not event-signaled.
     An error during recheck fails closed (revokes). Normal teardown (end()/close) auto-unregisters.
• resources/Resource.ts — registers the resulting subscription at the common auth chokepoint (authorizeActionOnResource) for both the subscribe (MQTT) and connect (SSE/WebSocket) actions. Subscriptions with no user principal (internal watchers, replication, local-bypass) are skipped.
• security/auth.ts — captures the bearer token's JWT exp on the authenticated user so a subscription opened with it can be revoked on expiry.

The recheck interval is overridable via HARPER_SUBSCRIPTION_REAUTH_INTERVAL_MS (used by the test).

Terminate semantics

terminate() calls the subscription's end(), which removes it from the broadcast notify loop — so no further events are delivered on either transport (the security-critical outcome). The transport connection may linger idle until the client disconnects; a clean socket close is a possible refinement.

Testing

integrationTests/security/subscription-revocation.test.ts opens an SSE collection subscription and asserts delivery stops after (1) drop_user (event-driven) and (2) bearer-token expiry (interval-driven), while an authorized stream keeps delivering. 2/2 pass. tsc clean.

Notes / review asks

• MQTT uses mTLS / username-password (no JWT exp), so token-expiry plumbing is HTTP-only; permission-loss covers all transports via the user re-fetch.
• Sibling to Subscription delivery bypasses row-level allowRead — all rows leak to any collection subscriber #1419/fix: enforce row-level allowRead during subscription delivery (#1419) #1524 (per-record subscription allowRead), which were closed — this is the table-level revocation that is the real subscription-security need.

Closes #1414.

🤖 Generated with Claude Code

[gemini-code-assist]

Code Review

This pull request implements continuous re-authorization for live subscriptions (SSE, WebSocket, MQTT) to ensure they are terminated if a user's permissions are revoked or if their bearer token expires. The implementation includes a new registry and background sweeper, along with comprehensive integration tests. The review feedback highlights two important issues: a critical security vulnerability where stale user data in the context could bypass authorization checks during re-evaluation, and a potential stream cleanup bug where arguments are not forwarded when overriding the subscription's end method.

[claude]

Reviewed; no blockers found.

[cb1kenobi]

This is an epic feature, love it!