github/penpot
1
0
Fork 0
mirror of https://github.com/penpot/penpot.git synced 2026-10-30 06:20:32 +00:00
Code Issues Packages Projects Releases Wiki Activity

🐛 Revoke websocket subscriptions when access is revoked

Browse Source 
GHSA-m53j-2766-6jqw. A subscription is authorized once, when it is
opened, and that decision is trusted for the rest of the connection's
life. A member whose access is revoked keeps receiving the live file
diff for as long as their tab stays open.

Closing the connection from the mutation does not work: the registry
is local to one backend instance, while the RPC that revokes access
runs on whichever instance received the request. So the revocation
travels over the message bus instead, which every instance shares, and
each instance re-verifies the subscriptions it owns.

`revalidate-profile-subscriptions` looks the profile up in the new
index, then re-checks every file and team subscription with the same
`has-read-permissions?` predicates that `:subscribe-file` and
`:subscribe-team` authorize with. The event is only the trigger; the
fresh check is the decision, so the watcher keeps no second copy of the
permission rules. That is what makes a downgrade to viewer keep its
subscription, and keeps the read-only access of a non-member
organization owner.

`close-team-subscription` mirrors `close-file-subscription` for team
traffic. Both now forget the subscription instead of leaving a stale
one behind, which makes them idempotent and stops `:pointer-update`
from publishing to a topic the connection no longer relays.

The internal topic is separate from the `uuid/zero` system topic, which
is piped straight to the client. No client can reach it:
`:subscribe-file` and `:subscribe-team` only accept ids that already
passed a permission check.

Closes #12155

AI-assisted-by: space-bunny-free
This commit is contained in:
Andrey Antukh 2026-10-07 09:43:02 +00:00
parent 58178d2383
commit 6060a7140f
3 changed files with 328 additions and 8 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

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

@ -295,7 +295,25 @@
:profile-id profile-id})
(let [ch (:channel subs)]
(sp/close! ch)
(mbus/purge! msgbus [ch])))))
(mbus/purge! msgbus [ch])
(swap! state dissoc ::file-subscription)))))
(defn close-team-subscription
"Tears down the team subscription held by the connection `wsp`, if it
is subscribed to `team-id`.
Closing the channel is what stops the pipe into the client output, so
no further team or organization traffic reaches this connection.
Does nothing when the connection is subscribed to a different team,
so it is safe to call for a subscription that is already gone."
[{:keys [::mbus/msgbus]} {:keys [::ws/state]} team-id]
(let [subs (::team-subscription @state)]
(when (= (:team-id subs) team-id)
(let [ch (:channel subs)]
(sp/close! ch)
(mbus/purge! msgbus [ch])
(swap! state dissoc ::team-subscription)))))
(defmethod handle-message :unsubscribe-file
[cfg {:keys [::ws/id] :as wsp} {:keys [file-id] :as params}]
@ -331,6 +349,114 @@
:message message
:conn-id id))
;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;
;; SUBSCRIPTION REVOCATION
;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;
;; 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
;; 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}))
(defn revalidate-profile-subscriptions
"Re-checks every subscription held by the connections of `profile-id`
and closes the ones that are no longer authorized.
The event is the trigger, the fresh permission check is the decision.
Reusing the same predicates that `:subscribe-file` and
`:subscribe-team` authorize with means the watcher needs no second copy
of the permission rules: a downgrade to viewer keeps read access and
so keeps the subscription, and a non-member organization owner keeps
the read-only access it is entitled to.
Returns the number of subscriptions it closed."
[cfg profile-id]
(let [closed (volatile! 0)]
(doseq [id (connections-for-profile profile-id)]
(when-let [wsp (get-connection id)]
(let [subs (some-> wsp ::ws/state deref)
fsub (get subs ::file-subscription)
tsub (get subs ::team-subscription)]
(when (and fsub
(not (files/has-read-permissions? cfg profile-id
(:file-id fsub))))
(close-file-subscription cfg wsp (:file-id fsub))
(vswap! closed inc))
(when (and tsub
(not (teams/has-read-permissions? cfg profile-id
(:team-id tsub))))
(close-team-subscription cfg wsp (:team-id tsub))
(vswap! closed inc)))))
@closed))
(defn- watch-revocations
"Consumes revocation events and applies them to the connections this
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)
(sp/go-loop []
(when-let [{:keys [type profile-id]} (sp/take! ch)]
(when (= :profile-permissions-changed type)
(try
(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)))))
;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;
;; INTEGRANT
;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;
(def ^:private schema:revocation-watcher
[:map
::mbus/msgbus
::db/pool
:app.nitrate/client])
(defmethod ig/assert-key ::revocation-watcher
[_ params]
(assert (sm/valid? schema:revocation-watcher params)))
(defmethod ig/init-key ::revocation-watcher
[_ cfg]
(watch-revocations cfg))
;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;
;; HTTP HANDLER
;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;

5
backend/src/app/main.clj
View File

