🐛 Announce access changes so subscriptions get revoked

The revocation watcher only helps if the mutations that can take
access away actually announce it. Seven of them now do, after the write
that changed access:

- delete-team-member and leave-team announce the affected member.
- update-team-member-role announces too. A role change never removes
  read on its own, but announcing lets the watcher decide from fresh
  data instead of assuming, which is what keeps a downgrade to viewer
  from being mistaken for a revocation.
- delete-file, delete-project, move-files and move-project announce to
  the whole team. Which profiles had the content open is only known to
  their own connections, and over-announcing is cheap: the watcher
  re-checks and closes nothing for members who keep access.

delete-share-link is deliberately excluded. 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, so there is no subscription to revoke.

leave-team also gains the notification it never had: it published
nothing at all on the leaver's own topic, so a session that left kept
its editor open with no idea, while a removed member was told.

The topic and the publish helper move to `app.rpc.notifications`, next
to the other bus notifications. The command namespaces cannot require
the watcher itself: `app.http.websocket` already requires these command
namespaces, so that would be a cycle. The watcher keeps the subscription
registry, which is the part it owns.

Closes #12155

AI-assisted-by: space-bunny-free
This commit is contained in:
Andrey Antukh 2026-10-07 10:12:53 +00:00
parent 6060a7140f
commit e6e6b20cbb
7 changed files with 350 additions and 87 deletions

View File

