From f7f4cbb301883d07e2696f9127cc6a8b4095fd2f Mon Sep 17 00:00:00 2001 From: Andrey Antukh Date: Wed, 7 Oct 2026 08:48:10 +0000 Subject: [PATCH] :bug: Revoke websocket subscriptions when access is revoked GHSA-m53j-2766-6jqw. A websocket subscription was authorized once, when it opened, and that decision was trusted for the rest of the connection's life, so a member whose access was revoked kept receiving live file content as long as the tab stayed open. The revocation travels over the message bus instead of closing the socket from the mutation: the connection registry is local to one backend instance, while the RPC that revokes access runs on whichever instance received the request. Every instance re-verifies the subscriptions it owns with the same `has-read-permissions?` predicates that `:subscribe-file` and `:subscribe-team` authorize with, so a downgrade to viewer keeps its subscription and a non-member organization owner keeps read-only access. Two mechanisms cover each other: - Access-changing mutations announce after the write. delete-team- member, leave-team, update-team-member-role, delete-file, delete-project, move-files, move-project and delete-team announce. delete-share-link is deliberately excluded: a share-link visitor never passes the subscribe permission check, so there is no subscription to revoke. Team-wide changes publish one event per team; the watcher resolves the members itself, and both publishers defer through `db/after-commit` so the watcher never decides from pre-commit state. - Every relay re-checks access every 5 minutes by default (`subscription-revalidation-interval`), driven from the relay loop with `alts!` over the subscription channel and a one-shot tick channel, so a steady stream of edits cannot push the check away forever. Supporting changes: - The registry holds `:connections` and `:by-profile` in one atom, kept consistent by register/unregister in a single swap!; a profile lookup is one map access, and a profile whose last connection leaves is dropped from the index. - `close-file-subscription` and `close-team-subscription` announce the :leave-* message, close the relay channel and purge the bus subscription, and are idempotent. Closing the channel ends the relay loop and its re-check task. - The permission query the re-check uses filters soft-deleted files, projects and teams, matching its sibling queries. - A transient database error fails open: it logs and keeps the subscription; the next tick retries. - The revocation watcher's params schema declared the nitrate client as a bare keyword inside [:map ...], which malli resolved as a registry lookup and which took the whole system build down under enable-backend-asserts. It is now an explicit pair, optional because init-key yields nil with the admin-console flag off. The tests reproduce the leak against the real message bus, including the negative control for the interval re-check. The written records live in mem:backend/websocket-subscriptions. Closes #12155 AI-assisted-by: space-bunny-free AI-assisted-by: muse-spark-1.3-contributor --- .serena/memories/backend/core.md | 1 + .serena/memories/backend/subtleties.md | 4 +- .serena/memories/backend/testing.md | 3 +- .../backend/websocket-subscriptions.md | 44 + backend/src/app/binfile/common.clj | 16 +- backend/src/app/config.clj | 4 + backend/src/app/http/websocket.clj | 373 ++++++- backend/src/app/main.clj | 5 + backend/src/app/rpc/commands/files.clj | 7 + backend/src/app/rpc/commands/management.clj | 25 + backend/src/app/rpc/commands/projects.clj | 7 + backend/src/app/rpc/commands/teams.clj | 33 +- backend/src/app/rpc/notifications.clj | 67 ++ backend/test/backend_tests/binfile_test.clj | 42 + .../backend_tests/http_websocket_test.clj | 983 +++++++++++++++++- 15 files changed, 1559 insertions(+), 55 deletions(-) create mode 100644 .serena/memories/backend/websocket-subscriptions.md 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))))))))