Connect events and next= on the OAuth door

backend/druks/services/routes.py

@@ -1,5 +1,5 @@
 from fastapi import APIRouter, Depends, HTTPException, Request
-from fastapi.responses import HTMLResponse, RedirectResponse
+from fastapi.responses import HTMLResponse, RedirectResponse, Response
 
 from druks.accounts.context import current_account_id
 from druks.accounts.dependencies import current_session_account
@@ -13,6 +13,7 @@
 from druks.services.models import OauthConnection, ServiceIdentity
 from druks.services.oauth import OauthClient, complete_connect
 from druks.services.schemas import ConnectionResponse, ServiceResponse
+from druks.signals import publish
 
 router = APIRouter(prefix="/api/services", tags=["services"])
 oauth_router = APIRouter(prefix="/api/oauth", tags=["oauth"])
@@ -53,7 +54,14 @@ async def connect_service(name: str, payload: dict[str, str]) -> ServiceResponse
         # A replaced client can never refresh the old client's connections.
         client = OauthClient(provider=name)
         for connection in OauthConnection.list_for_provider(name):
+            connection_id, account_id = connection.id, connection.account_id
             await client.disconnect(connection)
+            await publish(
+                "oauth.disconnected",
+                provider=name,
+                connection_id=connection_id,
+                account_id=account_id,
+            )
     return ServiceResponse.from_row(service, row)
 
 
@@ -66,7 +74,7 @@ def _get_oauth_service(name: str):
 
 @oauth_router.get("/{name}/connect", dependencies=[Depends(current_session_account)])
 async def connect_oauth_service(
-    name: str, request: Request, connection: str = ""
+    name: str, request: Request, connection: str = "", next: str = ""
 ) -> RedirectResponse:
     service = _get_oauth_service(name)
     account_id = current_account_id.get()