@ -22,6 +22,7 @@
[app.nitrate :as nitrate]
[app.rpc.commands.files :as files]
[app.rpc.commands.teams :as teams]
[app.rpc.notifications :as notifications]
[app.util.websocket :as ws]
[integrant.core :as ig]
[promesa.exec.csp :as sp]
@ -354,40 +355,15 @@
;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;
;; A subscription is authorized once, when it is opened, and that single
;; decision is then trusted for as long as the connection lives. When a
;; mutation can take access away, it announces it here and every backend
;; decision is then trusted for as long as the connection lives. The
;; mutation that changes access announces it on
;; `app.rpc.notifications/internal-revocation-topic`, and every backend
;; re-verifies the subscriptions it owns for that profile.
;;
;; The announcement travels over the message bus precisely because the
;; connection registry is local: the RPC that revokes access runs on
;; whichever instance received the request, which is usually not the one
;; holding the socket.
(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.
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")
(defn notify-permissions-changed
"Asks every backend to re-verify the subscriptions currently held by
`profile-id`, because something may have changed its access.
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."
[{:keys [::mbus/msgbus]} profile-id]
(mbus/pub! msgbus
:topic internal-revocation-topic
:message {:type :profile-permissions-changed
:profile-id profile-id}))
;; The announcement travels over the message bus precisely because this
;; registry is local: the RPC that revokes access runs on whichever
;; instance received the request, which is usually not the one holding
;; the socket.
(defn revalidate-profile-subscriptions
"Re-checks every subscription held by the connections of `profile-id`
@ -425,7 +401,9 @@
instance owns."
[{:keys [::mbus/msgbus] :as cfg}]
(let [ch (sp/chan :buf (sp/dropping-buffer 64))]
(mbus/sub! msgbus :topic internal-revocation-topic :chan ch)
(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)

View File

@ -41,6 +41,7 @@
[app.rpc.cond :as-alias cond]
[app.rpc.doc :as-alias doc]
[app.rpc.helpers :as rph]
[app.rpc.notifications :as ntf]
[app.rpc.permissions :as perms]
[app.util.blob :as blob]
[app.util.events :as events]
@ -1094,6 +1095,13 @@
:file-id id
:profile-id profile-id})
;; Deleting the file ends read access for whoever had it open, and
;; 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))))
(rph/with-meta (rph/wrap)
{::audit/props {:project-id (:project-id file)
:name (:name file)

View File

@ -25,6 +25,8 @@
[app.rpc.commands.projects :as proj]
[app.rpc.commands.teams :as teams]
[app.rpc.doc :as-alias doc]
[app.rpc.notifications :as ntf]
[app.rpc.permissions :as perms]
[app.setup :as-alias setup]
[app.setup.templates :as tmpl]
[app.storage.tmp :as tmp]
@ -253,6 +255,19 @@
;; --- COMMAND: Move file
(defn- notify-team-access-changed
"Announces a possible read-access change to every member of `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]
(doseq [team-id (into #{} (remove nil?) team-ids)]
(ntf/notify-team-permissions-changed
cfg (map :id (teams/get-team-members conn team-id)))))
(def sql:get-files
"select id, features, project_id from file where id = ANY(?)")
@ -323,6 +338,14 @@
(recur (rest project-ids)
(ct/plus modified-at 10))))
;; Moving files can take them out of the reach of the team they came
;; from. Only the open connections know who was relying on them, so
;; 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
(keep (partial perms/get-team-id-for-project conn) source))
nil))
(def ^:private
@ -392,6 +415,10 @@
;; delete possible broken relations on moved files
(db/exec-one! conn [sql:delete-broken-relations pids])
;; 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)])
nil))
(def ^:private

View File

@ -22,6 +22,7 @@
[app.rpc.commands.teams :as teams]
[app.rpc.doc :as-alias doc]
[app.rpc.helpers :as rph]
[app.rpc.notifications :as ntf]
[app.rpc.permissions :as perms]
[app.rpc.quotes :as quotes]
[app.util.services :as sv]))
@ -318,6 +319,13 @@
:profile-id profile-id
:project-id id)
project (delete-project cfg team id)]
;; 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))))
(rph/with-meta (rph/wrap)
{::audit/props {:team-id (:team-id project)
:name (:name project)

View File

@ -29,6 +29,7 @@
[app.rpc :as-alias rpc]
[app.rpc.commands.profile :as profile]
[app.rpc.doc :as-alias doc]
[app.rpc.notifications :as ntf]
[app.rpc.permissions :as perms]
[app.rpc.quotes :as quotes]
[app.setup :as-alias setup]
@ -767,9 +768,10 @@
:team-id team-id}))
(defn leave-team
[{:keys [::db/conn ::mbus/msgbus]} {:keys [profile-id id reassign-to]}]
[{:keys [::db/conn ::mbus/msgbus] :as cfg} {:keys [profile-id id reassign-to]}]
(let [perms (get-permissions conn profile-id id)
members (get-team-members conn id)]
members (get-team-members conn id)
team (db/get conn :team {:id id})]
(cond
;; we can only proceed if there are more members in the team
@ -818,6 +820,16 @@
(delete-team-rels conn profile-id id)
;; The leaver is told so, exactly as a removed member is, so the
;; session stops asking for content it can no longer read.
(mbus/pub! msgbus
:topic profile-id
:message {:type :team-membership-change
:change :removed
:team-id id
:team-name (:name team)})
(ntf/notify-permissions-changed cfg profile-id)
nil))
(def ^:private schema:leave-team
@ -900,7 +912,7 @@
;; --- Mutation: Team Update Role
(defn update-team-member-role
[{:keys [::db/conn ::mbus/msgbus]} {:keys [profile-id team-id member-id role] :as params}]
[{:keys [::db/conn ::mbus/msgbus] :as cfg} {:keys [profile-id team-id member-id role] :as params}]
;; We retrieve all team members instead of query the
;; database for a single member. This is just for
;; convenience, if this becomes a bottleneck or problematic,
@ -950,6 +962,12 @@
params
{:team-id team-id
:profile-id member-id})
;; A role change never removes read access on its own, but the open
;; subscriptions are re-checked rather than assumed: the watcher
;; decides from fresh data, which is also what keeps a downgrade to
;; viewer from being mistaken for a revocation.
(ntf/notify-permissions-changed cfg member-id)
nil)))
(def ^:private schema:update-team-member-role
@ -1017,6 +1035,11 @@
:team-id team-id
:team-name (:name team)}))
;; Whatever this member may still read through another path (an
;; organization owner keeps read-only access), the subscriptions they
;; already hold have to be re-checked, not trusted from here on.
(ntf/notify-permissions-changed cfg member-id)
nil))
;; --- Mutation: Update Team Photo

View File

