Node page Discord picker: org's linked servers + voice channels, not raw IDs (#213 item 3, #109 item 13)

- c2-core routers/discord_web.py:
  PUT  /discord/guilds/{id}/directory (service key): the server bot pushes each
       server's name + voice channels; merged onto discord_guilds, never touches org_id.
  GET  /discord/nodes/{id}/join-options: the node org's servers (same
       resolve rule as the bot), their voice channels with any live stream,
       radio-bot readiness + invite links, and this node's streams. No tokens.
  POST /discord/nodes/{id}/streams: stream pinned to the node; server must be
       the node org's, channel must be in the bot's directory, guild + node caps.
  POST /discord/nodes/{id}/streams/stop: one stream on this node, or all.
  Control = platform admin, org owner, or operator owning the node.
- nodes.py: legacy raw-ID discord_join now refuses a server outside the node's org.
- discord bot: DirectorySync cog pushes on ready (15-min loop), join/leave,
  rename, voice/category channel changes (debounced).
- frontend: NodeDiscordRadio panel + picker replaces DiscordJoinModal and the
  admin-only Join/Leave buttons.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

drb-c2-core/app/main.py

@@ -20,6 +20,7 @@ from app.internal.auth import (
 from app.routers import nodes, systems, calls, upload, tokens, incidents, alerts, admin, trips, places, links, users
 from app.routers import enrollment, media, org, waitlist, telemetry, replay, traffic, cameras
 from app.routers import discord as discord_router
+from app.routers import discord_web
 from app.internal import dynsec
 from app.internal import firestore as fstore
 
@@ -158,6 +159,9 @@ app.include_router(media.router)
 # auth is per-endpoint: bot routes take the service key (or a platform admin);
 # the token-fetch route takes the calling node's own API key.
 app.include_router(discord_router.router)
+# auth is per-endpoint: directory push takes the service key; the node-page
+# picker takes a Firebase token scoped to the node's org (server-26#213).
+app.include_router(discord_web.router)
 # 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_web.py

@@ -0,0 +1,283 @@
+"""
+Discord radio channels from the web node page (server-26#213 item 3, #109 item 13).
+
+The node page used to take raw guild/channel IDs. It now picks from:
+
+  * the Discord servers the caller's org owns — ``discord_guilds`` docs that
+    resolve to the node's org (an explicit link, or the unlinked-server
+    fallback org, same rule the bot uses: streams.resolve_guild_org), and
+  * their voice channels, from a directory the DRB server bot pushes
+    (``PUT /discord/guilds/{id}/directory``). The bot already holds its
+    gateway cache of every server it is in; C2 never calls Discord with a
+    token on the browser's behalf and no token string ever reaches the browser.
+
+A pick becomes a ``discord_streams`` desired-state doc **pinned to this node**
+(streams.start_stream(node_id=...)), exactly what the old raw-ID path did, with
+the checks that path skipped: the server must belong to the node's org, the
+channel must be a voice channel the bot reported in that server, and the
+per-server and per-node stream caps apply.
+
+Auth:
+  * PUT  /discord/guilds/{id}/directory          — service key (the server bot) only.
+  * GET  /discord/nodes/{node_id}/join-options   — any signed-in member of the node's org
+                                                   (or a platform admin); ``can_control`` says
+                                                   whether the caller may start/stop.
+  * POST /discord/nodes/{node_id}/streams        — platform admin, the org's owner, or an
+  * POST /discord/nodes/{node_id}/streams/stop     operator whose owned_node_ids lists the node.
+"""
+from __future__ import annotations
+
+from datetime import datetime, timezone
+from typing import Optional
+
+from fastapi import APIRouter, Depends, HTTPException
+from fastapi.responses import JSONResponse
+from pydantic import BaseModel, Field
+
+from app.config import settings
+from app.internal import discord_leases as leases
+from app.internal import discord_streams as streams
+from app.internal import firestore as fstore
+from app.internal.auth import (
+    get_org_role,
+    get_role,
+    require_firebase_token,
+    require_service_key,
+)
+from app.internal.logger import logger
+
+router = APIRouter(prefix="/discord", tags=["discord"])
+
+MAX_CHANNELS = 200
+_NAME_MAX = 100
+
+
+def _now_iso() -> str:
+    return datetime.now(timezone.utc).isoformat()
+
+
+# ---------------------------------------------------------------------------
+# Directory — pushed by the DRB server bot from its gateway cache
+# ---------------------------------------------------------------------------
+
+class VoiceChannelIn(BaseModel):
+    channel_id: str = Field(..., pattern=r"^[0-9]{1,25}$")
+    name: str = Field(..., max_length=_NAME_MAX)
+    category: Optional[str] = Field(None, max_length=_NAME_MAX)
+    position: int = 0
+
+
+class GuildDirectory(BaseModel):
+    name: Optional[str] = Field(None, max_length=_NAME_MAX)
+    present: bool = True
+    voice_channels: list[VoiceChannelIn] = Field(default_factory=list, max_length=MAX_CHANNELS)
+
+
+@router.put("/guilds/{guild_id}/directory")
+async def put_guild_directory(guild_id: str, body: GuildDirectory, _: dict = Depends(require_service_key)):
+    """The server bot's view of one Discord server: its name and voice channels.
+    Merged onto ``discord_guilds/{id}`` — never touches ``org_id`` (the link)."""
+    if not guild_id.isdigit():
+        raise HTTPException(400, "guild_id must be a Discord snowflake.")
+    data: dict = {"guild_id": guild_id, "bot_present": body.present, "directory_at": _now_iso()}
+    if body.present:
+        data["voice_channels"] = [c.model_dump() for c in body.voice_channels]
+        if body.name:
+            data["name"] = body.name
+    await fstore.doc_set(streams.GUILDS, guild_id, data)
+    return {"ok": True, "guild_id": guild_id, "voice_channels": len(body.voice_channels)}
+
+
+# ---------------------------------------------------------------------------
+# Node page — picker
+# ---------------------------------------------------------------------------
+
+async def _node_for_caller(node_id: str, decoded: dict) -> tuple[dict, bool]:
+    """The node, and whether this caller may start/stop its radio channels.
+    A node outside the caller's org is a 404, never a 403 (no existence leak)."""
+    node = await fstore.doc_get("nodes", node_id)
+    if not node:
+        raise HTTPException(404, f"Node '{node_id}' not found.")
+    if get_role(decoded) == "admin":
+        return node, True
+    org_id = decoded.get("org_id")
+    if not org_id or node.get("org_id") != org_id:
+        raise HTTPException(404, f"Node '{node_id}' not found.")
+    can = get_org_role(decoded) == "owner" or (
+        get_role(decoded) == "operator" and node_id in (decoded.get("owned_node_ids") or [])
+    )
+    return node, can
+
+
+def _require_control(can: bool) -> None:
+    if not can:
+        raise HTTPException(403, "Only the organization owner, an admin, or this node's operator can do that.")
+
+
+async def _org_guilds(org_id: str) -> list[dict]:
+    """``discord_guilds`` docs this org may stream into, bot still present."""
+    fallback = settings.discord_unlinked_guild_org_id or None
+    out = []
+    for g in await fstore.collection_list(streams.GUILDS):
+        gid = str(g.get("guild_id") or "")
+        if not gid or g.get("bot_present") is False:
+            continue
+        owner = g.get("org_id") or fallback
+        if owner and owner == org_id:
+            out.append(g)
+    out.sort(key=lambda g: ((g.get("name") or "").lower(), g["guild_id"]))
+    return out
+
+
+def _bot_status(tokens: dict, *, node_id: str, guild_id: str, org_id: str) -> dict:
+    res = leases.pick_token(tokens, node_id=node_id, guild_id=guild_id, org_id=org_id)
+    if res.ok:
+        return {"ok": True, "code": None, "message": None, "invite": []}
+    return {"ok": False, "code": res.error, "message": streams.ERROR_TEXT.get(res.error or "", res.error),
+            "invite": [i for i in res.invite if i.get("invite_url")][:5]}
+
+
+def _stream_view(s: dict, node_names: dict) -> dict:
+    return {
+        "stream_id": s.get("stream_id"),
+        "guild_id": s.get("guild_id"),
+        "channel_id": s.get("channel_id"),
+        "guild_name": s.get("guild_name"),
+        "channel_name": s.get("channel_name"),
+        "system_name": s.get("system_name"),
+        "mode": s.get("mode"),
+        "node_id": s.get("node_id"),
+        "node_name": node_names.get(s.get("node_id") or "") or s.get("node_id"),
+        "actual": s.get("actual"),
+        "updated_at": s.get("updated_at"),
+    }
+
+
+@router.get("/nodes/{node_id}/join-options")
+async def join_options(node_id: str, decoded: dict = Depends(require_firebase_token)):
+    node, can_control = await _node_for_caller(node_id, decoded)
+    org_id = node.get("org_id")
+    system_id = streams._effective_system(node)
+    system = (await fstore.doc_get_cached("systems", system_id)) if system_id else None
+
+    guilds = await _org_guilds(org_id) if org_id else []
+    live = [s for s in await fstore.collection_list(streams.STREAMS, desired="on")
+            if s.get("org_id") == org_id]
+    nodes = {n.get("node_id"): n.get("name") for n in await fstore.collection_list("nodes", org_id=org_id)
+             if n.get("node_id")} if org_id else {}
+    by_channel = {s.get("stream_id"): s for s in live}
+    tokens = {t["token_id"]: t for t in await fstore.collection_list(leases.TOKENS) if t.get("token_id")}
+
+    out_guilds = []
+    for g in guilds:
+        gid = g["guild_id"]
+        channels = []
+        for c in sorted(g.get("voice_channels") or [],
+                        key=lambda c: (c.get("position") or 0, (c.get("name") or "").lower())):
+            s = by_channel.get(leases.stream_id_for(gid, str(c.get("channel_id"))))
+            channels.append({
+                "channel_id": str(c.get("channel_id")),
+                "name": c.get("name"),
+                "category": c.get("category"),
+                "stream": _stream_view(s, nodes) if s else None,
+            })
+        out_guilds.append({
+            "guild_id": gid,
+            "name": g.get("name"),
+            "linked": bool(g.get("org_id")),
+            "synced": g.get("voice_channels") is not None,
+            "synced_at": g.get("directory_at"),
+            "voice_channels": channels,
+            "streams_on": sum(1 for s in live if s.get("guild_id") == gid),
+            "bot": _bot_status(tokens, node_id=node_id, guild_id=gid, org_id=org_id),
+        })
+
+    here = [s for s in live if s.get("node_id") == node_id]
+    here.sort(key=lambda s: (s.get("guild_name") or "", s.get("channel_name") or "", s.get("stream_id") or ""))
+    return {
+        "node_id": node_id,
+        "node_name": node.get("name"),
+        "node_online": node.get("status") in streams.ONLINE,
+        "system_id": system_id,
+        "system_name": (system or {}).get("name") or system_id,
+        "can_control": can_control,
+        "limits": {
+            "node_max": settings.discord_node_max_streams,
+            "node_used": len(here),
+            "guild_max": settings.discord_guild_max_streams,
+        },
+        "guilds": out_guilds,
+        "streams": [_stream_view(s, nodes) for s in here],
+    }
+
+
+class NodeStreamStart(BaseModel):
+    guild_id: str
+    channel_id: str
+    mode: str = "adhoc"
+
+
+@router.post("/nodes/{node_id}/streams")
+async def start_node_stream(node_id: str, body: NodeStreamStart, decoded: dict = Depends(require_firebase_token)):
+    node, can_control = await _node_for_caller(node_id, decoded)
+    _require_control(can_control)
+    org_id = node.get("org_id")
+    if not org_id:
+        raise HTTPException(409, "This node is not in an organization yet.")
+    if body.mode not in ("persistent", "adhoc"):
+        raise HTTPException(400, "mode must be persistent or adhoc.")
+    system_id = streams._effective_system(node)
+    if not system_id:
+        raise HTTPException(409, "This node is not listening to a system yet.")
+
+    guild_id, channel_id = str(body.guild_id), str(body.channel_id)
+    guild = next((g for g in await _org_guilds(org_id) if g["guild_id"] == guild_id), None)
+    if not guild:
+        raise HTTPException(404, "That Discord server is not linked to this organization.")
+    channel = next((c for c in guild.get("voice_channels") or [] if str(c.get("channel_id")) == channel_id), None)
+    if not channel:
+        raise HTTPException(400, "That is not a voice channel DRB can see in this server. "
+                                 "Refresh the list and pick again.")
+
+    stream_id = leases.stream_id_for(guild_id, channel_id)
+    existing = await fstore.doc_get(streams.STREAMS, stream_id) or {}
+    already_here = existing.get("desired") == "on" and existing.get("node_id") == node_id
+    if not already_here:
+        used = sum(1 for s in await fstore.collection_list(streams.STREAMS, node_id=node_id)
+                   if s.get("desired") == "on")
+        if used >= settings.discord_node_max_streams:
+            raise HTTPException(409, f"This node already carries {used} radio channels, its maximum. "
+                                     "Stop one first.")
+
+    try:
+        doc = await streams.start_stream(
+            org_id=org_id, guild_id=guild_id, channel_id=channel_id, system_id=system_id,
+            node_id=node_id, mode=body.mode,
+            channel_name=channel.get("name"), guild_name=guild.get("name"),
+            created_by=f"web:{decoded.get('uid') or 'unknown'}",
+            max_streams=settings.discord_guild_max_streams,
+        )
+    except streams.StreamError as e:
+        return JSONResponse(status_code=e.status,
+                            content={"detail": e.message, "code": e.code, "invite": e.invite})
+    logger.info(f"Web: radio channel {stream_id} pinned to node {node_id} ({body.mode})")
+    return {"ok": True, "stream_id": doc["stream_id"], "token_id": doc.get("token_id"),
+            "delivered": doc.get("delivered"), "reused": doc.get("reused"), "actual": doc.get("actual")}
+
+
+class NodeStreamStop(BaseModel):
+    stream_id: Optional[str] = None
+
+
+@router.post("/nodes/{node_id}/streams/stop")
+async def stop_node_streams(node_id: str, body: NodeStreamStop, decoded: dict = Depends(require_firebase_token)):
+    """Stop one radio channel this node carries, or (no stream_id) all of them."""
+    _, can_control = await _node_for_caller(node_id, decoded)
+    _require_control(can_control)
+    if body.stream_id:
+        doc = await fstore.doc_get(streams.STREAMS, body.stream_id)
+        if not doc or doc.get("node_id") != node_id or doc.get("desired") != "on":
+            raise HTTPException(404, "This node is not carrying that radio channel.")
+        await streams.stop_stream(body.stream_id, "stopped_from_web")
+        return {"ok": True, "stopped": [body.stream_id]}
+    return {"ok": True, "stopped": await streams.stop_node(node_id, "stopped_from_web")}

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

@@ -168,6 +168,11 @@ async def send_command(
     if cmd.action == "discord_join":
         if not cmd.guild_id or not cmd.channel_id:
             raise HTTPException(400, "guild_id and channel_id are required.")
+        # The raw-ID path never checked the server belongs to the node's org
+        # (server-26#213): one org's node could be streamed into another org's
+        # Discord server. The web picker uses /discord/nodes/{id}/streams now.
+        if await discord_streams.resolve_guild_org(cmd.guild_id) != node.get("org_id"):
+            raise HTTPException(404, "That Discord server is not linked to this node's organization.")
         system_id = discord_streams._effective_system(node)
         if not system_id:
             raise HTTPException(409, "This node is not listening to a system yet.")