@ -498,6 +498,11 @@
::session/manager (ig/ref ::session/manager)
:app.nitrate/client (ig/ref :app.nitrate/client)}
::http.ws/revocation-watcher
{::db/pool (ig/ref ::db/pool)
::mbus/msgbus (ig/ref ::mbus/msgbus)
:app.nitrate/client (ig/ref :app.nitrate/client)}
:app.http.assets/routes
{::http.assets/path (cf/get :assets-path)
::http.assets/cache-max-age (ct/duration {:hours 24})

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

@ -190,6 +190,36 @@
((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
drive it and inspect what it did."
[profile-id]
(let [id (uuid/next)
output (sp/chan :buf (sp/dropping-buffer 64))
ws-state (atom {})
wsp (make-wsp profile-id ws-state output)]
(ws/register-connection id wsp)
{:id id :wsp wsp :state ws-state :output output}))
(defn- subscribe
[{:keys [wsp] :as conn} handler params]
((get-method ws/handle-message handler) th/*system* wsp params)
conn)
(defn- subscribed-file [conn] (-> @(:state conn) ::ws/file-subscription :file-id))
(defn- subscribed-team [conn] (-> @(:state conn) ::ws/team-subscription :team-id))
(defn- publisher!
"Returns a fn that publishes `message` on `topic` through the system
msgbus, the same path `send-notifications!` uses."
@ -256,9 +286,10 @@
other (th/create-file* 2 {:profile-id (:id profile)
:project-id (:default-project-id profile)})
output (sp/chan :buf (sp/dropping-buffer 64))
relay (sp/chan :buf (sp/dropping-buffer 64))
ws-state (atom {::ws/file-subscription
{:file-id (:id file)
:channel (sp/chan :buf (sp/dropping-buffer 64))
:channel relay
:topic (:id file)}})
wsp (assoc (make-wsp (:id profile) ws-state output)
::ws/state ws-state)]
@ -276,13 +307,13 @@
(t/is (= 1 (count @purged))))))
(t/testing "the relay channel is closed so the go-loop stops"
(let [sub-ch (-> @ws-state ::ws/file-subscription :channel)]
(t/is (some? sub-ch))
(t/is (nil? (sp/poll! sub-ch)))
(t/is (true? (sp/closed? sub-ch)))))
(t/is (nil? (sp/poll! relay)))
(t/is (true? (sp/closed? relay))))
(t/testing "closing twice is safe"
(t/is (nil? (ws/close-file-subscription th/*system* wsp (:id file)))))))
(t/testing "the subscription is forgotten, so closing again is a no-op"
(t/is (nil? (::ws/file-subscription @ws-state)))
(with-redefs [mbus/pub! (fn [& _] (throw ::unexpected-publish))]
(t/is (nil? (ws/close-file-subscription th/*system* wsp (:id file))))))))
(t/deftest revoked-member-stops-receiving-file-changes
(let [owner (th/create-profile* 1 {:is-active true})
@ -383,3 +414,161 @@
(t/testing "viewer keeps receiving file changes"
(publish file-id change)
(t/is (some? (poll-msg! out :file-change revocation-timeout-ms)))))))
;; --- REVOCATION WATCHER
;;
;; The registry is local to a backend instance while the message bus is
;; shared, so a revocation travels over the bus and each instance
;; re-verifies the connections it owns.
(defn- editor-in-team!
"Builds an owner, an editor with read access to `team`, and a project
and file inside it."
[owner-i editor-i]
(let [owner (th/create-profile* owner-i {:is-active true})
editor (th/create-profile* editor-i {:is-active true})
team (th/create-team* owner-i {:profile-id (:id owner)})]
(th/create-team-role* {:team-id (:id team)
:profile-id (:id editor)
:role :editor})
(let [project (th/create-project* owner-i {:profile-id (:id editor)
:team-id (:id team)})
file (th/create-file* owner-i {:profile-id (:id editor)
:project-id (:id project)})]
{:owner owner :editor editor :team team :file file})))
(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)}))]
(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 "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 "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)}))]
(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 "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)
other-owner (th/create-profile* 3 {:is-active true})
other-editor (th/create-profile* 4 {:is-active true})
other-team (th/create-team* 3 {:profile-id (:id other-owner)})]
(th/create-team-role* {:team-id (:id other-team)
:profile-id (:id other-editor)
:role :editor})
(let [other-file (th/create-file* 3 {:profile-id (:id other-editor)
: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)}))]
(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))
(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 "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)}))]
(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))))
(ws/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/deftest internal-revocation-events-are-not-delivered-to-clients
(let [profile (th/create-profile* 1 {:is-active true})
id (uuid/next)
output (sp/chan :buf (sp/dropping-buffer 64))
wsp (assoc (make-wsp (:id profile) (atom {}) output)
::ws/id id)]
(ws/register-connection id wsp)
((get-method ws/handle-message :open) th/*system* wsp nil)
(t/testing "the client does receive its own profile traffic"
(mbus/pub! (::mbus/msgbus th/*system*)
:topic (:id profile)
:message {:type :notification :text "hello"})
(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))
(t/is (nil? (poll-msg! output :profile-permissions-changed
revocation-timeout-ms))))))