@@ -76,6 +84,9 @@ async def connect_oauth_service(
             raise HTTPException(
                 status_code=404, detail=f"No connection {connection!r} on {name!r}."
             )
+    if next and (not next.startswith("/") or next.startswith(("//", "/\\"))):
+        # A bare same-origin path only — anything host-shaped is an open redirect.
+        raise HTTPException(status_code=422, detail="next must be a path starting with '/'.")
     endpoint = request.app.state.settings.urls.endpoint
     if not endpoint:
         raise HTTPException(
@@ -90,13 +101,13 @@ async def connect_oauth_service(
     url = await client.begin_connect(
         redirect_uri=f"{endpoint.rstrip('/')}/api/oauth/callback",
         scopes=service.required_scopes(),
-        context={"account_id": account_id, "connection_id": connection},
+        context={"account_id": account_id, "connection_id": connection, "next": next},
     )
     return RedirectResponse(url)
 
 
 @oauth_router.get("/callback", response_class=HTMLResponse)
-async def oauth_callback(state: str = "", code: str = "", error: str = "") -> HTMLResponse:
+async def oauth_callback(state: str = "", code: str = "", error: str = "") -> Response:
     if error:
         raise HTTPException(
             status_code=400, detail=f"The authorization server denied the request: {error}"
@@ -112,7 +123,8 @@ async def oauth_callback(state: str = "", code: str = "", error: str = "") -> HT
         # A state begun by another door (an MCP connect) finishes at its own callback.
         raise HTTPException(status_code=400, detail=f"No OAuth service {provider!r}.")
     granted = tokens.get("scope", "").split() or pending["scopes"]
-    if pending["connection_id"]:
+    reconsent = bool(pending["connection_id"])
+    if reconsent:
         row = OauthConnection.get(pending["connection_id"])
         if not row:
             raise HTTPException(
@@ -122,12 +134,21 @@ async def oauth_callback(state: str = "", code: str = "", error: str = "") -> HT
         # A reconsent's narrower cached token must not serve until its TTL runs out.
         await OauthClient(provider=provider).evict_access_token(row.id)
     else:
-        OauthConnection.create(
+        row = OauthConnection.create(
             provider=provider,
             account_id=pending["account_id"],
             refresh_token=tokens["refresh_token"],
             scopes=granted,
         )
+    await publish(
+        "oauth.connected",
+        provider=provider,
+        connection_id=row.id,
+        account_id=row.account_id,
+        reconsent=reconsent,
+    )
+    if pending["next"]:
+        return RedirectResponse(pending["next"])
     return render_page("service_oauth_callback.html", name=provider)
 
 
@@ -146,4 +167,11 @@ async def disconnect_connection(connection_id: str) -> None:
     row = OauthConnection.get(connection_id)
     if not row:
         raise HTTPException(status_code=404, detail=f"No connection {connection_id!r}.")
-    await OauthClient(provider=row.provider).disconnect(row)
+    provider, account_id = row.provider, row.account_id
+    await OauthClient(provider=provider).disconnect(row)
+    await publish(
+        "oauth.disconnected",
+        provider=provider,
+        connection_id=connection_id,
+        account_id=account_id,
+    )

backend/tests/test_services.py

@@ -598,6 +598,12 @@ async def test_oauth_callback_creates_and_reconnects_a_connection(
     ServiceIdentity.connect(
         "acme", identity={"client_id": "id-1"}, secrets={"client_secret": "sec-1"}
     )
+    published = []
+
+    async def record(name, **kwargs):
+        published.append((name, kwargs))
+
+    monkeypatch.setattr("druks.services.routes.publish", record)
     await close_client()
     settings = make_settings(tmp_path, urls={"endpoint": "https://druks.example"})
     with TestClient(configure_app_for_test(settings=settings)) as client:
@@ -634,6 +640,16 @@ async def test_oauth_callback_creates_and_reconnects_a_connection(
         assert finish.status_code == 200
         assert len(OauthConnection.list_for_provider("acme")) == 1
 
+    assert [name for name, _ in published] == ["oauth.connected", "oauth.connected"]
+    fresh, reconsent = (kwargs for _, kwargs in published)
+    assert fresh == {
+        "provider": "acme",
+        "connection_id": connection.id,
+        "account_id": connection.account_id,
+        "reconsent": False,
+    }
+    assert reconsent["reconsent"] is True
+    assert reconsent["connection_id"] == connection.id
     druks.redis._client = None
     assert not await get_client().get(stale_key)
 
@@ -649,11 +665,17 @@ def test_oauth_connect_rejects_an_unknown_reconnect_target(tmp_path, acme, druks
         assert client.get("/api/oauth/acme/connect?connection=zzz").status_code == 404
 
 
-def test_connections_list_and_revoke(tmp_path, acme, druks_db):
+def test_connections_list_and_revoke(tmp_path, acme, druks_db, monkeypatch):
     from druks.accounts.models import Account
     from druks.services.models import OauthConnection
     from druks.testing import configure_app_for_test
 
+    published = []
+
+    async def record(name, **kwargs):
+        published.append((name, kwargs))
+
+    monkeypatch.setattr("druks.services.routes.publish", record)
     me = Account.get_or_create("op@example.com")
     with TestClient(configure_app_for_test(settings=make_settings(tmp_path))) as client:
         row = OauthConnection.create(
@@ -668,13 +690,28 @@ def test_connections_list_and_revoke(tmp_path, acme, druks_db):
         assert not OauthConnection.list_for_provider("acme")
         assert client.delete(f"/api/oauth/connections/{row.id}").status_code == 404
 
+    assert published == [
+        (
+            "oauth.disconnected",
+            {"provider": "acme", "connection_id": row.id, "account_id": me.id},
+        )
+    ]
+
 
-def test_replacing_the_client_credentials_deletes_its_connections(tmp_path, acme, druks_db):
+def test_replacing_the_client_credentials_deletes_its_connections(
+    tmp_path, acme, druks_db, monkeypatch
+):
     from druks.accounts.constants import SYSTEM_ACCOUNT_ID
     from druks.services.models import OauthConnection
     from druks.testing import configure_app_for_test
 
-    OauthConnection.create(
+    published = []
+
+    async def record(name, **kwargs):
+        published.append((name, kwargs))
+
+    monkeypatch.setattr("druks.services.routes.publish", record)
+    row = OauthConnection.create(
         provider="acme", account_id=SYSTEM_ACCOUNT_ID, refresh_token="rt-old", scopes=[]
     )
 
@@ -686,6 +723,12 @@ def test_replacing_the_client_credentials_deletes_its_connections(tmp_path, acme
 
     # The new client can never refresh the old client's connections.
     assert not OauthConnection.list_for_provider("acme")
+    assert published == [
+        (
+            "oauth.disconnected",
+            {"provider": "acme", "connection_id": row.id, "account_id": SYSTEM_ACCOUNT_ID},
+        )
+    ]
 
 
 def test_list_serves_the_connections_beside_the_declared_union(tmp_path, acme, druks_db):
@@ -714,3 +757,42 @@ def entry(client, name="acme"):
         assert connection["id"] == row.id
         assert connection["scopes"] == ["profile.read"]
         assert connection["connectedAt"]
+
+
+def test_next_lands_the_user_back_on_the_extension_page(tmp_path, acme, druks_db):
+    from urllib.parse import parse_qsl, urlparse
+
+    from druks.testing import configure_app_for_test
+
+    ServiceIdentity.connect(
+        "acme", identity={"client_id": "id-1"}, secrets={"client_secret": "sec-1"}
+    )
+    settings = make_settings(tmp_path, urls={"endpoint": "https://druks.example"})
+    with TestClient(configure_app_for_test(settings=settings)) as client:
+        consent = client.get(
+            "/api/oauth/acme/connect?next=/app/night_watch/accounts",
+            follow_redirects=False,
+        ).headers["location"]
+        state = dict(parse_qsl(urlparse(consent).query))["state"]
+
+        finish = client.get(
+            "/api/oauth/callback", params={"state": state, "code": "c-1"}, follow_redirects=False
+        )
+
+        assert finish.status_code == 307
+        assert finish.headers["location"] == "/app/night_watch/accounts"
+
+
+def test_next_rejects_anything_but_a_bare_path(tmp_path, acme, druks_db):
+    from druks.testing import configure_app_for_test
+
+    ServiceIdentity.connect(
+        "acme", identity={"client_id": "id-1"}, secrets={"client_secret": "sec-1"}
+    )
+    settings = make_settings(tmp_path, urls={"endpoint": "https://druks.example"})
+    with TestClient(configure_app_for_test(settings=settings)) as client:
+        for hostile in ("https://evil.test/x", "//evil.test/x", "/\\evil.test", "app/page"):
+            response = client.get(
+                "/api/oauth/acme/connect", params={"next": hostile}, follow_redirects=False
+            )
+            assert response.status_code == 422, hostile

docs/writing-an-extension.md

@@ -783,8 +783,35 @@ platform runs the consent with the union of every installed extension's
 declared scopes and stores the connection for the signed-in user. To widen
 an existing connection's scopes, open
 `/api/oauth/acme/connect?connection=<id>`; reconsent replaces its tokens.
-Register `https://<host>/api/oauth/callback` as the redirect URI at the
-provider; it serves every service.
+Add `?next=/app/night_watch/accounts` to land the user back on your
+page after consent instead of the generic "connected" page. `next`
+must be a bare path starting with `/` — a URL with a scheme or host is
+rejected, so the door can never redirect off the box. Register
+`https://<host>/api/oauth/callback` as the redirect URI at the provider; it
+serves every service.
+
+React to sign-ins with the signal machinery. The platform publishes
+`oauth.connected` when a consent completes — `reconsent` is true when it
+replaced an existing connection's tokens — and `oauth.disconnected` when a
+connection dies, whether the user revoked it or the service's client
+credentials were replaced. Subscribe in `subscribers.py`:
+
+```python
+from druks.signals import subscribe
+
+
+@subscribe("oauth.connected", provider="acme")
+async def adopt_sign_in(
+    provider: str, connection_id: str, account_id: str, reconsent: bool
+) -> None:
+    if not reconsent:
+        WatchedAccount.adopt(connection_id, account_id)
+
+
+@subscribe("oauth.disconnected", provider="acme")
+async def drop_sign_in(provider: str, connection_id: str, account_id: str) -> None:
+    WatchedAccount.drop(connection_id)
+```
 
 The user sees and revokes everything in Settings — every connection they
 hold, across services. Replacing a service's client credentials deletes its