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)))))) +