diff --git a/backend/src/app/http/websocket.clj b/backend/src/app/http/websocket.clj index 90f1f2c602..4bc96271bf 100644 --- a/backend/src/app/http/websocket.clj +++ b/backend/src/app/http/websocket.clj @@ -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) diff --git a/backend/src/app/rpc/commands/files.clj b/backend/src/app/rpc/commands/files.clj index 63e116391b..a0789e764f 100644 --- a/backend/src/app/rpc/commands/files.clj +++ b/backend/src/app/rpc/commands/files.clj @@ -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) diff --git a/backend/src/app/rpc/commands/management.clj b/backend/src/app/rpc/commands/management.clj index 62d5a88330..5812e5ef26 100644 --- a/backend/src/app/rpc/commands/management.clj +++ b/backend/src/app/rpc/commands/management.clj @@ -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 diff --git a/backend/src/app/rpc/commands/projects.clj b/backend/src/app/rpc/commands/projects.clj index db366020a2..ab4959631b 100644 --- a/backend/src/app/rpc/commands/projects.clj +++ b/backend/src/app/rpc/commands/projects.clj @@ -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) diff --git a/backend/src/app/rpc/commands/teams.clj b/backend/src/app/rpc/commands/teams.clj index 8c31676d3d..a87d45ec18 100644 --- a/backend/src/app/rpc/commands/teams.clj +++ b/backend/src/app/rpc/commands/teams.clj @@ -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 diff --git a/backend/src/app/rpc/notifications.clj b/backend/src/app/rpc/notifications.clj index 00d6b8548e..b93c846ede 100644 --- a/backend/src/app/rpc/notifications.clj +++ b/backend/src/app/rpc/notifications.clj @@ -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)) diff --git a/backend/test/backend_tests/http_websocket_test.clj b/backend/test/backend_tests/http_websocket_test.clj index 03faa659b8..117fa3b406 100644 --- a/backend/test/backend_tests/http_websocket_test.clj +++ b/backend/test/backend_tests/http_websocket_test.clj @@ -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)))))