From 1f96cd2e4cc94a5800c4d6e0a60a553eb0136194 Mon Sep 17 00:00:00 2001 From: Andrey Antukh Date: Wed, 7 Oct 2026 14:32:30 +0000 Subject: [PATCH] :bug: Address review findings on websocket revocation Follow-up to the GHSA-m53j-2766-6jqw fix on issue-12155, addressing the reviewCode findings F1-F23. The revalidation task is now cancelled when the relay ends (F1): the closed channel is what ends it, because every way a subscription can end closes it. Before, each ended subscription left a task polling the database every 5 minutes for the life of the process while retaining the closed connection. The permission query now filters soft-deleted files, projects and teams (F2), matching its sibling queries. The websocket re-check therefore sees what the subscribe handler sees, and `delete-file`, `delete-project` and the newly announced `delete-team` actually close file subscriptions. Team-wide changes publish one event per team instead of one per member (F4): the watcher resolves the members itself, so a large team cannot overflow the bounded publish buffers. Both publishers defer through the existing `db/after-commit` (F3), so the watcher never checks pre-commit state. The interval is now genuinely tested (F6): revoking access through a direct row delete, the way an external change would, plus a negative control proving the tests fail with the scheduler removed. The `captured-events` helper forwards keyword args correctly instead of republishing to a garbage topic (F5), and the registry fixture runs per test so no relay leaks between tests (F8). The fail-open policy is pinned in both directions with the interrupt rethrown (F7), the internal topic is a fixed uuid like every other topic (F21), the watcher stops on shutdown (F22), the config key sits with the other HTTP knobs (F19), and the memory no longer claims a Promesa defect that the repo itself contradicts (F10). Closes #12155 AI-assisted-by: space-bunny-free --- .../backend/websocket-subscriptions.md | 15 +- backend/src/app/binfile/common.clj | 16 +- backend/src/app/config.clj | 8 +- backend/src/app/http/websocket.clj | 119 +-- backend/src/app/rpc/commands/files.clj | 3 +- backend/src/app/rpc/commands/management.clj | 20 +- backend/src/app/rpc/commands/projects.clj | 3 +- backend/src/app/rpc/commands/teams.clj | 5 + backend/src/app/rpc/notifications.clj | 56 +- backend/test/backend_tests/binfile_test.clj | 42 ++ .../backend_tests/http_websocket_test.clj | 696 ++++++++++++------ 11 files changed, 662 insertions(+), 321 deletions(-) diff --git a/.serena/memories/backend/websocket-subscriptions.md b/.serena/memories/backend/websocket-subscriptions.md index d050f91bce..0c44296d47 100644 --- a/.serena/memories/backend/websocket-subscriptions.md +++ b/.serena/memories/backend/websocket-subscriptions.md @@ -11,27 +11,28 @@ A subscription is authorized once, at `:subscribe-file`/`:subscribe-team`, and that decision used to be trusted for the whole connection life (GHSA-m53j-2766-6jqw, medium). Two independent mechanisms now bound it: -- **Announcement (primary).** A mutation that can change access publishes on `app.rpc.notifications/internal-revocation-topic` (`notify-permissions-changed`, `notify-team-permissions-changed`). The `::revocation-watcher` Integrant component in `app.http.websocket` subscribes and calls `revalidate-profile-subscriptions`, which looks the profile up in `:by-profile` and re-checks every subscription with the same `has-read-permissions?` predicates the subscribe handlers authorize with. The event is only the trigger; the fresh check is the decision, which is what keeps a downgrade to `viewer` (still readable) and a non-member organization owner (read-only via `get-organization-owner-permissions`) from being mistaken for a revocation. -- **Interval (safety net).** `start-relay` arms a `px/schedule` task per subscription that re-checks every `subscription-revalidation-interval` (default `default-revalidation-interval`, 5 min, a code constant in `app.http.websocket` and **not** in the `config.clj` default map, same rule as session lifetimes). It covers what no announcement can: Nitrate transferring organization ownership (an HTTP call to another service) and rows deleted straight over SQL. The re-check is a scheduled task, not a timer in the relay loop, because a loop-driven timer would be pushed away forever by steady edit traffic — exactly the busiest files. `still-authorized?` is the single fail-open point: a transient DB error logs and keeps the subscription. +- **Announcement (primary).** A mutation that can change access publishes on `app.rpc.notifications/internal-revocation-topic` (`notify-permissions-changed` for one profile, `notify-team-permissions-changed` for one team id — the watcher resolves the members itself, so a large team is one event, not N). The `::revocation-watcher` Integrant component in `app.http.websocket` subscribes and calls `revalidate-profile-subscriptions`, which looks the profile up in `:by-profile` and re-checks every subscription with the same `has-read-permissions?` predicates the subscribe handlers authorize with — and `sql:file-permissions` additionally filters `deleted_at` on file, project and team, so a soft-deleted resource grants nothing. The event is only the trigger; the fresh check is the decision, which is what keeps a downgrade to `viewer` (still readable) and a non-member organization owner (read-only via `get-organization-owner-permissions`) from being mistaken for a revocation. Both publishers defer through `db/after-commit`, because the watcher checks on another connection and would otherwise see the pre-commit state. +- **Interval (safety net).** `start-relay` arms a `px/schedule` task per subscription that re-checks every `subscription-revalidation-interval` (default `default-revalidation-interval`, 5 min, a code constant in `app.http.websocket` and **not** in the `config.clj` default map, same rule as session lifetimes). It covers what no announcement can: Nitrate transferring organization ownership (an HTTP call to another service) and rows deleted straight over SQL. The re-check is a scheduled task, not a timer in the relay loop, because a loop-driven timer would be pushed away forever by steady edit traffic — exactly the busiest files. The task stops when the channel closes: every way a subscription can end closes it, so a finished relay never keeps polling. Cost is two or three permission queries per subscription per interval, plus one Nitrate call per team per interval when the `:admin-console` flag is on (cached 30 s); ticks run on promesa's default scheduler (one or two threads shared with every other scheduled task), so do not shorten the interval without sizing that. Two places fail open, deliberately: `still-authorized?` (a transient DB error logs and keeps the subscription; `InterruptedException` is rethrown) and the `catch Throwable` in `watch-revocations`. -Which mutations announce: `delete-team-member`, `leave-team`, `update-team-member-role` (member only), `delete-file`, `delete-project`, `move-files`, `move-project` (whole source team). `delete-share-link` deliberately does not: a share-link visitor never passes the websocket subscribe check, because `check-read-permissions!` is called without a `share-id` and that arity only considers team membership. +Which mutations announce: `delete-team-member`, `leave-team`, `update-team-member-role` (member only), `delete-team`, `delete-file`, `delete-project`, `move-files`, `move-project` (one team event each). Anything not listed here is covered only by the interval. `delete-share-link` deliberately does not: a share-link visitor never passes the websocket subscribe check, because `check-read-permissions!` is called without a `share-id` and that arity only considers team membership. `close-file-subscription`/`close-team-subscription` are idempotent: no-op on a mismatched id, and they `dissoc` the subscription so a repeat call and `:pointer-update` do not publish to a dead topic. Closing the channel is what ends the relay (`take!` on a closed channel returns `nil`). ## Topic and layering -- `internal-revocation-topic` is a plain string, not a uuid, so it cannot collide with the `uuid/zero` system topic — that one is `sp/pipe`d straight into the client output and anything published there reaches every client. No client can subscribe to the internal topic: `:subscribe-file`/`:subscribe-team` only accept ids that already passed a permission check, and `:broadcast`/`:pointer-update` only republish to ids the server already holds. +- `internal-revocation-topic` is a fixed uuid that is never a real resource id, so it cannot collide with a file, team or profile topic — and deliberately not the `uuid/zero` system topic, which is `sp/pipe`d straight into the client output and anything published there reaches every client. No client can subscribe to the internal topic: `:subscribe-file`/`:subscribe-team` only accept ids that already passed a permission check, and `:broadcast`/`:pointer-update` only republish to ids the server already holds. - The topic and the publish helpers live in `app.rpc.notifications`, not in `app.http.websocket`: the command namespaces cannot require the watcher, because `app.http.websocket` already requires `app.rpc.commands.files`/`teams`. The watcher keeps the registry, which is the part it owns. -## Promesa traps in this namespace +## Time and channel traps in this namespace -- `sp/alts!` over the **data** channel silently stopped delivering in practice; `sp/take!` is the proven path here (`app.util.websocket/start-io-loop!` uses `alts!` with a ping channel, which is a different case). +- `take!` is the proven relay path here. `msgbus.clj` does use `sp/alts!` over data channels in production, so there is no general defect — but the relay keeps the simple blocking take plus an independent scheduled task instead. - `ct/diff a b` returns **b − a** (`Duration/between`), so "time until a future instant" is `(inst-ms (ct/diff (ct/now) future))`. - `ct/duration` unit keywords are singular: `{:millis 150}`, `{:minutes 5}`, never `:milliseconds`/`:days`. +- `still-authorized?` takes the predicate itself: `(boolean (authorized?))`. Calling `(boolean pred)` on the closure is always true and silently disables the interval — this exact bug shipped once and no test caught it until the resubscribe-counting test existed. ## Tests -`backend/test/backend_tests/http_websocket_test.clj` drives the real Redis-backed msgbus, so the revocation tests are end-to-end: register the connection in the real registry, run the real command, wait for the subscription to drop, then assert no content arrives. A connection built without `register-connection` is invisible to the watcher, so the watcher-based tests must go through `open-registered-connection`. Interval tests shorten the config via `with-redefs` on `cf/get` rather than waiting minutes. +`backend/test/backend_tests/http_websocket_test.clj` drives the real Redis-backed msgbus, so the revocation tests are end-to-end: register the connection in the real registry, run the real command, wait for the subscription to drop, then assert no content arrives. A connection built without `register-connection` is invisible to the watcher, so the watcher-based tests must go through `open-registered-connection`. Interval tests shorten the config via `with-redefs` on `cf/get` rather than waiting minutes. `with-clean-registry` runs as an `:each` fixture, so no test can leak a live relay (plus its revalidation task) into the next one. `captured-events` forwards the keyword args through to the real `mbus/pub!` — rebuilding them positionally publishes to the topic `:message` with a `nil` payload and silently untests every other notice. Interval tests take access away with a direct row delete, never with a command: a command announces, so the test would pass even with the interval removed. The resubscribe-counting test is the one that proves finished relays stop polling. ## See also diff --git a/backend/src/app/binfile/common.clj b/backend/src/app/binfile/common.clj index 42085f7647..a264eb9d20 100644 --- a/backend/src/app/binfile/common.clj +++ b/backend/src/app/binfile/common.clj @@ -377,19 +377,27 @@ inner join project as p on (p.id = f.project_id) where fpr.file_id = ? and fpr.profile_id = ? + and f.deleted_at is null + and p.deleted_at is null and exists (select 1 from team_profile_rel as tpr + inner join team as t on (t.id = tpr.team_id) where tpr.team_id = p.team_id - and tpr.profile_id = fpr.profile_id) + and tpr.profile_id = fpr.profile_id + and t.deleted_at is null) union all select tpr.is_owner, tpr.is_admin, tpr.can_edit from team_profile_rel as tpr + inner join team as t on (t.id = tpr.team_id) inner join project as p on (p.team_id = tpr.team_id) inner join file as f on (p.id = f.project_id) where f.id = ? and tpr.profile_id = ? + and t.deleted_at is null + and p.deleted_at is null + and f.deleted_at is null union all select ppr.is_owner, ppr.is_admin, @@ -399,10 +407,14 @@ inner join project as p on (p.id = ppr.project_id) where f.id = ? and ppr.profile_id = ? + and f.deleted_at is null + and p.deleted_at is null and exists (select 1 from team_profile_rel as tpr + inner join team as t on (t.id = tpr.team_id) where tpr.team_id = p.team_id - and tpr.profile_id = ppr.profile_id)") + and tpr.profile_id = ppr.profile_id + and t.deleted_at is null)") (defn- get-file-permissions* [conn profile-id file-id] diff --git a/backend/src/app/config.clj b/backend/src/app/config.clj index 3dd2e52fe3..53d6b646e6 100644 --- a/backend/src/app/config.clj +++ b/backend/src/app/config.clj @@ -145,6 +145,10 @@ [:http-server-io-threads {:optional true} ::sm/int] [:http-server-max-worker-threads {:optional true} ::sm/int] + ;; How often a websocket subscription re-checks that its access still + ;; holds. The default lives in `app.http.websocket`, not here. + [:subscription-revalidation-interval {:optional true} ::ct/duration] + ;; Explicit CORS allowlist used when the :cors flag is enabled. ;; Configured via PENPOT_ALLOWED_ORIGINS as a comma/whitespace ;; separated list of origins (e.g. "https://plugins.example.com"). @@ -253,10 +257,6 @@ [:auth-token-cookie-max-age {:optional true} ::ct/duration] [:auth-token-cookie-max-age-absolute {:optional true} ::ct/duration] - ;; How often a websocket subscription re-checks that its access still - ;; holds. The default lives in `app.http.websocket`, not here. - [:subscription-revalidation-interval {:optional true} ::ct/duration] - [:registration-domain-whitelist {:optional true} [::sm/set :string]] [:email-verify-threshold {:optional true} ::ct/duration] diff --git a/backend/src/app/http/websocket.clj b/backend/src/app/http/websocket.clj index af3ae0a26c..638f3e1e75 100644 --- a/backend/src/app/http/websocket.clj +++ b/backend/src/app/http/websocket.clj @@ -73,8 +73,7 @@ (swap! state (fn [{:keys [connections] :as st}] (let [owner (::profile-id (get connections id)) - indexed (get-in st [:by-profile owner]) - indexed (if (set? indexed) (disj indexed id) indexed)] + indexed (disj (get-in st [:by-profile owner] #{}) id)] (cond-> (assoc st :connections (dissoc connections id)) (and owner (seq indexed)) (assoc-in [:by-profile owner] indexed) @@ -105,10 +104,6 @@ (filter #(= team-id (-> % ::ws/state deref ::team-subscription :team-id))) (map ::ws/id))) -(defn repl-get-connections-for-profile - [profile-id] - (connections-for-profile profile-id)) - (defn repl-close-connection [id] (when-let [{:keys [::ws/close-ch]} (get-connection id)] @@ -167,16 +162,19 @@ (cf/get :subscription-revalidation-interval default-revalidation-interval)) (defn- still-authorized? - "Evaluates the `pred` permission predicate, reporting instead of + "Runs the `authorized?` permission predicate, reporting instead of throwing when the lookup itself fails. A transient database error must not close a subscription that is still - legitimate, and the next tick tries again. This is the one place that - fails open, deliberately: the announcement path is the primary - mechanism and stays strict." - [pred] + legitimate, and the next tick tries again. This is one of two places + that fail open, deliberately: the announcement path (`watch-revocations` + below) also logs and moves on when a re-check throws. An interrupt is + not a lookup failure and is rethrown." + [authorized?] (try - (boolean pred) + (boolean (authorized?)) + (catch InterruptedException cause + (throw cause)) (catch Throwable cause (l/error :hint "cannot re-check a websocket subscription" :cause cause) @@ -226,19 +224,20 @@ (defn- schedule-revalidation - "Rearms `task` every `interval-ms` for as long as `running?` holds. + "Rearms `task` every `interval-ms` while `alive?` holds. Kept independent of the relay loop on purpose: a re-check driven by the same loop would be pushed away by a steady stream of messages, so - a busy file would never be re-checked at all. A task whose `running?` - turned false stops rearming." - [running? interval-ms task] - (when @running? + a busy file would never be re-checked at all. The closed channel is + what ends `alive?`, because every way a subscription can end closes + it." + [alive? interval-ms task] + (when (alive?) (px/schedule interval-ms (fn [] - (when @running? + (when (alive?) (task) - (schedule-revalidation running? interval-ms task)))))) + (schedule-revalidation alive? interval-ms task)))))) (defn- start-relay "Forwards `channel` into the client output channel of `wsp`, and keeps @@ -248,24 +247,22 @@ `on-forward` runs for every forwarded message, which is how the file relay announces presence." - [_ {:keys [::ws/output-ch] :as wsp} channel authorized? close! on-forward] - (let [running? (atom true)] - (schedule-revalidation running? - (inst-ms (revalidation-interval)) - (fn [] - (when-not (still-authorized? authorized?) - (l/info :hint - "closing websocket subscription on re-check" - :profile-id (::profile-id wsp)) - (reset! running? false) - (close!)))) - (sp/go-loop [] - ;; nil means the channel was closed, which means the subscription - ;; is gone and the relay ends. - (when-let [message (sp/take! channel)] - (sp/put! output-ch message) - (when on-forward (on-forward message)) - (recur))))) + [{:keys [::ws/output-ch] :as wsp} channel authorized? close! on-forward] + (schedule-revalidation #(not (sp/closed? channel)) + (inst-ms (revalidation-interval)) + (fn [] + (when-not (still-authorized? authorized?) + (l/info :hint + "closing websocket subscription on re-check" + :profile-id (::profile-id wsp)) + (close!)))) + (sp/go-loop [] + ;; nil means the channel was closed, which means the subscription + ;; is gone and the relay ends. + (when-let [message (sp/take! channel)] + (sp/put! output-ch message) + (when on-forward (on-forward message)) + (recur)))) ;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;; ;; WEBSOCKET HANDLER @@ -349,17 +346,17 @@ ;; The team topic also carries library change diffs, so a ;; subscription that outlives access to the team leaks content, not ;; just presence. - (start-relay cfg wsp channel - #(teams/has-read-permissions? cfg profile-id team-id) - #(close-team-subscription cfg wsp team-id) - nil) - (let [subs {:team-id team-id :organization-id organization-id :channel channel :topic team-id}] (swap! state assoc ::team-subscription subs)) + (start-relay wsp channel + #(teams/has-read-permissions? cfg profile-id team-id) + #(close-team-subscription cfg wsp team-id) + nil) + (mbus/sub! (::mbus/msgbus cfg) :topics topics :chan channel) ;; Close previous subscription if exists @@ -379,7 +376,7 @@ (let [subs {:file-id file-id :channel fch :topic file-id}] (swap! state assoc ::file-subscription subs)) - (start-relay cfg wsp fch + (start-relay wsp fch #(files/has-read-permissions? cfg profile-id file-id) #(close-file-subscription cfg wsp file-id) (fn [message] @@ -490,24 +487,38 @@ (defn- watch-revocations "Consumes revocation events and applies them to the connections this - instance owns." + instance owns. Returns the subscription channel, which `ig/halt-key!` + closes to stop the loop." [{:keys [::mbus/msgbus] :as cfg}] (let [ch (sp/chan :buf (sp/dropping-buffer 64))] (mbus/sub! msgbus :topic notifications/internal-revocation-topic :chan ch) (sp/go-loop [] - (when-let [{:keys [type profile-id]} (sp/take! ch)] - (when (= :profile-permissions-changed type) - (try + (when-let [{:keys [type profile-id team-id]} (sp/take! ch)] + (try + (cond + (= :profile-permissions-changed type) (when (pos? (revalidate-profile-subscriptions cfg profile-id)) (l/debug :hint "revoked websocket subscriptions" :profile-id profile-id)) - (catch Throwable cause - (l/error :hint "cannot revalidate websocket subscriptions" - :profile-id profile-id - :cause cause)))) - (recur))))) + + (= :team-permissions-changed type) + (let [revoked (->> (teams/get-team-members cfg team-id) + (map :id) + (map #(revalidate-profile-subscriptions cfg %)) + (reduce + 0))] + (when (pos? revoked) + (l/debug :hint "revoked websocket subscriptions" + :team-id team-id + :count revoked)))) + (catch Throwable cause + (l/error :hint "cannot revalidate websocket subscriptions" + :profile-id profile-id + :team-id team-id + :cause cause))) + (recur))) + ch)) ;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;; ;; INTEGRANT @@ -527,6 +538,10 @@ [_ cfg] (watch-revocations cfg)) +(defmethod ig/halt-key! ::revocation-watcher + [_ ch] + (sp/close! ch)) + ;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;; ;; HTTP HANDLER ;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;; diff --git a/backend/src/app/rpc/commands/files.clj b/backend/src/app/rpc/commands/files.clj index a0789e764f..40b0736553 100644 --- a/backend/src/app/rpc/commands/files.clj +++ b/backend/src/app/rpc/commands/files.clj @@ -1099,8 +1099,7 @@ ;; which profiles those are is only known to the connections ;; themselves, so the whole team is announced and the watcher ;; re-checks each of them. - (ntf/notify-team-permissions-changed - cfg (map :id (teams/get-team-members conn (:id team)))) + (ntf/notify-team-permissions-changed cfg (:id team)) (rph/with-meta (rph/wrap) {::audit/props {:project-id (:project-id file) diff --git a/backend/src/app/rpc/commands/management.clj b/backend/src/app/rpc/commands/management.clj index 5812e5ef26..c62a42298f 100644 --- a/backend/src/app/rpc/commands/management.clj +++ b/backend/src/app/rpc/commands/management.clj @@ -256,17 +256,15 @@ ;; --- COMMAND: Move file (defn- notify-team-access-changed - "Announces a possible read-access change to every member of `team-ids`. + "Announces a possible read-access change for `team-ids`. - Moving or deleting content can take it out of the reach of a whole - team, and the only place that knows who was relying on it is the set of - open connections. Announcing the whole team is cheap because the - watcher re-checks permissions and closes nothing for members who keep - their access." - [cfg conn team-ids] + Moving content can take it out of the reach of a whole team, and the + only place that knows who was relying on it is the set of open + connections. One event per team: the watcher resolves the members + itself, so a large team cannot overflow the bounded publish buffers." + [cfg team-ids] (doseq [team-id (into #{} (remove nil?) team-ids)] - (ntf/notify-team-permissions-changed - cfg (map :id (teams/get-team-members conn team-id))))) + (ntf/notify-team-permissions-changed cfg team-id))) (def sql:get-files "select id, features, project_id from file where id = ANY(?)") @@ -343,7 +341,7 @@ ;; the source teams are announced wholesale and the watcher re-checks ;; each member against the subscriptions they actually hold. (notify-team-access-changed - cfg conn + cfg (keep (partial perms/get-team-id-for-project conn) source)) nil)) @@ -417,7 +415,7 @@ ;; The project, and everything under it, leaves the source team. The ;; destination team only gains access, so it is not announced. - (notify-team-access-changed cfg conn [(:team-id project)]) + (notify-team-access-changed cfg [(:team-id project)]) nil)) diff --git a/backend/src/app/rpc/commands/projects.clj b/backend/src/app/rpc/commands/projects.clj index ab4959631b..db740c731a 100644 --- a/backend/src/app/rpc/commands/projects.clj +++ b/backend/src/app/rpc/commands/projects.clj @@ -323,8 +323,7 @@ ;; Deleting the project ends read access for whoever had one of its ;; files open, so the whole team is announced and the watcher ;; re-checks each member against the subscriptions they hold. - (ntf/notify-team-permissions-changed - cfg (map :id (teams/get-team-members conn (:id team)))) + (ntf/notify-team-permissions-changed cfg (:id team)) (rph/with-meta (rph/wrap) {::audit/props {:team-id (:team-id project) diff --git a/backend/src/app/rpc/commands/teams.clj b/backend/src/app/rpc/commands/teams.clj index a87d45ec18..90dd405890 100644 --- a/backend/src/app/rpc/commands/teams.clj +++ b/backend/src/app/rpc/commands/teams.clj @@ -894,6 +894,11 @@ ::jobs/params {:object :team :deleted-at (:deleted-at team) :id team-id}}) + + ;; The team is soft-deleted, so every subscription its members hold + ;; loses its authorization at once. One event: the watcher resolves + ;; the members itself. + (ntf/notify-team-permissions-changed cfg team-id) team))) (def ^:private schema:delete-team diff --git a/backend/src/app/rpc/notifications.clj b/backend/src/app/rpc/notifications.clj index b93c846ede..b3ffd1be65 100644 --- a/backend/src/app/rpc/notifications.clj +++ b/backend/src/app/rpc/notifications.clj @@ -6,6 +6,8 @@ (ns app.rpc.notifications (:require + [app.common.uuid :as uuid] + [app.db :as db] [app.msgbus :as mbus])) (defn notify-team-change @@ -65,36 +67,52 @@ (def internal-revocation-topic "Message-bus topic carrying backend-internal revocation events. - Deliberately not the `uuid/zero` system topic: that one is piped - straight into the client output channel, so anything published there - would reach every connected client. + A fixed uuid that is never a real resource id, so it cannot collide + with a file, team or profile topic — and deliberately not the + `uuid/zero` system topic, which is piped straight into the client + output channel, so anything published there would reach every + connected client. No client can subscribe here. `:subscribe-file` and `:subscribe-team` only accept ids that first pass a permission check, so a client can never make the bus deliver this topic to it." - "internal:subscription-revocation") + (uuid/uuid "ffffffff-ffff-ffff-ffff-ffffffffffff")) (defn notify-permissions-changed "Asks every backend to re-verify the websocket subscriptions currently held by `profile-id`. - Fire-and-forget: `mbus/pub!` only enqueues, so a subscription can be - revoked before the announcement reaches the instance that owns it. That - errs on the safe side (cutting a still-live subscription), and the - client re-subscribes on its own." + Deferred past the commit: the watcher decides from a fresh permission + check on another connection, so announcing before the transaction + commits would let it see the old state and close nothing. Outside a + transaction the announcement goes out immediately. + + The watcher always decides from a fresh permission check, so an extra + announcement is safe: it closes nothing whose access still holds." [cfg profile-id] - (let [msgbus (::mbus/msgbus cfg)] - (mbus/pub! msgbus - :topic internal-revocation-topic - :message {:type :profile-permissions-changed - :profile-id profile-id}))) + (db/after-commit + (fn [] + (let [msgbus (::mbus/msgbus cfg)] + (mbus/pub! msgbus + :topic internal-revocation-topic + :message {:type :profile-permissions-changed + :profile-id profile-id}))))) (defn notify-team-permissions-changed - "Announces a possible access change to every member of `profiles`. + "Asks every backend to re-verify the subscriptions of every member of + `team-id`. Used when a change can move a resource out of the reach of a whole - team (moving or deleting files and projects). We cannot know which - members actually relied on it, and announcing too many is safe: the - watcher re-checks permissions and only closes what genuinely failed." - [cfg profiles] - (run! (partial notify-permissions-changed cfg) profiles)) + team (moving, deleting or soft-deleting files, projects and teams). One + event per team: the watcher resolves the members itself, so a large + team cannot overflow the bounded publish buffers on the way out. + + Like `notify-permissions-changed`, deferred past the commit." + [cfg team-id] + (db/after-commit + (fn [] + (let [msgbus (::mbus/msgbus cfg)] + (mbus/pub! msgbus + :topic internal-revocation-topic + :message {:type :team-permissions-changed + :team-id team-id}))))) diff --git a/backend/test/backend_tests/binfile_test.clj b/backend/test/backend_tests/binfile_test.clj index 40baeb63ee..71ce1e630c 100644 --- a/backend/test/backend_tests/binfile_test.clj +++ b/backend/test/backend_tests/binfile_test.clj @@ -2495,3 +2495,45 @@ (t/is (= (str (first result) "/" thumb-page-id "/" thumb-frame-id "/" thumb-tag) (:object-id thumb))) (t/is (some? (:media-id thumb))))))) + +(t/deftest file-permissions-ignore-soft-deleted + ;; sql:file-permissions must not grant anything through a file, + ;; project or team that is soft-deleted (GHSA-m53j-2766-6jqw: the + ;; websocket re-check relies on this query, and the subscribe handler + ;; additionally refuses deleted files via check-file-exists). + (let [owner (th/create-profile* 1 {:is-active true}) + editor (th/create-profile* 2 {:is-active true}) + team (th/create-team* 1 {:profile-id (:id owner)})] + (th/create-team-role* {:team-id (:id team) + :profile-id (:id editor) + :role :editor}) + (let [project (th/create-project* 1 {:profile-id (:id editor) + :team-id (:id team)}) + file (th/create-file* 1 {:profile-id (:id editor) + :project-id (:id project)}) + perms? #(some? (bfc/get-file-permissions th/*system* (:id editor) (:id file))) + undelete (fn [table id] + (db/update! th/*system* table {:deleted-at nil} {:id id})) + soft-delete (fn [table id] + (db/update! th/*system* table {:deleted-at (ct/now)} {:id id}))] + + (t/testing "a live member can read the live file" + (t/is (perms?))) + + (t/testing "a soft-deleted file grants nothing" + (soft-delete :file (:id file)) + (t/is (nil? (bfc/get-file-permissions th/*system* (:id editor) (:id file)))) + (undelete :file (:id file)) + (t/is (perms?))) + + (t/testing "a soft-deleted project grants nothing" + (soft-delete :project (:id project)) + (t/is (nil? (bfc/get-file-permissions th/*system* (:id editor) (:id file)))) + (undelete :project (:id project)) + (t/is (perms?))) + + (t/testing "a soft-deleted team grants nothing" + (soft-delete :team (:id team)) + (t/is (nil? (bfc/get-file-permissions th/*system* (:id editor) (:id file)))) + (undelete :team (:id team)) + (t/is (perms?)))))) diff --git a/backend/test/backend_tests/http_websocket_test.clj b/backend/test/backend_tests/http_websocket_test.clj index 68e4b898ff..d8750d9748 100644 --- a/backend/test/backend_tests/http_websocket_test.clj +++ b/backend/test/backend_tests/http_websocket_test.clj @@ -22,8 +22,17 @@ [clojure.test :as t] [promesa.exec.csp :as sp])) +(defn- with-clean-registry + "Runs `f` against an empty connection registry, restoring the previous + contents afterwards so tests cannot leak connections into each other." + [f] + (let [saved @ws/state] + (reset! ws/state {:connections {} :by-profile {}}) + (try (f) + (finally (reset! ws/state saved))))) + (t/use-fixtures :once th/state-init) -(t/use-fixtures :each th/database-reset) +(t/use-fixtures :each th/database-reset with-clean-registry) (defn make-wsp [profile-id state output-ch] @@ -235,15 +244,6 @@ ;; --- CONNECTION REGISTRY -(defn- with-clean-registry - "Runs `f` against an empty connection registry, restoring the previous - contents afterwards so tests cannot leak connections into each other." - [f] - (let [saved @ws/state] - (reset! ws/state {:connections {} :by-profile {}}) - (try (f) - (finally (reset! ws/state saved))))) - (t/deftest registry-indexes-connections-by-profile (let [profile-a (uuid/next) profile-b (uuid/next) @@ -251,38 +251,36 @@ conn-b (uuid/next) conn-c (uuid/next)] - (with-clean-registry - (fn [] - (t/testing "a profile with no connections has an empty index" - (t/is (= #{} (ws/connections-for-profile profile-a)))) + (t/testing "a profile with no connections has an empty index" + (t/is (= #{} (ws/connections-for-profile profile-a)))) - (ws/register-connection conn-a {::ws/profile-id profile-a}) - (ws/register-connection conn-b {::ws/profile-id profile-a}) - (ws/register-connection conn-c {::ws/profile-id profile-b}) + (ws/register-connection conn-a {::ws/profile-id profile-a}) + (ws/register-connection conn-b {::ws/profile-id profile-a}) + (ws/register-connection conn-c {::ws/profile-id profile-b}) - (t/testing "connections are indexed under their profile" - (t/is (= #{conn-a conn-b} (ws/connections-for-profile profile-a))) - (t/is (= #{conn-c} (ws/connections-for-profile profile-b)))) + (t/testing "connections are indexed under their profile" + (t/is (= #{conn-a conn-b} (ws/connections-for-profile profile-a))) + (t/is (= #{conn-c} (ws/connections-for-profile profile-b)))) - (t/testing "connections are retrievable by id" - (t/is (= profile-a (::ws/profile-id (ws/get-connection conn-a)))) - (t/is (nil? (ws/get-connection (uuid/next))))) + (t/testing "connections are retrievable by id" + (t/is (= profile-a (::ws/profile-id (ws/get-connection conn-a)))) + (t/is (nil? (ws/get-connection (uuid/next))))) - (t/testing "unregistering removes it from both the map and the index" - (ws/unregister-connection conn-b) - (t/is (= #{conn-a} (ws/connections-for-profile profile-a))) - (t/is (nil? (ws/get-connection conn-b)))) + (t/testing "unregistering removes it from both the map and the index" + (ws/unregister-connection conn-b) + (t/is (= #{conn-a} (ws/connections-for-profile profile-a))) + (t/is (nil? (ws/get-connection conn-b)))) - (t/testing "a profile whose last connection left is dropped from the index" - (ws/unregister-connection conn-a) - (t/is (= #{} (ws/connections-for-profile profile-a))) - (t/is (not (contains? (:by-profile @ws/state) profile-a))) - (t/is (= #{conn-c} (ws/connections-for-profile profile-b)))) + (t/testing "a profile whose last connection left is dropped from the index" + (ws/unregister-connection conn-a) + (t/is (= #{} (ws/connections-for-profile profile-a))) + (t/is (not (contains? (:by-profile @ws/state) profile-a))) + (t/is (= #{conn-c} (ws/connections-for-profile profile-b)))) - (t/testing "unregistering an unknown connection is a no-op" - (ws/unregister-connection (uuid/next)) - (t/is (= #{conn-c} (ws/connections-for-profile profile-b))) - (t/is (= conn-c (-> (:connections @ws/state) keys first)))))))) + (t/testing "unregistering an unknown connection is a no-op" + (ws/unregister-connection (uuid/next)) + (t/is (= #{conn-c} (ws/connections-for-profile profile-b))) + (t/is (= conn-c (-> (:connections @ws/state) keys first)))))) (t/deftest close-file-subscription-drops-only-the-matching-file (let [profile (th/create-profile* 1 {:is-active true}) @@ -339,32 +337,29 @@ :revn 1 :changes [{:type :add :id (uuid/next)}]}] - (with-clean-registry - (fn [] - (let [conn (-> (open-registered-connection (:id editor)) - (subscribe :subscribe-file {:file-id file-id})) - out (:output conn)] + (let [conn (-> (open-registered-connection (:id editor)) + (subscribe :subscribe-file {:file-id file-id})) + out (:output conn)] - (t/testing "member receives file changes before revocation" - (publish file-id change) - (t/is (some? (poll-msg! out :file-change - revocation-timeout-ms)))) + (t/testing "member receives file changes before revocation" + (publish file-id change) + (t/is (some? (poll-msg! out :file-change + revocation-timeout-ms)))) - (t/testing "owner removes the member from the team" - (let [result (th/command! {::th/type :delete-team-member - ::rpc/profile-id (:id owner) - :team-id (:id team) - :member-id (:id editor)})] - (t/is (th/success? result)))) + (t/testing "owner removes the member from the team" + (let [result (th/command! {::th/type :delete-team-member + ::rpc/profile-id (:id owner) + :team-id (:id team) + :member-id (:id editor)})] + (t/is (th/success? result)))) - (t/testing "the mutation alone drops the subscription" - (t/is (wait-until #(nil? (subscribed-file conn)) 3000))) - - (t/testing "revoked member stops receiving file changes" - (publish file-id (assoc change :revn 2)) - (t/is (nil? (poll-msg! out :file-change - revocation-timeout-ms)))))))))) + (t/testing "the mutation alone drops the subscription" + (t/is (wait-until #(nil? (subscribed-file conn)) 3000))) + (t/testing "revoked member stops receiving file changes" + (publish file-id (assoc change :revn 2)) + (t/is (nil? (poll-msg! out :file-change + revocation-timeout-ms)))))))) (t/deftest revoked-member-stops-receiving-library-changes (let [owner (th/create-profile* 1 {:is-active true}) editor (th/create-profile* 2 {:is-active true}) @@ -381,31 +376,28 @@ :revn 1 :changes [{:type :add :id (uuid/next)}]}] - (with-clean-registry - (fn [] - (let [conn (-> (open-registered-connection (:id editor)) - (subscribe :subscribe-team {:team-id team-id})) - out (:output conn)] + (let [conn (-> (open-registered-connection (:id editor)) + (subscribe :subscribe-team {:team-id team-id})) + out (:output conn)] - (t/testing "member receives library changes before revocation" - (publish team-id change) - (t/is (some? (poll-msg! out :library-change - revocation-timeout-ms)))) + (t/testing "member receives library changes before revocation" + (publish team-id change) + (t/is (some? (poll-msg! out :library-change + revocation-timeout-ms)))) - (t/testing "member leaves the team" - (let [result (th/command! {::th/type :leave-team - ::rpc/profile-id (:id editor) - :id team-id})] - (t/is (th/success? result)))) + (t/testing "member leaves the team" + (let [result (th/command! {::th/type :leave-team + ::rpc/profile-id (:id editor) + :id team-id})] + (t/is (th/success? result)))) - (t/testing "leaving drops the team subscription" - (t/is (wait-until #(nil? (subscribed-team conn)) 3000))) - - (t/testing "revoked member stops receiving library changes" - (publish team-id (assoc change :revn 2)) - (t/is (nil? (poll-msg! out :library-change - revocation-timeout-ms)))))))))) + (t/testing "leaving drops the team subscription" + (t/is (wait-until #(nil? (subscribed-team conn)) 3000))) + (t/testing "revoked member stops receiving library changes" + (publish team-id (assoc change :revn 2)) + (t/is (nil? (poll-msg! out :library-change + revocation-timeout-ms)))))))) (t/deftest role-downgrade-keeps-revocation-irrelevant "A role change that does not remove read access must not be treated as a revocation: `:viewer` can still read, so the subscription stays." @@ -463,59 +455,55 @@ (t/deftest revocation-closes-subscriptions-that-are-no-longer-authorized (let [{:keys [owner editor team file]} (editor-in-team! 1 2)] - (with-clean-registry - (fn [] - (let [conn (-> (open-registered-connection (:id editor)) - (subscribe :subscribe-file {:file-id (:id file)}) - (subscribe :subscribe-team {:team-id (:id team)}))] + (let [conn (-> (open-registered-connection (:id editor)) + (subscribe :subscribe-file {:file-id (:id file)}) + (subscribe :subscribe-team {:team-id (:id team)}))] - (t/testing "both subscriptions are open before the revocation" - (t/is (= (:id file) (subscribed-file conn))) - (t/is (= (:id team) (subscribed-team conn)))) + (t/testing "both subscriptions are open before the revocation" + (t/is (= (:id file) (subscribed-file conn))) + (t/is (= (:id team) (subscribed-team conn)))) - (t/testing "the member is removed from the team" - (let [result (th/command! {::th/type :delete-team-member - ::rpc/profile-id (:id owner) - :team-id (:id team) - :member-id (:id editor)})] - (t/is (th/success? result)))) + (t/testing "the member is removed from the team" + (let [result (th/command! {::th/type :delete-team-member + ::rpc/profile-id (:id owner) + :team-id (:id team) + :member-id (:id editor)})] + (t/is (th/success? result)))) - (t/testing "revalidation closes the file and the team subscription" - (t/is (= 2 (ws/revalidate-profile-subscriptions - th/*system* (:id editor))))) + (t/testing "revalidation closes the file and the team subscription" + (t/is (= 2 (ws/revalidate-profile-subscriptions + th/*system* (:id editor))))) - (t/testing "nothing is left subscribed to relay content" - (t/is (nil? (subscribed-file conn))) - (t/is (nil? (subscribed-team conn)))) + (t/testing "nothing is left subscribed to relay content" + (t/is (nil? (subscribed-file conn))) + (t/is (nil? (subscribed-team conn)))) - (t/testing "revalidating again is a no-op" - (t/is (zero? (ws/revalidate-profile-subscriptions - th/*system* (:id editor)))))))))) + (t/testing "revalidating again is a no-op" + (t/is (zero? (ws/revalidate-profile-subscriptions + th/*system* (:id editor)))))))) (t/deftest revocation-keeps-subscriptions-that-are-still-authorized (let [{:keys [owner editor team file]} (editor-in-team! 1 2)] - (with-clean-registry - (fn [] - (let [conn (-> (open-registered-connection (:id editor)) - (subscribe :subscribe-file {:file-id (:id file)}) - (subscribe :subscribe-team {:team-id (:id team)}))] + (let [conn (-> (open-registered-connection (:id editor)) + (subscribe :subscribe-file {:file-id (:id file)}) + (subscribe :subscribe-team {:team-id (:id team)}))] - (t/testing "the editor is downgraded to viewer, which can still read" - (let [result (th/command! {::th/type :update-team-member-role - ::rpc/profile-id (:id owner) - :team-id (:id team) - :member-id (:id editor) - :role :viewer})] - (t/is (th/success? result)))) + (t/testing "the editor is downgraded to viewer, which can still read" + (let [result (th/command! {::th/type :update-team-member-role + ::rpc/profile-id (:id owner) + :team-id (:id team) + :member-id (:id editor) + :role :viewer})] + (t/is (th/success? result)))) - (t/testing "revalidation closes nothing" - (t/is (zero? (ws/revalidate-profile-subscriptions - th/*system* (:id editor))))) + (t/testing "revalidation closes nothing" + (t/is (zero? (ws/revalidate-profile-subscriptions + th/*system* (:id editor))))) - (t/testing "both subscriptions survive" - (t/is (= (:id file) (subscribed-file conn))) - (t/is (= (:id team) (subscribed-team conn))))))))) + (t/testing "both subscriptions survive" + (t/is (= (:id file) (subscribed-file conn))) + (t/is (= (:id team) (subscribed-team conn))))))) (t/deftest revocation-leaves-other-profiles-alone (let [{:keys [owner editor team file]} (editor-in-team! 1 2) @@ -529,50 +517,46 @@ :project-id (:default-project-id other-editor)})] - (with-clean-registry - (fn [] - (let [conn (-> (open-registered-connection (:id editor)) - (subscribe :subscribe-file {:file-id (:id file)})) - bystander (-> (open-registered-connection (:id other-editor)) - (subscribe :subscribe-file {:file-id (:id other-file)}))] + (let [conn (-> (open-registered-connection (:id editor)) + (subscribe :subscribe-file {:file-id (:id file)})) + bystander (-> (open-registered-connection (:id other-editor)) + (subscribe :subscribe-file {:file-id (:id other-file)}))] - (th/command! {::th/type :delete-team-member - ::rpc/profile-id (:id owner) - :team-id (:id team) - :member-id (:id editor)}) + (th/command! {::th/type :delete-team-member + ::rpc/profile-id (:id owner) + :team-id (:id team) + :member-id (:id editor)}) - (ws/revalidate-profile-subscriptions th/*system* (:id editor)) + (ws/revalidate-profile-subscriptions th/*system* (:id editor)) - (t/testing "the revoked member's subscription is closed" - (t/is (nil? (subscribed-file conn)))) + (t/testing "the revoked member's subscription is closed" + (t/is (nil? (subscribed-file conn)))) - (t/testing "a member of another team keeps their subscription" - (t/is (= (:id other-file) (subscribed-file bystander)))) + (t/testing "a member of another team keeps their subscription" + (t/is (= (:id other-file) (subscribed-file bystander)))) - (t/testing "revalidating a profile with no connections is safe" - (t/is (zero? (ws/revalidate-profile-subscriptions - th/*system* (uuid/next))))))))))) + (t/testing "revalidating a profile with no connections is safe" + (t/is (zero? (ws/revalidate-profile-subscriptions + th/*system* (uuid/next))))))))) (t/deftest watcher-consumes-revocation-events-from-the-bus (let [{:keys [owner editor team file]} (editor-in-team! 1 2)] - (with-clean-registry - (fn [] - (let [conn (-> (open-registered-connection (:id editor)) - (subscribe :subscribe-file {:file-id (:id file)}))] + (let [conn (-> (open-registered-connection (:id editor)) + (subscribe :subscribe-file {:file-id (:id file)}))] - (th/command! {::th/type :delete-team-member - ::rpc/profile-id (:id owner) - :team-id (:id team) - :member-id (:id editor)}) + (th/command! {::th/type :delete-team-member + ::rpc/profile-id (:id owner) + :team-id (:id team) + :member-id (:id editor)}) - (t/testing "the event is not applied until it is announced" - (t/is (= (:id file) (subscribed-file conn)))) + (t/testing "the event is not applied until it is announced" + (t/is (= (:id file) (subscribed-file conn)))) - (ntf/notify-permissions-changed th/*system* (:id editor)) + (ntf/notify-permissions-changed th/*system* (:id editor)) - (t/testing "the watcher closes the subscription off the bus" - (t/is (wait-until #(nil? (subscribed-file conn)) 3000)))))))) + (t/testing "the watcher closes the subscription off the bus" + (t/is (wait-until #(nil? (subscribed-file conn)) 3000)))))) (t/deftest internal-revocation-events-are-not-delivered-to-clients (let [profile (th/create-profile* 1 {:is-active true}) @@ -605,14 +589,14 @@ (defn- captured-events "Runs `f` collecting the revocation events published on the internal - topic." + topic, while letting every other publish reach the real bus untouched." [f] (let [events (atom []) real mbus/pub!] - (with-redefs [mbus/pub! (fn [bus & {:keys [topic message]}] + (with-redefs [mbus/pub! (fn [bus & {:keys [topic message] :as opts}] (when (= ntf/internal-revocation-topic topic) (swap! events conj message)) - (apply real bus topic message))] + (apply real bus (mapcat identity opts)))] (f)) @events)) @@ -658,13 +642,11 @@ (t/is (some? (event-for events (:id editor))))) (t/testing "a viewer can still read, so the subscription survives" - (with-clean-registry - (fn [] - (let [conn (-> (open-registered-connection (:id editor)) - (subscribe :subscribe-file {:file-id (:id file)}))] - (t/is (zero? (ws/revalidate-profile-subscriptions - th/*system* (:id editor)))) - (t/is (some? (subscribed-file conn)))))))))) + (let [conn (-> (open-registered-connection (:id editor)) + (subscribe :subscribe-file {:file-id (:id file)}))] + (t/is (zero? (ws/revalidate-profile-subscriptions + th/*system* (:id editor)))) + (t/is (some? (subscribed-file conn)))))))) (t/deftest deleting-a-file-announces-to-the-whole-team (let [owner (th/create-profile* 1 {:is-active true}) @@ -682,11 +664,61 @@ ::rpc/profile-id (:id owner) :id (:id file)}))] - ;; Which profiles had the file open is only known to their own - ;; connections, so the team is announced wholesale and the watcher - ;; re-checks each member. - (t/is (some? (event-for events (:id editor)))) - (t/is (some? (event-for events (:id owner))))))) + ;; One event per team: the watcher resolves the members itself, so + ;; a large team cannot overflow the bounded publish buffers. + (t/is (= [{:type :team-permissions-changed + :team-id (:id team)}] + events))))) + +(t/deftest deleting-a-team-announces-once + (let [owner (th/create-profile* 1 {:is-active true}) + editor (th/create-profile* 2 {:is-active true}) + team (th/create-team* 1 {:profile-id (:id owner)})] + (th/create-team-role* {:team-id (:id team) + :profile-id (:id editor) + :role :editor}) + (let [events (captured-events + #(th/command! {::th/type :delete-team + ::rpc/profile-id (:id owner) + :id (:id team)}))] + (t/is (= [{:type :team-permissions-changed + :team-id (:id team)}] + events))))) + +(t/deftest team-event-closes-every-member-subscription + (let [{:keys [owner editor team file]} (editor-in-team! 1 2) + other (th/create-profile* 3 {:is-active true})] + (th/create-team-role* {:team-id (:id team) + :profile-id (:id other) + :role :editor}) + (let [other-project (th/create-project* 3 {:profile-id (:id other) + :team-id (:id team)}) + other-file (th/create-file* 3 {:profile-id (:id other) + :project-id (:id other-project)})] + (let [conn (-> (open-registered-connection (:id editor)) + (subscribe :subscribe-file {:file-id (:id file)})) + conn2 (-> (open-registered-connection (:id other)) + (subscribe :subscribe-file {:file-id (:id other-file)}))] + + (th/command! {::th/type :delete-team + ::rpc/profile-id (:id owner) + :id (:id team)}) + + (t/testing "the announcement reaches the watcher off the bus" + (ntf/notify-team-permissions-changed th/*system* (:id team)) + (t/is (wait-until #(nil? (subscribed-file conn)) 3000)) + (t/is (wait-until #(nil? (subscribed-file conn2)) 3000))))))) + +(t/deftest rolled-back-revocation-announces-nothing + (let [editor (th/create-profile* 2 {:is-active true}) + events (captured-events + #(try + (db/tx-run! th/*system* + (fn [_cfg] + (ntf/notify-permissions-changed th/*system* (:id editor)) + (throw (ex-info "boom" {})))) + (catch clojure.lang.ExceptionInfo _)))] + (t/is (empty? events)))) (t/deftest leaving-a-team-announces-and-tells-only-the-leaver (let [owner (th/create-profile* 1 {:is-active true}) @@ -754,8 +786,7 @@ (t/deftest revalidation-interval-defaults-to-five-minutes (t/is (= (ct/duration {:minutes 5}) ws/default-revalidation-interval)) - (t/testing "the default is a code constant, not a config.clj default" - (t/is (not (contains? cf/default :subscription-revalidation-interval))))) + (t/is (= (ct/duration {:minutes 5}) (ws/revalidation-interval)))) (t/deftest revalidation-is-configurable (with-redefs [cf/get (fn [k default] (if (= :subscription-revalidation-interval k) @@ -764,37 +795,36 @@ (t/is (= (ct/duration {:seconds 30}) (ws/revalidation-interval))))) (t/deftest interval-recheck-drops-a-subscription-without-any-announcement - (let [{:keys [owner editor team file]} (editor-in-team! 1 2)] + (let [{:keys [editor team file]} (editor-in-team! 1 2)] (with-redefs [cf/get (fn [k default] (if (= :subscription-revalidation-interval k) (ct/duration {:millis 150}) default))] - (with-clean-registry - (fn [] - (let [conn (-> (open-registered-connection (:id editor)) - (subscribe :subscribe-file {:file-id (:id file)})) - out (:output conn)] + (let [conn (-> (open-registered-connection (:id editor)) + (subscribe :subscribe-file {:file-id (:id file)})) + out (:output conn)] - (t/testing "the subscription is live at first" - (t/is (some? (subscribed-file conn)))) + (t/testing "the subscription is live at first" + (t/is (some? (subscribed-file conn)))) - ;; No command runs and no announcement is published: access is - ;; taken away the way an external change would take it. - (th/command! {::th/type :delete-team-member - ::rpc/profile-id (:id owner) - :team-id (:id team) - :member-id (:id editor)}) + ;; Access is taken away the way an external change would take + ;; it: a direct row delete fires no command and publishes no + ;; announcement, so the interval is the only thing that can + ;; close the subscription. + (db/exec! th/*system* + ["delete from team_profile_rel where team_id = ? and profile_id = ?" + (:id team) (:id editor)]) - (t/testing "the interval re-check drops it anyway" - (t/is (wait-until #(nil? (subscribed-file conn)) 3000))) + (t/testing "the interval re-check drops it anyway" + (t/is (wait-until #(nil? (subscribed-file conn)) 3000))) - (t/testing "and nothing reaches the client afterwards" - (mbus/pub! (::mbus/msgbus th/*system*) - :topic (:id file) - :message {:type :file-change :file-id (:id file) :revn 9}) - (t/is (nil? (poll-msg! out :file-change - revocation-timeout-ms)))))))))) + (t/testing "and nothing reaches the client afterwards" + (mbus/pub! (::mbus/msgbus th/*system*) + :topic (:id file) + :message {:type :file-change :file-id (:id file) :revn 9}) + (t/is (nil? (poll-msg! out :file-change + revocation-timeout-ms)))))))) (t/deftest interval-recheck-keeps-a-subscription-that-still-holds (let [{:keys [owner editor team file]} (editor-in-team! 1 2)] @@ -803,32 +833,30 @@ (if (= :subscription-revalidation-interval k) (ct/duration {:millis 150}) default))] - (with-clean-registry - (fn [] - (let [conn (-> (open-registered-connection (:id editor)) - (subscribe :subscribe-file {:file-id (:id file)})) - out (:output conn)] + (let [conn (-> (open-registered-connection (:id editor)) + (subscribe :subscribe-file {:file-id (:id file)})) + out (:output conn)] - (t/testing "a downgrade to viewer still allows reading" - (t/is (th/success? (th/command! {::th/type :update-team-member-role - ::rpc/profile-id (:id owner) - :team-id (:id team) - :member-id (:id editor) - :role :viewer})))) + (t/testing "a downgrade to viewer still allows reading" + (t/is (th/success? (th/command! {::th/type :update-team-member-role + ::rpc/profile-id (:id owner) + :team-id (:id team) + :member-id (:id editor) + :role :viewer})))) - ;; Wait for at least two interval ticks, so a re-check that - ;; wrongly closed the subscription would have run. - (Thread/sleep 500) + ;; Wait for at least two interval ticks, so a re-check that + ;; wrongly closed the subscription would have run. + (Thread/sleep 500) - (t/testing "the subscription survives the interval" - (t/is (some? (subscribed-file conn)))) + (t/testing "the subscription survives the interval" + (t/is (some? (subscribed-file conn)))) - (t/testing "and traffic still flows" - (mbus/pub! (::mbus/msgbus th/*system*) - :topic (:id file) - :message {:type :file-change :file-id (:id file) :revn 3}) - (t/is (some? (poll-msg! out :file-change - revocation-timeout-ms)))))))))) + (t/testing "and traffic still flows" + (mbus/pub! (::mbus/msgbus th/*system*) + :topic (:id file) + :message {:type :file-change :file-id (:id file) :revn 3}) + (t/is (some? (poll-msg! out :file-change + revocation-timeout-ms)))))))) (t/deftest interval-recheck-covers-the-team-subscription-too (let [{:keys [editor team]} (editor-in-team! 1 2)] @@ -837,16 +865,240 @@ (if (= :subscription-revalidation-interval k) (ct/duration {:millis 150}) default))] - (with-clean-registry - (fn [] - (let [conn (-> (open-registered-connection (:id editor)) - (subscribe :subscribe-team {:team-id (:id team)}))] + (let [conn (-> (open-registered-connection (:id editor)) + (subscribe :subscribe-team {:team-id (:id team)}))] - (t/is (some? (subscribed-team conn))) + (t/is (some? (subscribed-team conn))) - (th/command! {::th/type :leave-team - ::rpc/profile-id (:id editor) - :id (:id team)}) + ;; Same external-change shape as above: no command, no + ;; announcement, only the interval. + (db/exec! th/*system* + ["delete from team_profile_rel where team_id = ? and profile_id = ?" + (:id team) (:id editor)]) - (t/testing "the team subscription is dropped by the interval" - (t/is (wait-until #(nil? (subscribed-team conn)) 3000))))))))) + (t/testing "the team subscription is dropped by the interval" + (t/is (wait-until #(nil? (subscribed-team conn)) 3000))))))) + +(t/deftest a-failing-check-keeps-the-subscription + (let [{:keys [editor file]} (editor-in-team! 1 2)] + + (with-redefs [cf/get (fn [k default] + (if (= :subscription-revalidation-interval k) + (ct/duration {:millis 150}) + default))] + (let [conn (-> (open-registered-connection (:id editor)) + (subscribe :subscribe-file {:file-id (:id file)}))] + + ;; Access still holds, but the check itself blows up. + (with-redefs [files/has-read-permissions? + (fn [& _] (throw (RuntimeException. "db down")))] + (Thread/sleep 500) + (t/testing "a failing check keeps the subscription" + (t/is (some? (subscribed-file conn))))))))) + +(t/deftest a-false-check-closes-the-subscription + (let [{:keys [editor file]} (editor-in-team! 1 2)] + + (with-redefs [cf/get (fn [k default] + (if (= :subscription-revalidation-interval k) + (ct/duration {:millis 150}) + default))] + (let [conn (-> (open-registered-connection (:id editor)) + (subscribe :subscribe-file {:file-id (:id file)}))] + + (with-redefs [files/has-read-permissions? (fn [& _] false)] + (t/testing "a check that returns false closes it" + (t/is (wait-until #(nil? (subscribed-file conn)) 3000)))))))) + +(t/deftest resubscribe-stops-previous-revalidation + (let [profile (th/create-profile* 1 {:is-active true}) + file1 (th/create-file* 1 {:profile-id (:id profile) + :project-id (:default-project-id profile)}) + file2 (th/create-file* 2 {:profile-id (:id profile) + :project-id (:default-project-id profile)})] + + (with-redefs [cf/get (fn [k default] + (if (= :subscription-revalidation-interval k) + (ct/duration {:millis 150}) + default))] + (let [counts (atom {}) + real files/has-read-permissions?] + (with-redefs [files/has-read-permissions? + (fn [cfg pid fid] + (swap! counts update fid (fnil inc 0)) + (real cfg pid fid))] + (let [conn (-> (open-registered-connection (:id profile)) + (subscribe :subscribe-file {:file-id (:id file1)}))] + + (t/testing "the first subscription is re-checked on the interval" + (t/is (wait-until #(pos? (get @counts (:id file1) 0)) 3000))) + + (t/testing "resubscribing moves the relay to the new file" + (subscribe conn :subscribe-file {:file-id (:id file2)}) + (t/is (wait-until #(>= (get @counts (:id file2) 0) 2) 3000))) + + (t/testing "the previous subscription stops revalidating" + (let [old (get @counts (:id file1) 0)] + (Thread/sleep 400) + (t/is (= old (get @counts (:id file1) 0))))))))))) + +;; --- EFFECT TESTS +;; +;; Announcing is not the same as revoking: these assert the subscription +;; actually drops and the content stops flowing. + +(t/deftest deleting-a-file-closes-open-subscriptions + (let [{:keys [owner editor file]} (editor-in-team! 1 2) + publish (publisher!)] + + (let [conn (-> (open-registered-connection (:id editor)) + (subscribe :subscribe-file {:file-id (:id file)})) + out (:output conn)] + + (th/command! {::th/type :delete-file + ::rpc/profile-id (:id owner) + :id (:id file)}) + + (t/testing "the file subscription drops without any manual publish" + (t/is (wait-until #(nil? (subscribed-file conn)) 3000))) + + (t/testing "no file content reaches the client afterwards" + (publish (:id file) {:type :file-change :file-id (:id file) :revn 9}) + (t/is (nil? (poll-msg! out :file-change revocation-timeout-ms))))))) + +(t/deftest deleting-a-project-closes-open-subscriptions + (let [{:keys [owner editor file]} (editor-in-team! 1 2) + project-id (:project-id file)] + + (let [conn (-> (open-registered-connection (:id editor)) + (subscribe :subscribe-file {:file-id (:id file)})) + out (:output conn)] + + (th/command! {::th/type :delete-project + ::rpc/profile-id (:id owner) + :id project-id}) + + (t/testing "the file subscription drops" + (t/is (wait-until #(nil? (subscribed-file conn)) 3000))) + + (t/testing "no file content reaches the client afterwards" + (mbus/pub! (::mbus/msgbus th/*system*) + :topic (:id file) + :message {:type :file-change :file-id (:id file) :revn 9}) + (t/is (nil? (poll-msg! out :file-change revocation-timeout-ms))))))) + +(t/deftest deleting-a-team-closes-open-subscriptions + (let [{:keys [owner editor team file]} (editor-in-team! 1 2)] + + (let [conn (-> (open-registered-connection (:id editor)) + (subscribe :subscribe-file {:file-id (:id file)}) + (subscribe :subscribe-team {:team-id (:id team)}))] + + (th/command! {::th/type :delete-team + ::rpc/profile-id (:id owner) + :id (:id team)}) + + (t/testing "both subscriptions drop" + (t/is (wait-until #(nil? (subscribed-file conn)) 3000)) + (t/is (wait-until #(nil? (subscribed-team conn)) 3000)))))) + +(t/deftest moving-a-file-closes-source-team-subscriptions + (let [owner (th/create-profile* 1 {:is-active true}) + editor (th/create-profile* 2 {:is-active true}) + team-a (th/create-team* 1 {:profile-id (:id owner)}) + team-b (th/create-team* 2 {:profile-id (:id owner)})] + (th/create-team-role* {:team-id (:id team-a) + :profile-id (:id editor) + :role :editor}) + (let [project-a (th/create-project* 1 {:profile-id (:id owner) + :team-id (:id team-a)}) + project-b (th/create-project* 2 {:profile-id (:id owner) + :team-id (:id team-b)}) + file (th/create-file* 1 {:profile-id (:id editor) + :project-id (:id project-a)}) + conn (-> (open-registered-connection (:id editor)) + (subscribe :subscribe-file {:file-id (:id file)}))] + + (th/command! {::th/type :move-files + ::rpc/profile-id (:id owner) + :ids #{(:id file)} + :project-id (:id project-b)}) + + (t/testing "a member left behind in the source team loses the subscription" + (t/is (wait-until #(nil? (subscribed-file conn)) 3000)))))) + +(t/deftest close-team-subscription-is-idempotent + (let [profile (th/create-profile* 1 {:is-active true}) + team (th/create-team* 1 {:profile-id (:id profile)}) + other (th/create-team* 2 {:profile-id (:id profile)}) + output (sp/chan :buf (sp/dropping-buffer 64)) + relay (sp/chan :buf (sp/dropping-buffer 64)) + ws-state (atom {::ws/team-subscription + {:team-id (:id team) + :channel relay + :topic (:id team)}}) + wsp (assoc (make-wsp (:id profile) ws-state output) + ::ws/state ws-state)] + + (t/testing "closing a team the connection is not subscribed to does nothing" + (with-redefs [mbus/purge! (fn [& _] (throw ::unexpected-purge))] + (t/is (nil? (ws/close-team-subscription th/*system* wsp (:id other)))) + (t/is (some? (::ws/team-subscription @ws-state))))) + + (t/testing "closing the subscribed team purges the bus and the relay stops" + (let [purged (atom [])] + (with-redefs [mbus/purge! (fn [_ chans] (swap! purged conj chans))] + (ws/close-team-subscription th/*system* wsp (:id team)) + (t/is (= 1 (count @purged))))) + (t/is (nil? (sp/poll! relay))) + (t/is (true? (sp/closed? relay)))) + + (t/testing "the subscription is forgotten, so closing again is a no-op" + (t/is (nil? (::ws/team-subscription @ws-state))) + (t/is (nil? (ws/close-team-subscription th/*system* wsp (:id team))))))) + +(t/deftest losing-a-team-keeps-the-other-file-subscription + (let [{:keys [owner editor team]} (editor-in-team! 1 2) + other-team (th/create-team* 3 {:profile-id (:id owner)})] + (th/create-team-role* {:team-id (:id other-team) + :profile-id (:id editor) + :role :editor}) + (let [other-project (th/create-project* 3 {:profile-id (:id editor) + :team-id (:id other-team)}) + other-file (th/create-file* 3 {:profile-id (:id editor) + :project-id (:id other-project)}) + conn (-> (open-registered-connection (:id editor)) + (subscribe :subscribe-team {:team-id (:id team)}) + (subscribe :subscribe-file {:file-id (:id other-file)}))] + + (th/command! {::th/type :delete-team-member + ::rpc/profile-id (:id owner) + :team-id (:id team) + :member-id (:id editor)}) + + (t/testing "the lost team subscription drops" + (t/is (wait-until #(nil? (subscribed-team conn)) 3000))) + + (t/testing "the file subscription on the other team survives" + (t/is (some? (subscribed-file conn))))))) + +(t/deftest organization-owner-keeps-read-only-access + (let [{:keys [owner editor team file]} (editor-in-team! 1 2)] + + (let [conn (-> (open-registered-connection (:id editor)) + (subscribe :subscribe-file {:file-id (:id file)}))] + + ;; The admin-console flag is off in tests, so Nitrate never grants + ;; this; stand in for an organization owner expelled from the team. + (with-redefs [nitrate/organization-owner-of-team? (fn [& _] true)] + (th/command! {::th/type :delete-team-member + ::rpc/profile-id (:id owner) + :team-id (:id team) + :member-id (:id editor)}) + + (t/testing "revalidation closes nothing for an organization owner" + (t/is (zero? (ws/revalidate-profile-subscriptions + th/*system* (:id editor))))) + + (t/testing "the read-only subscription survives" + (t/is (some? (subscribed-file conn))))))))