diff --git a/drb-c2-core/app/internal/discord_guild_link.py b/drb-c2-core/app/internal/discord_guild_link.py index c846457..af9da2d 100644 --- a/drb-c2-core/app/internal/discord_guild_link.py +++ b/drb-c2-core/app/internal/discord_guild_link.py @@ -132,6 +132,9 @@ def _redeem_txn(code: str, guild_id: str, guild_name: Optional[str], discord_use "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")} diff --git a/drb-c2-core/app/internal/discord_leases.py b/drb-c2-core/app/internal/discord_leases.py index 1c4c31b..aca4b53 100644 --- a/drb-c2-core/app/internal/discord_leases.py +++ b/drb-c2-core/app/internal/discord_leases.py @@ -257,6 +257,30 @@ async def detach_stream(stream_id: str, stream_updates: Optional[dict] = None) - return await fstore.run_transaction(_txn) +async def delete_stopped_stream(stream_id: str) -> bool: + """Delete a stream doc, but only while it is stopped and holds no token. + + True when the doc is gone (deleted now, or already absent). False when it + is desired-on again or names a token — a concurrent start re-leased it, + and deleting it would leave that token's lease listing a stream nobody can + stop. ``lease_for_stream`` writes the stream doc in its own transaction, + so reading it here is enough to serialise against it.""" + db = fstore.db + + def _txn(txn): + ref = db.collection(STREAMS).document(stream_id) + snap = ref.get(transaction=txn) + if not snap.exists: + return True + stream = snap.to_dict() or {} + if stream.get("desired") == "on" or stream.get("token_id"): + return False + txn.delete(ref) + return True + + return await fstore.run_transaction(_txn) + + async def stamp_release_seq(node_id: str, seq: int) -> None: """Record the desired-state version that dropped each releasing token, so a stale actual report (from before the node saw the drop) can never free it.""" @@ -411,3 +435,42 @@ async def refresh_membership(token_ids: list[str]) -> int: changed += 1 await fstore.doc_update(TOKENS, tok_id, updates) return changed + + +async def leave_guild(guild_id: str, org_id: Optional[str]) -> list[dict]: + """Make the radio bots DRB holds tokens for leave one Discord server + (Discord REST ``DELETE /users/@me/guilds/{id}``, the bot's own token). + + Only platform-pool bots and ``org_id``'s own BYO bots — never another + org's branded bot. Bots whose cached membership says they are not in the + server are skipped; unknown membership is tried (404 = not a member). + The DRB *server* bot (the discord-bot service) is not in this pool, so it + stays in the server until someone kicks it. Returns one row per bot tried: + ``{token_id, name, left, error}``. Never raises for a Discord failure.""" + import httpx + guild_id = str(guild_id) + out: list[dict] = [] + for tok in await fstore.collection_list(TOKENS): + tok_id = tok.get("token_id") + if not tok_id or not tok.get("token"): + continue + if tok.get("org_id") is not None and tok.get("org_id") != org_id: + continue + members = tok.get("guild_ids") + if members is not None and guild_id not in members: + continue + row = {"token_id": tok_id, "name": tok.get("name") or tok_id, "left": False, "error": None} + try: + async with httpx.AsyncClient(timeout=10, base_url=DISCORD_API, + headers={"Authorization": f"Bot {tok['token']}"}) as client: + r = await client.delete(f"/users/@me/guilds/{guild_id}") + if r.status_code in (200, 204, 404): # 404: Unknown Guild — not a member + row["left"] = True + else: + row["error"] = f"Discord answered {r.status_code}" + except Exception as e: # network — report, do not fail the removal + row["error"] = str(e)[:200] + if row["left"] and members is not None: + await fstore.doc_update(TOKENS, tok_id, {"guild_ids": [g for g in members if g != guild_id]}) + out.append(row) + return out diff --git a/drb-c2-core/app/internal/discord_streams.py b/drb-c2-core/app/internal/discord_streams.py index dbe7556..4786a71 100644 --- a/drb-c2-core/app/internal/discord_streams.py +++ b/drb-c2-core/app/internal/discord_streams.py @@ -13,9 +13,11 @@ that should carry radio. Fields: (Idle auto-dismiss is not built yet — see the PR.) desired: "on" | "off" node_id, token_id, lease_id — placement (written with the lease, atomically) - actual: {state, error, since, at} — reported by the node, never guessed here + actual: {state, error, code, since, at} — reported by the node, never guessed here state: pending | connecting | connected | error | stopping | node_offline | paused_locally + code: the node's machine-readable reason on error (login_failed, + connect_failed, not_in_guild, bad_channel, no_audio) or None tg_filter: None — phase-2 seam (per-channel talkgroup gate); always None now now_playing_message_id, created_by_discord_id, created_at, updated_at @@ -324,10 +326,75 @@ async def stop_stream(stream_id: str, reason: str = "stopped") -> bool: }) if node_id: await publish_desired(node_id) - logger.info(f"Discord stream {stream_id} stopped ({reason})") + logger.info(f"Discord stream {stream_id} stopped ({reason}) node={node_id or existing.get('node_id')}" + f" guild={existing.get('guild_id')} channel={existing.get('channel_id')}") return True +async def delete_stream(stream_id: str, reason: str = "removed") -> Optional[dict]: + """Permanently remove a radio channel (server-26#213 follow-up). + + Order matters (server-26#216 lease semantics): the stream is first stopped + through the normal desired-state path — detached from its token (token -> + ``releasing``) and the node's list republished without it — and only then + is the doc deleted. The token returns to the pool exactly as for /leave, + when the node acks a list without it and no longer has a session on it + (``settle_releases`` reads the token lease, never the stream doc). + + The delete itself is a transaction that refuses if the stream was started + again in between (desired on, or a token re-leased onto it): deleting it + then would strand that token's lease on a stream that no longer exists. + Returns the doc as it was before removal, or None if it did not exist. + """ + existing = await fstore.doc_get(STREAMS, stream_id) + if not existing: + return None + if existing.get("desired") == "on" or existing.get("token_id"): + await stop_stream(stream_id, reason) + if not await leases.delete_stopped_stream(stream_id): + raise StreamError("restarted", "That radio channel was started again while it was being removed. " + "Stop it and try again.") + logger.info(f"Discord stream {stream_id} removed ({reason}) node={existing.get('node_id')}" + f" guild={existing.get('guild_id')} channel={existing.get('channel_id')}" + f" org={existing.get('org_id')}") + return existing + + +async def remove_guild(guild_id: str, by: str) -> dict: + """Forget a Discord server: remove every radio channel in it (stop first, + then delete — ``delete_stream``) and tombstone its ``discord_guilds`` doc. + + The doc is NOT deleted: with no doc, ``resolve_guild_org`` would hand the + server to ``DISCORD_UNLINKED_GUILD_ORG_ID`` the next time anyone typed a + command there. It is unlinked the way PR #218's unlink does it + (``org_id`` None + ``unlinked_at``, so no fallback) and stamped + ``removed_at`` so the admin page stops listing it. Linking it again (``/drb + link`` or the admin link) clears ``removed_at``.""" + guild_id = str(guild_id) + link = await fstore.doc_get(GUILDS, guild_id) or {} + removed = [] + for s in await fstore.collection_list(STREAMS, guild_id=guild_id): + try: + if s.get("stream_id") and await delete_stream(s["stream_id"], "server_removed"): + removed.append(s["stream_id"]) + except StreamError as e: + # A channel restarted mid-removal: stop here, leave the server + # linked, and tell the caller which channels are already gone so + # the partial removal is still audited. + e.removed = removed + raise + now = _now_iso() + await fstore.doc_set(GUILDS, guild_id, { + "guild_id": guild_id, "org_id": None, + "unlinked_at": now, "unlinked_by": by, + "removed_at": now, "removed_by": by, + }, merge=True) + logger.info(f"Discord guild {guild_id} removed by {by}: {len(removed)} radio channel(s) deleted," + f" was org={link.get('org_id')!r}") + return {"guild_id": guild_id, "previous_org_id": link.get("org_id"), "streams_removed": removed, + "guild_name": link.get("guild_name")} + + async def stop_guild(guild_id: str, reason: str, channel_id: Optional[str] = None) -> list[str]: """Stop every live stream in ONE Discord server (optionally one channel). The /leave fix for server-26#203: never touches another server's streams.""" @@ -358,6 +425,26 @@ async def stop_node(node_id: str, reason: str) -> list[str]: _VALID_STATES = {"connecting", "connected", "error", "stopped", "paused_locally"} +def _log_transition(node_id: str, doc: dict, prev_state: Optional[str], state: str, + error: Optional[str], code: Optional[str]) -> None: + """One log line per node-reported state CHANGE (never per heartbeat — the + edge re-reports every ~60s). Errors are warnings so they surface in the + c2-core log without turning debug on; a radio bot that cannot log in gets + its own line naming the token, since that is the one failure that takes + every channel on the token down at once.""" + where = (f"stream={doc.get('stream_id')} node={node_id} guild={doc.get('guild_id')}" + f" channel={doc.get('channel_id')} token={doc.get('token_id')}") + if state == "error": + if code == "login_failed": + logger.warning(f"Radio bot {doc.get('token_id')} failed to log in on node {node_id}: {error} ({where})") + else: + logger.warning(f"Discord stream {prev_state or 'pending'} -> error [{code or 'unknown'}]:" + f" {error} ({where})") + else: + logger.info(f"Discord stream {prev_state or 'pending'} -> {state} ({where})" + + (f": {error}" if error else "")) + + async def handle_actual(node_id: str, payload: dict) -> None: """``nodes/{node_id}/discord/actual``. A node may only report on streams currently placed on it — anything else is ignored, so a compromised node @@ -377,15 +464,18 @@ async def handle_actual(node_id: str, payload: dict) -> None: state = st.get("state") if st.get("state") in _VALID_STATES else "error" error = st.get("error") error = str(error)[:300] if error else None - if st.get("code") == "not_in_guild" and doc.get("token_id"): + code = st.get("code") + code = str(code)[:40] if code else None + if code == "not_in_guild" and doc.get("token_id"): recheck.add(doc["token_id"]) prev = doc.get("actual") or {} - if prev.get("state") == state and prev.get("error") == error: + if prev.get("state") == state and prev.get("error") == error and prev.get("code") == code: since = prev.get("since") or now else: since = now + _log_transition(node_id, doc, prev.get("state"), state, error, code) await fstore.doc_update(STREAMS, str(stream_id), { - "actual": {"state": state, "error": error, "since": since, "at": now}, + "actual": {"state": state, "error": error, "code": code, "since": since, "at": now}, }) if recheck: await leases.refresh_membership(sorted(recheck)) diff --git a/drb-c2-core/app/routers/discord.py b/drb-c2-core/app/routers/discord.py index 3ae3258..5e0e9b6 100644 --- a/drb-c2-core/app/routers/discord.py +++ b/drb-c2-core/app/routers/discord.py @@ -9,6 +9,8 @@ system/node/bot lists a server sees are its own org's. Auth: * bot routes — the Discord bot's service key, or a platform admin. * PUT /discord/guilds/{id}/link — platform admin only. + * DELETE /discord/streams/{stream_id}, DELETE /discord/guilds/{id} — platform + admin only: permanently remove a radio channel / a whole server. * GET /discord/node/{node_id}/tokens/{token_id} — that node's own API key; the token string is handed only to the node currently holding its lease. """ @@ -18,7 +20,7 @@ import secrets from datetime import datetime, timezone from typing import Optional -from fastapi import APIRouter, Depends, HTTPException, Security +from fastapi import APIRouter, Depends, HTTPException, Path, Security from fastapi.responses import JSONResponse from fastapi.security import HTTPAuthorizationCredentials, HTTPBearer from pydantic import BaseModel @@ -27,11 +29,15 @@ 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 require_admin_token, require_service_key_or_admin +from app.internal.audit import write_audit +from app.internal.auth import describe_actor, require_admin_token, require_service_key_or_admin from app.internal.logger import logger router = APIRouter(prefix="/discord", tags=["discord"]) _bearer = HTTPBearer(auto_error=False) +# guild:channel snowflakes (leases.stream_id_for) — anything else is a 422, +# never a Firestore path. +STREAM_ID = r"^[0-9]{1,25}:[0-9]{1,25}$" def _error(e: streams.StreamError) -> JSONResponse: @@ -162,6 +168,8 @@ async def link_guild(guild_id: str, body: GuildLink, decoded: dict = Depends(req doc = { "guild_id": guild_id, "org_id": body.org_id, "linked_by": decoded.get("uid"), "linked_at": datetime.now(timezone.utc).isoformat(), + # Re-linking a server that was unlinked or removed brings it back. + "unlinked_at": None, "unlinked_by": None, "removed_at": None, "removed_by": None, } if body.guild_name and body.guild_name.strip(): doc["guild_name"] = body.guild_name.strip()[:100] @@ -175,6 +183,64 @@ async def link_guild(guild_id: str, body: GuildLink, decoded: dict = Depends(req return {"ok": True, "guild_id": guild_id, "org_id": body.org_id, "stopped": stopped} +@router.delete("/streams/{stream_id}") +async def delete_stream(stream_id: str = Path(..., pattern=STREAM_ID), + decoded: dict = Depends(require_admin_token)): + """Permanently remove one radio channel: stopped through the desired-state + path first (node logs the bot out, lease released on its ack), then the + doc is deleted. Platform admin only; org owners use the node page route.""" + try: + doc = await streams.delete_stream(stream_id, "removed_from_web_admin") + except streams.StreamError as e: + return _error(e) + if not doc: + raise HTTPException(404, "Radio channel not found.") + uid, email = describe_actor(decoded) + await write_audit(uid, email, "discord.stream.delete", details={ + "stream_id": stream_id, "guild_id": doc.get("guild_id"), "channel_id": doc.get("channel_id"), + "org_id": doc.get("org_id"), "node_id": doc.get("node_id"), "was_desired": doc.get("desired"), + "via": "settings_discord", + }) + return {"ok": True, "stream_id": stream_id} + + +@router.delete("/guilds/{guild_id}") +async def remove_guild(guild_id: str, leave_radio_bots: bool = False, + decoded: dict = Depends(require_admin_token)): + """Remove a Discord server from DRB: delete its radio channels (each + stopped first), then unlink it with no fallback org (streams.remove_guild). + ``leave_radio_bots`` also makes the pool radio bots (and the server's org's + own BYO bots) leave it via Discord's REST API. The DRB server bot is not + one of those and stays until someone kicks it in Discord.""" + if not guild_id.isdigit(): + raise HTTPException(400, "guild_id must be a Discord server id (digits only).") + known = await fstore.doc_get(streams.GUILDS, guild_id) + if not known and not await fstore.collection_list(streams.STREAMS, guild_id=guild_id): + raise HTTPException(404, "Discord server not found.") + uid, email = describe_actor(decoded) + acting_org = await streams.resolve_guild_org(guild_id) + try: + out = await streams.remove_guild(guild_id, uid) + except streams.StreamError as e: + # Channels deleted before the race are gone for good — audit them. + await write_audit(uid, email, "discord.guild.remove", details={ + "guild_id": guild_id, "acting_org_id": acting_org, "outcome": "incomplete", "error": e.code, + "streams_removed": getattr(e, "removed", []), "leave_radio_bots": leave_radio_bots, + }) + return _error(e) + left: list = [] + if leave_radio_bots: + left = await leases.leave_guild(guild_id, out["previous_org_id"] or acting_org) + await write_audit(uid, email, "discord.guild.remove", details={ + "guild_id": guild_id, "previous_org_id": out["previous_org_id"], "acting_org_id": acting_org, + "streams_removed": out["streams_removed"], "leave_radio_bots": leave_radio_bots, + "radio_bots_left": [r["token_id"] for r in left if r["left"]], + "radio_bots_failed": [r["token_id"] for r in left if not r["left"]], + }) + return {"ok": True, "guild_id": guild_id, "streams_removed": out["streams_removed"], + "radio_bots": [{"name": r["name"], "left": r["left"], "error": r["error"]} for r in left]} + + # --------------------------------------------------------------------------- # Admin view — the web Discord page (server-26#213 item 2) # --------------------------------------------------------------------------- @@ -201,6 +267,8 @@ async def admin_overview(_: dict = Depends(require_admin_token)): guilds: dict[str, dict] = {} for gid, link in links.items(): + if link.get("removed_at"): + continue # removed from DRB (DELETE /discord/guilds/{id}); re-listed if it gets a channel again guilds[gid] = {"guild_id": gid, "guild_name": link.get("guild_name"), "linked": bool(link.get("org_id")), "org_id": link.get("org_id"), "linked_at": link.get("linked_at"), # Explicitly unlinked servers get no fallback org (mirrors diff --git a/drb-c2-core/app/routers/discord_web.py b/drb-c2-core/app/routers/discord_web.py index 8174590..22957bc 100644 --- a/drb-c2-core/app/routers/discord_web.py +++ b/drb-c2-core/app/routers/discord_web.py @@ -24,13 +24,15 @@ Auth: 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. + * DELETE /discord/nodes/{node_id}/streams/{id} — platform admin or the org's owner only + (permanent: stop first, then delete). """ from __future__ import annotations from datetime import datetime, timezone from typing import Optional -from fastapi import APIRouter, Depends, HTTPException +from fastapi import APIRouter, Depends, HTTPException, Path from fastapi.responses import JSONResponse from pydantic import BaseModel, Field @@ -38,7 +40,9 @@ 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.audit import write_audit from app.internal.auth import ( + describe_actor, get_org_role, get_role, require_firebase_token, @@ -110,6 +114,13 @@ async def _node_for_caller(node_id: str, decoded: dict) -> tuple[dict, bool]: return node, can +def _can_remove(decoded: dict) -> bool: + """Permanent removal is narrower than start/stop: platform admin or the + org's owner (``_node_for_caller`` has already pinned the org), never a + node operator.""" + return get_role(decoded) == "admin" or get_org_role(decoded) == "owner" + + def _guild_name(g: dict) -> Optional[str]: return g.get("guild_name") or g.get("name") @@ -209,6 +220,7 @@ async def join_options(node_id: str, decoded: dict = Depends(require_firebase_to "system_id": system_id, "system_name": (system or {}).get("name") or system_id, "can_control": can_control, + "can_remove": _can_remove(decoded), "limits": { "node_max": settings.discord_node_max_streams, "node_used": len(here), @@ -291,3 +303,31 @@ async def stop_node_streams(node_id: str, body: NodeStreamStop, decoded: dict = 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")} + + +@router.delete("/nodes/{node_id}/streams/{stream_id}") +async def remove_node_stream(node_id: str, + stream_id: str = Path(..., pattern=r"^[0-9]{1,25}:[0-9]{1,25}$"), + decoded: dict = Depends(require_firebase_token)): + """Permanently remove a radio channel this node carries (or last carried): + stopped through desired state first so the node logs the bot out and the + lease is released on its ack (server-26#216), then the doc is deleted.""" + node, _ = await _node_for_caller(node_id, decoded) + if not _can_remove(decoded): + raise HTTPException(403, "Only the organization owner or an admin can remove a radio channel.") + doc = await fstore.doc_get(streams.STREAMS, stream_id) + # Same org AND this node: an owner can never reach another org's channel + # through their own node's URL. + if not doc or doc.get("node_id") != node_id or doc.get("org_id") != node.get("org_id"): + raise HTTPException(404, "This node is not carrying that radio channel.") + try: + await streams.delete_stream(stream_id, "removed_from_web") + except streams.StreamError as e: + return JSONResponse(status_code=e.status, content={"detail": e.message, "code": e.code}) + uid, email = describe_actor(decoded) + await write_audit(uid, email, "discord.stream.delete", details={ + "stream_id": stream_id, "guild_id": doc.get("guild_id"), "channel_id": doc.get("channel_id"), + "org_id": doc.get("org_id"), "node_id": node_id, "was_desired": doc.get("desired"), + "via": "node_page", + }) + return {"ok": True, "stream_id": stream_id} diff --git a/drb-c2-core/tests/fake_firestore.py b/drb-c2-core/tests/fake_firestore.py index 56cdccc..8f1685f 100644 --- a/drb-c2-core/tests/fake_firestore.py +++ b/drb-c2-core/tests/fake_firestore.py @@ -75,6 +75,9 @@ class _Txn: def update(self, ref: _DocRef, data: dict): self.writes.append(("update", ref.coll, ref.id, copy.deepcopy(data), True)) + def delete(self, ref: _DocRef): + self.writes.append(("delete", ref.coll, ref.id, {}, False)) + class FakeFirestore: def __init__(self): @@ -91,6 +94,10 @@ class FakeFirestore: def _write(self, kind: str, coll: str, doc_id: str, data: dict, merge: bool): docs = self.data.setdefault(coll, {}) + if kind == "delete": + docs.pop(doc_id, None) + self.versions[(coll, doc_id)] = self.versions.get((coll, doc_id), 0) + 1 + return if kind == "update" and doc_id not in docs: raise KeyError(f"No document to update: {coll}/{doc_id}") if merge and doc_id in docs: diff --git a/drb-c2-core/tests/test_discord_remove_and_logging.py b/drb-c2-core/tests/test_discord_remove_and_logging.py new file mode 100644 index 0000000..b010c0c --- /dev/null +++ b/drb-c2-core/tests/test_discord_remove_and_logging.py @@ -0,0 +1,383 @@ +""" +Discord radio channels: server-side state logging, permanent removal of a +radio channel, and removal of a whole Discord server (server-26#213 / #221 +follow-up). + +Removal must go through desired state first — the node is told to drop the +channel (and logs the bot out), the token is only ``releasing`` until the node +acks — and only then is the doc deleted. Deleting first would leave the +token's lease naming a stream nobody can stop. +""" +import pytest +from fastapi import HTTPException +from fastapi.testclient import TestClient + +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_streams as streams +from app.internal.auth import require_admin_token, require_firebase_token +from app.main import app +from app.routers import discord_web as web +from tests.fake_firestore import FakeFirestore + +ORG, OTHER, SYS = "org-a", "org-b", "sys-1" +ADMIN = {"uid": "u-admin", "role": "admin", "email": "admin@example.test"} +OWNER = {"uid": "u-own", "org_id": ORG, "org_role": "owner"} +OPERATOR = {"uid": "u-op", "org_id": ORG, "org_role": "member", "role": "operator", "owned_node_ids": ["n1"]} +STRANGER = {"uid": "u-x", "org_id": OTHER, "org_role": "owner"} + +client = TestClient(app) + + +class _Log: + def __init__(self): + self.lines: list[tuple[str, str]] = [] + + def __getattr__(self, level): + return lambda msg, *a, **k: self.lines.append((level, str(msg))) + + +@pytest.fixture +def fs(monkeypatch): + fake = FakeFirestore.install(monkeypatch) + fake.published = [] + fake.on_publish = None + + def _publish(node_id, payload): + if fake.on_publish: + fake.on_publish(node_id, payload) + fake.published.append((node_id, payload)) + return True + + from app.internal.mqtt_handler import mqtt_handler + monkeypatch.setattr(mqtt_handler, "publish_discord_desired", _publish) + streams._publish_locks.clear() + + async def _no_discord(token_ids): + return 0 + monkeypatch.setattr(leases, "refresh_membership", _no_discord) + monkeypatch.setattr(settings, "discord_unlinked_guild_org_id", "") + + for nid, org in (("n1", ORG), ("n2", ORG), ("nx", OTHER)): + fake.put("nodes", nid, {"node_id": nid, "name": f"Node {nid}", "org_id": org, "status": "online", + "approval_status": "approved", "assigned_system_id": SYS}) + fake.put("systems", SYS, {"system_id": SYS, "name": "County P25", "org_id": ORG}) + fake.put("organizations", ORG, {"org_id": ORG, "name": "Alpha"}) + fake.put("discord_guilds", "100", {"guild_id": "100", "org_id": ORG, "guild_name": "Fire Co. 7", + "bot_present": True, + "voice_channels": [{"channel_id": "1001", "name": "Dispatch"}, + {"channel_id": "1002", "name": "Fireground"}]}) + for tid, org in (("t1", None), ("t2", None)): + fake.put("bot_tokens", tid, {"token_id": tid, "name": f"Radio {tid}", "token": f"secret-{tid}", + "org_id": org, "guild_ids": ["100", "200"], "valid": True, + "application_id": f"app-{tid}", "lease": None}) + yield fake + app.dependency_overrides.clear() + + +async def _start(channel="1001", node_id="n1", guild="100", org=ORG): + return await streams.start_stream(org_id=org, guild_id=guild, channel_id=channel, system_id=SYS, + node_id=node_id) + + +def _lease(fs, tid): + return (fs.get("bot_tokens", tid) or {}).get("lease") + + +def _audit(fs, action): + return [a for a in (fs.data.get("audit_log") or {}).values() if a["action"] == action] + + +# --------------------------------------------------------------------------- +# 1. Logging node-reported state transitions +# --------------------------------------------------------------------------- + +async def test_actual_logs_transitions_not_heartbeats_and_keeps_the_code(fs, monkeypatch): + log = _Log() + monkeypatch.setattr(streams, "logger", log) + doc = await _start() + sid, tok = doc["stream_id"], doc["token_id"] + + report = {"streams": {sid: {"state": "connected", "error": None, "code": None}}} + await streams.handle_actual("n1", report) + await streams.handle_actual("n1", report) # heartbeat: same state + infos = [m for lvl, m in log.lines if lvl == "info" and "pending -> connected" in m] + assert len(infos) == 1 + assert f"stream={sid}" in infos[0] and "node=n1" in infos[0] + assert "guild=100" in infos[0] and "channel=1001" in infos[0] and f"token={tok}" in infos[0] + + err = {"streams": {sid: {"state": "error", "code": "connect_failed", + "error": "Could not connect to the voice channel (TimeoutError: )."}}} + await streams.handle_actual("n1", err) + await streams.handle_actual("n1", err) + warns = [m for lvl, m in log.lines if lvl == "warning"] + assert len(warns) == 1 + assert "connected -> error [connect_failed]" in warns[0] and "TimeoutError" in warns[0] + actual = fs.get("discord_streams", sid)["actual"] + assert actual["code"] == "connect_failed" and actual["error"].startswith("Could not connect") + + +async def test_login_failure_reason_is_logged_against_the_token(fs, monkeypatch): + log = _Log() + monkeypatch.setattr(streams, "logger", log) + doc = await _start() + await streams.handle_actual("n1", {"streams": {doc["stream_id"]: { + "state": "error", "code": "login_failed", + "error": "The radio bot could not log in to Discord (LoginFailure: Improper token has been passed.)."}}}) + warns = [m for lvl, m in log.lines if lvl == "warning"] + assert len(warns) == 1 + assert warns[0].startswith(f"Radio bot {doc['token_id']} failed to log in on node n1:") + assert "Improper token" in warns[0] + + +async def test_actual_from_a_node_not_carrying_the_stream_is_neither_stored_nor_logged(fs, monkeypatch): + log = _Log() + monkeypatch.setattr(streams, "logger", log) + doc = await _start(node_id="n1") + await streams.handle_actual("n2", {"streams": {doc["stream_id"]: {"state": "error", "error": "spoof"}}}) + assert fs.get("discord_streams", doc["stream_id"])["actual"]["state"] == "pending" + assert not [m for _, m in log.lines if "spoof" in m] + + +# --------------------------------------------------------------------------- +# 2. Remove a radio channel +# --------------------------------------------------------------------------- + +async def test_delete_stops_through_desired_state_before_deleting(fs): + doc = await _start() + sid, tok = doc["stream_id"], doc["token_id"] + seen = [] + + def _on_publish(node_id, payload): + # The node must be told to drop the channel while the doc still exists. + cur = fs.get("discord_streams", sid) + seen.append((node_id, [s["stream_id"] for s in payload["streams"]], cur and cur["desired"])) + fs.on_publish = _on_publish + + before = await streams.delete_stream(sid, "removed") + assert before["stream_id"] == sid + assert seen == [("n1", [], "off")] # stop published first, doc still there + assert fs.get("discord_streams", sid) is None # then deleted + + # Lease follows #216: releasing until the node acks a list without it. + lease = _lease(fs, tok) + assert lease["releasing"] is True and lease["streams"] == [] + await streams.handle_actual("n1", {"applied_seq": lease["release_seq"], "sessions": [tok], "streams": {}}) + assert _lease(fs, tok) is not None # still logged in on the node + await streams.handle_actual("n1", {"applied_seq": lease["release_seq"], "sessions": [], "streams": {}}) + assert _lease(fs, tok) is None # node logged out -> back in the pool + + +async def test_delete_of_an_already_stopped_stream_does_not_republish(fs): + doc = await _start() + await streams.stop_stream(doc["stream_id"]) + n = len(fs.published) + assert await streams.delete_stream(doc["stream_id"]) + assert len(fs.published) == n + assert fs.get("discord_streams", doc["stream_id"]) is None + assert await streams.delete_stream(doc["stream_id"]) is None + + +async def test_delete_refuses_when_the_stream_is_restarted_in_between(fs, monkeypatch): + doc = await _start() + real_stop = streams.stop_stream + + async def _stop_then_restart(stream_id, reason="stopped"): + await real_stop(stream_id, reason) + await _start() # /join lands between stop and delete + monkeypatch.setattr(streams, "stop_stream", _stop_then_restart) + + with pytest.raises(streams.StreamError) as e: + await streams.delete_stream(doc["stream_id"]) + assert e.value.code == "restarted" and e.value.status == 409 + cur = fs.get("discord_streams", doc["stream_id"]) + assert cur["desired"] == "on" and doc["stream_id"] in _lease(fs, cur["token_id"])["streams"] + + +def test_admin_route_deletes_and_audits(fs): + import asyncio + doc = asyncio.run(_start()) + app.dependency_overrides[require_admin_token] = lambda: ADMIN + assert client.delete("/discord/streams/not-a-stream").status_code == 422 + assert client.delete("/discord/streams/100:9999").status_code == 404 + r = client.delete(f"/discord/streams/{doc['stream_id']}") + assert r.status_code == 200 and r.json()["ok"] is True + assert fs.get("discord_streams", doc["stream_id"]) is None + (entry,) = _audit(fs, "discord.stream.delete") + assert entry["actor_uid"] == "u-admin" + assert entry["details"]["stream_id"] == doc["stream_id"] and entry["details"]["was_desired"] == "on" + assert entry["details"]["org_id"] == ORG and entry["details"]["node_id"] == "n1" + + +def test_admin_route_requires_platform_admin(fs): + def deny(): + raise HTTPException(403, "Admin access required") + app.dependency_overrides[require_admin_token] = deny + assert client.delete("/discord/streams/100:1001").status_code == 403 + assert client.delete("/discord/guilds/100").status_code == 403 + + +async def test_node_page_remove_is_owner_or_admin_and_org_scoped(fs): + a = await _start("1001", node_id="n1") + b = await _start("1002", node_id="n2") + + # An operator can stop but not permanently remove. + with pytest.raises(HTTPException) as e: + await web.remove_node_stream("n1", a["stream_id"], OPERATOR) + assert e.value.status_code == 403 + # Another org's owner cannot even see the node. + with pytest.raises(HTTPException) as e: + await web.remove_node_stream("n1", a["stream_id"], STRANGER) + assert e.value.status_code == 404 + # A stream on another node of the same org is not reachable through n1's URL. + with pytest.raises(HTTPException) as e: + await web.remove_node_stream("n1", b["stream_id"], OWNER) + assert e.value.status_code == 404 + + assert (await web.remove_node_stream("n1", a["stream_id"], OWNER))["ok"] is True + assert fs.get("discord_streams", a["stream_id"]) is None + assert fs.get("discord_streams", b["stream_id"])["desired"] == "on" # untouched + (entry,) = _audit(fs, "discord.stream.delete") + assert entry["actor_uid"] == "u-own" and entry["details"]["via"] == "node_page" + + assert (await web.remove_node_stream("n2", b["stream_id"], ADMIN))["ok"] is True + + +async def test_join_options_says_who_may_remove(fs): + assert (await web.join_options("n1", OWNER))["can_remove"] is True + assert (await web.join_options("n1", ADMIN))["can_remove"] is True + opts = await web.join_options("n1", OPERATOR) + assert opts["can_control"] is True and opts["can_remove"] is False + + +# --------------------------------------------------------------------------- +# 3. Remove a Discord server +# --------------------------------------------------------------------------- + +async def test_remove_guild_deletes_streams_and_unlinks_without_fallback(fs, monkeypatch): + monkeypatch.setattr(settings, "discord_unlinked_guild_org_id", ORG) + a = await _start("1001", node_id="n1") + b = await _start("1002", node_id="n1") + await streams.stop_stream(b["stream_id"]) # one already stopped + out = await streams.remove_guild("100", "u-admin") + + assert sorted(out["streams_removed"]) == sorted([a["stream_id"], b["stream_id"]]) + assert fs.data["discord_streams"] == {} + g = fs.get("discord_guilds", "100") + assert g["org_id"] is None and g["unlinked_at"] and g["removed_at"] and g["removed_by"] == "u-admin" + # No fallback even though DISCORD_UNLINKED_GUILD_ORG_ID is set. + assert await streams.resolve_guild_org("100") is None + # Leases follow #216: releasing, freed on the node's ack. + toks = {a["token_id"], b["token_id"]} + assert len(toks) == 2 # one bot per guild voice connection + leased = [_lease(fs, t) for t in toks] + assert all(lease["releasing"] for lease in leased) + seq = max(lease["release_seq"] for lease in leased) + await streams.handle_actual("n1", {"applied_seq": seq, "sessions": [], "streams": {}}) + assert [_lease(fs, t) for t in toks] == [None, None] + + +def test_remove_guild_interrupted_by_a_restart_is_still_audited(fs, monkeypatch): + import asyncio + a = asyncio.run(_start("1001")) + b = asyncio.run(_start("1002", node_id="n2")) + real_stop = streams.stop_stream + + async def _stop(stream_id, reason="stopped"): + await real_stop(stream_id, reason) + if stream_id == b["stream_id"]: + await _start("1002", node_id="n2") # restarted between stop and delete + monkeypatch.setattr(streams, "stop_stream", _stop) + app.dependency_overrides[require_admin_token] = lambda: ADMIN + + r = client.delete("/discord/guilds/100") + assert r.status_code == 409 and r.json()["code"] == "restarted" + (entry,) = _audit(fs, "discord.guild.remove") + assert entry["details"]["outcome"] == "incomplete" + assert entry["details"]["streams_removed"] == [a["stream_id"]] + assert fs.get("discord_guilds", "100")["org_id"] == ORG # left linked: a live channel remains + + +def test_remove_guild_route_hides_server_and_relink_restores_it(fs): + import asyncio + asyncio.run(_start("1001")) + app.dependency_overrides[require_admin_token] = lambda: ADMIN + assert client.delete("/discord/guilds/abc").status_code == 400 + assert client.delete("/discord/guilds/999").status_code == 404 + + r = client.delete("/discord/guilds/100") + assert r.status_code == 200 and r.json()["streams_removed"] == ["100:1001"] + assert r.json()["radio_bots"] == [] # not asked to leave + body = client.get("/discord/admin/overview").json() + assert body["guilds"] == [] and body["streams"] == [] + (entry,) = _audit(fs, "discord.guild.remove") + assert entry["details"]["previous_org_id"] == ORG and entry["details"]["streams_removed"] == ["100:1001"] + + # Admin link brings it back. + assert client.put("/discord/guilds/100/link", json={"org_id": ORG}).status_code == 200 + g = fs.get("discord_guilds", "100") + assert g["removed_at"] is None and g["unlinked_at"] is None + assert [x["guild_id"] for x in client.get("/discord/admin/overview").json()["guilds"]] == ["100"] + + +async def test_redeeming_a_link_code_restores_a_removed_server(fs): + await streams.remove_guild("100", "u-admin") + minted = await glink.mint_code(ORG, "u-own") + await glink.redeem_code(minted["code"], "100", "Fire Co. 7", "4242") + g = fs.get("discord_guilds", "100") + assert g["org_id"] == ORG and g["removed_at"] is None + + +def test_remove_guild_can_make_pool_and_own_radio_bots_leave(fs, monkeypatch): + import httpx + fs.put("bot_tokens", "byo-mine", {"token_id": "byo-mine", "name": "Ours", "token": "secret-mine", + "org_id": ORG, "guild_ids": ["100"], "valid": True}) + fs.put("bot_tokens", "byo-theirs", {"token_id": "byo-theirs", "name": "Theirs", "token": "secret-theirs", + "org_id": OTHER, "guild_ids": ["100"], "valid": True}) + fs.put("bot_tokens", "elsewhere", {"token_id": "elsewhere", "name": "Elsewhere", "token": "secret-else", + "org_id": None, "guild_ids": ["300"], "valid": True}) + calls = [] + + class _Resp: + def __init__(self, code): + self.status_code = code + + class _Client: + def __init__(self, *, headers, **kw): + self.auth = headers["Authorization"] + + async def __aenter__(self): + return self + + async def __aexit__(self, *exc): + return False + + async def delete(self, url): + calls.append((self.auth, url)) + return _Resp(500 if self.auth == "Bot secret-t2" else 204) + + monkeypatch.setattr(httpx, "AsyncClient", _Client) + app.dependency_overrides[require_admin_token] = lambda: ADMIN + r = client.delete("/discord/guilds/100?leave_radio_bots=true") + assert r.status_code == 200 + tried = sorted(a for a, _ in calls) + # pool bots in the server + this org's BYO; never another org's bot, never a bot not in it + assert tried == ["Bot secret-mine", "Bot secret-t1", "Bot secret-t2"] + assert all(u == "/users/@me/guilds/100" for _, u in calls) + rows = {x["name"]: x for x in r.json()["radio_bots"]} + assert rows["Radio t1"]["left"] is True and rows["Radio t2"]["left"] is False + assert "secret" not in r.text + assert fs.get("bot_tokens", "t1")["guild_ids"] == ["200"] + assert fs.get("bot_tokens", "t2")["guild_ids"] == ["100", "200"] # failed: membership unchanged + assert fs.get("bot_tokens", "byo-theirs")["guild_ids"] == ["100"] + (entry,) = _audit(fs, "discord.guild.remove") + assert sorted(entry["details"]["radio_bots_left"]) == ["byo-mine", "t1"] + assert entry["details"]["radio_bots_failed"] == ["t2"] + + +def test_node_page_routes_need_a_signed_in_user(fs): + def deny(): + raise HTTPException(401, "Missing authorization") + app.dependency_overrides[require_firebase_token] = deny + assert client.delete("/discord/nodes/n1/streams/100:1001").status_code == 401 diff --git a/drb-frontend/components/NodeDiscordRadio.tsx b/drb-frontend/components/NodeDiscordRadio.tsx index 7e65002..0409d93 100644 --- a/drb-frontend/components/NodeDiscordRadio.tsx +++ b/drb-frontend/components/NodeDiscordRadio.tsx @@ -354,8 +354,24 @@ export function NodeDiscordRadio({ nodeId }: { nodeId: string }) { } } + async function remove(s: DiscordStreamView) { + const name = s.channel_name ?? "this channel"; + if (!confirm(`Remove radio from ${name} permanently? It is stopped first (the node logs its radio bot out), then deleted from DRB.`)) return; + setBusy(`rm:${s.stream_id}`); + setActionError(null); + try { + await c2api.removeNodeDiscordStream(nodeId, s.stream_id); + await load(); + } catch (err) { + setActionError(parseC2Error(err).message); + } finally { + setBusy(null); + } + } + const streams = options?.streams ?? []; const canControl = options?.can_control ?? false; + const canRemove = options?.can_remove ?? false; return (
@@ -405,7 +421,7 @@ export function NodeDiscordRadio({ nodeId }: { nodeId: string }) { Server · channel Mode Desired → actual - {canControl && } + {(canControl || canRemove) && } @@ -419,19 +435,34 @@ export function NodeDiscordRadio({ nodeId }: { nodeId: string }) { {s.actual?.error && ( -
{s.actual.error}
+
+ {s.actual.error} + {s.actual.code && [{s.actual.code}]} +
)} - {canControl && ( - - + {(canControl || canRemove) && ( + + {canControl && ( + + )} + {canRemove && ( + + )} )} diff --git a/drb-frontend/components/discord/ServersPanel.tsx b/drb-frontend/components/discord/ServersPanel.tsx index b7932b6..351f8a6 100644 --- a/drb-frontend/components/discord/ServersPanel.tsx +++ b/drb-frontend/components/discord/ServersPanel.tsx @@ -10,6 +10,8 @@ import { DIM, Field, INPUT, Panel, Status, Table, TD } from "./ui"; // radio channel without being linked (it acts for DISCORD_UNLINKED_GUILD_ORG_ID). // Linking is PUT /discord/guilds/{id}/link (platform admin). The owner-side // "/drb link code:" flow is server-26#213 item 1 and is not built here. +// Remove is DELETE /discord/guilds/{id}: its radio channels are stopped then +// deleted, and the server is unlinked with no fallback org. function LinkForm({ overview, @@ -83,6 +85,63 @@ function LinkForm({ ); } +type RemoveResult = { guild: string; streams: number; bots: { name: string; left: boolean; error: string | null }[] }; + +function RemoveForm({ + guild, + onDone, + onCancel, +}: { + guild: DiscordGuild; + onDone: (r: RemoveResult) => void; + onCancel: () => void; +}) { + const [leaveBots, setLeaveBots] = useState(false); + const [saving, setSaving] = useState(false); + const [error, setError] = useState(null); + const n = guild.live_streams; + + async function submit() { + setSaving(true); + setError(null); + try { + const r = await c2api.removeDiscordGuild(guild.guild_id, leaveBots); + onDone({ guild: guildLabel(guild), streams: r.streams_removed.length, bots: r.radio_bots }); + } catch (err) { + setError(parseC2Error(err).message); + } finally { + setSaving(false); + } + } + + return ( +
+

+ Remove {guildLabel(guild)} from DRB?{" "} + {n > 0 + ? `Its ${n} live radio channel${n === 1 ? " is" : "s are"} stopped first, then every radio channel in it is deleted.` + : "Every radio channel in it is deleted."}{" "} + It is unlinked and will not fall back to the default organization; linking it again brings it back. +

+ +

+ The DRB bot itself stays in the Discord server until someone kicks it there. +

+ {error &&

{error}

} +
+ + +
+
+ ); +} + export function ServersPanel({ overview, onChanged, @@ -92,6 +151,8 @@ export function ServersPanel({ }) { // "new" = blank link form; a guild id = editing that row's link. const [editing, setEditing] = useState(null); + const [removing, setRemoving] = useState(null); + const [removed, setRemoved] = useState(null); const fallback = overview.limits.unlinked_guild_org_id; const fallbackName = overview.limits.unlinked_guild_org_name || fallback; const max = overview.limits.guild_max_streams; @@ -119,6 +180,19 @@ export function ServersPanel({ } > {editing === "new" && setEditing(null)} />} + {removed && ( +
+ Removed {removed.guild}: {removed.streams} radio channel{removed.streams === 1 ? "" : "s"} deleted. + {removed.bots.length > 0 && ( + <> + {" "}Radio bots:{" "} + {removed.bots.map((b) => `${b.name} ${b.left ? "left" : `did not leave (${b.error ?? "unknown error"})`}`).join("; ")}. + + )}{" "} + The DRB bot is still in that Discord server until someone kicks it. + +
+ )} {overview.guilds.length === 0 ? (

No Discord servers yet. Link one, or run a radio command in a server the DRB bot is in.

) : ( @@ -147,13 +221,50 @@ export function ServersPanel({ Unlinked )} - - + ); + if (removing === g.guild_id) { + return [ + row, + + + setRemoving(null)} + onDone={async (r) => { + setRemoving(null); + setRemoved(r); + await onChanged(); + }} + /> + + , + ]; + } if (editing !== g.guild_id) return [row]; return [ row, diff --git a/drb-frontend/components/discord/StreamsPanel.tsx b/drb-frontend/components/discord/StreamsPanel.tsx index 7acb8d9..5eced8a 100644 --- a/drb-frontend/components/discord/StreamsPanel.tsx +++ b/drb-frontend/components/discord/StreamsPanel.tsx @@ -18,6 +18,8 @@ import { DIM, Field, INPUT, InviteLinks, Panel, Status, Table, TD } from "./ui"; // Radio channels = discord_streams docs, desired vs actual. Start/Stop/Add go // through the same routes the Discord bot uses (POST /discord/streams, // POST /discord/guilds/{id}/stop), which accept a platform-admin token. +// Remove is DELETE /discord/streams/{id} (admin): c2-core stops the channel +// through desired state first, then deletes it. type Invite = { name: string; invite_url: string | null }[]; @@ -214,7 +216,12 @@ export function StreamsPanel({ {s.mode === "adhoc" ? "ad-hoc" : s.mode ?? "—"} {label} - {isOn && s.actual?.error &&
{s.actual.error}
} + {isOn && s.actual?.error && ( +
+ {s.actual.error} + {s.actual.code && [{s.actual.code}]} +
+ )} {!isOn && s.stop_reason &&
{s.stop_reason}
} {err && (
@@ -224,7 +231,7 @@ export function StreamsPanel({ )} {s.token_name ?? —} - + {isOn ? ( ); diff --git a/drb-frontend/lib/c2api.ts b/drb-frontend/lib/c2api.ts index d8a5a4e..bfb3067 100644 --- a/drb-frontend/lib/c2api.ts +++ b/drb-frontend/lib/c2api.ts @@ -84,6 +84,21 @@ export const c2api = { channel_name?: string; preferred_token_id?: string; }) => request("/discord/streams", { method: "POST", body: JSON.stringify(body) }), + /** Permanently remove a radio channel (stopped through desired state first). Admin only. */ + deleteDiscordStream: (streamId: string) => + request<{ ok: boolean; stream_id: string }>(`/discord/streams/${encodeURIComponent(streamId)}`, { + method: "DELETE", + }), + /** Remove a Discord server from DRB: its radio channels are deleted and it is unlinked with no fallback org. */ + removeDiscordGuild: (guildId: string, leaveRadioBots: boolean) => + request<{ + ok: boolean; + guild_id: string; + streams_removed: string[]; + radio_bots: { name: string; left: boolean; error: string | null }[]; + }>(`/discord/guilds/${encodeURIComponent(guildId)}?leave_radio_bots=${leaveRadioBots ? "true" : "false"}`, { + method: "DELETE", + }), stopDiscordStream: (guildId: string, channelId: string) => request<{ stopped: string[] }>(`/discord/guilds/${encodeURIComponent(guildId)}/stop`, { method: "POST", @@ -116,6 +131,12 @@ export const c2api = { method: "POST", body: JSON.stringify(streamId ? { stream_id: streamId } : {}), }), + /** Permanently remove one radio channel from this node (org owner or admin). */ + removeNodeDiscordStream: (nodeId: string, streamId: string) => + request<{ ok: boolean; stream_id: string }>( + `/discord/nodes/${nodeId}/streams/${encodeURIComponent(streamId)}`, + { method: "DELETE" }, + ), deleteNode: (id: string) => request(`/nodes/${id}`, { method: "DELETE" }), diff --git a/drb-frontend/lib/discord.ts b/drb-frontend/lib/discord.ts index e86639d..59bf385 100644 --- a/drb-frontend/lib/discord.ts +++ b/drb-frontend/lib/discord.ts @@ -41,6 +41,8 @@ export interface RadioBotAddResult { export interface DiscordStreamActual { state?: string; error?: string | null; + /** The node's reason code on error: login_failed, connect_failed, not_in_guild, bad_channel, no_audio. */ + code?: string | null; since?: string | null; at?: string | null; } diff --git a/drb-frontend/lib/discordPicker.ts b/drb-frontend/lib/discordPicker.ts index 025cbc6..7079f98 100644 --- a/drb-frontend/lib/discordPicker.ts +++ b/drb-frontend/lib/discordPicker.ts @@ -22,7 +22,7 @@ export interface DiscordStreamView { mode: "persistent" | "adhoc" | null; node_id: string | null; node_name: string | null; - actual: { state: StreamState; error: string | null; since?: string; at?: string } | null; + actual: { state: StreamState; error: string | null; code?: string | null; since?: string; at?: string } | null; updated_at: string | null; } @@ -56,6 +56,8 @@ export interface DiscordJoinOptions { system_id: string | null; system_name: string | null; can_control: boolean; + /** Permanent removal: platform admin or the org owner only (operators can stop, not remove). */ + can_remove?: boolean; limits: { node_max: number; node_used: number; guild_max: number }; guilds: DiscordGuildOption[]; streams: DiscordStreamView[];