@ -49,3 +49,52 @@
:topic organization-id
:message {:type :organization-change-sso
:organization-id organization-id})))
;; --- Subscription revocation
;; A websocket subscription is authorized once, when it is opened, and
;; that single decision is trusted for as long as the connection lives.
;; A mutation that can take access away announces it here, and every
;; backend re-verifies the subscriptions it owns for that profile.
;;
;; The announcement travels over the message bus precisely because the
;; connection registry is local to one backend instance, while the RPC
;; that revokes access runs on whichever instance received the request,
;; which is usually not the one holding the socket.
(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.
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")
(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."
[cfg profile-id]
(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`.
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))

View File

@ -14,6 +14,7 @@
[app.rpc :as-alias rpc]
[app.rpc.commands.files :as files]
[app.rpc.commands.teams :as teams]
[app.rpc.notifications :as ntf]
[app.util.websocket :as util-ws]
[backend-tests.helpers :as th]
[clojure.test :as t]
@ -180,26 +181,6 @@
(recur)))))))
(deref result (inc ms) nil)))
(defn- open-subscription!
"Runs `handler` for `profile-id` against a live system, returning the
output channel that receives whatever the subscription forwards."
[handler profile-id params]
(let [output-ch (sp/chan :buf (sp/dropping-buffer 64))
state (atom {})
wsp (make-wsp profile-id state output-ch)]
((get-method ws/handle-message handler) th/*system* wsp params)
output-ch))
(defn- wait-until
"Polls `pred` until it holds or `ms` elapses. Returns whether it held."
[pred ms]
(let [deadline (+ (System/currentTimeMillis) ms)]
(loop []
(cond
(pred) true
(>= (System/currentTimeMillis) deadline) false
:else (do (Thread/sleep 25) (recur))))))
(defn- open-registered-connection
"Registers a connection for `profile-id` in the real registry, so the
revocation watcher can find it, and returns the pieces a test needs to
@ -212,6 +193,28 @@
(ws/register-connection id wsp)
{:id id :wsp wsp :state ws-state :output output}))
(defn- open-subscription!
"Opens a registered subscription for `profile-id` and returns the
output channel that receives whatever it forwards.
The connection goes through the real registry on purpose: these tests
assert the end-to-end path, where a mutation announces the change and
the watcher closes the subscription."
[handler profile-id params]
(let [{:keys [wsp output]} (open-registered-connection profile-id)]
((get-method ws/handle-message handler) th/*system* wsp params)
output))
(defn- wait-until
"Polls `pred` until it holds or `ms` elapses. Returns whether it held."
[pred ms]
(let [deadline (+ (System/currentTimeMillis) ms)]
(loop []
(cond
(pred) true
(>= (System/currentTimeMillis) deadline) false
:else (do (Thread/sleep 25) (recur))))))
(defn- subscribe
[{:keys [wsp] :as conn} handler params]
((get-method ws/handle-message handler) th/*system* wsp params)
@ -332,24 +335,33 @@
change {:type :file-change
:file-id file-id
:revn 1
:changes [{:type :add :id (uuid/next)}]}
out (open-subscription! :subscribe-file (:id editor)
{:file-id file-id})]
:changes [{:type :add :id (uuid/next)}]}]
(t/testing "member receives file changes before revocation"
(publish file-id change)
(t/is (some? (poll-msg! out :file-change revocation-timeout-ms))))
(with-clean-registry
(fn []
(let [conn (-> (open-registered-connection (:id editor))
(subscribe :subscribe-file {:file-id file-id}))
out (:output conn)]
(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 "member receives file changes before revocation"
(publish file-id change)
(t/is (some? (poll-msg! out :file-change
revocation-timeout-ms))))
(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 "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/deftest revoked-member-stops-receiving-library-changes
(let [owner (th/create-profile* 1 {:is-active true})
@ -365,23 +377,32 @@
:team-id team-id
:file-id (uuid/next)
:revn 1
:changes [{:type :add :id (uuid/next)}]}
out (open-subscription! :subscribe-team (:id editor)
{:team-id team-id})]
:changes [{:type :add :id (uuid/next)}]}]
(t/testing "member receives library changes before revocation"
(publish team-id change)
(t/is (some? (poll-msg! out :library-change revocation-timeout-ms))))
(with-clean-registry
(fn []
(let [conn (-> (open-registered-connection (:id editor))
(subscribe :subscribe-team {:team-id team-id}))
out (:output conn)]
(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 receives library changes before revocation"
(publish team-id change)
(t/is (some? (poll-msg! out :library-change
revocation-timeout-ms))))
(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 "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/deftest role-downgrade-keeps-revocation-irrelevant
"A role change that does not remove read access must not be treated as
@ -546,7 +567,7 @@
(t/testing "the event is not applied until it is announced"
(t/is (= (:id file) (subscribed-file conn))))
(ws/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))))))))
@ -568,7 +589,156 @@
(t/is (some? (poll-msg! output :notification revocation-timeout-ms))))
(t/testing "an internal revocation event never reaches the client"
(ws/notify-permissions-changed th/*system* (:id profile))
(ntf/notify-permissions-changed th/*system* (:id profile))
(t/is (nil? (poll-msg! output :profile-permissions-changed
revocation-timeout-ms))))))
;; --- MUTATIONS THAT ANNOUNCE THE CHANGE
;;
;; The watcher only helps if the mutations that can revoke access
;; actually announce it. Each of these asserts the announcement reaches
;; the bus, and that a change which does not remove access announces
;; nothing harmful.
(defn- captured-events
"Runs `f` collecting the revocation events published on the internal
topic."
[f]
(let [events (atom [])
real mbus/pub!]
(with-redefs [mbus/pub! (fn [bus & {:keys [topic message]}]
(when (= ntf/internal-revocation-topic topic)
(swap! events conj message))
(apply real bus topic message))]
(f))
@events))
(defn- event-for
[events profile-id]
(first (filter #(= profile-id (:profile-id %)) events)))
(t/deftest removing-a-member-announces-the-revocation
(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-member
::rpc/profile-id (:id owner)
:team-id (:id team)
:member-id (:id editor)}))]
(t/is (= {:type :profile-permissions-changed
:profile-id (:id editor)}
(event-for events (:id editor)))))))
(t/deftest a-role-change-announces-and-still-closes-nothing
(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)})
events (captured-events
#(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 "the change is announced so subscriptions get re-checked"
(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))))))))))
(t/deftest deleting-a-file-announces-to-the-whole-team
(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)})
events (captured-events
#(th/command! {::th/type :delete-file
::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)))))))
(t/deftest leaving-a-team-announces-and-tells-only-the-leaver
(let [owner (th/create-profile* 1 {:is-active true})
editor (th/create-profile* 2 {:is-active true})
other (th/create-profile* 3 {: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})
(th/create-team-role* {:team-id (:id team)
:profile-id (:id other)
:role :editor})
(let [published (atom [])
real mbus/pub!]
(with-redefs [mbus/pub! (fn [bus & args]
(let [opts (apply hash-map args)]
(swap! published conj opts))
(apply real bus args))]
(th/command! {::th/type :leave-team
::rpc/profile-id (:id editor)
:id (:id team)}))
(t/testing "the leaver is told they were removed, so the client leaves"
(let [msg (->> @published
(filter #(= (:id editor) (:topic %)))
(map :message)
first)]
(t/is (= :team-membership-change (:type msg)))
(t/is (= :removed (:change msg)))
(t/is (= (:id team) (:team-id msg)))))
(t/testing "no other member is told they were removed"
(t/is (empty? (filter #(= (:id other) (:topic %)) @published)))))))
(t/deftest deleting-a-share-link-announces-nothing
;; A share-link visitor never passes the websocket subscribe check:
;; `check-read-permissions!` is called without a share-id, and that
;; arity only considers team membership. There is no subscription to
;; revoke, so announcing would be noise.
(let [owner (th/create-profile* 1 {:is-active true})
file (th/create-file* 1 {:profile-id (:id owner)
:project-id (:default-project-id owner)})]
(th/command! {::th/type :create-share-link
::rpc/profile-id (:id owner)
:file-id (:id file)
:who-comment "none"
:who-inspect "none"
:pages #{}})
(let [link-id (:id (db/exec-one! th/*system*
["select id from share_link where file_id = ?"
(:id file)]))
events (captured-events
#(th/command! {::th/type :delete-share-link
::rpc/profile-id (:id owner)
:id link-id}))]
(t/is (some? link-id))
(t/is (empty? events)))))