drb-c2-core/app/internal/discord_guild_link.py

@@ -27,6 +27,16 @@ Firestore layout:
     unlinked_at|None, unlinked_by|None
 
 Person names are never stored: the redeemer is kept as a Discord user id.
+
+Every way a server gets linked — ``/drb link code:`` (``redeem_and_settle``),
+the platform admin's ``PUT /discord/guilds/{id}/link`` and the per-bot OAuth
+invite (server-26#225 S6) (both ``link_and_settle``) — writes the guild doc
+through ONE transaction body (``_apply_link``) and then runs ONE settle step
+(``_settle``): a server that was acting for a different org (in practice the
+``DISCORD_UNLINKED_GUILD_ORG_ID`` fallback) has its live radio channels
+stopped, because they were placed on that org's nodes. The org it acted for
+before is read inside the same transaction as the write, so the stop decision
+can't race a concurrent link.
 """
 from __future__ import annotations
 
@@ -35,6 +45,7 @@ import secrets
 from datetime import datetime, timedelta, timezone
 from typing import Optional
 
+from app.internal import discord_streams as streams
 from app.internal import firestore as fstore
 
 CODES = "discord_link_codes"
@@ -91,6 +102,63 @@ async def mint_code(org_id: str, uid: str) -> dict:
             "expires_minutes": CODE_TTL_MINUTES}
 
 
+# Stop reasons the settle step stamps on the streams it stops.
+RELINK_STOP_REASON = "relinked"
+ADMIN_RELINK_STOP_REASON = "server linked to another organization"
+
+
+def _apply_link(txn, guild_ref, guild: Optional[dict], *, org_id: str, guild_id: str,
+                guild_name: Optional[str], linked_by: str, via: str, now: datetime,
+                allow_move: bool = False, extra: Optional[dict] = None):
+    """The guild-side half of every link, inside a transaction whose reads are
+    all done. Returns a LinkError (nothing written) or the result dict:
+    ``{org_id, already_linked, previous_org_id}`` — ``previous_org_id`` is the
+    org the server acted for before (its link, else the fallback org).
+
+    ``allow_move`` is the platform admin's support override; every self-serve
+    path refuses to move a server another org has linked."""
+    current = (guild or {}).get("org_id")
+    if current and current != org_id and not allow_move:
+        return LinkError(409, "linked_elsewhere",
+                         "This server is already linked to a different DRB organization. "
+                         "That organization has to unlink it first.")
+    iso = now.isoformat()
+    name = (guild_name or "").strip()[:100]
+    txn.set(guild_ref, {
+        "guild_id": guild_id,
+        "org_id": org_id,
+        "guild_name": name or (guild or {}).get("guild_name"),
+        "linked_by": linked_by,
+        "linked_via": via,
+        "linked_at": ((guild or {}).get("linked_at") if current == org_id else None) or iso,
+        "unlinked_at": None,
+        "unlinked_by": None,
+        # A server an admin removed (streams.remove_guild) comes back on link.
+        "removed_at": None,
+        "removed_by": None,
+        **(extra or {}),
+    }, merge=True)
+    return {"org_id": org_id, "already_linked": current == org_id,
+            "previous_org_id": streams.acting_org_of(guild)}
+
+
+def link_guild_txn(org_id: str, guild_id: str, guild_name: Optional[str], linked_by: str, via: str, *,
+                   allow_move: bool = False, extra: Optional[dict] = None, now: Optional[datetime] = None):
+    """Transaction body that links ``guild_id`` to ``org_id`` directly (no code)."""
+    db = fstore.db
+    when = now or _now()
+
+    def _txn(txn):
+        guild_ref = db.collection(GUILDS).document(guild_id)
+        snap = guild_ref.get(transaction=txn)
+        guild = snap.to_dict() if snap.exists else None
+        out = _apply_link(txn, guild_ref, guild, org_id=org_id, guild_id=guild_id, guild_name=guild_name,
+                          linked_by=linked_by, via=via, now=when, allow_move=allow_move, extra=extra)
+        return {"error": out} if isinstance(out, LinkError) else out
+
+    return _txn
+
+
 def _redeem_txn(code: str, guild_id: str, guild_name: Optional[str], discord_user_id: str, now: datetime):
     """Build the transaction body. Every read happens before any write."""
     db = fstore.db
@@ -113,37 +181,20 @@ def _redeem_txn(code: str, guild_id: str, guild_name: Optional[str], discord_use
             return {"error": LinkError(410, "code_expired",
                                        "That link code has expired. Generate a new one in DRB → "
                                        "Settings → Organization.")}
-        org_id = doc["org_id"]
+        out = _apply_link(txn, guild_ref, guild, org_id=doc["org_id"], guild_id=guild_id,
-        current = (guild or {}).get("org_id")
+                          guild_name=guild_name, linked_by=f"discord:{discord_user_id}", via="code", now=now)
-        if current and current != org_id:
+        if isinstance(out, LinkError):
-            return {"error": LinkError(409, "linked_elsewhere",
+            return {"error": out}       # the code is NOT consumed: its org can use it elsewhere
-                                       "This server is already linked to a different DRB organization. "
+        txn.update(code_ref, {"redeemed_at": now.isoformat(), "redeemed_guild_id": guild_id,
-                                       "That organization has to unlink it first.")}
-
-        iso = now.isoformat()
-        txn.update(code_ref, {"redeemed_at": iso, "redeemed_guild_id": guild_id,
                               "redeemed_by_discord_id": discord_user_id})
-        txn.set(guild_ref, {
+        return {**out, "created_by_uid": doc.get("created_by_uid")}
-            "guild_id": guild_id,
-            "org_id": org_id,
-            "guild_name": (guild_name or "")[:100] or (guild or {}).get("guild_name"),
-            "linked_by": f"discord:{discord_user_id}",
-            "linked_via": "code",
-            "linked_at": ((guild or {}).get("linked_at") if current == org_id else None) or iso,
-            "unlinked_at": None,
-            "unlinked_by": None,
-            # A server an admin removed (streams.remove_guild) comes back on link.
-            "removed_at": None,
-            "removed_by": None,
-        }, merge=True)
-        return {"org_id": org_id, "already_linked": current == org_id,
-                "created_by_uid": doc.get("created_by_uid")}
 
     return _txn
 
 
 async def redeem_code(raw_code: str, guild_id: str, guild_name: Optional[str], discord_user_id: str) -> dict:
-    """Link ``guild_id`` to the code's org. Raises LinkError on refusal."""
+    """Link ``guild_id`` to the code's org. Raises LinkError on refusal.
+    Does NOT stop anything — routes call ``redeem_and_settle``."""
     code = normalize(raw_code)
     if len(code) != _CODE_LEN:
         raise LinkError(404, "invalid_code",
@@ -155,6 +206,39 @@ async def redeem_code(raw_code: str, guild_id: str, guild_name: Optional[str], d
     return result
 
 
+async def _settle(guild_id: str, result: dict, stop_reason: str) -> dict:
+    """After a link committed: a server that was acting for another org (the
+    fallback org, or — admin override only — another linked org) has its live
+    radio channels stopped; they were placed on that org's nodes, and left
+    running the reconciler would keep re-placing them there
+    (server-26#225 plan finding 3: this lived in two routers before)."""
+    before = result.get("previous_org_id")
+    stopped: list = []
+    if before and before != result["org_id"]:
+        stopped = await streams.stop_guild(str(guild_id), stop_reason)
+    return {**result, "streams_stopped": stopped}
+
+
+async def link_and_settle(org_id: str, guild_id: str, guild_name: Optional[str], linked_by: str, via: str, *,
+                          allow_move: bool = False, extra: Optional[dict] = None,
+                          stop_reason: str = RELINK_STOP_REASON) -> dict:
+    """Link a server to an org directly (admin, OAuth) and settle. Raises
+    LinkError (``linked_elsewhere``) unless ``allow_move``. Returns
+    ``{org_id, already_linked, previous_org_id, streams_stopped}``."""
+    result = await fstore.run_transaction(
+        link_guild_txn(org_id, str(guild_id), guild_name, linked_by, via, allow_move=allow_move, extra=extra))
+    if "error" in result:
+        raise result["error"]
+    return await _settle(guild_id, result, stop_reason)
+
+
+async def redeem_and_settle(raw_code: str, guild_id: str, guild_name: Optional[str],
+                            discord_user_id: str) -> dict:
+    """``/drb link code:`` — redeem, then settle. Raises LinkError."""
+    result = await redeem_code(raw_code, guild_id, guild_name, discord_user_id)
+    return await _settle(guild_id, result, RELINK_STOP_REASON)
+
+
 async def list_org_guilds(org_id: str) -> list[dict]:
     return [
         {"guild_id": g.get("guild_id"), "guild_name": g.get("guild_name"),

drb-c2-core/app/internal/discord_oauth.py

@@ -0,0 +1,246 @@
+"""
+Per-bot OAuth "Connect a Discord server" (server-26#225 S6, owner decision Q1).
+
+An org owner links a Discord server by inviting ONE radio bot through
+Discord's OAuth2 bot invite. The invite's ``state`` carries a single-use,
+short-lived nonce bound to the org and the user who started it; Discord
+redirects back with ``code`` (and ``guild_id``), C2 exchanges the code with
+THAT bot application's client secret, and the exchange's ``guild`` object —
+never the query string — says which server to link.
+
+Secrets:
+
+  bot_oauth/{token_id}            server-only (Firestore rules deny all
+    client_secret, enabled,       client reads; c2-core's admin SDK only)
+    set_by, set_at, checked_at
+
+  The secret is never on ``bot_tokens`` (GET /tokens returns an allowlist of
+  those fields), never returned by any route, never logged, never put in an
+  error string. Firestore encrypts it at rest like every other document; the
+  codebase has no application-level encryption pattern (bot tokens are
+  stored the same way), so none is invented here.
+
+States:
+
+  discord_oauth_states/{sha256(nonce)}   the nonce itself is never stored
+    org_id, uid, token_id, node_id, return_to, created_at, expires_at,
+    used_at|None, used_by|None
+
+OAuth is off while ``settings.discord_oauth_redirect_uri`` is empty: every
+route 404s and the UI hides the button.
+
+HTTP goes through discord_rest's shared client (User-Agent, transport for
+tests) and counts 401/403/429 answers against its process-wide breaker, so a
+wrong secret retried in a loop can't contribute to an IP ban either.
+"""
+from __future__ import annotations
+
+import hashlib
+import secrets
+import time
+from datetime import datetime, timedelta, timezone
+from typing import Any, Optional
+from urllib.parse import urlencode
+
+import httpx
+
+from app.config import settings
+from app.internal import discord_leases as leases
+from app.internal import discord_rest
+from app.internal import firestore as fstore
+
+OAUTH = "bot_oauth"
+STATES = "discord_oauth_states"
+AUTHORIZE_URL = "https://discord.com/oauth2/authorize"
+# Discord client secrets are 32 url-safe characters today; accept a margin.
+SECRET_MIN, SECRET_MAX = 16, 128
+RETURN_TO_MAX = 200
+
+
+def enabled() -> bool:
+    return bool((settings.discord_oauth_redirect_uri or "").strip())
+
+
+def _now() -> datetime:
+    return datetime.now(timezone.utc)
+
+
+def state_key(nonce: str) -> str:
+    return hashlib.sha256((nonce or "").encode()).hexdigest()
+
+
+def safe_return_to(value: Optional[str]) -> Optional[str]:
+    """A same-site path only (``/nodes/abc``), never an absolute or
+    protocol-relative URL — the callback page redirects to it."""
+    v = (value or "").strip()
+    if not v or len(v) > RETURN_TO_MAX or not v.startswith("/") or v.startswith("//") or "\\" in v:
+        return None
+    if any(ord(ch) < 0x20 for ch in v):
+        return None
+    return v
+
+
+# ---------------------------------------------------------------------------
+# Secrets
+# ---------------------------------------------------------------------------
+
+async def enabled_token_ids() -> set[str]:
+    """Bots whose OAuth secret is set and enabled."""
+    return {d.get("token_id") for d in await fstore.collection_list(OAUTH)
+            if d.get("token_id") and d.get("enabled") and d.get("client_secret")}
+
+
+async def load_secret(token_id: str) -> Optional[str]:
+    doc = await fstore.doc_get(OAUTH, token_id)
+    if not doc or not doc.get("enabled") or not doc.get("client_secret"):
+        return None
+    return doc["client_secret"]
+
+
+async def store_secret(token_id: str, client_secret: str, uid: str) -> None:
+    now = _now().isoformat()
+    await fstore.doc_set(OAUTH, token_id, {
+        "token_id": token_id, "client_secret": client_secret, "enabled": True,
+        "set_by": uid, "set_at": now, "checked_at": now,
+    }, merge=False)
+
+
+async def clear_secret(token_id: str) -> bool:
+    if not await fstore.doc_get(OAUTH, token_id):
+        return False
+    await fstore.doc_delete(OAUTH, token_id)
+    return True
+
+
+# ---------------------------------------------------------------------------
+# Discord OAuth2 HTTP (client_id + client_secret, basic auth)
+# ---------------------------------------------------------------------------
+
+async def _post_form(path: str, data: dict, client_id: str, client_secret: str) -> discord_rest.RestResult:
+    """POST an application-authenticated form. Never raises for a Discord or
+    network failure; the error string never holds the secret, the code or a
+    token (only Discord's short ``error`` field)."""
+    now = time.monotonic()
+    if discord_rest._breaker.tripped(now):
+        return discord_rest.RestResult(None, error="Discord REST paused (invalid-request budget reached)")
+    try:
+        r = await discord_rest._client().post(path, data=data, auth=(client_id, client_secret))
+    except httpx.HTTPError as e:
+        return discord_rest.RestResult(None, error=type(e).__name__)
+    status = r.status_code
+    if status in (401, 403) or (status == 429 and r.headers.get("x-ratelimit-scope") != "shared"):
+        discord_rest._breaker.record(time.monotonic())
+    body: Any = None
+    try:
+        body = r.json() if r.content else None
+    except ValueError:
+        body = None
+    if 200 <= status < 300:
+        return discord_rest.RestResult(status, body)
+    err = body.get("error") if isinstance(body, dict) and isinstance(body.get("error"), str) else None
+    return discord_rest.RestResult(status, None, error=f"Discord answered {status}" + (f" ({err[:40]})" if err else ""))
+
+
+async def exchange_code(client_id: str, client_secret: str, code: str) -> discord_rest.RestResult:
+    return await _post_form("/oauth2/token", {
+        "grant_type": "authorization_code", "code": code,
+        "redirect_uri": settings.discord_oauth_redirect_uri.strip(),
+    }, client_id, client_secret)
+
+
+async def revoke(client_id: str, client_secret: str, access_token: Optional[str]) -> None:
+    """Best effort: C2 never uses the user's access token, so it is revoked
+    as soon as the guild has been read from the exchange."""
+    if access_token:
+        await _post_form("/oauth2/token/revoke", {"token": access_token, "token_type_hint": "access_token"},
+                         client_id, client_secret)
+
+
+async def check_secret(client_id: str, client_secret: str) -> discord_rest.RestResult:
+    """Prove ``client_secret`` belongs to ``client_id`` with a client-
+    credentials grant (scope ``identify`` — the bot owner's own token), then
+    revoke the token it returns. Nothing is stored from the answer."""
+    r = await _post_form("/oauth2/token", {"grant_type": "client_credentials", "scope": "identify"},
+                         client_id, client_secret)
+    if r.ok and isinstance(r.data, dict):
+        await revoke(client_id, client_secret, r.data.get("access_token"))
+    return r
+
+
+# ---------------------------------------------------------------------------
+# State + authorize URL
+# ---------------------------------------------------------------------------
+
+def authorize_url(application_id: str, nonce: str) -> str:
+    return AUTHORIZE_URL + "?" + urlencode({
+        "client_id": application_id,
+        "scope": "bot",
+        "permissions": str(leases.RADIO_BOT_PERMISSIONS),
+        "response_type": "code",
+        "redirect_uri": settings.discord_oauth_redirect_uri.strip(),
+        "state": nonce,
+    })
+
+
+async def create_state(*, org_id: str, uid: str, token_id: str, node_id: Optional[str],
+                       return_to: Optional[str]) -> tuple[str, datetime]:
+    nonce = secrets.token_urlsafe(32)
+    now = _now()
+    expires_at = now + timedelta(seconds=max(60, settings.discord_oauth_state_ttl_seconds))
+    await fstore.doc_set(STATES, state_key(nonce), {
+        "org_id": org_id, "uid": uid, "token_id": token_id, "node_id": node_id,
+        "return_to": return_to, "created_at": now.isoformat(), "expires_at": expires_at.isoformat(),
+        "used_at": None, "used_by": None,
+    }, merge=False)
+    return nonce, expires_at
+
+
+class StateError(Exception):
+    def __init__(self, status: int, code: str, message: str):
+        super().__init__(message)
+        self.status, self.code, self.message = status, code, message
+
+
+def consume_state_txn(nonce: str, uid: str, now: datetime):
+    """Transaction body: the state must exist, be unexpired and unused, and
+    belong to ``uid`` (login-CSRF: someone else's state never links into the
+    caller's org, and the caller's own code can't be replayed by another
+    account). Any presentation of a live state burns it — a failure later in
+    the flow means starting again, never retrying the same state."""
+    db = fstore.db
+    key = state_key(nonce)
+
+    def _txn(txn):
+        ref = db.collection(STATES).document(key)
+        snap = ref.get(transaction=txn)
+        doc = snap.to_dict() if snap.exists else None
+        if not doc:
+            return {"error": StateError(400, "invalid_state",
+                                        "That Discord link is not valid. Start again from DRB.")}
+        if doc.get("used_at"):
+            return {"error": StateError(409, "state_used",
+                                        "That Discord link was already used. Start again from DRB.")}
+        txn.update(ref, {"used_at": now.isoformat(), "used_by": uid})
+        try:
+            expired = now > datetime.fromisoformat(doc.get("expires_at") or "")
+        except ValueError:
+            expired = True
+        if expired:
+            return {"error": StateError(410, "state_expired",
+                                        "That Discord link expired. Start again from DRB.")}
+        if doc.get("uid") != uid:
+            return {"error": StateError(403, "wrong_user",
+                                        "That Discord link was started by a different DRB account. "
+                                        "Sign in as that account, or start again.")}
+        return {"state": doc}
+
+    return _txn
+
+
+async def consume_state(nonce: str, uid: str) -> dict:
+    if not nonce or len(nonce) > 200:
+        raise StateError(400, "invalid_state", "That Discord link is not valid. Start again from DRB.")
+    out = await fstore.run_transaction(consume_state_txn(nonce, uid, _now()))
+    if "error" in out:
+        raise out["error"]
+    return out["state"]

drb-c2-core/app/internal/discord_streams.py

@@ -93,6 +93,17 @@ def _parse_ts(value) -> Optional[datetime]:
 # Guild -> org
 # ---------------------------------------------------------------------------
 
+def acting_org_of(link: Optional[dict]) -> Optional[str]:
+    """``resolve_guild_org`` for an already-read ``discord_guilds`` doc (None =
+    no doc). Pure, so a link transaction can decide from the snapshot it read
+    (discord_guild_link.link_guild_txn, server-26#225 S6)."""
+    if link and link.get("org_id"):
+        return link["org_id"]
+    if link and link.get("unlinked_at"):
+        return None
+    return settings.discord_unlinked_guild_org_id or None
+
+
 async def resolve_guild_org(guild_id: str) -> Optional[str]:
     """The org a Discord server acts for: its ``discord_guilds`` link, else
     ``settings.discord_unlinked_guild_org_id`` (None when that is empty).
@@ -100,12 +111,7 @@ async def resolve_guild_org(guild_id: str) -> Optional[str]:
     A server an org explicitly unlinked (``unlinked_at`` set, server-26#213)
     gets no fallback: otherwise unlinking org B's server would quietly hand it
     to the founding org."""
-    link = await fstore.doc_get(GUILDS, str(guild_id))
+    return acting_org_of(await fstore.doc_get(GUILDS, str(guild_id)))
-    if link and link.get("org_id"):
-        return link["org_id"]
-    if link and link.get("unlinked_at"):
-        return None
-    return settings.discord_unlinked_guild_org_id or None
 
 
 async def _node_stream_counts(exclude_stream_id: Optional[str] = None) -> dict[str, int]:

drb-c2-core/app/main.py

@@ -25,6 +25,7 @@ from app.internal import dynsec
 from app.internal import firestore as fstore
 from app.routers import discord_web
 from app.routers import discord_checks
+from app.routers import discord_oauth
 
 
 async def _clear_legacy_token_flags():
@@ -166,6 +167,7 @@ app.include_router(discord_web.router)
 app.include_router(discord_router.router)
 app.include_router(discord_link_router.router)  # auth per-endpoint (org owner / bot service key) — server-26#213
 app.include_router(discord_checks.router)  # auth per-endpoint (org member / service key or admin) — server-26#225
+app.include_router(discord_oauth.router)  # auth per-endpoint (org owner / signed-in starter); 404 while OAuth is off — #225 S6
 # NOTE: there used to be an app.routers.mqtt_auth router here (an HTTP
 # backend for the mosquitto-go-auth plugin). That plugin's upstream project
 # is archived (no CVE patches) and was rejected for an internet-facing

drb-c2-core/app/routers/discord.py

@@ -17,7 +17,6 @@ Auth:
 from __future__ import annotations
 
 import secrets
-from datetime import datetime, timezone
 from typing import Optional
 
 from fastapi import APIRouter, Depends, HTTPException, Path, Security
@@ -26,6 +25,7 @@ from fastapi.security import HTTPAuthorizationCredentials, HTTPBearer
 from pydantic import BaseModel, Field
 
 from app.config import settings
+from app.internal import discord_guild_link as glink
 from app.internal import discord_leases as leases
 from app.internal import discord_rest
 from app.internal import discord_streams as streams
@@ -181,23 +181,18 @@ async def link_guild(guild_id: str, body: GuildLink, decoded: dict = Depends(req
     org = await fstore.doc_get("organizations", body.org_id)
     if not org:
         raise HTTPException(404, "Organization not found.")
-    previous_org = await streams.resolve_guild_org(guild_id)
+    # Support override: may move a server between orgs. Re-linking a server
-    doc = {
+    # that was unlinked or removed brings it back. Live channels placed on the
-        "guild_id": guild_id, "org_id": body.org_id,
+    # previous org's nodes are stopped (glink._settle) — left running, the
-        "linked_by": decoded.get("uid"), "linked_at": datetime.now(timezone.utc).isoformat(),
+    # reconciler would keep re-placing them there.
-        # Re-linking a server that was unlinked or removed brings it back.
+    out = await glink.link_and_settle(body.org_id, guild_id, body.guild_name, decoded.get("uid") or "admin",
-        "unlinked_at": None, "unlinked_by": None, "removed_at": None, "removed_by": None,
+                                      "admin", allow_move=True, stop_reason=glink.ADMIN_RELINK_STOP_REASON)
-    }
+    uid, email = describe_actor(decoded)
-    if body.guild_name and body.guild_name.strip():
+    await write_audit(uid, email, "discord.guild.link", details={
-        doc["guild_name"] = body.guild_name.strip()[:100]
+        "guild_id": guild_id, "org_id": body.org_id, "previous_org_id": out["previous_org_id"],
-    await fstore.doc_set(streams.GUILDS, guild_id, doc)
+        "already_linked": out["already_linked"], "streams_stopped": len(out["streams_stopped"]), "via": "admin",
-    # Live channels in this server were placed on the previous org's nodes;
+    })
-    # left running, the reconciler would keep re-placing them there, so one
+    return {"ok": True, "guild_id": guild_id, "org_id": body.org_id, "stopped": out["streams_stopped"]}
-    # org's equipment would keep serving a server that now belongs to another.
-    stopped: list[str] = []
-    if previous_org and previous_org != body.org_id:
-        stopped = await streams.stop_guild(guild_id, "server linked to another organization")
-    return {"ok": True, "guild_id": guild_id, "org_id": body.org_id, "stopped": stopped}
 
 
 @router.delete("/streams/{stream_id}")

drb-c2-core/app/routers/discord_link.py

@@ -101,23 +101,21 @@ async def redeem_link_code(guild_id: str, body: RedeemBody, principal: dict = De
     redeem_guild_limiter.check(f"guild:{guild_id}")
     redeem_user_limiter.check(f"user:{body.discord_user_id}")
     actor = f"discord:{body.discord_user_id}"
-    before = await streams.resolve_guild_org(guild_id)
     try:
-        result = await glink.redeem_code(body.code, guild_id, body.guild_name, body.discord_user_id)
+        # Links, then stops anything the server had playing for the org it
+        # acted for before (the DISCORD_UNLINKED_GUILD_ORG_ID fallback).
+        result = await glink.redeem_and_settle(body.code, guild_id, body.guild_name, body.discord_user_id)
     except glink.LinkError as e:
         await write_audit(actor, "", "discord.guild.link_refused",
                           details={"guild_id": guild_id, "reason": e.code})
         return JSONResponse(status_code=e.status, content={"detail": e.message, "code": e.code})
 
     org_id = result["org_id"]
-    stopped: list = []
+    stopped = result["streams_stopped"]
-    if before and before != org_id:
-        # The server was acting for DISCORD_UNLINKED_GUILD_ORG_ID; anything it
-        # had playing belongs to that org, not the one it just linked to.
-        stopped = await streams.stop_guild(guild_id, "relinked")
     await write_audit(actor, "", "discord.guild.link", details={
         "guild_id": guild_id, "org_id": org_id, "already_linked": result["already_linked"],
         "code_created_by_uid": result.get("created_by_uid"), "streams_stopped": len(stopped),
+        "via": "code",
     })
     logger.info(f"Discord guild {guild_id} linked to org={org_id!r} by {actor}")
     org = await fstore.doc_get("organizations", org_id)

drb-c2-core/app/routers/discord_oauth.py

@@ -0,0 +1,289 @@
+"""
+"Connect a Discord server" through a radio bot's own OAuth2 invite
+(server-26#225 S6, owner decision Q1 = per-bot OAuth).
+
+  GET  /discord/oauth/options   org owner / platform admin — which bot Connect
+                                would invite (no state minted). 404 = off.
+  POST /discord/oauth/start     org owner / platform admin — mint a state,
+                                return Discord's authorize URL
+  POST /discord/oauth/complete  signed-in user who started it — exchange the
+                                code, link the server, check permissions
+
+Platform admins may pass ?org_id= to act for another org (resolve_org_scope).
+Every route is a 404 while ``discord_oauth_redirect_uri`` is empty or no
+eligible bot has an OAuth secret (PUT /tokens/{id}/oauth).
+
+Which bot (``_pick_bot``): an explicit ``token_id`` the org may use; else the
+radio bot preferred for the node's system (``node_id``); else the org's own
+bring-your-own bot; else the pool bot with the fewest servers that is under
+``discord_pool_bot_guild_cap`` (owner decision Q3). Only bots with an OAuth
+secret set count. A bot the caller's org can't use is a 404, never a 403.
+
+The org a server links to is the one the STATE names, re-checked against a
+fresh read of the caller's membership; the server is the one Discord's code
+exchange names. The redirect's ``guild_id`` query parameter is ignored.
+
+This reverses ``discord_web``'s old "C2 never calls Discord on the browser's
+behalf" on purpose (plan finding 5): the call is made with the bot
+application's credentials, after the org-scope checks and rate limits here.
+"""
+from __future__ import annotations
+
+import asyncio
+from typing import Optional
+
+from fastapi import APIRouter, Depends, HTTPException, Query
+from fastapi.responses import JSONResponse
+from pydantic import BaseModel, Field
+
+from app.config import settings
+from app.internal import discord_guild_link as glink
+from app.internal import discord_leases as leases
+from app.internal import discord_oauth as oauth
+from app.internal import discord_rest
+from app.internal import discord_streams as streams
+from app.internal import firestore as fstore
+from app.internal.audit import write_audit
+from app.internal.auth import (
+    _RateLimiter, describe_actor, get_role, require_firebase_token, require_org_owner_token, resolve_org_scope,
+)
+from app.internal.logger import logger
+from app.routers.discord_checks import voice_check
+
+router = APIRouter(prefix="/discord/oauth", tags=["discord"])
+
+start_limiter = _RateLimiter(max_calls=5, window_seconds=600)
+complete_limiter = _RateLimiter(max_calls=10, window_seconds=600)
+_ID = r"^[A-Za-z0-9_-]{1,64}$"
+_OFF = "Connecting a Discord server through DRB isn't available."
+
+
+def _require_enabled() -> None:
+    if not oauth.enabled():
+        raise HTTPException(404, _OFF)
+
+
+def _err(status: int, code: str, message: str, **extra) -> JSONResponse:
+    return JSONResponse(status_code=status, content={"detail": message, "code": code, **extra})
+
+
+def _server_count(tok: dict) -> Optional[int]:
+    members = tok.get("guild_ids")
+    return len(members) if isinstance(members, list) else None
+
+
+def _usable(tok: dict, org_id: str, oauth_ids: set) -> bool:
+    return (tok.get("token_id") in oauth_ids
+            and tok.get("valid") is not False
+            and bool(tok.get("application_id"))
+            and tok.get("org_id") in (None, org_id)
+            and leases.invitable(tok))
+
+
+async def _pick_bot(org_id: str, *, node_id: Optional[str] = None,
+                    token_id: Optional[str] = None) -> tuple[dict, str]:
+    """(token doc, why) — why is ``chosen`` | ``preferred`` | ``org_bot`` |
+    ``pool``. Raises HTTPException 404 when nothing fits."""
+    tokens = {t["token_id"]: t for t in await fstore.collection_list(leases.TOKENS) if t.get("token_id")}
+    oauth_ids = await oauth.enabled_token_ids()
+
+    if token_id:
+        tok = tokens.get(token_id)
+        if not tok or tok.get("org_id") not in (None, org_id):
+            raise HTTPException(404, "Radio bot not found.")
+        if not _usable(tok, org_id, oauth_ids):
+            raise HTTPException(404, f"{tok.get('name') or 'That radio bot'} can't be used to connect a server "
+                                     "right now.")
+        return tok, "chosen"
+
+    if node_id:
+        node = await fstore.doc_get("nodes", node_id)
+        if not node or node.get("org_id") != org_id:
+            raise HTTPException(404, "Node not found.")
+        system_id = streams._effective_system(node)
+        system = await fstore.doc_get("systems", system_id) if system_id else None
+        pref = tokens.get((system or {}).get("preferred_token_id") or "")
+        if pref and _usable(pref, org_id, oauth_ids):
+            return pref, "preferred"
+
+    own = sorted((t for t in tokens.values() if t.get("org_id") == org_id and _usable(t, org_id, oauth_ids)),
+                 key=lambda t: (str(t.get("name") or "").lower(), t["token_id"]))
+    if own:
+        return own[0], "org_bot"
+
+    cap = settings.discord_pool_bot_guild_cap
+    pool = sorted((t for t in tokens.values() if t.get("org_id") is None and _usable(t, org_id, oauth_ids)),
+                  key=lambda t: (_server_count(t) if _server_count(t) is not None else cap,
+                                 str(t.get("name") or "").lower(), t["token_id"]))
+    if pool:
+        return pool[0], "pool"
+    raise HTTPException(404, _OFF)
+
+
+def _bot_summary(tok: dict, why: str) -> dict:
+    return {"name": tok.get("name") or "DRB radio bot", "reason": why,
+            "own_bot": tok.get("org_id") is not None}
+
+
+# ---------------------------------------------------------------------------
+# Routes
+# ---------------------------------------------------------------------------
+
+@router.get("/options")
+async def oauth_options(node_id: Optional[str] = Query(None, pattern=_ID),
+                        token_id: Optional[str] = Query(None, pattern=_ID),
+                        org_id: Optional[str] = None,
+                        decoded: dict = Depends(require_org_owner_token)):
+    """Whether Connect is offered here, and which bot it would invite."""
+    _require_enabled()
+    scope = resolve_org_scope(decoded, org_id)
+    tok, why = await _pick_bot(scope, node_id=node_id, token_id=token_id)
+    return {"available": True, "bot": _bot_summary(tok, why)}
+
+
+class StartBody(BaseModel):
+    node_id: Optional[str] = Field(None, pattern=_ID)
+    token_id: Optional[str] = Field(None, pattern=_ID)
+    return_to: Optional[str] = None
+
+
+@router.post("/start")
+async def oauth_start(body: StartBody, org_id: Optional[str] = None,
+                      decoded: dict = Depends(require_org_owner_token)):
+    _require_enabled()
+    scope = resolve_org_scope(decoded, org_id)
+    uid, email = describe_actor(decoded)
+    start_limiter.check(f"{uid}:discord_oauth_start")
+    tok, why = await _pick_bot(scope, node_id=body.node_id, token_id=body.token_id)
+    return_to = oauth.safe_return_to(body.return_to)
+    nonce, expires_at = await oauth.create_state(org_id=scope, uid=uid, token_id=tok["token_id"],
+                                                 node_id=body.node_id, return_to=return_to)
+    await write_audit(uid, email, "discord.oauth.start", details={
+        "org_id": scope, "token_id": tok["token_id"], "reason": why, "node_id": body.node_id})
+    return {"authorize_url": oauth.authorize_url(str(tok["application_id"]), nonce),
+            "expires_at": expires_at.isoformat(), "bot": _bot_summary(tok, why)}
+
+
+class CompleteBody(BaseModel):
+    # Lengths checked in the handler: a pydantic 422 would echo the code back.
+    code: str
+    state: str
+
+
+async def _still_owner(decoded: dict, uid: str, org_id: str) -> bool:
+    """Fresh read: the claims in an ID token can be up to an hour old."""
+    if get_role(decoded) == "admin":
+        return True
+    member = await fstore.doc_get("org_members", uid)
+    return bool(member and member.get("org_id") == org_id and member.get("org_role") == "owner")
+
+
+async def _channel_summary(guild_id: str, org_id: str, node_id: Optional[str]) -> Optional[dict]:
+    """Permissions right after the link, within discord_oauth_link_check_seconds.
+    None = not answered in time (the picker's badges will say)."""
+    link = await fstore.doc_get(streams.GUILDS, guild_id) or {}
+    try:
+        out = await asyncio.wait_for(
+            voice_check(guild_id=guild_id, org_id=org_id, node_id=node_id,
+                        channels=link.get("voice_channels") or [], channel_id=None, refresh=True,
+                        with_token_ids=False),
+            timeout=settings.discord_oauth_link_check_seconds)
+    except Exception as e:   # timeout or anything else: the link already stands
+        logger.info(f"Discord OAuth link {guild_id}: permission check skipped ({type(e).__name__})")
+        return None
+    channels = [{"channel_id": c["channel_id"], "name": c.get("name"), "state": c["state"],
+                 "fix": next((b.get("fix") for b in c.get("bots") or [] if b.get("fix")), None)}
+                for c in out.get("channels") or []]
+    return {
+        "enabled": out.get("enabled"),
+        "channels_ready": sum(1 for c in channels if c["state"] == "ready"),
+        "channels_blocked": sum(1 for c in channels if c["state"] == "needs_fix"),
+        "channels": channels,
+    }
+
+
+@router.post("/complete")
+async def oauth_complete(body: CompleteBody, decoded: dict = Depends(require_firebase_token)):
+    _require_enabled()
+    uid, email = describe_actor(decoded)
+    complete_limiter.check(f"{uid}:discord_oauth_complete")
+    if not (0 < len(body.code) <= 200):
+        return _err(400, "code_rejected", "Discord didn't accept that sign-in. Start again from DRB.")
+
+    # 1. Consume the state (single use; must be this caller's).
+    try:
+        state = await oauth.consume_state(body.state, uid)
+    except oauth.StateError as e:
+        await write_audit(uid, email, "discord.oauth.refused", details={"reason": e.code})
+        return _err(e.status, e.code, e.message)
+    org_id, token_id = state["org_id"], state["token_id"]
+    return_to = state.get("return_to")
+
+    # 2. Still the owner of that org (fresh read, not the token's claims).
+    if not await _still_owner(decoded, uid, org_id):
+        await write_audit(uid, email, "discord.oauth.refused", details={"org_id": org_id, "reason": "not_owner"})
+        return _err(403, "not_owner", "Only the organization owner can connect a Discord server.")
+
+    # 3. Exchange the code with that bot application's own secret.
+    tok = await fstore.doc_get(leases.TOKENS, token_id)
+    secret = await oauth.load_secret(token_id)
+    if (not tok or not tok.get("application_id") or not secret
+            or tok.get("org_id") not in (None, org_id) or tok.get("valid") is False):
+        return _err(409, "bot_unavailable", "That radio bot can no longer connect servers. Start again from DRB.")
+    client_id = str(tok["application_id"])
+    r = await oauth.exchange_code(client_id, secret, body.code)
+    if not r.ok or not isinstance(r.data, dict):
+        await write_audit(uid, email, "discord.oauth.refused", details={
+            "org_id": org_id, "token_id": token_id, "reason": "exchange_failed", "status": r.status})
+        if r.status is not None and 400 <= r.status < 500 and r.status != 429:
+            return _err(400, "code_rejected", "Discord didn't accept that sign-in. Start again from DRB.")
+        return _err(502, "discord_unavailable", "Couldn't reach Discord to finish connecting. Start again from DRB.")
+    await oauth.revoke(client_id, secret, r.data.get("access_token"))
+
+    # 4. The server is the one Discord's exchange names — never the query string.
+    guild = r.data.get("guild") if isinstance(r.data.get("guild"), dict) else None
+    guild_id = str((guild or {}).get("id") or "")
+    if not guild_id.isdigit():
+        await write_audit(uid, email, "discord.oauth.refused", details={
+            "org_id": org_id, "token_id": token_id, "reason": "no_guild"})
+        return _err(400, "no_guild", "Discord didn't say which server the bot was added to. Start again from DRB.")
+    guild_name = guild.get("name") if isinstance(guild.get("name"), str) else None
+
+    # 5. Link + settle (a server another org linked is never moved: 409).
+    try:
+        result = await glink.link_and_settle(org_id, guild_id, guild_name, uid, "oauth",
+                                             extra={"linked_with_token_id": token_id})
+    except glink.LinkError as e:
+        await write_audit(uid, email, "discord.guild.link_refused", details={
+            "guild_id": guild_id, "org_id": org_id, "reason": e.code, "via": "oauth", "token_id": token_id})
+        return _err(e.status, e.code, e.message, guild_name=guild_name, bot_name=tok.get("name"))
+
+    # 6. The bot is in the server now: record it, drop stale cached answers, audit.
+    members = tok.get("guild_ids")
+    if isinstance(members, list) and guild_id not in members:
+        await fstore.doc_update(leases.TOKENS, token_id, {"guild_ids": [*members, guild_id]})
+    discord_rest.invalidate(guild_id=guild_id)
+    await write_audit(uid, email, "discord.guild.link", details={
+        "guild_id": guild_id, "org_id": org_id, "already_linked": result["already_linked"],
+        "previous_org_id": result["previous_org_id"], "streams_stopped": len(result["streams_stopped"]),
+        "via": "oauth", "token_id": token_id,
+    })
+    logger.info(f"Discord guild {guild_id} linked to org={org_id!r} via OAuth (bot {token_id}) by uid={uid}")
+
+    # 7. Which voice channels are ready, in the same request (capped).
+    check = await _channel_summary(guild_id, org_id, state.get("node_id"))
+    return {
+        "ok": True,
+        "guild_id": guild_id,
+        "guild_name": guild_name,
+        "org_id": org_id,
+        "bot_name": tok.get("name"),
+        "already_linked": result["already_linked"],
+        "streams_stopped": len(result["streams_stopped"]),
+        "node_id": state.get("node_id"),
+        "return_to": return_to,
+        "checked": check is not None,
+        "channels_ready": check["channels_ready"] if check else None,
+        "channels_blocked": check["channels_blocked"] if check else None,
+        "channels": check["channels"] if check else [],
+    }

drb-c2-core/app/routers/tokens.py

@@ -2,8 +2,10 @@ import uuid
 from fastapi import APIRouter, HTTPException, Depends, Query
 from pydantic import BaseModel
 from typing import Optional
+from app.config import settings
 from app.internal import firestore as fstore
 from app.internal import discord_leases as leases
+from app.internal import discord_oauth
 from app.internal import discord_rest
 from app.internal.audit import write_audit
 from app.internal.auth import (
@@ -27,9 +29,27 @@ class TokenCreate(BaseModel):
     org_id: Optional[str] = None
 
 
-def _masked(t: dict) -> dict:
+# What GET /tokens may return of a bot_tokens doc (server-26#225 S6: an
-    out = {**t, "token": "•••" + (t.get("token") or "")[-4:]}
+# allowlist, so a field added to the doc later — a secret, say — never
+# leaks by default). The token string itself is masked; OAuth client secrets
+# live in bot_oauth/{id}, never on this doc, and are never returned.
+_PUBLIC_TOKEN_FIELDS = (
+    "token_id", "name", "org_id", "valid", "guild_ids", "check_error", "validated_at",
+    "application_id", "bot_user_id", "bot_name", "lease", "in_use", "assigned_node_id", "assigned_at",
+    "preferred_for_system_id",
+)
+
+
+def _masked(t: dict, oauth_ids: Optional[set] = None) -> dict:
+    out = {k: t[k] for k in _PUBLIC_TOKEN_FIELDS if k in t}
+    out["token"] = "•••" + (t.get("token") or "")[-4:]
     out["invite_url"] = leases.invite_url(t.get("application_id"))
+    members = t.get("guild_ids")
+    out["server_count"] = len(members) if isinstance(members, list) else None
+    # Pool bots stop being offered at the cap (owner decision Q3); BYO bots have none.
+    out["server_cap"] = settings.discord_pool_bot_guild_cap if t.get("org_id") is None else None
+    if oauth_ids is not None:
+        out["oauth_enabled"] = t.get("token_id") in oauth_ids
     return out
 
 
@@ -47,7 +67,8 @@ async def _republish_after_release(lease: Optional[dict]) -> None:
 @router.get("")
 async def list_tokens(_: dict = Depends(require_admin_token)):
     """List all tokens. The actual token string is masked for safety."""
-    return [_masked(t) for t in await fstore.collection_list("bot_tokens")]
+    oauth_ids = await discord_oauth.enabled_token_ids()
+    return [_masked(t, oauth_ids) for t in await fstore.collection_list("bot_tokens")]
 
 
 @router.post("", status_code=201)
@@ -83,7 +104,7 @@ async def refresh_token(token_id: str, _: dict = Depends(require_admin_token)):
     if not await fstore.doc_get("bot_tokens", token_id):
         raise HTTPException(404, "Token not found.")
     await discord_rest.refresh_membership([token_id])
-    return _masked(await fstore.doc_get("bot_tokens", token_id) or {})
+    return _masked(await fstore.doc_get("bot_tokens", token_id) or {}, await discord_oauth.enabled_token_ids())
 
 
 @router.post("/flush", status_code=200)
@@ -265,3 +286,57 @@ async def delete_token(token_id: str, _: dict = Depends(require_admin_token)):
     if existing.get("lease"):
         raise HTTPException(409, "Token is currently leased to a node. Stop its radio channels (or flush) first.")
     await fstore.doc_delete("bot_tokens", token_id)
+    await discord_oauth.clear_secret(token_id)
+
+
+class OAuthSecret(BaseModel):
+    # Length is checked in the handler, not here: a pydantic 422 echoes the
+    # rejected input back in its body.
+    client_secret: str
+
+
+@router.put("/{token_id}/oauth", status_code=200)
+async def set_oauth_secret(token_id: str, body: OAuthSecret, decoded: dict = Depends(require_admin_token)):
+    """Set (or replace) this radio bot application's OAuth2 client secret, so
+    "Connect a Discord server" can invite it (server-26#225 S6). Checked with
+    Discord first (client-credentials grant) — a wrong secret is refused and
+    nothing is stored. The secret goes to bot_oauth/{id}, server-side only:
+    it is never echoed, never returned by any route and never logged."""
+    if "/" in token_id:
+        raise HTTPException(400, "Bad id.")
+    tok = await fstore.doc_get("bot_tokens", token_id)
+    if not tok:
+        raise HTTPException(404, "Token not found.")
+    if not tok.get("application_id"):
+        raise HTTPException(409, "DRB doesn't know this bot's application id yet. Re-check the bot, then try again.")
+    secret = (body.client_secret or "").strip()
+    if not (discord_oauth.SECRET_MIN <= len(secret) <= discord_oauth.SECRET_MAX):
+        raise HTTPException(400, "That doesn't look like a Discord client secret. Copy it from the Discord "
+                                 "Developer Portal → OAuth2.")
+    r = await discord_oauth.check_secret(str(tok["application_id"]), secret)
+    uid, email = describe_actor(decoded)
+    if not r.ok:
+        await write_audit(uid, email, "discord.token.oauth_refused", details={
+            "token_id": token_id, "status": r.status})
+        if r.status in (400, 401):
+            raise HTTPException(400, "Discord didn't accept that client secret for this bot's application. "
+                                     "Copy it again from the Discord Developer Portal → OAuth2.")
+        raise HTTPException(503, "Couldn't check the secret with Discord right now. Nothing was saved; "
+                                 "try again in a minute.")
+    await discord_oauth.store_secret(token_id, secret, uid)
+    await write_audit(uid, email, "discord.token.oauth_set", details={"token_id": token_id})
+    logger.info(f"Radio bot {token_id}: OAuth secret set by uid={uid}")
+    return {"ok": True, "token_id": token_id, "oauth_enabled": True,
+            "redirect_uri_configured": discord_oauth.enabled()}
+
+
+@router.delete("/{token_id}/oauth", status_code=200)
+async def clear_oauth_secret(token_id: str, decoded: dict = Depends(require_admin_token)):
+    """Forget this bot's OAuth secret: Connect stops offering it."""
+    if "/" in token_id:
+        raise HTTPException(400, "Bad id.")
+    removed = await discord_oauth.clear_secret(token_id)
+    uid, email = describe_actor(decoded)
+    if removed:
+        await write_audit(uid, email, "discord.token.oauth_clear", details={"token_id": token_id})
+    return {"ok": True, "token_id": token_id, "oauth_enabled": False}

drb-c2-core/tests/test_discord_oauth.py

@@ -0,0 +1,412 @@
+"""
+Per-bot OAuth "Connect a Discord server" (server-26#225 S6).
+
+Discord is an httpx.MockTransport; Firestore is tests/fake_firestore.py.
+"""
+import asyncio
+import base64
+import json
+import logging
+from datetime import datetime, timedelta, timezone
+from urllib.parse import parse_qs, urlparse
+
+import httpx
+import pytest
+from fastapi import HTTPException
+
+from app.config import settings
+from app.internal import discord_guild_link as glink
+from app.internal import discord_oauth as oauth
+from app.internal import discord_rest
+from app.internal import discord_streams as streams
+from app.routers import discord_oauth as router
+from app.routers import tokens as tokens_router
+from tests.fake_firestore import FakeFirestore
+
+REDIRECT = "https://drb.example.test/discord/connected"
+SECRET_A = "s3cr3t-app-a-0123456789abcdefXYZ"
+SECRET_POOL = "s3cr3t-pool-0123456789abcdefXYZ"
+GUILD = "100000000000000001"
+OWNER_A = {"uid": "owner-a", "email": "a@example.test", "org_id": "org-a", "org_role": "owner"}
+OWNER_B = {"uid": "owner-b", "email": "b@example.test", "org_id": "org-b", "org_role": "owner"}
+ADMIN = {"uid": "admin-1", "email": "admin@example.test", "role": "admin", "org_id": "founding"}
+
+
+class Discord:
+    """Scripted OAuth endpoints. ``token_reply`` answers POST /oauth2/token."""
+
+    def __init__(self):
+        self.calls: list[httpx.Request] = []
+        self.token_reply = httpx.Response(200, json={
+            "access_token": "user-access-token", "token_type": "Bearer", "scope": "bot",
+            "guild": {"id": GUILD, "name": "Station 7"}})
+
+    def __call__(self, request: httpx.Request) -> httpx.Response:
+        self.calls.append(request)
+        path = request.url.path.removeprefix("/api/v10")
+        if path == "/oauth2/token":
+            r = self.token_reply
+            return r(request) if callable(r) else r
+        if path == "/oauth2/token/revoke":
+            return httpx.Response(200, json={})
+        return httpx.Response(404, json={"code": 0})
+
+    def form(self, i: int) -> dict:
+        return {k: v[0] for k, v in parse_qs(self.calls[i].content.decode()).items()}
+
+    def basic(self, i: int) -> str:
+        raw = self.calls[i].headers["authorization"].split(" ", 1)[1]
+        return base64.b64decode(raw).decode()
+
+    def paths(self):
+        return [r.url.path.removeprefix("/api/v10") for r in self.calls]
+
+
+@pytest.fixture
+def discord(monkeypatch):
+    d = Discord()
+    discord_rest.reset()
+    monkeypatch.setattr(discord_rest, "_transport", httpx.MockTransport(d))
+    return d
+
+
+@pytest.fixture
+def fs(monkeypatch, discord):
+    fake = FakeFirestore.install(monkeypatch)
+    monkeypatch.setattr(settings, "discord_oauth_redirect_uri", REDIRECT)
+    monkeypatch.setattr(settings, "discord_unlinked_guild_org_id", "")
+    monkeypatch.setattr(settings, "discord_pool_bot_guild_cap", 90)
+    for org in ("org-a", "org-b", "founding"):
+        fake.put("organizations", org, {"org_id": org, "name": org})
+    fake.put("org_members", "owner-a", {"uid": "owner-a", "org_id": "org-a", "org_role": "owner"})
+    fake.put("org_members", "owner-b", {"uid": "owner-b", "org_id": "org-b", "org_role": "owner"})
+    # org-a's own bot, two pool bots (one full), org-b's own bot.
+    fake.put("bot_tokens", "own-a", {"token_id": "own-a", "name": "Riverton Radio", "token": "tok-own-a",
+                                     "org_id": "org-a", "application_id": "111", "guild_ids": []})
+    fake.put("bot_tokens", "pool-1", {"token_id": "pool-1", "name": "DRB Radio 1", "token": "tok-pool-1",
+                                      "org_id": None, "application_id": "201", "guild_ids": ["9"] * 40})
+    fake.put("bot_tokens", "pool-2", {"token_id": "pool-2", "name": "DRB Radio 2", "token": "tok-pool-2",
+                                      "org_id": None, "application_id": "202", "guild_ids": ["9"] * 90})
+    fake.put("bot_tokens", "own-b", {"token_id": "own-b", "name": "Harbor Radio", "token": "tok-own-b",
+                                     "org_id": "org-b", "application_id": "333", "guild_ids": []})
+    for tid, secret in (("own-a", SECRET_A), ("pool-1", SECRET_POOL), ("pool-2", SECRET_POOL),
+                        ("own-b", SECRET_A)):
+        fake.put(oauth.OAUTH, tid, {"token_id": tid, "client_secret": secret, "enabled": True})
+    for lim in (router.start_limiter, router.complete_limiter):
+        lim._log.clear()
+    stopped: list = []
+
+    async def _stop_guild(guild_id, reason, channel_id=None, stream_id=None):
+        stopped.append((guild_id, reason))
+        return [f"{guild_id}:1"]
+    monkeypatch.setattr(streams, "stop_guild", _stop_guild)
+    fake.stopped = stopped
+
+    async def _voice_check(**kw):
+        return {"enabled": True, "channels": [
+            {"channel_id": "1", "name": "dispatch", "state": "ready", "bots": [{"fix": None}]},
+            {"channel_id": "2", "name": "fireground", "state": "needs_fix",
+             "bots": [{"fix": "In Discord: Edit Channel → Permissions → add Riverton Radio → allow Speak, "
+                              "then try again."}]},
+            {"channel_id": "3", "name": "ops", "state": "unknown", "bots": []},
+        ]}
+    monkeypatch.setattr(router, "voice_check", _voice_check)
+    return fake
+
+
+def _audit(fake, action):
+    return [e for e in (fake.data.get("audit_log") or {}).values() if e["action"] == action]
+
+
+def _status(resp):
+    return getattr(resp, "status_code", 200)
+
+
+def _body(resp):
+    return json.loads(resp.body) if hasattr(resp, "body") else resp
+
+
+async def _start(decoded=OWNER_A, **kw):
+    out = await router.oauth_start(router.StartBody(**kw), org_id=None, decoded=decoded)
+    q = parse_qs(urlparse(out["authorize_url"]).query)
+    return out, {k: v[0] for k, v in q.items()}
+
+
+async def _complete(state, decoded=OWNER_A, code="the-code"):
+    return await router.oauth_complete(router.CompleteBody(code=code, state=state), decoded=decoded)
+
+
+# ---------------------------------------------------------------------------
+# Off switch
+# ---------------------------------------------------------------------------
+
+async def test_every_route_is_404_while_redirect_uri_is_empty(fs, monkeypatch):
+    monkeypatch.setattr(settings, "discord_oauth_redirect_uri", "")
+    for call in (router.oauth_options(node_id=None, token_id=None, org_id=None, decoded=OWNER_A),
+                 router.oauth_start(router.StartBody(), org_id=None, decoded=OWNER_A),
+                 _complete("anything")):
+        with pytest.raises(HTTPException) as e:
+            await call
+        assert e.value.status_code == 404
+
+
+async def test_404_when_no_usable_bot_has_a_secret(fs):
+    for tid in ("own-a", "pool-1", "pool-2"):
+        fs.data[oauth.OAUTH].pop(tid)
+    with pytest.raises(HTTPException) as e:
+        await router.oauth_options(node_id=None, token_id=None, org_id=None, decoded=OWNER_A)
+    assert e.value.status_code == 404
+
+
+# ---------------------------------------------------------------------------
+# Which bot
+# ---------------------------------------------------------------------------
+
+async def test_org_bot_first_then_least_full_pool_bot_under_cap(fs):
+    out = await router.oauth_options(node_id=None, token_id=None, org_id=None, decoded=OWNER_A)
+    assert out["bot"] == {"name": "Riverton Radio", "reason": "org_bot", "own_bot": True}
+    fs.data[oauth.OAUTH].pop("own-a")
+    out = await router.oauth_options(node_id=None, token_id=None, org_id=None, decoded=OWNER_A)
+    assert out["bot"]["name"] == "DRB Radio 1" and out["bot"]["reason"] == "pool"
+    # pool-2 is at the cap (90): never offered, even when it's the only one left
+    fs.data[oauth.OAUTH].pop("pool-1")
+    with pytest.raises(HTTPException) as e:
+        await router.oauth_options(node_id=None, token_id=None, org_id=None, decoded=OWNER_A)
+    assert e.value.status_code == 404
+
+
+async def test_preferred_bot_for_the_nodes_system_wins(fs):
+    fs.put("nodes", "n1", {"node_id": "n1", "org_id": "org-a", "assigned_system_id": "s1"})
+    fs.put("systems", "s1", {"system_id": "s1", "org_id": "org-a", "preferred_token_id": "pool-1"})
+    out = await router.oauth_options(node_id="n1", token_id=None, org_id=None, decoded=OWNER_A)
+    assert out["bot"]["name"] == "DRB Radio 1" and out["bot"]["reason"] == "preferred"
+    # another org's node is a 404
+    fs.put("nodes", "nb", {"node_id": "nb", "org_id": "org-b", "assigned_system_id": "s1"})
+    with pytest.raises(HTTPException) as e:
+        await router.oauth_options(node_id="nb", token_id=None, org_id=None, decoded=OWNER_A)
+    assert e.value.status_code == 404
+
+
+async def test_explicit_bot_must_be_usable_by_the_org(fs):
+    out = await router.oauth_options(node_id=None, token_id="pool-1", org_id=None, decoded=OWNER_A)
+    assert out["bot"]["reason"] == "chosen"
+    for tid in ("own-b", "pool-2", "nope"):     # other org's bot, full pool bot, unknown
+        with pytest.raises(HTTPException) as e:
+            await router.oauth_options(node_id=None, token_id=tid, org_id=None, decoded=OWNER_A)
+        assert e.value.status_code == 404
+
+
+# ---------------------------------------------------------------------------
+# Start
+# ---------------------------------------------------------------------------
+
+async def test_start_returns_the_bot_invite_and_stores_only_a_hash_of_the_state(fs):
+    out, q = await _start(return_to="/nodes/n1")
+    assert urlparse(out["authorize_url"]).netloc == "discord.com"
+    assert q["client_id"] == "111" and q["scope"] == "bot" and q["response_type"] == "code"
+    assert q["permissions"] == str((1 << 10) | (1 << 20) | (1 << 21))
+    assert q["redirect_uri"] == REDIRECT
+    nonce = q["state"]
+    assert len(nonce) >= 40
+    assert fs.get(oauth.STATES, nonce) is None                       # the nonce itself is never stored
+    st = fs.get(oauth.STATES, oauth.state_key(nonce))
+    assert st["org_id"] == "org-a" and st["uid"] == "owner-a" and st["token_id"] == "own-a"
+    assert st["return_to"] == "/nodes/n1" and st["used_at"] is None
+    assert len(_audit(fs, "discord.oauth.start")) == 1
+
+
+@pytest.mark.parametrize("bad", ["https://evil.test/x", "//evil.test/x", "/\\evil.test", "nodes", "/a\nb"])
+async def test_return_to_is_a_same_site_path_only(fs, bad):
+    _, q = await _start(return_to=bad)
+    assert fs.get(oauth.STATES, oauth.state_key(q["state"]))["return_to"] is None
+
+
+async def test_start_rate_limited_per_user(fs):
+    for _ in range(5):
+        await _start()
+    with pytest.raises(HTTPException) as e:
+        await _start()
+    assert e.value.status_code == 429
+
+
+# ---------------------------------------------------------------------------
+# Complete
+# ---------------------------------------------------------------------------
+
+async def test_complete_links_the_server_from_the_exchange_and_reports_channels(fs, discord):
+    _, q = await _start(return_to="/settings/discord")
+    out = await _complete(q["state"])
+    assert out["ok"] and out["guild_id"] == GUILD and out["guild_name"] == "Station 7"
+    assert out["org_id"] == "org-a" and out["bot_name"] == "Riverton Radio"
+    assert (out["channels_ready"], out["channels_blocked"]) == (1, 1)
+    assert out["channels"][1]["fix"].startswith("In Discord")
+    assert out["return_to"] == "/settings/discord"
+
+    g = fs.get(streams.GUILDS, GUILD)
+    assert g["org_id"] == "org-a" and g["linked_via"] == "oauth" and g["linked_by"] == "owner-a"
+    assert g["linked_with_token_id"] == "own-a"
+    assert fs.get("bot_tokens", "own-a")["guild_ids"] == [GUILD]
+    # the exchange used that bot application's own credentials and the exact redirect URI
+    assert discord.paths() == ["/oauth2/token", "/oauth2/token/revoke"]
+    assert discord.basic(0) == f"111:{SECRET_A}"
+    assert discord.form(0) == {"grant_type": "authorization_code", "code": "the-code", "redirect_uri": REDIRECT}
+    assert discord.form(1)["token"] == "user-access-token"           # revoked straight away
+    assert fs.get(oauth.STATES, oauth.state_key(q["state"]))["used_at"]
+    [entry] = _audit(fs, "discord.guild.link")
+    assert entry["details"]["via"] == "oauth" and entry["details"]["guild_id"] == GUILD
+
+
+async def test_state_is_single_use(fs):
+    _, q = await _start()
+    assert _status(await _complete(q["state"])) == 200
+    again = await _complete(q["state"])
+    assert _status(again) == 409 and _body(again)["code"] == "state_used"
+
+
+async def test_another_users_state_is_refused_and_burned(fs):
+    _, q = await _start(OWNER_A)
+    resp = await _complete(q["state"], decoded=OWNER_B)              # login-CSRF shape
+    assert _status(resp) == 403 and _body(resp)["code"] == "wrong_user"
+    assert fs.get(streams.GUILDS, GUILD) is None
+    assert _status(await _complete(q["state"], decoded=OWNER_A)) == 409   # consumed
+
+
+async def test_expired_and_unknown_states(fs):
+    _, q = await _start()
+    key = oauth.state_key(q["state"])
+    past = (datetime.now(timezone.utc) - timedelta(seconds=1)).isoformat()
+    fs.put(oauth.STATES, key, {**fs.get(oauth.STATES, key), "expires_at": past})
+    resp = await _complete(q["state"])
+    assert _status(resp) == 410 and fs.get(streams.GUILDS, GUILD) is None
+    assert _status(await _complete("not-a-state")) == 400
+
+
+async def test_caller_who_is_no_longer_owner_is_refused(fs, discord):
+    _, q = await _start()
+    fs.put("org_members", "owner-a", {"uid": "owner-a", "org_id": "org-a", "org_role": "member"})
+    resp = await _complete(q["state"])                               # token claims still say owner
+    assert _status(resp) == 403 and _body(resp)["code"] == "not_owner"
+    assert discord.calls == [] and fs.get(streams.GUILDS, GUILD) is None
+
+
+async def test_exchange_without_a_guild_links_nothing(fs, discord):
+    discord.token_reply = httpx.Response(200, json={"access_token": "x", "scope": "bot"})
+    _, q = await _start()
+    resp = await _complete(q["state"])
+    assert _status(resp) == 400 and _body(resp)["code"] == "no_guild"
+    assert fs.get(streams.GUILDS, GUILD) is None
+
+
+async def test_rejected_code_and_unreachable_discord(fs, discord):
+    discord.token_reply = httpx.Response(400, json={"error": "invalid_grant"})
+    _, q = await _start()
+    resp = await _complete(q["state"])
+    assert _status(resp) == 400 and _body(resp)["code"] == "code_rejected"
+
+    def _down(request):
+        raise httpx.ConnectError("down", request=request)
+    discord.token_reply = _down
+    _, q = await _start()
+    resp = await _complete(q["state"])
+    assert _status(resp) == 502 and fs.get(streams.GUILDS, GUILD) is None
+
+
+async def test_server_linked_to_another_org_is_never_moved(fs):
+    fs.put(streams.GUILDS, GUILD, {"guild_id": GUILD, "org_id": "org-b", "guild_name": "Harbor"})
+    _, q = await _start()
+    resp = await _complete(q["state"])
+    assert _status(resp) == 409 and _body(resp)["code"] == "linked_elsewhere"
+    assert fs.get(streams.GUILDS, GUILD)["org_id"] == "org-b"
+    assert fs.stopped == []
+    assert [e["details"]["via"] for e in _audit(fs, "discord.guild.link_refused")] == ["oauth"]
+
+
+async def test_linking_away_from_the_fallback_org_stops_its_streams(fs, monkeypatch):
+    monkeypatch.setattr(settings, "discord_unlinked_guild_org_id", "founding")
+    _, q = await _start()
+    out = await _complete(q["state"])
+    assert out["streams_stopped"] == 1 and fs.stopped == [(GUILD, glink.RELINK_STOP_REASON)]
+
+
+async def test_permission_check_timeout_leaves_the_link_standing(fs, monkeypatch):
+    monkeypatch.setattr(settings, "discord_oauth_link_check_seconds", 0.05)
+
+    async def _slow(**kw):
+        await asyncio.sleep(1)
+    monkeypatch.setattr(router, "voice_check", _slow)
+    _, q = await _start()
+    out = await _complete(q["state"])
+    assert out["ok"] and out["checked"] is False and out["channels_ready"] is None
+    assert fs.get(streams.GUILDS, GUILD)["org_id"] == "org-a"
+
+
+async def test_bot_handed_to_another_org_after_start_is_refused(fs):
+    _, q = await _start()
+    fs.put("bot_tokens", "own-a", {**fs.get("bot_tokens", "own-a"), "org_id": "org-b"})
+    resp = await _complete(q["state"])
+    assert _status(resp) == 409 and _body(resp)["code"] == "bot_unavailable"
+
+
+# ---------------------------------------------------------------------------
+# Secrets: PUT /tokens/{id}/oauth, never returned, never logged
+# ---------------------------------------------------------------------------
+
+async def test_set_secret_checks_it_with_discord_and_never_echoes_it(fs, discord, caplog):
+    caplog.set_level(logging.DEBUG)
+    fs.data[oauth.OAUTH].clear()
+    discord.token_reply = httpx.Response(401, json={"error": "invalid_client"})
+    with pytest.raises(HTTPException) as e:
+        await tokens_router.set_oauth_secret("pool-1", tokens_router.OAuthSecret(client_secret=SECRET_POOL),
+                                             decoded=ADMIN)
+    assert e.value.status_code == 400 and fs.get(oauth.OAUTH, "pool-1") is None
+
+    discord.token_reply = httpx.Response(200, json={"access_token": "cc-token", "scope": "identify"})
+    out = await tokens_router.set_oauth_secret("pool-1", tokens_router.OAuthSecret(client_secret=SECRET_POOL),
+                                               decoded=ADMIN)
+    assert out["oauth_enabled"] is True
+    assert discord.form(1) == {"grant_type": "client_credentials", "scope": "identify"}
+    assert discord.basic(1) == f"201:{SECRET_POOL}"
+    assert discord.form(2)["token"] == "cc-token"                    # the check's token is revoked
+    assert fs.get(oauth.OAUTH, "pool-1")["client_secret"] == SECRET_POOL
+    assert "client_secret" not in fs.get("bot_tokens", "pool-1")
+
+    listed = await tokens_router.list_tokens(_=ADMIN)
+    everything = json.dumps([listed, out, fs.data.get("audit_log")]) + caplog.text
+    assert SECRET_POOL not in everything and "tok-pool-1" not in everything
+    row = next(t for t in listed if t["token_id"] == "pool-1")
+    assert row["oauth_enabled"] is True and row["server_count"] == 40 and row["server_cap"] == 90
+
+
+async def test_unreachable_discord_saves_nothing(fs, discord):
+    fs.data[oauth.OAUTH].clear()
+
+    def _down(request):
+        raise httpx.ConnectError("down", request=request)
+    discord.token_reply = _down
+    with pytest.raises(HTTPException) as e:
+        await tokens_router.set_oauth_secret("pool-1", tokens_router.OAuthSecret(client_secret=SECRET_POOL),
+                                             decoded=ADMIN)
+    assert e.value.status_code == 503 and fs.get(oauth.OAUTH, "pool-1") is None
+
+
+async def test_token_list_is_an_allowlist(fs):
+    fs.put("bot_tokens", "pool-1", {**fs.get("bot_tokens", "pool-1"), "client_secret": "LEAK", "future": "x"})
+    row = next(t for t in await tokens_router.list_tokens(_=ADMIN) if t["token_id"] == "pool-1")
+    assert "client_secret" not in row and "future" not in row
+    assert row["token"] == "•••ol-1" and row["name"] == "DRB Radio 1"
+
+
+async def test_clear_and_delete_remove_the_secret(fs):
+    await tokens_router.clear_oauth_secret("pool-1", decoded=ADMIN)
+    assert fs.get(oauth.OAUTH, "pool-1") is None
+    await tokens_router.delete_token("own-a", _=ADMIN)
+    assert fs.get(oauth.OAUTH, "own-a") is None
+
+
+async def test_malformed_secret_is_refused_without_echo(fs, discord):
+    fs.data[oauth.OAUTH].clear()
+    with pytest.raises(HTTPException) as e:
+        await tokens_router.set_oauth_secret("pool-1", tokens_router.OAuthSecret(client_secret="short-secret"),
+                                             decoded=ADMIN)
+    assert e.value.status_code == 400 and "short-secret" not in str(e.value.detail)
+    assert discord.calls == [] and fs.get(oauth.OAUTH, "pool-1") is None

infra/firestore/firestore.rules

@@ -154,6 +154,16 @@ service cloud.firestore {
       allow read, write: if false;
     }
 
+    // Radio-bot OAuth2 client secrets and in-flight "Connect a Discord
+    // server" states (server-26#225 S6). Secrets: c2-core only, never any client.
+    match /bot_oauth/{tokenId} {
+      allow read, write: if false;
+    }
+
+    match /discord_oauth_states/{stateHash} {
+      allow read, write: if false;
+    }
+
     match /waitlist/{entryId} {
       allow read, write: if false; // POST /waitlist writes via the admin SDK
     }