diff --git a/.serena/memories/backend/core.md b/.serena/memories/backend/core.md index 16978ac1ff..00708b1265 100644 --- a/.serena/memories/backend/core.md +++ b/.serena/memories/backend/core.md @@ -9,6 +9,7 @@ Backend: JVM Clojure; Integrant; PostgreSQL; Redis/Valkey; RPC; HTTP; storage; m - Embedded Ladybug graph experiment, projection, incremental sync, console, and risks: `mem:backend/graph-experiment` - Session lifetime config, token `:exp`, and idle/absolute GC: `mem:backend/session-expiration`. - Auth flows, permission model, teams, projects, invitations, comments, webhooks, audit: `mem:backend/auth-permissions-product-domains` +- WebSocket subscriptions, the connection registry (local per backend, unlike the bus), and how a lost permission gets revoked: `mem:backend/websocket-subscriptions`. - Audit-log event collection (RPC wrapper, frontend ingestion), telemetry duality, webhook fan-out, error reporters, Nexus archival and retention: `mem:audit-log` (guía completa, también para perfiles junior; el stub `mem:backend/audit-log` solo redirige aquí) - Services, task-queue/Pub-Sub topology constraints -> `mem:prod-infra/core`. diff --git a/.serena/memories/backend/subtleties.md b/.serena/memories/backend/subtleties.md index f7bdc3020a..efe4b5f900 100644 --- a/.serena/memories/backend/subtleties.md +++ b/.serena/memories/backend/subtleties.md @@ -45,7 +45,9 @@ - `job` is deletion-protected at the SQL level (`deletion_protection__tgr`): a stray `DELETE FROM job` raises instead of dropping the row and its `job_event` cascade without touching the referenced `storage_object`. The only intended deleters, the jobs GC and the submit dedupe, disable the guard with `SET LOCAL rules.deletion_protection TO off` inside their transaction; tests use `th/db-force-delete`. - A missing job-def raises (`:no-job-definition`) instead of completing. Throwing with `ex-data :type ::retry` still controls retry behavior; `:strategy ::noop` retries without incrementing retry count. - Cron entries claim their `scheduled_task` row with `FOR UPDATE SKIP LOCKED`, disable statement/idle-in-transaction timeouts locally, submit one `job` row per entry when no active instance exists (no-overlap), and reschedule themselves in `finally` unless interrupted. Worker, dispatcher, and cron components do not start when the DB pool is read-only. -- Integrant params schemas name injected components as bare `::ns/key` entries (pool, metrics, msgbus, job defs), and a bare entry is a malli registry lookup, not `any?`: the owning namespace must `sm/register!` it at load, or the backend crashes at boot with `:malli.core/invalid-schema` instead of asserting. Unit tests never build those keys, so `backend-tests.worker-test` compiles the three worker schemas to catch it. +- Integrant params schemas name injected components as bare `::ns/key` entries (pool, metrics, msgbus, job defs), and a bare entry is a malli registry lookup, not `any?`: the owning namespace must `sm/register!` it at load, or the backend crashes at boot with `:malli.core/invalid-schema` instead of asserting. Unit tests never build those keys, so `backend-tests.worker-test` compiles the three worker schemas and `backend-tests.http-websocket-test` compiles the revocation-watcher one to catch it. The assertion only runs with `PENPOT_FLAGS=enable-backend-asserts`, so without it a broken schema reaches production and only crashes at boot. +- Whether a bare entry works has nothing to do with its position or with how many of them there are: it is a registry lookup, so it works only if that exact keyword was `sm/register!`'d. `[:map ::db/pool ::metrics/metrics]` is fine, `[:map :app.aaa/unregistered :app.ccc/unregistered]` throws. When a component is **not** registered as a malli type, spell it as an explicit pair instead of registering a type just to satisfy the lookup — `[:app.nitrate/client {:optional true} [:maybe :map]]` is how `auth.oidc/schema:routes-params` and `loggers.audit.archive-task/schema:audit-log-archive-params` declare it. `{:optional true}` matters because `init-key` yields `nil` when the owning feature flag (`:admin-console`) is off, and a required key would then fail the assert on a perfectly normal boot. +- `sm/valid?` catches every `Throwable` and returns `false`, so an uncompilable schema and a genuinely invalid value both surface as a bare `Assert failed:` with no cause. When an `assert-key` fires with no message, compile the schema by hand to see the real exception. ## Config and HTTP/session middleware diff --git a/.serena/memories/backend/testing.md b/.serena/memories/backend/testing.md index e06f9ace32..2678b720e8 100644 --- a/.serena/memories/backend/testing.md +++ b/.serena/memories/backend/testing.md @@ -8,7 +8,8 @@ JVM `clojure.test` (kaocha runner) under `backend/test/backend_tests/`. - **Preferred way to run them: `scripts/ci --test`**, which runs lint, tests and format per module and writes a log per task under `.ci-logs/`. It exports `PENPOT_FLAGS` with `enable-backend-asserts` appended, so use it rather than calling `clojure -M:dev:test` by hand. - **Why the flag matters:** `clojure.core/assert` is elided at compile time when `*assert*` is false, and `app.config` only sets that var from the flags when it loads. The test fixture redefines the flags after that, so they never reach it: without the flag the suite runs with every assertion compiled out, and a test that depends on an assert only means something with it set. `scripts/ci` and `tests-backend.yml` export the flag (GitHub Actions sets it too). When running the suite directly — including a `--focus` run — the command is `PENPOT_FLAGS="enable-backend-asserts" clojure -M:dev:test`. - Coverage: if code is added or modified in `src/`, corresponding tests in `test/backend_tests/` must be added or updated. -- Isolated run: `clojure -M:dev:test --focus backend-tests.my-ns-test` for a specific test namespace, or `clojure -M:dev:test --focus backend-tests.my-ns-test/my-test-var` for a specific test var. +- Isolated run: `clojure -M:dev:test --focus backend-tests.my-ns-test` for a specific test namespace, or `clojure -M:dev:test --focus backend-tests.my-ns-test/my-test-var` for a specific test var. Keep the `PENPOT_FLAGS="enable-backend-asserts"` prefix on these too; a focused run without it hides assert-dependent failures exactly like a full one. +- A **whole-suite** crash reported as `Execution error ... Assert failed: ...` with no `FAIL in` line is not a test failure: an `ig/assert-key` ran while the test system was being built, which happens outside `clojure.test`, so kaocha reports it as a load/build error and the summary has no counts. Fix the offending schema (see `mem:backend/subtleties`) rather than hunting for a test. - Multiple focus symbols are supported by REPEATING the flag (kaocha accumulates them, last-wins does not apply): `clojure -M:dev:test --focus backend-tests.ns-one --focus backend-tests.ns-two` — preferred over a shell loop of isolated runs. - Regression run: `clojure -M:dev:test` to ensure no regressions in related functional areas. - If you need to filter output, tee to a temp file first: `clojure -M:dev:test 2>&1 | tee /tmp/penpot-test-output.txt`. diff --git a/.serena/memories/backend/websocket-subscriptions.md b/.serena/memories/backend/websocket-subscriptions.md new file mode 100644 index 0000000000..8b84cdfb8b --- /dev/null +++ b/.serena/memories/backend/websocket-subscriptions.md @@ -0,0 +1,44 @@ +# Backend WebSocket Subscriptions + +## Registry + +- `app.http.websocket/state` holds `{:connections {conn-id → wsp} :by-profile {profile-id → #{conn-id}}}` in **one** atom, so a single `swap!` keeps both consistent; `register-connection`/`unregister-connection` maintain them together and drop a profile from the index once its last connection leaves. +- Each connection's own `::ws/state` atom holds its subscriptions (`::file-subscription`, `::team-subscription`). `state` is shadowed by `::ws/state` in every `handle-message` destructuring, so only `on-connect`, `on-disconnect` and the `repl-*` helpers touch the registry — reading the registry from a handler is a bug. +- `::profile-id` and `::session-id` on a connection live in the `app.http.websocket` namespace, while `::ws/id`, `::ws/state`, `::ws/output-ch`, `::ws/session-id` live in `app.util.websocket`. Mixing them up yields silently `nil` values (a `nil` `session-id` makes the channel's `:xf` drop **every** message). +- The registry is local to one backend instance. The message bus is shared (`mem:prod-infra/core`). Anything that must reach connections owned by another instance has to go through the bus — closing a connection from an RPC handler only ever works in a single-backend deployment. + +## Revocation model + +A subscription is authorized once, at `:subscribe-file`/`:subscribe-team`, and that decision used to be trusted for the whole connection life (GHSA-m53j-2766-6jqw, medium). Two independent mechanisms now bound it: + +- **Announcement (primary).** A mutation that can change access publishes on `app.rpc.notifications/internal-revocation-topic` (`notify-permissions-changed` for one profile, `notify-team-permissions-changed` for one team id — the watcher resolves the members itself, so a large team is one event, not N). The `::revocation-watcher` Integrant component in `app.http.websocket` subscribes and calls `revalidate-profile-subscriptions`, which looks the profile up in `:by-profile` and re-checks every subscription with the same `has-read-permissions?` predicates the subscribe handlers authorize with — and `sql:file-permissions` additionally filters `deleted_at` on file, project and team, so a soft-deleted resource grants nothing. The event is only the trigger; the fresh check is the decision, which is what keeps a downgrade to `viewer` (still readable) and a non-member organization owner (read-only via `get-organization-owner-permissions`) from being mistaken for a revocation. Both publishers defer through `db/after-commit`, because the watcher checks on another connection and would otherwise see the pre-commit state. +- **Interval (safety net).** `start-relay` runs the relay as a `go-loop` that waits on the subscription channel and a one-shot tick channel together via `sp/alts!` with `:priority true` on the tick: when both hold something, the tick wins, so a steady stream of messages can never push the re-check away. The cadence is `subscription-revalidation-interval` (default `default-revalidation-interval`, 5 min, a code constant in `app.http.websocket` and **not** in the `config.clj` default map, same rule as session lifetimes). The deadline is fixed per cadence: a single tick channel covers a whole cadence and message turns reuse it (minting a fresh timeout per turn would both churn timers and starve under load exactly like an in-loop timer does); a new one is minted only after a tick fires. It covers what no announcement can: Nitrate transferring organization ownership (an HTTP call to another service) and rows deleted straight over SQL. Closing the channel ends the loop either way, because a take on a closed channel is always ready with nil — every way a subscription can end closes it, so a finished relay never keeps polling. Cost is two or three permission queries per subscription per interval, plus one Nitrate call per team per interval when the `:admin-console` flag is on (cached 30 s); one `timeout-chan` per subscription per cadence on promesa's default scheduler (one or two threads shared with every other scheduled task) — do not shorten the interval without sizing that. A tick abandoned at teardown (the subscription closed while it was pending) still fires its scheduled close harmlessly. Two places fail open, deliberately: `still-authorized?` (a transient DB error logs and keeps the subscription; `InterruptedException` is rethrown) and the `catch Throwable` in `watch-revocations`. + +Which mutations announce: `delete-team-member`, `leave-team`, `update-team-member-role` (member only), `delete-team`, `delete-file`, `delete-project`, `move-files`, `move-project` (one team event each). Anything not listed here is covered only by the interval. `delete-share-link` deliberately does not: a share-link visitor never passes the websocket subscribe check, because `check-read-permissions!` is called without a `share-id` and that arity only considers team membership. + +`close-file-subscription`/`close-team-subscription` are idempotent: no-op on a mismatched id, and they `dissoc` the subscription so a repeat call and `:pointer-update` do not publish to a dead topic. Closing the channel is what ends the relay (`take!` on a closed channel returns `nil`). + +## Topic and layering + +- `internal-revocation-topic` is a fixed uuid that is never a real resource id, so it cannot collide with a file, team or profile topic — and deliberately not the `uuid/zero` system topic, which is `sp/pipe`d straight into the client output and anything published there reaches every client. No client can subscribe to the internal topic: `:subscribe-file`/`:subscribe-team` only accept ids that already passed a permission check, and `:broadcast`/`:pointer-update` only republish to ids the server already holds. +- The topic and the publish helpers live in `app.rpc.notifications`, not in `app.http.websocket`: the command namespaces cannot require the watcher, because `app.http.websocket` already requires `app.rpc.commands.files`/`teams`. The watcher keeps the registry, which is the part it owns. + +## Time and channel traps in this namespace + +- The relay waits with `sp/alts!` over `[tick-ch channel]` with `:priority true`, following the `msgbus.clj` pattern (`start-io-loop` destructures the same `[val port]` shape and compares ports with `identical?`). A take on a closed channel is always ready with nil, which is what ends the loop — and what makes closing the channel from any teardown path prompt even while parked in `alts!`. `sp/timeout-chan` is one-shot: it closes once after the delay, and one abandoned at teardown still fires its scheduled close harmlessly. +- `ct/diff a b` returns **b − a** (`Duration/between`), so "time until a future instant" is `(inst-ms (ct/diff (ct/now) future))`. +- `ct/duration` unit keywords are singular: `{:millis 150}`, `{:minutes 5}`, never `:milliseconds`/`:days`. +- `still-authorized?` takes the check itself: `(boolean (check-access))`. Calling `(boolean pred)` on the closure is always true and silently disables the interval — this exact bug shipped once and no test caught it until the resubscribe-counting test existed. Naming rule here: no `?` for the action that queries and can fail (`check-access`), `?` only for the safe yes/no answer (`still-authorized?`). + +## Tests + +`backend/test/backend_tests/http_websocket_test.clj` drives the real Redis-backed msgbus, so the revocation tests are end-to-end: register the connection in the real registry, run the real command, wait for the subscription to drop, then assert no content arrives. A connection built without `register-connection` is invisible to the watcher, so the watcher-based tests must go through `open-registered-connection`. Interval tests shorten the config via `with-redefs` on `cf/get` rather than waiting minutes. `with-clean-registry` runs as an `:each` fixture, so no test can leak a live relay loop into the next one. `captured-events` forwards the keyword args through to the real `mbus/pub!` — rebuilding them positionally publishes to the topic `:message` with a `nil` payload and silently untests every other notice. Interval tests take access away with a direct row delete, never with a command: a command announces, so the test would pass even with the interval removed. The resubscribe-counting test is the one that proves finished relays stop polling. + +`revocation-watcher-schema-resolves-every-ref` covers the `::revocation-watcher` params schema, which no other test compiles. It existed broken for a while: the three injected keys were written as bare keywords, which is a valid malli *property* position and not a key, so the schema raised `:malli.core/invalid-schema` — invisible unless `PENPOT_FLAGS=enable-backend-asserts` was set, and a crash of the whole system build rather than one test (see `mem:backend/testing` on the flag). + +## See also + +- `mem:backend/auth-permissions-product-domains` for the permission predicates and the not-found semantics that the re-checks reuse. +- `mem:prod-infra/core` for why the bus crosses backends and the registry does not. +- `mem:frontend/routing-app-shell-subtleties` for the client side: the frontend already reacts to `team-membership-change` by navigating away, which is why revocation needs no new client message. +- `.agents/plans/2026-10-07-websocket-subscription-revocation.md` for the original plan, and the sibling `.agents/plans/2026-10-05-leave-team-file-access-cleanup.md` whose hardened permission queries this re-verification depends on (without live team membership required in `sql:file-permissions`, the re-check would keep reporting access for a member who already left). diff --git a/backend/src/app/binfile/common.clj b/backend/src/app/binfile/common.clj index efc6174d4b..8c84378c2d 100644 --- a/backend/src/app/binfile/common.clj +++ b/backend/src/app/binfile/common.clj @@ -384,19 +384,27 @@ inner join project as p on (p.id = f.project_id) where fpr.file_id = ? and fpr.profile_id = ? + and f.deleted_at is null + and p.deleted_at is null and exists (select 1 from team_profile_rel as tpr + inner join team as t on (t.id = tpr.team_id) where tpr.team_id = p.team_id - and tpr.profile_id = fpr.profile_id) + and tpr.profile_id = fpr.profile_id + and t.deleted_at is null) union all select tpr.is_owner, tpr.is_admin, tpr.can_edit from team_profile_rel as tpr + inner join team as t on (t.id = tpr.team_id) inner join project as p on (p.team_id = tpr.team_id) inner join file as f on (p.id = f.project_id) where f.id = ? and tpr.profile_id = ? + and t.deleted_at is null + and p.deleted_at is null + and f.deleted_at is null union all select ppr.is_owner, ppr.is_admin, @@ -406,10 +414,14 @@ inner join project as p on (p.id = ppr.project_id) where f.id = ? and ppr.profile_id = ? + and f.deleted_at is null + and p.deleted_at is null and exists (select 1 from team_profile_rel as tpr + inner join team as t on (t.id = tpr.team_id) where tpr.team_id = p.team_id - and tpr.profile_id = ppr.profile_id)") + and tpr.profile_id = ppr.profile_id + and t.deleted_at is null)") (defn- get-file-permissions* [conn profile-id file-id] diff --git a/backend/src/app/config.clj b/backend/src/app/config.clj index 99f47ea40c..615a009a71 100644 --- a/backend/src/app/config.clj +++ b/backend/src/app/config.clj @@ -158,6 +158,10 @@ [:http-server-io-threads {:optional true} ::sm/int] [:http-server-max-worker-threads {:optional true} ::sm/int] + ;; 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] + ;; Explicit CORS allowlist used when the :cors flag is enabled. ;; Configured via PENPOT_ALLOWED_ORIGINS as a comma/whitespace ;; separated list of origins (e.g. "https://plugins.example.com"). diff --git a/backend/src/app/http/websocket.clj b/backend/src/app/http/websocket.clj index 50bd64f115..44b58a8a22 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] @@ -22,6 +23,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] @@ -37,31 +39,79 @@ ;; WEBSOCKET HOOKS ;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;; -(def state (atom {})) +(def state + "Registry of the websocket connections held by this backend instance. + + Holds `:connections` (connection id -> connection data) and + `:by-profile` (profile id -> set of connection ids). Both live in the + same atom so a single `swap!` keeps them consistent, and so the + profile index never has to be rebuilt by scanning every connection. + + NOTE: the registry is local to one backend instance, unlike the + message bus, which is shared. Code that must reach connections + regardless of which instance owns them has to go through the bus." + (atom {:connections {} + :by-profile {}})) + +(defn register-connection + "Adds `wsp` to the registry under `id`, indexing it by its profile." + [id wsp] + (let [profile-id (::profile-id wsp)] + (swap! state + (fn [st] + (-> st + (update :connections assoc id wsp) + (cond-> profile-id + (update-in [:by-profile profile-id] + (fnil conj #{}) id))))))) + +(defn unregister-connection + "Removes the connection `id` from the registry and from the profile + index." + [id] + (swap! state + (fn [{:keys [connections] :as st}] + (let [owner (::profile-id (get connections id)) + indexed (disj (get-in st [:by-profile owner] #{}) id)] + (cond-> (assoc st :connections (dissoc connections id)) + (and owner (seq indexed)) + (update :by-profile assoc owner indexed) + (and owner (empty? indexed)) + (update :by-profile dissoc owner)))))) + +(defn get-connection + "Returns the connection data registered under `id`, or nil." + [id] + (get-in @state [:connections id])) + +(defn connections-for-profile + "Returns the set of connection ids currently held by `profile-id`." + [profile-id] + (get-in @state [:by-profile profile-id] #{})) ;; REPL HELPERS (defn repl-get-connections-for-file [file-id] - (->> (vals @state) + (->> (vals (:connections @state)) (filter #(= file-id (-> % ::ws/state deref ::file-subscription :file-id))) (map ::ws/id))) (defn repl-get-connections-for-team [team-id] - (->> (vals @state) + (->> (vals (:connections @state)) (filter #(= team-id (-> % ::ws/state deref ::team-subscription :team-id))) (map ::ws/id))) (defn repl-close-connection [id] - (when-let [{:keys [::ws/close-ch] :as wsp} (get @state id)] + (when-let [{:keys [::ws/close-ch]} (get-connection id)] (sp/put! close-ch [8899 "closed from server"]) (sp/close! close-ch))) (defn repl-get-connection-info [id] - (when-let [wsp (get @state id)] + (when-let [wsp (get-connection id)] (let [subs (some-> wsp ::ws/state deref)] {:id id :created-at (::created-at wsp) @@ -85,6 +135,137 @@ (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? + "Runs the `check-access` permission check, 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 one of two places + that fail open, deliberately: the announcement path (`watch-revocations` + below) also logs and moves on when a re-check throws. An interrupt is + not a lookup failure and is rethrown." + [check-access] + (try + (boolean (check-access)) + (catch InterruptedException cause + (throw cause)) + (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- start-relay + "Forwards `channel` into the client output channel of `wsp`, and keeps + a re-check of `check-access` running on a fixed cadence. When access no + longer holds, `close-fn` tears the subscription down, which closes the + channel and ends the loop. + + The loop waits on the subscription channel and a one-shot tick channel + together, with priority on the tick: when both hold something, the tick + wins, so a steady stream of messages can never push the re-check away. + A single tick channel covers a whole cadence and message turns reuse + it, so busy files don't churn timers; the cadence holds under load + because nothing moves the tick. Closing the channel ends the loop + either way, because a take on a closed channel is always ready + with nil. + + `on-forward` runs for every forwarded message, which is how the file + relay announces presence." + [{:keys [::ws/output-ch] :as wsp} channel check-access close-fn on-forward] + (let [interval-ms (inst-ms (revalidation-interval))] + (sp/go-loop [tick-ch (sp/timeout-chan interval-ms)] + (let [[message port] (sp/alts! [tick-ch channel] :priority true)] + (cond + (identical? port tick-ch) + (do + (when-not (still-authorized? check-access) + (l/info :hint + "closing websocket subscription on re-check" + :profile-id (::profile-id wsp)) + (close-fn)) + (when-not (sp/closed? channel) + (recur (sp/timeout-chan interval-ms)))) + + (nil? message) + ;; The channel was closed, which means the subscription is gone + ;; and the relay ends. + nil + + :else + (do + (sp/put! output-ch message) + (when on-forward (on-forward message)) + (recur tick-ch))))))) + +;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;; ;; WEBSOCKET HANDLER ;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;; @@ -151,10 +332,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] @@ -163,23 +344,29 @@ 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. (let [subs {:team-id team-id :organization-id organization-id :channel channel :topic team-id}] (swap! state assoc ::team-subscription subs)) + (start-relay wsp channel + #(teams/has-read-permissions? cfg profile-id team-id) + #(close-team-subscription cfg wsp team-id) + nil) + + (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) @@ -190,51 +377,40 @@ (let [subs {:file-id file-id :channel fch :topic file-id}] (swap! state assoc ::file-subscription subs)) + (start-relay wsp fch + #(files/has-read-permissions? cfg profile-id file-id) + #(close-file-subscription cfg wsp file-id) + (fn [{:keys [type]}] + (when (or (= :join-file type) + (= :leave-file type) + (= :disconnect type)) + (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)))) + (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 - [{:keys [::mbus/msgbus]} {:keys [::ws/id ::ws/state ::session-id ::profile-id]} {:keys [file-id] :as params}] + [cfg {:keys [::ws/id] :as wsp} {:keys [file-id] :as params}] (l/trace :fn "handle-message" :event "unsubscribe-file" :file-id file-id :conn-id id) - - (let [subs (::file-subscription @state) - message {:type :leave-file - :file-id file-id - :session-id session-id - :profile-id profile-id}] - - (when (= (:file-id subs) file-id) - (mbus/pub! msgbus :topic file-id :message message) - (let [ch (:channel subs)] - (sp/close! ch) - (mbus/purge! msgbus [ch]))))) + (close-file-subscription cfg wsp file-id)) (defmethod handle-message :keepalive [_ _ _] @@ -265,6 +441,109 @@ :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. 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 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` + 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. Returns the subscription channel, which `ig/halt-key!` + closes to stop the loop." + [{:keys [::mbus/msgbus] :as cfg}] + (let [ch (sp/chan :buf (sp/dropping-buffer 64))] + (mbus/sub! msgbus + :topic notifications/internal-revocation-topic + :chan ch) + (sp/go-loop [] + (when-let [{:keys [type profile-id team-id]} (sp/take! ch)] + (try + (cond + (= :profile-permissions-changed type) + (when (pos? (revalidate-profile-subscriptions cfg profile-id)) + (l/debug :hint "revoked websocket subscriptions" + :profile-id profile-id)) + + (= :team-permissions-changed type) + (let [revoked (->> (teams/get-team-members cfg team-id) + (map :id) + (map #(revalidate-profile-subscriptions cfg %)) + (reduce + 0))] + (when (pos? revoked) + (l/debug :hint "revoked websocket subscriptions" + :team-id team-id + :count revoked)))) + (catch Throwable cause + (l/error :hint "cannot revalidate websocket subscriptions" + :profile-id profile-id + :team-id team-id + :cause cause))) + (recur))) + ch)) + +;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;; +;; INTEGRANT +;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;; + +(def ^:private schema:revocation-watcher + [:map + ::mbus/msgbus + ::db/pool + [:app.nitrate/client {:optional true} [:maybe :map]]]) + +(defmethod ig/assert-key ::revocation-watcher + [_ params] + (assert (sm/valid? schema:revocation-watcher params))) + +(defmethod ig/init-key ::revocation-watcher + [_ cfg] + (watch-revocations cfg)) + +(defmethod ig/halt-key! ::revocation-watcher + [_ ch] + (sp/close! ch)) + ;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;; ;; HTTP HANDLER ;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;; @@ -273,7 +552,7 @@ [{:keys [::mtx/metrics]} {:keys [::ws/id] :as wsp}] (let [created-at (ct/now)] (l/trace :fn "on-connect" :conn-id id) - (swap! state assoc id wsp) + (register-connection id wsp) (mtx/run! metrics :id :websocket-active-connections :inc 1) @@ -281,7 +560,7 @@ (assoc wsp ::ws/on-disconnect (fn [] (l/trace :fn "on-disconnect" :conn-id id) - (swap! state dissoc id) + (unregister-connection id) (mtx/run! metrics :id :websocket-active-connections :dec 1) (mtx/run! metrics :id :websocket-session-timing diff --git a/backend/src/app/main.clj b/backend/src/app/main.clj index 71ddafd9bf..26b6552e92 100644 --- a/backend/src/app/main.clj +++ b/backend/src/app/main.clj @@ -571,6 +571,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/src/app/rpc/commands/files.clj b/backend/src/app/rpc/commands/files.clj index b2b465cf78..b174734867 100644 --- a/backend/src/app/rpc/commands/files.clj +++ b/backend/src/app/rpc/commands/files.clj @@ -42,6 +42,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] @@ -1168,6 +1169,12 @@ :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 (:id team)) + (rph/with-meta (rph/wrap) {::audit/props {:team-id (:id team) :project-id (:project-id file) diff --git a/backend/src/app/rpc/commands/management.clj b/backend/src/app/rpc/commands/management.clj index b0ac390b57..25ddf61fc8 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] @@ -270,6 +272,17 @@ ;; --- COMMAND: Move file +(defn- notify-team-access-changed + "Announces a possible read-access change for `team-ids`. + + Moving 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. One event per team: the watcher resolves the members + itself, so a large team cannot overflow the bounded publish buffers." + [cfg team-ids] + (doseq [team-id (into #{} (remove nil?) team-ids)] + (ntf/notify-team-permissions-changed cfg team-id))) + (def sql:get-files "select id, features, project_id from file where id = ANY(?)") @@ -340,6 +353,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 + (keep (partial perms/get-team-id-for-project conn) source)) + nil)) (def ^:private @@ -409,6 +430,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 [(: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 18e5c282f0..fa2e883926 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])) @@ -319,6 +320,12 @@ :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 (: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 d96c9a9cf1..789e0b82a1 100644 --- a/backend/src/app/rpc/commands/teams.clj +++ b/backend/src/app/rpc/commands/teams.clj @@ -31,6 +31,7 @@ [app.rpc.commands.profile :as profile] [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.setup :as-alias setup] @@ -810,9 +811,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 @@ -861,6 +863,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 @@ -926,6 +938,10 @@ ::jobs/params {:object :team :deleted-at (:deleted-at team) :id team-id}}) + ;; The team is soft-deleted, so every subscription its members hold + ;; loses its authorization at once. One event: the watcher resolves + ;; the members itself. + (ntf/notify-team-permissions-changed cfg team-id) (rph/with-meta (rph/wrap team) {::audit/props {:organization-id organization-id :team-id team-id @@ -958,7 +974,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, @@ -1008,6 +1024,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 @@ -1081,6 +1103,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..b3ffd1be65 100644 --- a/backend/src/app/rpc/notifications.clj +++ b/backend/src/app/rpc/notifications.clj @@ -6,6 +6,8 @@ (ns app.rpc.notifications (:require + [app.common.uuid :as uuid] + [app.db :as db] [app.msgbus :as mbus])) (defn notify-team-change @@ -49,3 +51,68 @@ :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. + + A fixed uuid that is never a real resource id, so it cannot collide + with a file, team or profile topic — and deliberately not the + `uuid/zero` system topic, which 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." + (uuid/uuid "ffffffff-ffff-ffff-ffff-ffffffffffff")) + +(defn notify-permissions-changed + "Asks every backend to re-verify the websocket subscriptions currently + held by `profile-id`. + + Deferred past the commit: the watcher decides from a fresh permission + check on another connection, so announcing before the transaction + commits would let it see the old state and close nothing. Outside a + transaction the announcement goes out immediately. + + The watcher always decides from a fresh permission check, so an extra + announcement is safe: it closes nothing whose access still holds." + [cfg profile-id] + (db/after-commit + (fn [] + (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 + "Asks every backend to re-verify the subscriptions of every member of + `team-id`. + + Used when a change can move a resource out of the reach of a whole + team (moving, deleting or soft-deleting files, projects and teams). One + event per team: the watcher resolves the members itself, so a large + team cannot overflow the bounded publish buffers on the way out. + + Like `notify-permissions-changed`, deferred past the commit." + [cfg team-id] + (db/after-commit + (fn [] + (let [msgbus (::mbus/msgbus cfg)] + (mbus/pub! msgbus + :topic internal-revocation-topic + :message {:type :team-permissions-changed + :team-id team-id}))))) diff --git a/backend/test/backend_tests/binfile_test.clj b/backend/test/backend_tests/binfile_test.clj index 9e525af2aa..9a5484eaaf 100644 --- a/backend/test/backend_tests/binfile_test.clj +++ b/backend/test/backend_tests/binfile_test.clj @@ -3271,3 +3271,45 @@ (t/is (empty? (th/db-query :file-library-sync {:library-file-id (:id victim)}))) (t/is (empty? (th/db-query :file-library-rel {:library-file-id (:id file)}))) (t/is (empty? (th/db-query :file-library-sync {:library-file-id (:id file)})))))) + +(t/deftest file-permissions-ignore-soft-deleted + ;; sql:file-permissions must not grant anything through a file, + ;; project or team that is soft-deleted (GHSA-m53j-2766-6jqw: the + ;; websocket re-check relies on this query, and the subscribe handler + ;; additionally refuses deleted files via check-file-exists). + (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)}) + perms? #(some? (bfc/get-file-permissions th/*system* (:id editor) (:id file))) + undelete (fn [table id] + (db/update! th/*system* table {:deleted-at nil} {:id id})) + soft-delete (fn [table id] + (db/update! th/*system* table {:deleted-at (ct/now)} {:id id}))] + + (t/testing "a live member can read the live file" + (t/is (perms?))) + + (t/testing "a soft-deleted file grants nothing" + (soft-delete :file (:id file)) + (t/is (nil? (bfc/get-file-permissions th/*system* (:id editor) (:id file)))) + (undelete :file (:id file)) + (t/is (perms?))) + + (t/testing "a soft-deleted project grants nothing" + (soft-delete :project (:id project)) + (t/is (nil? (bfc/get-file-permissions th/*system* (:id editor) (:id file)))) + (undelete :project (:id project)) + (t/is (perms?))) + + (t/testing "a soft-deleted team grants nothing" + (soft-delete :team (:id team)) + (t/is (nil? (bfc/get-file-permissions th/*system* (:id editor) (:id file)))) + (undelete :team (:id team)) + (t/is (perms?)))))) diff --git a/backend/test/backend_tests/http_websocket_test.clj b/backend/test/backend_tests/http_websocket_test.clj index 6ffc10e744..e11ba9b150 100644 --- a/backend/test/backend_tests/http_websocket_test.clj +++ b/backend/test/backend_tests/http_websocket_test.clj @@ -6,7 +6,10 @@ (ns backend-tests.http-websocket-test (:require + [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.websocket :as ws] [app.msgbus :as mbus] @@ -14,13 +17,23 @@ [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] [promesa.exec.csp :as sp])) +(defn- with-clean-registry + "Runs `f` against an empty connection registry, restoring the previous + contents afterwards so tests cannot leak connections into each other." + [f] + (let [saved @ws/state] + (reset! ws/state {:connections {} :by-profile {}}) + (try (f) + (finally (reset! ws/state saved))))) + (t/use-fixtures :once th/state-init) -(t/use-fixtures :each th/database-reset) +(t/use-fixtures :each th/database-reset with-clean-registry) (defn make-wsp [profile-id state output-ch] @@ -145,3 +158,971 @@ (t/is (nil? ((get-method ws/handle-message :pointer-update) cfg empty-wsp msg))))))) + +;; --- SUBSCRIPTION REVOCATION +;; +;; GHSA-m53j-2766-6jqw: a websocket subscription checks read permission +;; once, at subscribe time. Once the connection is subscribed to the +;; message-bus topic it keeps receiving live content for the rest of its +;; life, even after access is revoked. + +(def ^:private revocation-timeout-ms + "How long to wait for a message that should never arrive. Long enough + to absorb the round trip through redis, short enough to keep the test + fast." + 500) + +(defn- poll-msg! + "Drains `ch` for up to `ms`, returning the first message whose `:type` + is `type`, or nil when none arrives. + + Draining rather than reading a single message matters for the negative + assertions: a live subscription also carries presence traffic, so + \"no `:file-change` arrived\" is the property under test, not \"the + channel stayed empty\"." + [ch type ms] + (let [result (promise) + deadline (+ (System/currentTimeMillis) ms)] + (sp/go + (loop [] + (let [remain (- deadline (System/currentTimeMillis))] + (when (pos? remain) + (let [[msg _] (sp/alts! [ch (sp/timeout-chan remain)])] + (if (= type (:type msg)) + (deliver result msg) + (recur))))))) + (deref result (inc ms) nil))) + +(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- 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) + 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." + [] + (let [msgbus (::mbus/msgbus th/*system*)] + (fn [topic message] + (mbus/pub! msgbus :topic topic :message message)))) + +;; --- CONNECTION REGISTRY + +(t/deftest registry-indexes-connections-by-profile + (let [profile-a (uuid/next) + profile-b (uuid/next) + conn-a (uuid/next) + conn-b (uuid/next) + conn-c (uuid/next)] + + (t/testing "a profile with no connections has an empty index" + (t/is (= #{} (ws/connections-for-profile profile-a)))) + + (ws/register-connection conn-a {::ws/profile-id profile-a}) + (ws/register-connection conn-b {::ws/profile-id profile-a}) + (ws/register-connection conn-c {::ws/profile-id profile-b}) + + (t/testing "connections are indexed under their profile" + (t/is (= #{conn-a conn-b} (ws/connections-for-profile profile-a))) + (t/is (= #{conn-c} (ws/connections-for-profile profile-b)))) + + (t/testing "connections are retrievable by id" + (t/is (= profile-a (::ws/profile-id (ws/get-connection conn-a)))) + (t/is (nil? (ws/get-connection (uuid/next))))) + + (t/testing "unregistering removes it from both the map and the index" + (ws/unregister-connection conn-b) + (t/is (= #{conn-a} (ws/connections-for-profile profile-a))) + (t/is (nil? (ws/get-connection conn-b)))) + + (t/testing "a profile whose last connection left is dropped from the index" + (ws/unregister-connection conn-a) + (t/is (= #{} (ws/connections-for-profile profile-a))) + (t/is (not (contains? (:by-profile @ws/state) profile-a))) + (t/is (= #{conn-c} (ws/connections-for-profile profile-b)))) + + (t/testing "unregistering an unknown connection is a no-op" + (ws/unregister-connection (uuid/next)) + (t/is (= #{conn-c} (ws/connections-for-profile profile-b))) + (t/is (= conn-c (-> (:connections @ws/state) keys first)))))) + +(t/deftest close-file-subscription-drops-only-the-matching-file + (let [profile (th/create-profile* 1 {:is-active true}) + file (th/create-file* 1 {:profile-id (:id profile) + :project-id (:default-project-id profile)}) + 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 relay + :topic (:id file)}}) + wsp (assoc (make-wsp (:id profile) ws-state output) + ::ws/state ws-state)] + + (t/testing "closing a file the connection is not subscribed to does nothing" + (with-redefs [mbus/pub! (fn [& _] (throw ::unexpected-publish)) + mbus/purge! (fn [& _] (throw ::unexpected-purge))] + (t/is (nil? (ws/close-file-subscription th/*system* wsp (:id other)))) + (t/is (some? (::ws/file-subscription @ws-state))))) + + (t/testing "closing the subscribed file purges the bus" + (let [purged (atom [])] + (with-redefs [mbus/purge! (fn [_ chans] (swap! purged conj chans))] + (ws/close-file-subscription th/*system* wsp (:id file)) + (t/is (= 1 (count @purged)))))) + + (t/testing "the relay channel is closed so the go-loop stops" + (t/is (nil? (sp/poll! relay))) + (t/is (true? (sp/closed? relay)))) + + (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}) + 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)}) + file-id (:id file) + publish (publisher!) + change {:type :file-change + :file-id file-id + :revn 1 + :changes [{:type :add :id (uuid/next)}]}] + + (let [conn (-> (open-registered-connection (:id editor)) + (subscribe :subscribe-file {:file-id file-id})) + out (:output conn)] + + (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 "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}) + 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 [team-id (:id team) + publish (publisher!) + change {:type :library-change + :team-id team-id + :file-id (uuid/next) + :revn 1 + :changes [{:type :add :id (uuid/next)}]}] + + (let [conn (-> (open-registered-connection (:id editor)) + (subscribe :subscribe-team {:team-id team-id})) + out (:output conn)] + + (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 "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 + a revocation: `:viewer` can still read, so the subscription stays." + (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)}) + file-id (:id file) + publish (publisher!) + change {:type :file-change :file-id file-id :revn 1 :changes []} + out (open-subscription! :subscribe-file (:id editor) + {:file-id file-id})] + + (t/testing "editor is downgraded to viewer" + (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 "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)] + + (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)] + + (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)})] + + (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)] + + (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)))) + + (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)))))) + +(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" + (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, while letting every other publish reach the real bus untouched." + [f] + (let [events (atom []) + real mbus/pub!] + (with-redefs [mbus/pub! (fn [bus & {:keys [topic message] :as opts}] + (when (= ntf/internal-revocation-topic topic) + (swap! events conj message)) + (apply real bus (mapcat identity opts)))] + (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" + (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)}))] + + ;; One event per team: the watcher resolves the members itself, so + ;; a large team cannot overflow the bounded publish buffers. + (t/is (= [{:type :team-permissions-changed + :team-id (:id team)}] + events))))) + +(t/deftest deleting-a-team-announces-once + (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 + ::rpc/profile-id (:id owner) + :id (:id team)}))] + (t/is (= [{:type :team-permissions-changed + :team-id (:id team)}] + events))))) + +(t/deftest team-event-closes-every-member-subscription + (let [{:keys [owner editor team file]} (editor-in-team! 1 2) + other (th/create-profile* 3 {:is-active true})] + (th/create-team-role* {:team-id (:id team) + :profile-id (:id other) + :role :editor}) + (let [other-project (th/create-project* 3 {:profile-id (:id other) + :team-id (:id team)}) + other-file (th/create-file* 3 {:profile-id (:id other) + :project-id (:id other-project)})] + (let [conn (-> (open-registered-connection (:id editor)) + (subscribe :subscribe-file {:file-id (:id file)})) + conn2 (-> (open-registered-connection (:id other)) + (subscribe :subscribe-file {:file-id (:id other-file)}))] + + (th/command! {::th/type :delete-team + ::rpc/profile-id (:id owner) + :id (:id team)}) + + (t/testing "the announcement reaches the watcher off the bus" + (ntf/notify-team-permissions-changed th/*system* (:id team)) + (t/is (wait-until #(nil? (subscribed-file conn)) 3000)) + (t/is (wait-until #(nil? (subscribed-file conn2)) 3000))))))) + +(t/deftest rolled-back-revocation-announces-nothing + (let [editor (th/create-profile* 2 {:is-active true}) + events (captured-events + #(try + (db/tx-run! th/*system* + (fn [_cfg] + (ntf/notify-permissions-changed th/*system* (:id editor)) + (throw (ex-info "boom" {})))) + (catch clojure.lang.ExceptionInfo _)))] + (t/is (empty? events)))) + +(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))))) + +;; --- 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/is (= (ct/duration {:minutes 5}) (ws/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 [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))] + (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)))) + + ;; Access is taken away the way an external change would take + ;; it: a direct row delete fires no command and publishes no + ;; announcement, so the interval is the only thing that can + ;; close the subscription. + (db/exec! th/*system* + ["delete from team_profile_rel where team_id = ? and profile_id = ?" + (:id team) (: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))] + (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))] + (let [conn (-> (open-registered-connection (:id editor)) + (subscribe :subscribe-team {:team-id (:id team)}))] + + (t/is (some? (subscribed-team conn))) + + ;; Same external-change shape as above: no command, no + ;; announcement, only the interval. + (db/exec! th/*system* + ["delete from team_profile_rel where team_id = ? and profile_id = ?" + (:id team) (:id editor)]) + + (t/testing "the team subscription is dropped by the interval" + (t/is (wait-until #(nil? (subscribed-team conn)) 3000))))))) + +(t/deftest a-failing-check-keeps-the-subscription + (let [{:keys [editor file]} (editor-in-team! 1 2)] + + (with-redefs [cf/get (fn [k default] + (if (= :subscription-revalidation-interval k) + (ct/duration {:millis 150}) + default))] + (let [conn (-> (open-registered-connection (:id editor)) + (subscribe :subscribe-file {:file-id (:id file)}))] + + ;; Access still holds, but the check itself blows up. + (with-redefs [files/has-read-permissions? + (fn [& _] (throw (RuntimeException. "db down")))] + (Thread/sleep 500) + (t/testing "a failing check keeps the subscription" + (t/is (some? (subscribed-file conn))))))))) + +(t/deftest a-false-check-closes-the-subscription + (let [{:keys [editor file]} (editor-in-team! 1 2)] + + (with-redefs [cf/get (fn [k default] + (if (= :subscription-revalidation-interval k) + (ct/duration {:millis 150}) + default))] + (let [conn (-> (open-registered-connection (:id editor)) + (subscribe :subscribe-file {:file-id (:id file)}))] + + (with-redefs [files/has-read-permissions? (fn [& _] false)] + (t/testing "a check that returns false closes it" + (t/is (wait-until #(nil? (subscribed-file conn)) 3000)))))))) + +(t/deftest resubscribe-stops-previous-revalidation + (let [profile (th/create-profile* 1 {:is-active true}) + file1 (th/create-file* 1 {:profile-id (:id profile) + :project-id (:default-project-id profile)}) + file2 (th/create-file* 2 {:profile-id (:id profile) + :project-id (:default-project-id profile)})] + + (with-redefs [cf/get (fn [k default] + (if (= :subscription-revalidation-interval k) + (ct/duration {:millis 150}) + default))] + (let [counts (atom {}) + real files/has-read-permissions?] + (with-redefs [files/has-read-permissions? + (fn [cfg pid fid] + (swap! counts update fid (fnil inc 0)) + (real cfg pid fid))] + (let [conn (-> (open-registered-connection (:id profile)) + (subscribe :subscribe-file {:file-id (:id file1)}))] + + (t/testing "the first subscription is re-checked on the interval" + (t/is (wait-until #(pos? (get @counts (:id file1) 0)) 3000))) + + (t/testing "resubscribing moves the relay to the new file" + (subscribe conn :subscribe-file {:file-id (:id file2)}) + (t/is (wait-until #(>= (get @counts (:id file2) 0) 2) 3000))) + + (t/testing "the previous subscription stops revalidating" + (let [old (get @counts (:id file1) 0)] + (Thread/sleep 400) + (t/is (= old (get @counts (:id file1) 0))))))))))) + +;; --- EFFECT TESTS +;; +;; Announcing is not the same as revoking: these assert the subscription +;; actually drops and the content stops flowing. + +(t/deftest deleting-a-file-closes-open-subscriptions + (let [{:keys [owner editor file]} (editor-in-team! 1 2) + publish (publisher!)] + + (let [conn (-> (open-registered-connection (:id editor)) + (subscribe :subscribe-file {:file-id (:id file)})) + out (:output conn)] + + (th/command! {::th/type :delete-file + ::rpc/profile-id (:id owner) + :id (:id file)}) + + (t/testing "the file subscription drops without any manual publish" + (t/is (wait-until #(nil? (subscribed-file conn)) 3000))) + + (t/testing "no file content reaches the client afterwards" + (publish (:id file) {:type :file-change :file-id (:id file) :revn 9}) + (t/is (nil? (poll-msg! out :file-change revocation-timeout-ms))))))) + +(t/deftest deleting-a-project-closes-open-subscriptions + (let [{:keys [owner editor file]} (editor-in-team! 1 2) + project-id (:project-id file)] + + (let [conn (-> (open-registered-connection (:id editor)) + (subscribe :subscribe-file {:file-id (:id file)})) + out (:output conn)] + + (th/command! {::th/type :delete-project + ::rpc/profile-id (:id owner) + :id project-id}) + + (t/testing "the file subscription drops" + (t/is (wait-until #(nil? (subscribed-file conn)) 3000))) + + (t/testing "no file content 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 deleting-a-team-closes-open-subscriptions + (let [{:keys [owner editor team file]} (editor-in-team! 1 2)] + + (let [conn (-> (open-registered-connection (:id editor)) + (subscribe :subscribe-file {:file-id (:id file)}) + (subscribe :subscribe-team {:team-id (:id team)}))] + + (th/command! {::th/type :delete-team + ::rpc/profile-id (:id owner) + :id (:id team)}) + + (t/testing "both subscriptions drop" + (t/is (wait-until #(nil? (subscribed-file conn)) 3000)) + (t/is (wait-until #(nil? (subscribed-team conn)) 3000)))))) + +(t/deftest moving-a-file-closes-source-team-subscriptions + (let [owner (th/create-profile* 1 {:is-active true}) + editor (th/create-profile* 2 {:is-active true}) + team-a (th/create-team* 1 {:profile-id (:id owner)}) + team-b (th/create-team* 2 {:profile-id (:id owner)})] + (th/create-team-role* {:team-id (:id team-a) + :profile-id (:id editor) + :role :editor}) + (let [project-a (th/create-project* 1 {:profile-id (:id owner) + :team-id (:id team-a)}) + project-b (th/create-project* 2 {:profile-id (:id owner) + :team-id (:id team-b)}) + file (th/create-file* 1 {:profile-id (:id editor) + :project-id (:id project-a)}) + conn (-> (open-registered-connection (:id editor)) + (subscribe :subscribe-file {:file-id (:id file)}))] + + (th/command! {::th/type :move-files + ::rpc/profile-id (:id owner) + :ids #{(:id file)} + :project-id (:id project-b)}) + + (t/testing "a member left behind in the source team loses the subscription" + (t/is (wait-until #(nil? (subscribed-file conn)) 3000)))))) + +(t/deftest close-team-subscription-is-idempotent + (let [profile (th/create-profile* 1 {:is-active true}) + team (th/create-team* 1 {:profile-id (:id profile)}) + other (th/create-team* 2 {:profile-id (:id profile)}) + output (sp/chan :buf (sp/dropping-buffer 64)) + relay (sp/chan :buf (sp/dropping-buffer 64)) + ws-state (atom {::ws/team-subscription + {:team-id (:id team) + :channel relay + :topic (:id team)}}) + wsp (assoc (make-wsp (:id profile) ws-state output) + ::ws/state ws-state)] + + (t/testing "closing a team the connection is not subscribed to does nothing" + (with-redefs [mbus/purge! (fn [& _] (throw ::unexpected-purge))] + (t/is (nil? (ws/close-team-subscription th/*system* wsp (:id other)))) + (t/is (some? (::ws/team-subscription @ws-state))))) + + (t/testing "closing the subscribed team purges the bus and the relay stops" + (let [purged (atom [])] + (with-redefs [mbus/purge! (fn [_ chans] (swap! purged conj chans))] + (ws/close-team-subscription th/*system* wsp (:id team)) + (t/is (= 1 (count @purged))))) + (t/is (nil? (sp/poll! relay))) + (t/is (true? (sp/closed? relay)))) + + (t/testing "the subscription is forgotten, so closing again is a no-op" + (t/is (nil? (::ws/team-subscription @ws-state))) + (t/is (nil? (ws/close-team-subscription th/*system* wsp (:id team))))))) + +(t/deftest losing-a-team-keeps-the-other-file-subscription + (let [{:keys [owner editor team]} (editor-in-team! 1 2) + other-team (th/create-team* 3 {:profile-id (:id owner)})] + (th/create-team-role* {:team-id (:id other-team) + :profile-id (:id editor) + :role :editor}) + (let [other-project (th/create-project* 3 {:profile-id (:id editor) + :team-id (:id other-team)}) + other-file (th/create-file* 3 {:profile-id (:id editor) + :project-id (:id other-project)}) + conn (-> (open-registered-connection (:id editor)) + (subscribe :subscribe-team {:team-id (:id team)}) + (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)}) + + (t/testing "the lost team subscription drops" + (t/is (wait-until #(nil? (subscribed-team conn)) 3000))) + + (t/testing "the file subscription on the other team survives" + (t/is (some? (subscribed-file conn))))))) + +;; The revocation watcher is the only integrant key here, so it is the +;; only one that compiles this schema at boot: a ref with no registered +;; schema crashes the boot with :malli.core/invalid-schema instead of +;; asserting. The test system builds the component, but assert-key only +;; runs its check with *assert* set, so pin the compilation here. +(t/deftest revocation-watcher-schema-resolves-every-ref + (t/testing "the schema compiles" + (t/is (sm/schema? (sm/schema @#'ws/schema:revocation-watcher)))) + + (t/testing "every ref the key injects is registered" + (t/is (= #{::db/pool ::mbus/msgbus} + (disj (set (sm/keys @#'ws/schema:revocation-watcher)) + :app.nitrate/client)))) + (t/testing "the nitrate client is optional, so a nil one passes" + (t/is (sm/valid? @#'ws/schema:revocation-watcher + {::db/pool (:app.db/pool th/*system*) + ::mbus/msgbus (:app.msgbus/msgbus th/*system*)}))) + (t/testing "a non-map nitrate client is rejected" + (t/is (not (sm/valid? @#'ws/schema:revocation-watcher + {::db/pool (:app.db/pool th/*system*) + ::mbus/msgbus (:app.msgbus/msgbus th/*system*) + :app.nitrate/client "not-a-client"}))))) + +(t/deftest organization-owner-keeps-read-only-access + (let [{:keys [owner editor team file]} (editor-in-team! 1 2)] + + (let [conn (-> (open-registered-connection (:id editor)) + (subscribe :subscribe-file {:file-id (:id file)}))] + + ;; The admin-console flag is off in tests, so Nitrate never grants + ;; this; stand in for an organization owner expelled from the team. + (with-redefs [nitrate/organization-owner-of-team? (fn [& _] true)] + (th/command! {::th/type :delete-team-member + ::rpc/profile-id (:id owner) + :team-id (:id team) + :member-id (:id editor)}) + + (t/testing "revalidation closes nothing for an organization owner" + (t/is (zero? (ws/revalidate-profile-subscriptions + th/*system* (:id editor))))) + + (t/testing "the read-only subscription survives" + (t/is (some? (subscribed-file conn))))))))