From d608f3b89811dc8c4e95bd417ebd0a45a3ee0eb2 Mon Sep 17 00:00:00 2001 From: Andrey Antukh Date: Wed, 7 Oct 2026 10:54:48 +0000 Subject: [PATCH] :bug: Re-check websocket subscriptions on an interval The announcement path is the primary mechanism, but it cannot see an access change that fires no command here: Nitrate transferring the ownership of an organization is an HTTP call to another service, and rows deleted straight over SQL publish nothing. Both keep a subscription alive until the connection happens to close. Each relay now re-checks access every `subscription-revalidation-interval` (5 minutes by default, overridable) and drops the subscription when it no longer holds. The default is a code constant in `app.http.websocket` rather than the `config.clj` default map, matching how session lifetimes are configured. The re-check is a scheduled task rather than a timer inside the relay loop, on purpose: driven by the loop, a steady stream of edits would push the check away forever and a busy file would never be re-checked at all. `start-relay` also replaces the `sp/pipe` of the team subscription and the ad-hoc `go-loop` of the file one, so both now share the same forwarding and re-check path. `still-authorized?` is the one place that fails open: a transient database error logs and keeps the subscription rather than kicking a user out of an editor, and the next tick tries again. Closes #12155 AI-assisted-by: space-bunny-free --- backend/src/app/config.clj | 4 + backend/src/app/http/websocket.clj | 228 ++++++++++++------ .../backend_tests/http_websocket_test.clj | 108 +++++++++ 3 files changed, 272 insertions(+), 68 deletions(-) diff --git a/backend/src/app/config.clj b/backend/src/app/config.clj index a4375c2184..3dd2e52fe3 100644 --- a/backend/src/app/config.clj +++ b/backend/src/app/config.clj @@ -253,6 +253,10 @@ [:auth-token-cookie-max-age {:optional true} ::ct/duration] [:auth-token-cookie-max-age-absolute {:optional true} ::ct/duration] + ;; How often a websocket subscription re-checks that its access still + ;; holds. The default lives in `app.http.websocket`, not here. + [:subscription-revalidation-interval {:optional true} ::ct/duration] + [:registration-domain-whitelist {:optional true} [::sm/set :string]] [:email-verify-threshold {:optional true} ::ct/duration] diff --git a/backend/src/app/http/websocket.clj b/backend/src/app/http/websocket.clj index 4bc96271bf..af3ae0a26c 100644 --- a/backend/src/app/http/websocket.clj +++ b/backend/src/app/http/websocket.clj @@ -15,6 +15,7 @@ [app.common.schema :as sm] [app.common.time :as ct] [app.common.uuid :as uuid] + [app.config :as cf] [app.db :as db] [app.http.session :as session] [app.metrics :as mtx] @@ -25,6 +26,7 @@ [app.rpc.notifications :as notifications] [app.util.websocket :as ws] [integrant.core :as ig] + [promesa.exec :as px] [promesa.exec.csp :as sp] [yetti.websocket :as yws])) @@ -139,6 +141,133 @@ (pp/pprint))) ;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;; +;; SUBSCRIPTIONS +;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;; + +;; A subscription is authorized once, when it is opened, and that +;; single decision is then trusted for as long as the connection +;; lives. Two things bound that: a mutation that changes access +;; announces it (see the revocation section below), and the relay +;; re-checks on an interval. + +(def default-revalidation-interval + "How long a subscription may rely on the authorization it was granted + with before its relay loop re-checks it. + + This is the safety net for what the announcement cannot cover: an + access change made outside a Penpot command. Nitrate transferring the + ownership of an organization is an HTTP call to another service that + fires no RPC here, and rows deleted straight over SQL fire nothing + either. Both are rare, so the interval is generous; the cost is one + permission query per subscription per interval." + (ct/duration {:minutes 5})) + +(defn revalidation-interval + [] + (cf/get :subscription-revalidation-interval default-revalidation-interval)) + +(defn- still-authorized? + "Evaluates the `pred` 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] + (try + (boolean pred) + (catch Throwable cause + (l/error :hint "cannot re-check a websocket subscription" + :cause cause) + true))) + +(defn close-file-subscription + "Tears down the file subscription held by the connection `wsp`, if it + is subscribed to `file-id`. + + Announces the departure so the remaining participants drop the + presence of this session, closes the relay channel (which in turn + stops the `:subscribe-file` go-loop, because `take!` on a closed + channel returns nil) and removes the subscription from the bus. + + Does nothing when the connection is subscribed to a different file, + so it is safe to call for a subscription that is already gone." + [{:keys [::mbus/msgbus]} {:keys [::ws/state ::session-id ::profile-id]} file-id] + (let [subs (::file-subscription @state)] + (when (= (:file-id subs) file-id) + (mbus/pub! msgbus + :topic file-id + :message {:type :leave-file + :file-id file-id + :session-id session-id + :profile-id profile-id}) + (let [ch (:channel subs)] + (sp/close! 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 relay, 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))))) + + +(defn- schedule-revalidation + "Rearms `task` every `interval-ms` for as long as `running?` 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? + (px/schedule interval-ms + (fn [] + (when @running? + (task) + (schedule-revalidation running? interval-ms task)))))) + +(defn- start-relay + "Forwards `channel` into the client output channel of `wsp`, and keeps + a re-check of `authorized?` running on the interval. When access no + longer holds, `close!` tears the subscription down, which closes the + channel and ends the loop. + + `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))))) + +;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;; ;; WEBSOCKET HANDLER ;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;; @@ -205,10 +334,10 @@ nil))) (defmethod handle-message :subscribe-team - [cfg {:keys [::ws/id ::ws/state ::ws/output-ch ::session-id ::profile-id]} {:keys [team-id] :as params}] + [cfg {:keys [::ws/id ::ws/state ::session-id ::profile-id] :as wsp} {:keys [team-id] :as params}] (l/trace :fn "handle-message" :event "subscribe-team" :team-id team-id :conn-id id) (teams/check-read-permissions! cfg profile-id team-id) - (let [prev-subs (get @state ::team-subscription) + (let [prev-subs (::team-subscription @state) organization-id (get-team-organization-id cfg team-id) ;; Resolved server-side so a client only hears its readable team's org topics (cond-> [team-id] @@ -217,8 +346,13 @@ channel (sp/chan :buf (sp/dropping-buffer 64) :xf (remove #(= (:session-id %) session-id)))] - (sp/pipe channel output-ch false) - (mbus/sub! (::mbus/msgbus cfg) :topics topics :chan channel) + ;; 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 @@ -226,14 +360,15 @@ :topic team-id}] (swap! state assoc ::team-subscription subs)) + (mbus/sub! (::mbus/msgbus cfg) :topics topics :chan channel) + ;; Close previous subscription if exists (when-let [ch (:channel prev-subs)] (sp/close! ch) (mbus/purge! (::mbus/msgbus cfg) [ch])))) - (defmethod handle-message :subscribe-file - [cfg {:keys [::ws/id ::ws/state ::ws/output-ch ::session-id ::profile-id]} {:keys [file-id] :as params}] + [cfg {:keys [::ws/id ::ws/state ::session-id ::profile-id] :as wsp} {:keys [file-id] :as params}] (l/trace :fn "handle-message" :event "subscribe-file" :file-id file-id :conn-id id) (bfc/check-file-exists cfg file-id) (files/check-read-permissions! cfg profile-id file-id) @@ -244,77 +379,34 @@ (let [subs {:file-id file-id :channel fch :topic file-id}] (swap! state assoc ::file-subscription subs)) + (start-relay cfg wsp fch + #(files/has-read-permissions? cfg profile-id file-id) + #(close-file-subscription cfg wsp file-id) + (fn [message] + (when (contains? #{:join-file :leave-file :disconnect} + (:type message)) + (mbus/pub! (::mbus/msgbus cfg) + :topic file-id + :message {:type :presence + :file-id file-id + :session-id session-id + :profile-id profile-id})))) + ;; Close previous subscription if exists (when-let [ch (:channel psub)] (sp/close! ch) (mbus/purge! (::mbus/msgbus cfg) [ch])) - (sp/go-loop [] - (when-let [{:keys [type] :as message} (sp/take! fch)] - (sp/put! output-ch message) - (when (or (= :join-file type) - (= :leave-file type) - (= :disconnect type)) - (let [message {:type :presence - :file-id file-id - :session-id session-id - :profile-id profile-id}] - (mbus/pub! (::mbus/msgbus cfg) - :topic file-id - :message message))) - (recur))) - ;; Subscribe to file topic (mbus/sub! (::mbus/msgbus cfg) :topic file-id :chan fch) ;; Notifify the rest of participants of the new connection. - (let [message {:type :join-file - :file-id file-id - :session-id session-id - :profile-id profile-id}] - (mbus/pub! (::mbus/msgbus cfg) :topic file-id :message message)))) - -(defn close-file-subscription - "Tears down the file subscription held by the connection `wsp`, if it - is subscribed to `file-id`. - - Announces the departure so the remaining participants drop the - presence of this session, closes the relay channel (which in turn - stops the `:subscribe-file` go-loop, because `take!` on a closed - channel returns nil) and removes the subscription from the bus. - - Does nothing when the connection is subscribed to a different file, - so it is safe to call for a subscription that is already gone." - [{:keys [::mbus/msgbus]} {:keys [::ws/state ::session-id ::profile-id]} file-id] - (let [subs (::file-subscription @state)] - (when (= (:file-id subs) file-id) - (mbus/pub! msgbus - :topic file-id - :message {:type :leave-file - :file-id file-id - :session-id session-id - :profile-id profile-id}) - (let [ch (:channel subs)] - (sp/close! 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))))) + (mbus/pub! (::mbus/msgbus cfg) + :topic file-id + :message {:type :join-file + :file-id file-id + :session-id session-id + :profile-id profile-id}))) (defmethod handle-message :unsubscribe-file [cfg {:keys [::ws/id] :as wsp} {:keys [file-id] :as params}] diff --git a/backend/test/backend_tests/http_websocket_test.clj b/backend/test/backend_tests/http_websocket_test.clj index 117fa3b406..68e4b898ff 100644 --- a/backend/test/backend_tests/http_websocket_test.clj +++ b/backend/test/backend_tests/http_websocket_test.clj @@ -6,7 +6,9 @@ (ns backend-tests.http-websocket-test (:require + [app.common.time :as ct] [app.common.uuid :as uuid] + [app.config :as cf] [app.db :as db] [app.http.websocket :as ws] [app.msgbus :as mbus] @@ -742,3 +744,109 @@ :id link-id}))] (t/is (some? link-id)) (t/is (empty? events))))) + +;; --- PERIODIC REVALIDATION +;; +;; The announcement path is the primary mechanism. This interval is the +;; safety net for access changes that fire no command here, such as +;; Nitrate transferring the ownership of an organization, or rows +;; deleted straight over SQL. + +(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/deftest revalidation-is-configurable + (with-redefs [cf/get (fn [k default] (if (= :subscription-revalidation-interval k) + (ct/duration {:seconds 30}) + default))] + (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)] + + (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)] + + (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)}) + + (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/deftest interval-recheck-keeps-a-subscription-that-still-holds + (let [{:keys [owner 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)] + + (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) + + (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/deftest interval-recheck-covers-the-team-subscription-too + (let [{:keys [editor team]} (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-team {:team-id (:id team)}))] + + (t/is (some? (subscribed-team conn))) + + (th/command! {::th/type :leave-team + ::rpc/profile-id (:id editor) + :id (:id team)}) + + (t/testing "the team subscription is dropped by the interval" + (t/is (wait-until #(nil? (subscribed-team conn)) 3000)))))))))