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_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 55c37b5bc..bacbb1fbe 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 @@ -68,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. """ @@ -82,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.""" @@ -229,6 +265,135 @@ 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: + """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 @@ -263,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/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/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..714e9c0c2 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. @@ -34,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 @@ -103,17 +125,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,37 +150,109 @@ 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. + """The bound tenant's live 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. + 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() + 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/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/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/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/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 == [] 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] 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}" )