From 0d800519209e58b5508a1e1c9a8347d3764aaf17 Mon Sep 17 00:00:00 2001 From: Petr Bauch Date: Tue, 15 Sep 2026 09:36:49 -0400 Subject: [PATCH 1/3] feat(db): let an install end, and a workspace be installed again (CHOO-2626) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit `messaging_installs` made `(platform, external_workspace_id)` unique across the deployment so that an inbound event from a workspace has exactly one tenant to go to. That is still the guarantee. The constraint enforcing it was too strong: nothing deletes an install row, so the claim outlived the install and a workspace could be connected once, ever. The error a customer got told them to remove the existing install, through a path that did not exist. What has to be unique is the set of installs that are serving. The constraint becomes a unique index over `status = 'active'`, under the same name — the store reads that name out of the integrity error to tell a claimed workspace from any other failed write, and Postgres reports a unique index by the name a constraint would have carried. `tenant_of_messaging_install` is redefined against the same predicate, and that part is load-bearing rather than tidy. It resolves a workspace to a tenant for traffic nobody has authenticated, and it was written relying on the old constraint to answer at most once; the caller refuses an ambiguous answer. Left alone, the first workspace to be installed, released and installed again would make it answer twice, and the customer's live install would stop receiving events because of one they had themselves ended. `get_for_workspace` takes the same predicate for the same reason. `ended_at` records when, alongside the `status` that says which of the two ways it ended — a decision here, or news from the platform. An operator looking at a bridge that stopped working needs to know which. `encrypted_bot_token` becomes nullable so an install that has ended can keep its record without keeping its secret. Nothing yet writes any of this: no store method ends an install and no route calls one. This is the schema that makes those possible. `_REDEFINED_SINCE` is new bookkeeping in the lookup test. Creating and dropping a function both show up as a function that is there or is not; a redefinition leaves one of the right name and signature answering a different question, which nothing about the shape of the schema reveals. Co-Authored-By: Claude Opus 5 --- core/switch_core/db/models.py | 56 +++++--- .../db/stores/messaging_install_store.py | 51 ++++++-- core/switch_core/db/tenant_lookup.py | 16 ++- ...a7f2c3e9b481_messaging_installs_can_end.py | 108 ++++++++++++++++ .../db/test_messaging_install_claim.py | 121 +++++++++++++++++- .../switch_core/db/test_tenant_lookup.py | 22 +++- 6 files changed, 335 insertions(+), 39 deletions(-) create mode 100644 core/switch_core/migrations/versions/a7f2c3e9b481_messaging_installs_can_end.py diff --git a/core/switch_core/db/models.py b/core/switch_core/db/models.py index ddc33ca2d..2099d6ed6 100644 --- a/core/switch_core/db/models.py +++ b/core/switch_core/db/models.py @@ -1145,19 +1145,33 @@ class MessagingInstall(TenantScoped, Base): token, and what it may do. **`(platform, external_workspace_id)` is unique across the whole - deployment, not per tenant**, and that is the single most important line - here. Inbound events arrive over one public endpoint carrying a workspace - id and no tenant, so a workspace claimed by two tenants is a message with - two possible destinations and no way to choose — which is the failure this - whole phase exists to make unrepresentable. The database decides it rather - than a read-then-insert in application code, because the check and the - write cannot be made atomic from outside. - - That constraint is also the one place a tenant learns something about - another: claiming a workspace somebody else already claimed fails, and the - failure says so. It is the right answer — the alternative is a silent - second claim — and what it discloses is that *some* tenant holds a - workspace the caller was already able to name. + deployment among rows that are still `active`**, and that is the single + most important line here. Inbound events arrive over one public endpoint + carrying a workspace id and no tenant, so a workspace claimed by two + tenants is a message with two possible destinations and no way to choose — + which is the failure this whole phase exists to make unrepresentable. The + database decides it rather than a read-then-insert in application code, + because the check and the write cannot be made atomic from outside. + + It is a *partial* index rather than a plain constraint because an install + has to be able to end. A customer who removes the app in Slack, or an + operator who disconnects it here, leaves a row behind — and a row that + still occupied the workspace would mean nobody could ever install that + workspace again, including the customer who just removed it. Ending an + install therefore frees the workspace, and keeps the record of the one + that ended. + + That index is also the one place a tenant learns something about another: + claiming a workspace somebody else already holds fails, and the failure + says so. It is the right answer — the alternative is a silent second claim + — and what it discloses is that *some* tenant holds a workspace the caller + was already able to name. + + `status` is `active`, `disconnected` (an operator here ended it) or + `revoked` (the platform told us it was over). The two endings are recorded + apart because they call for different things: one is somebody's decision + and the other is news, and an operator looking at a bridge that stopped + working needs to know which. `bridge_id` is nullable because the install row is written before anything is built on it, and because removing a bridge should not force the @@ -1167,7 +1181,10 @@ class MessagingInstall(TenantScoped, Base): `encrypted_bot_token` uses the same key as every other credential this schema stores (`crypto.encrypt_token` over the configured secret), so it is protected against a stolen dump and not against a compromised process. - A per-tenant key is a stronger boundary and a later decision. + A per-tenant key is a stronger boundary and a later decision. It is + nullable so that an install which has ended can keep its record without + keeping its secret: the token is worthless by then, and a worthless + credential still reads like a credential to whoever finds the dump. `scopes` is the platform's own spelling of what was granted, stored verbatim rather than parsed into a list — a scope string that means @@ -1177,10 +1194,12 @@ class MessagingInstall(TenantScoped, Base): __tablename__ = "messaging_installs" __table_args__ = ( - UniqueConstraint( + Index( + "uq_messaging_installs_workspace", "platform", "external_workspace_id", - name="uq_messaging_installs_workspace", + unique=True, + postgresql_where=text("status = 'active'"), ), UniqueConstraint("id", "tenant_id", name="uq_messaging_installs_id_tenant"), ForeignKeyConstraint( @@ -1193,7 +1212,7 @@ class MessagingInstall(TenantScoped, Base): id: Mapped[str] = mapped_column(Text, primary_key=True, default=_uuid) platform: Mapped[str] = mapped_column(Text, nullable=False) external_workspace_id: Mapped[str] = mapped_column(Text, nullable=False) - encrypted_bot_token: Mapped[str] = mapped_column(Text, nullable=False) + encrypted_bot_token: Mapped[str | None] = mapped_column(Text, nullable=True) scopes: Mapped[str] = mapped_column(Text, nullable=False) status: Mapped[str] = mapped_column(Text, nullable=False) installed_by_user_id: Mapped[str] = mapped_column( @@ -1203,6 +1222,9 @@ class MessagingInstall(TenantScoped, Base): installed_at: Mapped[str] = mapped_column( DateTime(timezone=True), server_default=func.now(), nullable=False ) + ended_at: Mapped[datetime | None] = mapped_column( + DateTime(timezone=True), nullable=True + ) class MessagingInstallState(TenantScoped, Base): diff --git a/core/switch_core/db/stores/messaging_install_store.py b/core/switch_core/db/stores/messaging_install_store.py index 269560850..b91a6d2a5 100644 --- a/core/switch_core/db/stores/messaging_install_store.py +++ b/core/switch_core/db/stores/messaging_install_store.py @@ -18,6 +18,18 @@ #: is worthless by the time anyone reads it. STATE_TTL = timedelta(minutes=10) +#: An install that is serving. The only status the workspace uniqueness index +#: covers, so it is also the answer to "who holds this workspace". +INSTALL_ACTIVE = "active" + +#: Ended here, by someone who decided to. The bridge is gone with it. +INSTALL_DISCONNECTED = "disconnected" + +#: Ended there: the platform told us the app was removed or its token killed. +#: Distinct from `disconnected` because an operator whose bridge stopped +#: working needs to know whether it was news or a decision. +INSTALL_REVOKED = "revoked" + class MessagingInstallClaimedError(RuntimeError): """Another tenant has already installed the app into this workspace. @@ -103,17 +115,21 @@ async def record_install( """Claim a workspace for the bound tenant. The claim is the insert: `(platform, external_workspace_id)` is unique - across the deployment, so the database decides who holds a workspace - rather than a read followed by a write that cannot be made atomic with - it. A second tenant's install therefore fails here, loudly, instead of - producing an event with two possible destinations. + across the deployment among active rows, so the database decides who + holds a workspace rather than a read followed by a write that cannot be + made atomic with it. A second tenant's install therefore fails here, + loudly, instead of producing an event with two possible destinations. + + Installs that have ended are not in the index, so re-installing a + workspace somebody released is an ordinary insert and needs no check of + its own. """ install = MessagingInstall( platform=platform, external_workspace_id=external_workspace_id, encrypted_bot_token=encrypted_bot_token, scopes=scopes, - status="active", + status=INSTALL_ACTIVE, installed_by_user_id=user_id, ) session.add(install) @@ -124,26 +140,33 @@ async def record_install( raise raise MessagingInstallClaimedError( f"the {platform} workspace {external_workspace_id} is already " - "connected to Switch. Remove the existing install before " - "connecting it again." + "connected to Switch. Disconnect the existing install — from " + "the organisation that holds it, which may not be yours — " + "before connecting it again." ) from exc return install async def get_for_workspace( self, session: AsyncSession, *, platform: str, external_workspace_id: str ) -> MessagingInstall | None: - """The bound tenant's install of one workspace, if it is theirs. - - Deliberately still scoped, even though the unique constraint means at - most one row exists deployment-wide: the caller is an inbound webhook - that resolved a tenant from the workspace a moment ago, and this - re-reading it under RLS is what makes a mistake there a miss rather - than a cross-tenant read. + """The bound tenant's live install of one workspace, if it is theirs. + + Deliberately still scoped, even though the unique index means at most + one active row exists deployment-wide: the caller is an inbound webhook + that resolved a tenant from the workspace a moment ago, and re-reading + it under RLS is what makes a mistake there a miss rather than a + cross-tenant read. + + The status predicate matches the index's and `tenant_of_messaging_ + install`'s. Without it a workspace installed, removed and installed + again returns two rows and this raises — an outage for the live + install, caused by the ended one. """ result = await session.execute( select(MessagingInstall).where( MessagingInstall.platform == platform, MessagingInstall.external_workspace_id == external_workspace_id, + MessagingInstall.status == INSTALL_ACTIVE, ) ) return result.scalars().one_or_none() diff --git a/core/switch_core/db/tenant_lookup.py b/core/switch_core/db/tenant_lookup.py index adcd78dfd..371a457da 100644 --- a/core/switch_core/db/tenant_lookup.py +++ b/core/switch_core/db/tenant_lookup.py @@ -122,6 +122,13 @@ happened to mint the same string, and refuse a customer's traffic for a reason in someone else's account. +It also filters on `status = 'active'`, which is not a refinement but the +other half of the same uniqueness. Only active rows are unique by index, so a +workspace that was installed, removed and installed again has two rows and +this would otherwise answer twice — refusing a customer's live traffic on the +strength of an install they themselves ended. The predicate here and the one +on the index are the same predicate, and have to stay that way. + Why not the obvious alternatives is argued in `docs/old/multi-tenancy-phase1-db.md`, "The bootstrap problem"; the short version is that returning rows instead of tenant ids would put a second copy @@ -306,14 +313,17 @@ def signature(self) -> str: query=( "SELECT tenant_id FROM messaging_installs " "WHERE platform = p_platform " - "AND external_workspace_id = p_external_workspace_id" + "AND external_workspace_id = p_external_workspace_id " + "AND status = 'active'" ), purpose=( "Which tenant an inbound event from an installed workspace belongs " "to. The public webhook is unauthenticated by nature and knows only " "the platform it was posted to and the workspace the payload names, " - "so this runs before anything else the request does. Unique by " - "constraint on exactly this pair, so it answers at most once." + "so this runs before anything else the request does. The status " + "predicate is the index's own: only active rows are unique, and an " + "install that ended must stop answering for its workspace rather " + "than compete with the one that replaced it." ), ), ) diff --git a/core/switch_core/migrations/versions/a7f2c3e9b481_messaging_installs_can_end.py b/core/switch_core/migrations/versions/a7f2c3e9b481_messaging_installs_can_end.py new file mode 100644 index 000000000..98a1c950d --- /dev/null +++ b/core/switch_core/migrations/versions/a7f2c3e9b481_messaging_installs_can_end.py @@ -0,0 +1,108 @@ +"""an install that can end, and a workspace that can be installed again + +`c8a4e21f6d30` gave `messaging_installs` a plain unique constraint on +`(platform, external_workspace_id)`, on the grounds that a workspace claimed +by two tenants is an event with two destinations. That grounds is still right; +the constraint was too strong for it. It made a claim permanent — the row it +protects is never deleted, so a customer who removed the Switch app from their +own Slack could not put it back, and the error they got told them to "remove +the existing install" through a path that did not exist. + +What actually has to be unique is the set of installs that are *serving*. So +the constraint becomes a unique index over `status = 'active'` under the same +name, and ending an install frees the workspace while keeping the record of +what happened to it. + +`tenant_of_messaging_install` is redefined against the same predicate, and +that is not a tidy-up. The function resolves a workspace to a tenant for +traffic nobody has authenticated, and it was written to rely on the old +constraint answering at most once. Left alone, the first workspace to be +installed twice would make it answer twice and the caller refuses an ambiguous +answer — so the customer's live install would stop receiving events because of +one they had themselves ended. + +Two column changes come with it. `ended_at` records when, alongside the +`status` that says which of the two ways. And `encrypted_bot_token` becomes +nullable so an ended install can keep its record without keeping its secret: +the token is dead by then, and a dead credential still reads like a credential +to whoever finds the dump. + +The lookup DDL below is a verbatim copy of `switch_core/db/tenant_lookup.py` +as it stood when this migration was written, copied rather than imported for +the reason every revision in this chain copies: a migration records a change +that already happened, and importing the live module would let a later edit +silently change what this one means. +`tests/switch_core/db/test_frozen_ddl_matches_create_all.py` is what keeps the +copy from drifting. + +Revision ID: a7f2c3e9b481 +Revises: d3f6b0c95a17 +Create Date: 2026-09-15 00:00:00.000000 + +""" + +from collections.abc import Sequence + +import sqlalchemy as sa +from alembic import op + +revision: str = "a7f2c3e9b481" +down_revision: str | None = "d3f6b0c95a17" +branch_labels: str | Sequence[str] | None = None +depends_on: str | Sequence[str] | None = None + +TABLE = "messaging_installs" +INDEX_NAME = "uq_messaging_installs_workspace" + +SECURE_SEARCH_PATH = "pg_catalog, public, pg_temp" + +CREATE_TENANT_OF_MESSAGING_INSTALL = f"""CREATE OR REPLACE FUNCTION tenant_of_messaging_install(p_platform text, p_external_workspace_id text) + RETURNS SETOF text + LANGUAGE sql STABLE SECURITY DEFINER + SET search_path = {SECURE_SEARCH_PATH} +AS $$SELECT tenant_id FROM messaging_installs WHERE platform = p_platform AND external_workspace_id = p_external_workspace_id AND status = 'active'$$""" + +RESTORE_TENANT_OF_MESSAGING_INSTALL = f"""CREATE OR REPLACE FUNCTION tenant_of_messaging_install(p_platform text, p_external_workspace_id text) + RETURNS SETOF text + LANGUAGE sql STABLE SECURITY DEFINER + SET search_path = {SECURE_SEARCH_PATH} +AS $$SELECT tenant_id FROM messaging_installs WHERE platform = p_platform AND external_workspace_id = p_external_workspace_id$$""" + + +def upgrade() -> None: + # Same name as the constraint it replaces, deliberately: the store reads + # the name out of the integrity error to tell a claimed workspace from any + # other write that failed, and Postgres reports a unique index by the same + # name a unique constraint would have carried. + op.drop_constraint(INDEX_NAME, TABLE, type_="unique") + op.create_index( + INDEX_NAME, + TABLE, + ["platform", "external_workspace_id"], + unique=True, + postgresql_where=sa.text("status = 'active'"), + ) + op.add_column( + TABLE, sa.Column("ended_at", sa.DateTime(timezone=True), nullable=True) + ) + op.alter_column( + TABLE, "encrypted_bot_token", existing_type=sa.Text(), nullable=True + ) + op.execute(CREATE_TENANT_OF_MESSAGING_INSTALL) + + +def downgrade() -> None: + op.execute(RESTORE_TENANT_OF_MESSAGING_INSTALL) + op.alter_column( + TABLE, "encrypted_bot_token", existing_type=sa.Text(), nullable=False + ) + op.drop_column(TABLE, "ended_at") + op.drop_index(INDEX_NAME, table_name=TABLE) + # Both of the statements above and this one fail on data the old schema + # cannot hold: an ended install whose token was discarded, and a workspace + # holding more than one row. That is the honest outcome — going back means + # choosing which of those installs survives, and a migration is not the + # thing that gets to choose. + op.create_unique_constraint( + INDEX_NAME, TABLE, ["platform", "external_workspace_id"] + ) diff --git a/core/tests/switch_core/db/test_messaging_install_claim.py b/core/tests/switch_core/db/test_messaging_install_claim.py index 1319744ae..c1af5cd94 100644 --- a/core/tests/switch_core/db/test_messaging_install_claim.py +++ b/core/tests/switch_core/db/test_messaging_install_claim.py @@ -19,16 +19,26 @@ who could already name it. That is deliberate, and the alternative is worse: a silent second claim, discovered when a customer's messages start arriving in somebody else's rooms. + +The claim covers **active** installs only, and the second half of this file is +about that. A claim that outlived the install would mean a workspace could be +connected once ever — including by the customer who had just disconnected it — +so ending an install releases the workspace. The lookup has to agree with the +index about which rows count, and the tests below measure that agreement from +both directions: a released workspace can be claimed again, and a workspace +with history resolves to whoever holds it now rather than refusing because +two rows name it. """ from __future__ import annotations import uuid +from datetime import UTC, datetime import pytest -from sqlalchemy import select +from sqlalchemy import select, update from sqlalchemy.exc import IntegrityError -from sqlalchemy.ext.asyncio import async_sessionmaker +from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker from switch_core.db.models import MessagingInstall, Tenant, User from switch_core.db.session_scope import tenant_session @@ -76,6 +86,20 @@ def _install(fixture: _Fixture, tenant_id: str) -> MessagingInstall: ) +async def _end(session: AsyncSession, install_id: str, status: str) -> None: + """End an install the way the store will, without depending on it yet. + + A direct write, because what is under test here is the index and the + lookup: the properties have to hold for any row in that state, not only + for rows a particular method happened to produce. + """ + await session.execute( + update(MessagingInstall) + .where(MessagingInstall.id == install_id) + .values(status=status, ended_at=datetime.now(UTC), encrypted_bot_token=None) + ) + + async def test_a_second_tenant_cannot_claim_a_claimed_workspace( rls_harness: RLSHarness, ) -> None: @@ -149,3 +173,96 @@ async def test_the_same_workspace_on_another_platform_is_a_separate_claim( ) == fixture.tenant_b ) + + +async def test_an_install_that_ended_releases_the_workspace( + rls_harness: RLSHarness, +) -> None: + """The claim lasts as long as the install and not a moment longer. + + A permanent claim is not a stricter version of this guarantee, it is a + different and wrong one: a customer who tries Switch, disconnects, and + comes back finds their own workspace held by a row nobody can see and + nobody can release. + """ + fixture = await _two_tenants_and_a_workspace(rls_harness.owner) + + async with tenant_session(rls_harness.restricted, fixture.tenant_a) as session: + install = _install(fixture, fixture.tenant_a) + session.add(install) + await session.flush() + await _end(session, install.id, "disconnected") + await session.commit() + + async with tenant_session(rls_harness.restricted, fixture.tenant_b) as session: + session.add(_install(fixture, fixture.tenant_b)) + await session.commit() + + assert ( + await tenant_of_messaging_install( + rls_harness.restricted, "slack", fixture.workspace + ) + == fixture.tenant_b + ) + + +async def test_the_lookup_does_not_answer_twice_for_a_workspace_with_history( + rls_harness: RLSHarness, +) -> None: + """The failure the status predicate exists to prevent. + + The lookup refuses an ambiguous answer rather than picking one, which is + right — but it makes a second row for the same workspace an outage for the + live install rather than a stale record. So the predicate on the function + has to be the index's predicate, and this measures it with enough history + to catch a lookup that merely takes the first row: two ended installs and + one live one, inserted in that order. + """ + fixture = await _two_tenants_and_a_workspace(rls_harness.owner) + + for _ in range(2): + async with tenant_session(rls_harness.restricted, fixture.tenant_a) as session: + install = _install(fixture, fixture.tenant_a) + session.add(install) + await session.flush() + await _end(session, install.id, "revoked") + await session.commit() + + async with tenant_session(rls_harness.restricted, fixture.tenant_a) as session: + session.add(_install(fixture, fixture.tenant_a)) + await session.commit() + + assert ( + await tenant_of_messaging_install( + rls_harness.restricted, "slack", fixture.workspace + ) + == fixture.tenant_a + ) + + +async def test_a_workspace_nobody_holds_any_longer_resolves_to_nobody( + rls_harness: RLSHarness, +) -> None: + """An ended install must not go on routing the workspace's traffic. + + The app can still be sitting in the customer's Slack after a disconnect + here, posting events at us for as long as someone leaves it there. Those + events belong to no tenant now, and the lookup saying so is what turns + them into a refusal instead of a delivery into rooms the customer has + stopped paying for. + """ + fixture = await _two_tenants_and_a_workspace(rls_harness.owner) + + async with tenant_session(rls_harness.restricted, fixture.tenant_a) as session: + install = _install(fixture, fixture.tenant_a) + session.add(install) + await session.flush() + await _end(session, install.id, "disconnected") + await session.commit() + + assert ( + await tenant_of_messaging_install( + rls_harness.restricted, "slack", fixture.workspace + ) + is None + ) diff --git a/core/tests/switch_core/db/test_tenant_lookup.py b/core/tests/switch_core/db/test_tenant_lookup.py index 5e39f0bf9..a68e939ef 100644 --- a/core/tests/switch_core/db/test_tenant_lookup.py +++ b/core/tests/switch_core/db/test_tenant_lookup.py @@ -103,6 +103,15 @@ "tenant_of_messaging_install": "c8a4e21f6d30", } +# A third kind of change, and the quietest: a lookup whose body a later +# revision replaced. Creation and removal both show up as a function that is +# there or is not; a redefinition leaves a function of the right name and +# signature answering a different question, which nothing about the shape of +# the schema reveals. So the revision that last wrote the body is named here +# and compared against the module, while the creating revision above goes on +# owning the drop. +_REDEFINED_SINCE = {"tenant_of_messaging_install": "a7f2c3e9b481"} + def _revision_module(revision: str) -> ModuleType: """One revision module, loaded through Alembic. @@ -722,17 +731,24 @@ def test_the_added_lookup_is_installed_by_a_revision_and_not_only_here( Driven by `_ADDED_SINCE` rather than naming one lookup, so the next entry is covered by adding it there and nowhere else. + + The body is compared against `_REDEFINED_SINCE` where there is an + entry, because the revision that creates a function is not always the + one that last says what it does. The drop is still the creating + revision's: a redefinition replaces a body and leaves the function it + replaced nothing to undo. """ for name, revision in _ADDED_SINCE.items(): lookup = TENANT_LOOKUPS_BY_NAME[name] - module = _revision_module(revision) + defining = _REDEFINED_SINCE.get(name, revision) + module = _revision_module(defining) assert getattr(module, f"CREATE_{name.upper()}") == create_lookup_ddl( lookup ), ( - f"revision {revision} would install {name} with different DDL " + f"revision {defining} would install {name} with different DDL " "from the one db/tenant_lookup.py builds." ) - assert getattr(module, f"DROP_{name.upper()}") == ( + assert getattr(_revision_module(revision), f"DROP_{name.upper()}") == ( f"DROP FUNCTION IF EXISTS {lookup.signature}" ) From 7371f1e5f943cfd3ee589e0ca828ce2d73625ae3 Mon Sep 17 00:00:00 2001 From: Petr Bauch Date: Tue, 15 Sep 2026 09:47:58 -0400 Subject: [PATCH 2/3] feat(messaging): let an install end (CHOO-2626) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit An install could be made and never unmade. Disconnecting one now revokes the credential at the platform, removes the bridge, and marks the row ended so the workspace is free to be installed again — and the two events by which a platform says an install is over are read the same way. The order is deliberate. The platform is told first, so a refusal nobody understands leaves the install exactly as it was and the operator can try again rather than finding the bridge gone and the token still live. The row is ended next, which also releases its pointer at the bridge — the foreign key has no ON DELETE, so nothing could delete the bridge while the install still named it. The bridge goes last: if that fails, what is left is a credential-less bridge an operator can see and remove, not an install still claiming a workspace it has been thrown out of. Ending twice is success on both paths. Slack redelivers its uninstall event, and an operator can click disconnect on a row a redelivery ended a second earlier; neither is a fault to report. Co-Authored-By: Claude Opus 5 --- .../bridges/collaboration/install.py | 43 +++ .../bridges/collaboration/install_service.py | 151 +++++++++- .../bridges/collaboration/slack/install.py | 55 ++++ .../db/stores/messaging_install_store.py | 81 +++++- .../collaboration/test_install_service.py | 262 ++++++++++++++++++ 5 files changed, 587 insertions(+), 5 deletions(-) diff --git a/core/switch_core/bridges/collaboration/install.py b/core/switch_core/bridges/collaboration/install.py index 8abdcffee..aa1b2d6fa 100644 --- a/core/switch_core/bridges/collaboration/install.py +++ b/core/switch_core/bridges/collaboration/install.py @@ -205,6 +205,31 @@ async def redeem(self, *, code: str, redirect_uri: str) -> InstallGrant: grant, including a well-formed response the platform marked as failed. """ + @abstractmethod + async def revoke(self, *, bot_token: str) -> None: + """Tell the platform the credential it granted us is finished with. + + Called when an operator disconnects an install. Deleting our copy is + not the same act: the token stays valid at the platform, so a dump + taken before the disconnect would still hold a working key into a + customer's workspace. This is the half only the platform can do. + + **A token the platform already considers dead is a success, not a + failure.** The common reason to be disconnecting at all is that the + customer removed the app on their side, and an implementation that + raised on "this token is already invalid" would make exactly that + install impossible to disconnect, forever. + + Raise :class:`MessagingInstallError` for anything else — a refusal we + do not understand leaves a live credential behind and the operator + should hear about it rather than see a disconnect that reports + success. + + Not the same thing as the platform's own uninstall. Revoking a token + does not remove the app from the workspace; it ends this deployment's + access with it. + """ + @abstractmethod def verify_webhook(self, *, headers: Mapping[str, str], body: bytes) -> None: """Prove an inbound event came from the platform, or raise. @@ -245,6 +270,24 @@ def workspace_of_event(self, payload: Mapping[str, object]) -> str: event we cannot route is not an event we may guess at. """ + @abstractmethod + def revocation_of_event(self, payload: Mapping[str, object]) -> str | None: + """Why this event says the install is over, or `None` if it does not. + + Platforms report the end of an install as an ordinary event on the + ordinary endpoint, which makes it easy to treat as one: the app is + removed from a workspace, an event arrives saying so, nothing reads + it, and the deployment goes on holding a dead token, a running bridge + and a claim on a workspace whose owner believes they have left. + + A reason rather than a flag because it is the thing worth logging — an + operator asking why their bridge stopped needs the platform's own + answer, and there is more than one way an install can end. + + Runs on a payload that has been authenticated and not yet routed, so + like :meth:`workspace_of_event` it must not touch the database. + """ + @abstractmethod def connection_config(self, grant: InstallGrant) -> dict[str, object]: """Render a grant as the connection config this platform's adapter takes. diff --git a/core/switch_core/bridges/collaboration/install_service.py b/core/switch_core/bridges/collaboration/install_service.py index 55c37b5bc..ba83e198e 100644 --- a/core/switch_core/bridges/collaboration/install_service.py +++ b/core/switch_core/bridges/collaboration/install_service.py @@ -26,6 +26,27 @@ it is an orphan nothing can revoke, whereas an install row with no bridge is a recorded credential waiting to be used, which is a state the schema already allows for. + +**Ending one runs the same steps backwards, and there are two ways in.** An +operator disconnects here, or the platform tells us the app is gone. They +differ in exactly one step — whether there is a live token to revoke — and in +nothing else, because both have to leave the same state behind: no bridge, no +stored credential, a released workspace and a row saying what happened. A +deployment that handled only the first would go on holding a dead token and a +claim on a workspace whose owner believes they have left. + +The revoking order is: tell the platform first, then destroy things. A +revocation we do not understand then leaves the install exactly as it was and +the operator can try again, where the other order would have removed the +bridge and left a working key into a customer's workspace in whatever dump was +taken next. + +Then the row, and the bridge last — the reverse of the building order and for +the same reason read backwards. The install's pointer at the bridge is a real +foreign key, so the row has to let go before the bridge can be deleted at all; +and if the deletion then fails, what is left is a bridge with no credential +that an operator can see and remove, rather than an install still claiming a +workspace it has already been thrown out of. """ from __future__ import annotations @@ -53,10 +74,15 @@ from switch_core.bridges.collaboration.lifecycle_service import ( CollaborationBridgeLifecycleService, ) -from switch_core.crypto import encrypt_token +from switch_core.crypto import decrypt_token, encrypt_token from switch_core.db.models import MessagingInstall from switch_core.db.session_scope import tenant_session -from switch_core.db.stores.messaging_install_store import MessagingInstallStore +from switch_core.db.stores.messaging_install_store import ( + INSTALL_ACTIVE, + INSTALL_DISCONNECTED, + INSTALL_REVOKED, + MessagingInstallStore, +) from switch_core.db.tenant_lookup import tenant_of_messaging_install from switch_core.tenant_context import no_tenant, tenant_scope @@ -229,6 +255,127 @@ async def complete( ) return attached + # ── Ending an install ──────────────────────────────────────────────────── + + async def disconnect(self, *, tenant_id: str, install_id: str) -> MessagingInstall: + """End an install because somebody here said to. + + Idempotent: disconnecting an install that has already ended returns it + unchanged. An operator can reach this for a row the platform's own + `app_uninstalled` ended a second earlier, and that is not an error to + show them. + + **Removing the bridge detaches every room that used it**, which become + internal-only. That is the honest consequence of disconnecting a + messaging app and it is not softened here — but it is the reason this + is an explicit action with a confirmation in front of it rather than + something inferred. + """ + async with tenant_session(self._session_factory, tenant_id) as session: + install = await self._store.get(session, install_id=install_id) + platform = install.platform + workspace_id = install.external_workspace_id + bridge_id = install.bridge_id + token = install.encrypted_bot_token + already_ended = install.status != INSTALL_ACTIVE + + if already_ended: + logger.info( + "Install %s of %s workspace %s had already ended; nothing to do", + install_id, + platform, + workspace_id, + ) + async with tenant_session(self._session_factory, tenant_id) as session: + return await self._store.get(session, install_id=install_id) + + with tenant_scope(tenant_id): + if token is not None: + await self._installers.get(platform).revoke( + bot_token=decrypt_token(token, self._secret) + ) + + async with tenant_session(self._session_factory, tenant_id) as session: + ended = await self._store.end( + session, install_id=install_id, status=INSTALL_DISCONNECTED + ) + await session.commit() + + if bridge_id is not None: + await self._lifecycle.remove(bridge_id) + + logger.info( + "Disconnected %s workspace %s for tenant %s", + platform, + workspace_id, + tenant_id, + ) + return ended + + async def revoked(self, *, platform: str, workspace_id: str, reason: str) -> None: + """End an install because the platform said it is over. + + Resolves the workspace itself rather than going through `resolve`, + which is built for delivering an event and insists on a running bridge. + Here a missing bridge is beside the point: the news is that the install + is finished, and a deployment that could only record that while the + bridge happened to be up would keep the dead ones it most needs to + clear. + + A workspace that resolves to nobody is the ordinary case, not a fault. + The platform retries these events, so the second delivery arrives after + the first has already ended the install. + + No revocation call: the token this would revoke is the one the platform + has just told us it killed. + """ + tenant_id = await tenant_of_messaging_install( + self._session_factory, platform, workspace_id + ) + if tenant_id is None: + logger.info( + "Ignoring end-of-install for %s workspace %s (%s): no tenant " + "holds it, so it has already ended", + platform, + workspace_id, + reason, + ) + return + + with tenant_scope(tenant_id): + async with tenant_session(self._session_factory, tenant_id) as session: + install = await self._store.get_for_workspace( + session, platform=platform, external_workspace_id=workspace_id + ) + if install is None: + logger.warning( + "End-of-install for %s workspace %s resolved to tenant " + "%s and could not then be read as that tenant", + platform, + workspace_id, + tenant_id, + ) + return + install_id = install.id + bridge_id = install.bridge_id + + async with tenant_session(self._session_factory, tenant_id) as session: + await self._store.end( + session, install_id=install_id, status=INSTALL_REVOKED + ) + await session.commit() + + if bridge_id is not None: + await self._lifecycle.remove(bridge_id) + + logger.warning( + "Ended the install of %s workspace %s for tenant %s: %s", + platform, + workspace_id, + tenant_id, + reason, + ) + # ── Inbound events ─────────────────────────────────────────────────────── # # The other direction, and the one that runs constantly. Three steps, kept diff --git a/core/switch_core/bridges/collaboration/slack/install.py b/core/switch_core/bridges/collaboration/slack/install.py index 492e8891b..848aada72 100644 --- a/core/switch_core/bridges/collaboration/slack/install.py +++ b/core/switch_core/bridges/collaboration/slack/install.py @@ -39,6 +39,21 @@ AUTHORIZE_URL = "https://slack.com/oauth/v2/authorize" +#: Slack's ways of saying the token is finished already. Revoking one of these +#: is the outcome the caller wanted, so it is a success — and it is the usual +#: case, because the reason to be disconnecting is often that the customer +#: removed the app first. +_ALREADY_DEAD = frozenset({"invalid_auth", "token_revoked", "account_inactive"}) + +#: The two events by which Slack says an install is over, and what each means +#: in words an operator can read. `app_uninstalled` is the customer removing +#: the app; `tokens_revoked` is the narrower case of the tokens being killed +#: while the app stays. Both leave us unable to act in the workspace. +_REVOCATION_EVENTS: dict[str, str] = { + "app_uninstalled": "the app was removed from the Slack workspace", + "tokens_revoked": "Slack revoked this workspace's tokens", +} + def _json_object(raw: bytes) -> dict[str, Any]: try: @@ -163,6 +178,31 @@ async def redeem(self, *, code: str, redirect_uri: str) -> InstallGrant: scopes=response.get("scope") or "", ) + async def revoke(self, *, bot_token: str) -> None: + try: + response = await AsyncWebClient(token=bot_token).auth_revoke() + except SlackApiError as error: + reason = error.response.get("error", "") + if reason in _ALREADY_DEAD: + logger.info( + "Slack reports the bot token was already invalid (%s); " + "treating the revocation as done", + reason, + ) + return + raise MessagingInstallError( + f"Slack refused to revoke the bot token: {reason or error}" + ) from error + + # `revoked: false` with `ok: true` is Slack accepting the call and + # telling you it did nothing — which for a disconnect is the whole of + # what was asked for, so it cannot be read off `ok` alone. + if not response.get("revoked"): + raise MessagingInstallError( + "Slack accepted the revocation request and reported the token " + "was not revoked, so it is still valid." + ) + def verify_webhook(self, *, headers: Mapping[str, str], body: bytes) -> None: try: valid = self._verifier.is_valid_request(body, dict(headers)) @@ -236,6 +276,21 @@ def workspace_of_event(self, payload: Mapping[str, object]) -> str: raise WebhookPayloadError("Slack event names no workspace") return workspace_id + def revocation_of_event(self, payload: Mapping[str, object]) -> str | None: + """Read the two end-of-install events out of an Events API envelope. + + Only `event_callback` envelopes carry one. A slash command or an + interaction cannot say the app was uninstalled — there would be nobody + left to press the button — so the inner `event.type` is the only place + worth looking, and reading a `type` from anywhere else would let an + interaction payload with the right-looking field disconnect a + customer's workspace. + """ + event = payload.get("event") + if not isinstance(event, dict): + return None + return _REVOCATION_EVENTS.get(event.get("type", "")) + def connection_config(self, grant: InstallGrant) -> dict[str, object]: return { "bot_token": grant.bot_token, diff --git a/core/switch_core/db/stores/messaging_install_store.py b/core/switch_core/db/stores/messaging_install_store.py index b91a6d2a5..714e9c0c2 100644 --- a/core/switch_core/db/stores/messaging_install_store.py +++ b/core/switch_core/db/stores/messaging_install_store.py @@ -46,6 +46,16 @@ class MessagingInstallStateError(RuntimeError): """A state could not be redeemed: already used, expired, or not ours.""" +class MessagingInstallNotFound(RuntimeError): + """No install of that id is visible to the tenant bound on this session. + + "Not visible" rather than "does not exist", and the two are deliberately + not distinguished: the reads here are scoped, so an id belonging to + another tenant misses exactly as an invented one does. Telling them apart + would be a way to ask whether an id exists elsewhere. + """ + + class MessagingInstallStore: async def start_install( self, session: AsyncSession, *, platform: str, user_id: str @@ -171,13 +181,78 @@ async def get_for_workspace( ) return result.scalars().one_or_none() + async def get(self, session: AsyncSession, *, install_id: str) -> MessagingInstall: + """One of the bound tenant's installs, by id, or raise.""" + install = await session.get(MessagingInstall, install_id) + if install is None: + raise MessagingInstallNotFound( + f"no install {install_id} belongs to this organisation" + ) + return install + + async def list_for_tenant(self, session: AsyncSession) -> list[MessagingInstall]: + """Every install this tenant has ever made, newest first. + + Ended ones included, and that is the point of the method rather than a + side effect. An operator looking at this list is usually looking at it + because something stopped working, and a list of only the live installs + answers "there is nothing here" to the question "what happened to the + one that was here yesterday". + """ + result = await session.execute( + select(MessagingInstall).order_by( + MessagingInstall.installed_at.desc(), MessagingInstall.id + ) + ) + return list(result.scalars()) + + async def end( + self, session: AsyncSession, *, install_id: str, status: str + ) -> MessagingInstall: + """Mark an install finished, and discard the credential and bridge with it. + + `status` says which of the two ways it ended — see + `INSTALL_DISCONNECTED` and `INSTALL_REVOKED`. Anything else is a + programming error rather than a state the column may hold: the + uniqueness index reads `status = 'active'`, so a typo here would leave + a row that is neither serving nor releasing its workspace. + + **Ending an install twice is success.** The platform retries the event + that says an app was uninstalled, and an operator can click disconnect + on a row a retry has already ended; both must reach the same place. + That is also why this is an ordinary read-then-write rather than the + single conditional statement `redeem_state` uses — there is no race to + lose here, because two writers racing to end the same install both want + what the other is doing. + """ + if status not in (INSTALL_DISCONNECTED, INSTALL_REVOKED): + raise ValueError( + f"{status!r} is not a way an install can end; expected " + f"{INSTALL_DISCONNECTED!r} or {INSTALL_REVOKED!r}" + ) + install = await self.get(session, install_id=install_id) + if install.status != INSTALL_ACTIVE: + return install + + install.status = status + install.ended_at = datetime.now(UTC) + # The token is worthless the moment the platform is told so, and a + # worthless credential still reads like a live one to whoever finds + # the dump. Keeping the row is the record; keeping the secret is not + # part of it. + install.encrypted_bot_token = None + # And the pointer goes with it, because the bridge is about to. The + # foreign key has no `ON DELETE`, so a row still naming the bridge is + # what would refuse its deletion. + install.bridge_id = None + await session.flush() + return install + async def attach_bridge( self, session: AsyncSession, *, install_id: str, bridge_id: str ) -> MessagingInstall: """Point an install at the bridge now serving it.""" - install = await session.get(MessagingInstall, install_id) - if install is None: - raise MessagingInstallStateError(f"install not found: {install_id}") + install = await self.get(session, install_id=install_id) install.bridge_id = bridge_id await session.flush() return install diff --git a/core/tests/switch_core/bridges/collaboration/test_install_service.py b/core/tests/switch_core/bridges/collaboration/test_install_service.py index 9f8897369..c7f0252f6 100644 --- a/core/tests/switch_core/bridges/collaboration/test_install_service.py +++ b/core/tests/switch_core/bridges/collaboration/test_install_service.py @@ -26,6 +26,7 @@ InstallGrant, MessagingAppInstaller, MessagingInstallerRegistry, + MessagingInstallError, WebhookEndpoint, ) from switch_core.bridges.collaboration.install_service import ( @@ -47,7 +48,11 @@ ) from switch_core.db.session_scope import tenant_session from switch_core.db.stores.messaging_install_store import ( + INSTALL_ACTIVE, + INSTALL_DISCONNECTED, + INSTALL_REVOKED, MessagingInstallClaimedError, + MessagingInstallNotFound, MessagingInstallStateError, MessagingInstallStore, ) @@ -67,6 +72,8 @@ class _FakeInstaller(MessagingAppInstaller): def __init__(self, workspace_id: str) -> None: self.workspace_id = workspace_id self.redeem_calls: list[str] = [] + self.revoked_tokens: list[str] = [] + self.revoke_error: Exception | None = None def authorize_url(self, *, state: str, redirect_uri: str) -> str: return f"https://platform.example/authorize?state={state}" @@ -80,6 +87,19 @@ async def redeem(self, *, code: str, redirect_uri: str) -> InstallGrant: scopes="chat:write", ) + async def revoke(self, *, bot_token: str) -> None: + if self.revoke_error is not None: + raise self.revoke_error + self.revoked_tokens.append(bot_token) + + def revocation_of_event(self, payload: Mapping[str, object]) -> str | None: + event = payload.get("event") + if not isinstance(event, dict): + return None + if event.get("type") == "app_uninstalled": + return "the app was removed" + return None + def verify_webhook(self, *, headers: Mapping[str, str], body: bytes) -> None: return None @@ -110,6 +130,7 @@ def __init__( self._tenant_id = tenant_id self._suffix = suffix self.registered: list[dict[str, object]] = [] + self.removed: list[str] = [] async def register(self, **kwargs: object) -> CollaborationBridge: self.registered.append(kwargs) @@ -134,6 +155,20 @@ async def register(self, **kwargs: object) -> CollaborationBridge: await session.commit() return CollaborationBridge(id=bridge_id) + async def remove(self, bridge_id: str) -> None: + """Delete the row, not just record the call. + + The install points at the bridge through a foreign key with no + `ON DELETE`, so a stub that only counted removals would let a caller + that forgot to release the pointer pass here and fail in production. + """ + self.removed.append(bridge_id) + async with tenant_session(self._factory, self._tenant_id) as session: + bridge = await session.get(CollaborationBridge, bridge_id) + if bridge is not None: + await session.delete(bridge) + await session.commit() + class _Fixture: def __init__(self) -> None: @@ -187,6 +222,23 @@ async def _begin(factory: async_sessionmaker, fixture: _Fixture, tenant_id: str) return parse_qs(urlparse(url).query)["state"][0] +async def _installed( + factory: async_sessionmaker, fixture: _Fixture, tenant_id: str +) -> MessagingInstall: + """Run both legs and return the install they produced.""" + state = await _begin(factory, fixture, tenant_id) + return await fixture.service.complete( + platform="slack", code="the-code", state_token=state + ) + + +async def _reread( + factory: async_sessionmaker, tenant_id: str, install_id: str +) -> MessagingInstall: + async with tenant_session(factory, tenant_id) as session: + return await MessagingInstallStore().get(session, install_id=install_id) + + class TestTheRoundTrip: async def test_an_install_lands_in_the_tenant_that_started_it( self, rls_harness: RLSHarness @@ -359,3 +411,213 @@ async def test_a_workspace_another_tenant_holds_is_refused( platform="slack", code="the-code", state_token=state ) assert fixture.lifecycle.registered == [] + + +class TestDisconnecting: + """An install an operator ended, and what has to be true afterwards.""" + + async def test_the_platform_is_told_before_anything_is_destroyed( + self, rls_harness: RLSHarness + ) -> None: + """Deleting our copy of a token does not stop it working. + + Revoking is the half only the platform can do, so a disconnect that + skipped it would leave a live key into a customer's workspace in every + backup taken before it. + """ + fixture = await _fixture(rls_harness) + install = await _installed(rls_harness.restricted, fixture, fixture.tenant_a) + + await fixture.service.disconnect( + tenant_id=fixture.tenant_a, install_id=install.id + ) + + assert fixture.installer.revoked_tokens == ["xoxb-granted"] + + async def test_it_leaves_no_credential_and_no_bridge( + self, rls_harness: RLSHarness + ) -> None: + fixture = await _fixture(rls_harness) + install = await _installed(rls_harness.restricted, fixture, fixture.tenant_a) + bridge_id = install.bridge_id + + ended = await fixture.service.disconnect( + tenant_id=fixture.tenant_a, install_id=install.id + ) + + assert ended.status == INSTALL_DISCONNECTED + assert ended.ended_at is not None + assert ended.encrypted_bot_token is None + assert ended.bridge_id is None + assert fixture.lifecycle.removed == [bridge_id] + + async def test_the_workspace_can_be_installed_again( + self, rls_harness: RLSHarness + ) -> None: + """The reason the uniqueness index is partial. + + A customer who disconnects and changes their mind must be able to + click Add to Slack again, and a row that went on occupying the + workspace would mean nobody ever could. + """ + fixture = await _fixture(rls_harness) + first = await _installed(rls_harness.restricted, fixture, fixture.tenant_a) + await fixture.service.disconnect( + tenant_id=fixture.tenant_a, install_id=first.id + ) + + second = await _installed(rls_harness.restricted, fixture, fixture.tenant_a) + + assert second.id != first.id + assert second.status == INSTALL_ACTIVE + assert second.bridge_id is not None + + async def test_disconnecting_twice_is_success( + self, rls_harness: RLSHarness + ) -> None: + """An operator can click it on a row the platform ended a second ago.""" + fixture = await _fixture(rls_harness) + install = await _installed(rls_harness.restricted, fixture, fixture.tenant_a) + await fixture.service.disconnect( + tenant_id=fixture.tenant_a, install_id=install.id + ) + + again = await fixture.service.disconnect( + tenant_id=fixture.tenant_a, install_id=install.id + ) + + assert again.status == INSTALL_DISCONNECTED + assert fixture.installer.revoked_tokens == ["xoxb-granted"] + assert len(fixture.lifecycle.removed) == 1 + + async def test_a_refusal_nobody_understands_leaves_the_install_intact( + self, rls_harness: RLSHarness + ) -> None: + """Which is the whole reason the platform is told first. + + The operator sees the failure and can try again; the alternative is a + bridge already gone and a token still valid. + """ + fixture = await _fixture(rls_harness) + install = await _installed(rls_harness.restricted, fixture, fixture.tenant_a) + fixture.installer.revoke_error = MessagingInstallError("Slack said no") + + with pytest.raises(MessagingInstallError): + await fixture.service.disconnect( + tenant_id=fixture.tenant_a, install_id=install.id + ) + + still = await _reread(rls_harness.restricted, fixture.tenant_a, install.id) + assert still.status == INSTALL_ACTIVE + assert still.encrypted_bot_token is not None + assert still.bridge_id is not None + assert fixture.lifecycle.removed == [] + + async def test_another_tenants_install_is_not_disconnectable( + self, rls_harness: RLSHarness + ) -> None: + """And misses the way an invented id would, telling the caller nothing.""" + fixture = await _fixture(rls_harness) + install = await _installed(rls_harness.restricted, fixture, fixture.tenant_a) + + with pytest.raises(MessagingInstallNotFound): + await fixture.service.disconnect( + tenant_id=fixture.tenant_b, install_id=install.id + ) + + assert fixture.installer.revoked_tokens == [] + + +class TestThePlatformEndingIt: + async def test_an_uninstall_event_ends_the_install( + self, rls_harness: RLSHarness + ) -> None: + fixture = await _fixture(rls_harness) + install = await _installed(rls_harness.restricted, fixture, fixture.tenant_a) + + await fixture.service.revoked( + platform="slack", + workspace_id=fixture.workspace, + reason="the app was removed", + ) + + ended = await _reread(rls_harness.restricted, fixture.tenant_a, install.id) + assert ended.status == INSTALL_REVOKED + assert ended.encrypted_bot_token is None + assert ended.bridge_id is None + assert fixture.lifecycle.removed == [install.bridge_id] + + async def test_nothing_is_revoked_at_a_platform_that_already_did_it( + self, rls_harness: RLSHarness + ) -> None: + fixture = await _fixture(rls_harness) + await _installed(rls_harness.restricted, fixture, fixture.tenant_a) + + await fixture.service.revoked( + platform="slack", + workspace_id=fixture.workspace, + reason="the app was removed", + ) + + assert fixture.installer.revoked_tokens == [] + + async def test_a_retried_event_is_not_an_error( + self, rls_harness: RLSHarness + ) -> None: + """Slack redelivers, so the second one arrives after the workspace is free.""" + fixture = await _fixture(rls_harness) + await _installed(rls_harness.restricted, fixture, fixture.tenant_a) + await fixture.service.revoked( + platform="slack", workspace_id=fixture.workspace, reason="removed" + ) + + await fixture.service.revoked( + platform="slack", workspace_id=fixture.workspace, reason="removed" + ) + + assert len(fixture.lifecycle.removed) == 1 + + async def test_a_workspace_nobody_installed_is_ignored( + self, rls_harness: RLSHarness + ) -> None: + fixture = await _fixture(rls_harness) + + await fixture.service.revoked( + platform="slack", workspace_id="T-nobody", reason="removed" + ) + + assert fixture.lifecycle.removed == [] + + +class TestWhatTheOperatorSees: + async def test_the_list_keeps_installs_that_ended( + self, rls_harness: RLSHarness + ) -> None: + """A list of only the live ones answers "nothing here" to "what happened + to the one that was here yesterday".""" + fixture = await _fixture(rls_harness) + first = await _installed(rls_harness.restricted, fixture, fixture.tenant_a) + await fixture.service.disconnect( + tenant_id=fixture.tenant_a, install_id=first.id + ) + second = await _installed(rls_harness.restricted, fixture, fixture.tenant_a) + + async with tenant_session(rls_harness.restricted, fixture.tenant_a) as session: + listed = await MessagingInstallStore().list_for_tenant(session) + + assert {install.id for install in listed} == {first.id, second.id} + assert {install.status for install in listed} == { + INSTALL_ACTIVE, + INSTALL_DISCONNECTED, + } + + async def test_the_list_is_the_bound_tenants_own( + self, rls_harness: RLSHarness + ) -> None: + fixture = await _fixture(rls_harness) + await _installed(rls_harness.restricted, fixture, fixture.tenant_a) + + async with tenant_session(rls_harness.restricted, fixture.tenant_b) as session: + listed = await MessagingInstallStore().list_for_tenant(session) + + assert listed == [] From 33e49929f2bfcdfec55388f5dd5b96b0470d883d Mon Sep 17 00:00:00 2001 From: Petr Bauch Date: Tue, 15 Sep 2026 09:54:29 -0400 Subject: [PATCH 3/3] feat(messaging): routes for ending an install (CHOO-2626) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The operator's two: list this organisation's installs, ended ones included, and disconnect one. The list keeps ended rows deliberately — somebody looking at it is usually looking because something stopped working, and "nothing here" is the wrong answer to "what happened to the bridge that was here yesterday". Neither carries a token or anything derived from one. On the webhook side, an uninstall is now read before the event is routed rather than after. It has to be: resolving insists on a running bridge, and this is the one event that arrives as the bridge goes away, so the ordinary path would drop it exactly when it mattered. An event for a workspace nobody holds now answers 200 instead of 404. That is the single place this endpoint says something other than what happened, and it is deliberate: an app left in a workspace whose install ended posts for as long as someone leaves it there, the platform cannot act on the refusal, and it counts refusals against the app as a whole — so the honest answer would be paid for by every other customer's delivery. The drop is in the log. Co-Authored-By: Claude Opus 5 --- .../bridges/collaboration/install_routes.py | 39 +++++- .../bridges/collaboration/install_service.py | 42 ++++++- .../switch_core/gateway/messaging_installs.py | 84 ++++++++++++- .../collaboration/test_install_webhook.py | 118 ++++++++++++++++-- 4 files changed, 269 insertions(+), 14 deletions(-) diff --git a/core/switch_core/bridges/collaboration/install_routes.py b/core/switch_core/bridges/collaboration/install_routes.py index 5c7029a7d..899ec55fe 100644 --- a/core/switch_core/bridges/collaboration/install_routes.py +++ b/core/switch_core/bridges/collaboration/install_routes.py @@ -42,6 +42,7 @@ from switch_core.bridges.collaboration.install_service import ( InstallPlatformMismatch, MessagingInstallService, + Revocation, WebhookBridgeUnavailable, WebhookTarget, WebhookWorkspaceUnknown, @@ -175,6 +176,29 @@ async def _deliver(target: WebhookTarget, event: InboundWebhook) -> None: target.tenant_id, ) + async def _end_install(platform: str, revocation: Revocation) -> None: + """Act on the platform's news after it has been acknowledged. + + Logged and not raised for the same reason as `_deliver`, and with more + at stake: the platform redelivers what it gets no answer to, and a + traceback out of a background task would leave a dead install claiming + a workspace with nothing in the log naming it. + """ + try: + await service.revoked( + platform=platform, + workspace_id=revocation.workspace_id, + reason=revocation.reason, + ) + except Exception: + logger.exception( + "Failed to end the install of %s workspace %s after the platform " + "reported it was over (%s)", + platform, + revocation.workspace_id, + revocation.reason, + ) + async def _inbound( platform: str, endpoint: WebhookEndpoint, @@ -217,6 +241,13 @@ async def _inbound( return PlainTextResponse(event.handshake) try: + # Before resolving, because this is the one event that arrives as + # the bridge it would be resolved to is going away. + revocation = service.revocation(platform=platform, event=event) + if revocation is not None: + background.add_task(_end_install, platform, revocation) + return Response(status_code=200) + target = await service.resolve(platform=platform, event=event) except WebhookPayloadError as failure: logger.error( @@ -224,8 +255,14 @@ async def _inbound( ) return Response(status_code=400) except WebhookWorkspaceUnknown as failure: + # A 200 for an event that reached nobody, which is the one place + # this file answers something other than what happened. The app + # left behind in a workspace whose install ended goes on posting, + # the platform cannot act on a 404, and it counts the refusals + # against the app as a whole — so the honest answer costs every + # other customer's delivery. The log is where it is visible. logger.warning("Dropped a %s event: %s", platform, failure) - return Response(status_code=404) + return Response(status_code=200) except WebhookBridgeUnavailable as failure: # Deliberately a 503: the platform retrying is the right behaviour # while a bridge restarts, and a 200 here would drop a real message diff --git a/core/switch_core/bridges/collaboration/install_service.py b/core/switch_core/bridges/collaboration/install_service.py index ba83e198e..bacbb1fbe 100644 --- a/core/switch_core/bridges/collaboration/install_service.py +++ b/core/switch_core/bridges/collaboration/install_service.py @@ -94,8 +94,10 @@ class WebhookWorkspaceUnknown(RuntimeError): Ordinary rather than alarming: an app left in a workspace whose install was removed goes on posting for as long as someone leaves it there. It is still - an error, because the alternative is answering "fine" to traffic that - reaches nobody. + an error here, because a resolver that returned nothing would make "nobody + holds this" and "here is where it goes" the same shape — but it is one the + route absorbs rather than reports, since the platform cannot fix it and + telling it repeatedly that its posts fail is held against the app itself. """ @@ -108,6 +110,14 @@ class WebhookBridgeUnavailable(RuntimeError): """ +@dataclass(frozen=True) +class Revocation: + """A platform saying, in an ordinary event, that an install is over.""" + + workspace_id: str + reason: str + + @dataclass(frozen=True) class WebhookTarget: """Where one verified event goes: a tenant, a bridge, and its live adapter.""" @@ -255,6 +265,14 @@ async def complete( ) return attached + async def list_installs(self, session: AsyncSession) -> list[MessagingInstall]: + """The bound tenant's installs, for the operator's own list. + + On the caller's session like `begin`, and scoped by it: there is no + tenant argument because there is no tenant to choose. + """ + return await self._store.list_for_tenant(session) + # ── Ending an install ──────────────────────────────────────────────────── async def disconnect(self, *, tenant_id: str, install_id: str) -> MessagingInstall: @@ -410,6 +428,26 @@ def authenticate( installer.verify_webhook(headers=headers, body=body) return installer.parse_webhook(endpoint=endpoint, body=body) + def revocation(self, *, platform: str, event: InboundWebhook) -> Revocation | None: + """Whether this event is the platform ending the install, and for whom. + + Asked before `resolve` and not after, because the two disagree about + what a missing bridge means. `resolve` insists on one, and the event + most worth reading here is precisely the one that arrives when the + bridge is on its way out — so an uninstall routed through the ordinary + path would be dropped exactly when it mattered. + + Pure, like `authenticate`: it reads the payload and nothing else, so + the route can answer the platform before any of the work begins. + """ + installer = self._installers.get(platform) + reason = installer.revocation_of_event(event.payload) + if reason is None: + return None + return Revocation( + workspace_id=installer.workspace_of_event(event.payload), reason=reason + ) + async def resolve(self, *, platform: str, event: InboundWebhook) -> WebhookTarget: """Turn a workspace id into the one bridge entitled to the event. diff --git a/core/switch_core/gateway/messaging_installs.py b/core/switch_core/gateway/messaging_installs.py index 16ee7ffab..ac730188d 100644 --- a/core/switch_core/gateway/messaging_installs.py +++ b/core/switch_core/gateway/messaging_installs.py @@ -10,6 +10,7 @@ from __future__ import annotations import logging +from datetime import datetime from typing import Annotated from fastapi import APIRouter, Depends, HTTPException @@ -18,7 +19,8 @@ from switch_core.bridges.collaboration.install import MessagingInstallError from switch_core.bridges.collaboration.install_service import MessagingInstallService -from switch_core.db.models import User +from switch_core.db.models import User, require_tenant_id +from switch_core.db.stores.messaging_install_store import MessagingInstallNotFound from switch_core.gateway.auth import require_admin from switch_core.gateway.dependencies import get_install_service, get_session @@ -35,6 +37,30 @@ class InstallablePlatforms(BaseModel): platforms: list[str] +class InstalledApp(BaseModel): + """One install as an operator needs to see it. + + No token and nothing derived from one. `status` and `ended_at` are the + point of the shape rather than decoration: an operator looks at this list + because a bridge stopped working, and "disconnected" and "revoked" are the + difference between a decision someone here made and news from the + platform. + """ + + id: str + platform: str + external_workspace_id: str + status: str + scopes: str + bridge_id: str | None + installed_at: datetime + ended_at: datetime | None + + +class InstalledApps(BaseModel): + installs: list[InstalledApp] + + def _require_installs( service: MessagingInstallService | None, ) -> MessagingInstallService: @@ -84,3 +110,59 @@ async def begin_install( await session.commit() logger.info("Started a %s install for user %s", platform, user.id) return InstallStart(authorize_url=authorize_url) + + +@router.get("/installs") +async def list_installs( + session: Annotated[AsyncSession, Depends(get_session)], + service: Annotated[MessagingInstallService | None, Depends(get_install_service)], + _user: Annotated[User, Depends(require_admin)], +) -> InstalledApps: + """This organisation's installs, ended ones included. + + An empty list when the deployment registered no app of its own, rather + than the 501 the install endpoints answer with: a deployment can have + installs recorded from before its credentials were removed, and a page + that cannot even show them is worse than one with nothing on it. + """ + if service is None: + return InstalledApps(installs=[]) + return InstalledApps( + installs=[ + InstalledApp.model_validate(install, from_attributes=True) + for install in await service.list_installs(session) + ] + ) + + +@router.delete("/installs/{install_id}") +async def disconnect_install( + install_id: str, + service: Annotated[MessagingInstallService | None, Depends(get_install_service)], + user: Annotated[User, Depends(require_admin)], +) -> InstalledApp: + """End an install: revoke the credential, remove the bridge, free the workspace. + + Not on the request's session, deliberately. Disconnecting revokes a token + at the platform and tears a bridge down, and holding this request's + transaction open across both would keep a row locked while somebody else's + API is slow. The service opens what it needs, in the order failure can be + recovered from. + + **Rooms that used the bridge become internal-only**, which is why this is a + delete an operator has to ask for rather than anything inferred. + """ + try: + ended = await _require_installs(service).disconnect( + tenant_id=require_tenant_id(), install_id=install_id + ) + except MessagingInstallNotFound as missing: + raise HTTPException(status_code=404, detail=str(missing)) from missing + except MessagingInstallError as failure: + # The platform refused to revoke and nothing was destroyed, so this is + # upstream's answer rather than a fault here — and it is retryable, + # which the operator needs to be told rather than left to guess. + raise HTTPException(status_code=502, detail=str(failure)) from failure + + logger.info("User %s disconnected messaging install %s", user.id, install_id) + return InstalledApp.model_validate(ended, from_attributes=True) diff --git a/core/tests/switch_core/bridges/collaboration/test_install_webhook.py b/core/tests/switch_core/bridges/collaboration/test_install_webhook.py index f7d346af0..fd4244e92 100644 --- a/core/tests/switch_core/bridges/collaboration/test_install_webhook.py +++ b/core/tests/switch_core/bridges/collaboration/test_install_webhook.py @@ -126,12 +126,29 @@ async def dispatch_event( class _FakeLifecycle: - def __init__(self) -> None: + def __init__(self, factory: async_sessionmaker) -> None: + self._factory = factory self.adapters: dict[str, CollaborationAdapter] = {} + self.removed: list[str] = [] def get_adapter(self, bridge_id: str) -> CollaborationAdapter | None: return self.adapters.get(bridge_id) + async def remove(self, bridge_id: str) -> None: + """Delete the row, on an unscoped session like the real one. + + The real lifecycle opens a plain session and lets the tenant bound + around the call decide what it can see, so a fake that took a tenant + argument would not be exercising the same thing. + """ + self.removed.append(bridge_id) + self.adapters.pop(bridge_id, None) + async with self._factory() as session: + bridge = await session.get(CollaborationBridge, bridge_id) + if bridge is not None: + await session.delete(bridge) + await session.commit() + @dataclass class _Workspace: @@ -144,7 +161,7 @@ class _Workspace: class _Fixture: def __init__(self, harness: RLSHarness) -> None: self.harness = harness - self.lifecycle = _FakeLifecycle() + self.lifecycle = _FakeLifecycle(harness.restricted) self.a: _Workspace self.b: _Workspace self.client: httpx.AsyncClient @@ -254,6 +271,16 @@ def _event(workspace_id: str, text: str) -> bytes: ).encode() +def _uninstalled(workspace_id: str) -> bytes: + return json.dumps( + { + "type": "event_callback", + "team_id": workspace_id, + "event": {"type": "app_uninstalled"}, + } + ).encode() + + def _texts(adapter: _RecordingAdapter) -> list[str]: return [payload["event"]["text"] for _, payload in adapter.dispatched] @@ -389,21 +416,24 @@ async def test_a_signature_over_a_different_body_is_refused( assert fixture.a.adapter.dispatched == [] assert fixture.b.adapter.dispatched == [] - async def test_an_unknown_workspace_is_not_reported_as_handled( + async def test_an_unknown_workspace_is_dropped_without_telling_the_platform( self, rls_harness: RLSHarness ) -> None: - """404 rather than a polite 200. - - The app is still installed somewhere we no longer serve, and the - honest answer is that this event reached nobody. A 200 would make it - invisible on both sides — a platform's own delivery log is often the - only place anyone would see it. + """200, and the one place this endpoint answers something other than + what happened. + + The app is still installed in a workspace we no longer serve, so it + posts for as long as someone leaves it there. The platform cannot act + on a refusal — there is nothing for it to fix — and it counts the + refusals against the app as a whole, so the honest 404 would be paid + for by every other customer's delivery. It is dropped, and it is in the + log. """ fixture = await _fixture(rls_harness) response = await _post(fixture, events_path("slack"), _event("T-nobody", "x")) - assert response.status_code == 404 + assert response.status_code == 200 assert fixture.a.adapter.dispatched == [] assert fixture.b.adapter.dispatched == [] @@ -535,3 +565,71 @@ async def test_the_refusal_is_not_swallowed(self, rls_harness: RLSHarness) -> No target = await fixture.service.resolve(platform="slack", event=event) with pytest.raises(Exception, match="does not receive events over HTTP"): await fixture.service.deliver(target, event) + + +class TestThePlatformSayingTheInstallIsOver: + """`app_uninstalled` arrives on the same URL as everything else.""" + + async def test_it_ends_the_install_instead_of_dispatching( + self, rls_harness: RLSHarness + ) -> None: + fixture = await _fixture(rls_harness) + + response = await _post( + fixture, events_path("slack"), _uninstalled(fixture.a.workspace_id) + ) + + assert response.status_code == 200 + assert fixture.a.adapter.dispatched == [] + assert fixture.lifecycle.removed == [fixture.a.bridge_id] + + async with tenant_session( + rls_harness.restricted, fixture.a.tenant_id + ) as session: + installs = await MessagingInstallStore().list_for_tenant(session) + assert [install.status for install in installs] == ["revoked"] + + async def test_it_ends_nobody_elses(self, rls_harness: RLSHarness) -> None: + """The payload decides here as well, and the blast radius is larger. + + A revocation routed by anything but the workspace in the event would + disconnect a customer who did nothing. + """ + fixture = await _fixture(rls_harness) + + await _post(fixture, events_path("slack"), _uninstalled(fixture.a.workspace_id)) + await _post(fixture, events_path("slack"), _event(fixture.b.workspace_id, "b")) + + assert _texts(fixture.b.adapter) == ["b"] + assert fixture.lifecycle.removed == [fixture.a.bridge_id] + + async def test_the_workspace_stops_resolving_to_anyone( + self, rls_harness: RLSHarness + ) -> None: + """Which is what frees it to be installed again. + + An app removed from a workspace can still post on its way out, and an + ended install that went on answering for the workspace would route + those to a bridge that no longer exists. + """ + fixture = await _fixture(rls_harness) + await _post(fixture, events_path("slack"), _uninstalled(fixture.a.workspace_id)) + + trailing = await _post( + fixture, events_path("slack"), _event(fixture.a.workspace_id, "late") + ) + + assert trailing.status_code == 200 + assert fixture.a.adapter.dispatched == [] + + async def test_a_redelivery_is_not_an_error(self, rls_harness: RLSHarness) -> None: + """Slack retries what it is slow to hear back from, and it hears 200 + from the first delivery only after the install has already gone.""" + fixture = await _fixture(rls_harness) + body = _uninstalled(fixture.a.workspace_id) + + first = await _post(fixture, events_path("slack"), body) + second = await _post(fixture, events_path("slack"), body) + + assert (first.status_code, second.status_code) == (200, 200) + assert fixture.lifecycle.removed == [fixture.a.bridge_id]