github/penpot
1
0
Fork 0
mirror of https://github.com/penpot/penpot.git synced 2026-10-29 14:00:37 +00:00
Code Issues Packages Projects Releases Wiki Activity

🐛 Address review findings on websocket revocation

Browse Source 
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
This commit is contained in:
Andrey Antukh 2026-10-07 14:32:30 +00:00
parent 5a25da8d6d
commit 1f96cd2e4c
11 changed files with 662 additions and 321 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

15
.serena/memories/backend/websocket-subscriptions.md
View File

@ -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

16
backend/src/app/binfile/common.clj
View File

@ -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]

8
backend/src/app/config.clj
View File

@ -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]

119
backend/src/app/http/websocket.clj
View File

@ -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
;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;

3
backend/src/app/rpc/commands/files.clj
View File

@ -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)

20
backend/src/app/rpc/commands/management.clj
View File

@ -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))

3
backend/src/app/rpc/commands/projects.clj
View File

@ -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)

5
backend/src/app/rpc/commands/teams.clj
View File

@ -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

56
backend/src/app/rpc/notifications.clj
View File

@ -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})))))

42
backend/test/backend_tests/binfile_test.clj
View File

@ -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?))))))

696
backend/test/backend_tests/http_websocket_test.clj
View File

@ -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))))))))