From 6060a7140fc8da3a9f09c0e6e54d69cf7aa480ee Mon Sep 17 00:00:00 2001 From: Andrey Antukh Date: Wed, 7 Oct 2026 09:43:02 +0000 Subject: [PATCH] :bug: Revoke websocket subscriptions when access is revoked 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 --- backend/src/app/http/websocket.clj | 128 ++++++++++- backend/src/app/main.clj | 5 + .../backend_tests/http_websocket_test.clj | 203 +++++++++++++++++- 3 files changed, 328 insertions(+), 8 deletions(-) diff --git a/backend/src/app/http/websocket.clj b/backend/src/app/http/websocket.clj index ddfd50455b..90f1f2c602 100644 --- a/backend/src/app/http/websocket.clj +++ b/backend/src/app/http/websocket.clj @@ -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 ;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;; diff --git a/backend/src/app/main.clj b/backend/src/app/main.clj index 87d0b31f29..6422e0f458 100644 --- a/backend/src/app/main.clj +++ b/backend/src/app/main.clj @@ -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}) diff --git a/backend/test/backend_tests/http_websocket_test.clj b/backend/test/backend_tests/http_websocket_test.clj index 74782c1054..03faa659b8 100644 --- a/backend/test/backend_tests/http_websocket_test.clj +++ b/backend/test/backend_tests/http_websocket_test.clj @@ -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)))))) +