Merge pull request 'Discord: channel permission math for radio bots (#225 S1)' (#230) from feat/225-s1-discord-perms into main
Build & Deploy / Build & push images (push) Failing after 13s
Build & Deploy / Deploy to VM (push) Skipped
Build & Deploy / Deploy Firestore rules & indexes (push) Skipped
Build & Deploy / Report a failed deploy (push) Successful in 1s

This commit was merged in pull request #230.
This commit is contained in:
2026-10-05 02:03:28 -04:00
2 changed files with 578 additions and 0 deletions
+290
View File
@@ -0,0 +1,290 @@
"""
Discord channel permission math for radio bots (server-26#225, slice S1).
Pure functions, no I/O. Input is the raw JSON Discord's REST API returns:
guild GET /guilds/{g} {id, owner_id, roles: [{id, name, permissions}]}
member GET /guilds/{g}/members/{bot_id} {user: {id}, roles: [role_id], communication_disabled_until}
channel GET /guilds/{g}/channels (one) {id, name, type, user_limit, permission_overwrites:
[{id, type, allow, deny}]}
``effective_permissions`` follows Discord's documented order
(https://discord.com/developers/docs/topics/permissions#permission-overwrites):
1. The guild owner has every permission.
2. Base = @everyone's permissions OR every role the member has.
ADMINISTRATOR there means every permission.
3. The channel's @everyone overwrite (deny, then allow).
4. The member's role overwrites: OR every deny, OR every allow, then apply
deny before allow, so a role that allows beats a role that denies.
5. The member's own overwrite, last.
6. No VIEW_CHANNEL in the channel means no permission there at all.
No CONNECT in a voice channel also takes the voice permissions (Speak,
etc.) — discord.py applies the same rule, so the edge's ``_join_refusal``
(node-26#38) reports the same labels.
7. A timed-out member keeps only View Channel and Read Message History.
It also returns a ``trace``: for each radio permission the bot ends up
without, a sentence naming what removed it ("denied by #dispatch's override
for @Muted"), so the customer is told exactly which Discord toggle to fix.
Permission integers are strings in the API and can exceed 2**53; they are
parsed with ``int`` and never pass through a float.
"""
from __future__ import annotations
from datetime import datetime, timezone
from typing import Any, Mapping, Optional
VIEW_CHANNEL = 1 << 10
READ_MESSAGE_HISTORY = 1 << 16
CONNECT = 1 << 20
SPEAK = 1 << 21
MOVE_MEMBERS = 1 << 24
ADMINISTRATOR = 1 << 3
# Every bit Discord defines today fits well under 64; owner/admin get them all.
ALL_PERMISSIONS = (1 << 64) - 1
# What a timed-out member keeps (Discord: "timed out members ... only View
# Channel and Read Message History").
TIMEOUT_MASK = VIEW_CHANNEL | READ_MESSAGE_HISTORY
# Permissions that only mean something to someone connected to voice; lost
# with Connect (discord.py ``Permissions.voice()``: connect, speak, mute,
# deafen, move, VAD, priority speaker, stream, soundboard...).
VOICE_IMPLIED_BY_CONNECT = (
(1 << 8) # PRIORITY_SPEAKER
| (1 << 9) # STREAM
| SPEAK
| (1 << 22) # MUTE_MEMBERS
| (1 << 23) # DEAFEN_MEMBERS
| MOVE_MEMBERS
| (1 << 25) # USE_VAD
| (1 << 32) # REQUEST_TO_SPEAK
| (1 << 39) # USE_EMBEDDED_ACTIVITIES
| (1 << 42) # USE_SOUNDBOARD
| (1 << 45) # USE_EXTERNAL_SOUNDS
)
# Same bits as discord_leases.RADIO_BOT_PERMISSIONS, in the order the edge
# lists them. Labels match node-26#38's ``_join_refusal`` strings exactly.
REQUIRED_VOICE = ((VIEW_CHANNEL, "View Channel"), (CONNECT, "Connect"), (SPEAK, "Speak"))
TRACKED = REQUIRED_VOICE + ((MOVE_MEMBERS, "Move Members"),)
GUILD_VOICE = 2 # Discord channel type for a normal voice channel
def parse_permissions(value: Any) -> int:
"""A Discord permission integer from its API string (or int / None).
Raises ValueError on anything else, so a malformed payload is "unknown"
to the caller rather than silently read as "no permissions".
"""
if value is None or value == "":
return 0
if isinstance(value, bool):
raise ValueError(f"not a permission integer: {value!r}")
if isinstance(value, int):
n = value
elif isinstance(value, str) and value.strip().isdigit():
n = int(value.strip())
else:
raise ValueError(f"not a permission integer: {value!r}")
if n < 0:
raise ValueError(f"not a permission integer: {value!r}")
return n
def _sid(value: Any) -> str:
return "" if value is None else str(value)
def _role_label(role: Optional[Mapping[str, Any]], role_id: str) -> str:
name = (role or {}).get("name") or f"role {role_id}"
return name if name.startswith("@") else f"@{name}"
def _channel_label(channel: Mapping[str, Any]) -> str:
name = channel.get("name")
return f"#{name}" if name else "the channel"
def _overwrite_kind(ow: Mapping[str, Any]) -> Optional[str]:
t = ow.get("type")
if t in (0, "0", "role"):
return "role"
if t in (1, "1", "member"):
return "member"
return None
def _timed_out(member: Mapping[str, Any], now: datetime) -> bool:
raw = member.get("communication_disabled_until")
if not raw:
return False
try:
until = datetime.fromisoformat(str(raw).replace("Z", "+00:00"))
except ValueError:
return False
if until.tzinfo is None:
until = until.replace(tzinfo=timezone.utc)
return until > now
class _Tracer:
"""Remembers, per tracked bit, the last step that removed it."""
def __init__(self) -> None:
self.reasons: dict[int, str] = {}
def step(self, before: int, after: int, reason: str) -> int:
for bit, _ in TRACKED:
if before & bit and not after & bit:
self.reasons[bit] = reason
elif after & bit:
self.reasons.pop(bit, None)
return after
def result(self, bits: int) -> dict[str, str]:
return {label: self.reasons.get(bit, "not granted")
for bit, label in TRACKED if not bits & bit}
def effective_permissions(
guild: Mapping[str, Any],
member: Mapping[str, Any],
channel: Mapping[str, Any],
*,
now: Optional[datetime] = None,
) -> tuple[int, dict[str, str]]:
"""The member's permission bits in ``channel``, and why tracked bits are missing.
``trace`` maps a label ("View Channel", "Connect", "Speak",
"Move Members") to a sentence naming what removed it; only labels the
member ends up WITHOUT appear. Raises ValueError on a malformed
permission value.
"""
now = now or datetime.now(timezone.utc)
guild_id = _sid(guild.get("id"))
user = member.get("user") or {}
member_id = _sid(user.get("id") or member.get("id"))
member_role_ids = {_sid(r) for r in (member.get("roles") or [])}
roles = {_sid(r.get("id")): r for r in (guild.get("roles") or [])}
chan = _channel_label(channel)
tracer = _Tracer()
# 1. Owner.
if member_id and member_id == _sid(guild.get("owner_id")):
return ALL_PERMISSIONS, {}
# 2. Base from @everyone + the member's roles. Every guild has an
# @everyone role (id == guild id) with a permissions value; without one
# the payload is partial, and reading it as "no permissions" would turn
# it into a definite deny instead of "unknown" (the plan is fail-open).
everyone_role = roles.get(guild_id)
if not guild_id or everyone_role is None or everyone_role.get("permissions") in (None, ""):
raise ValueError(f"guild payload has no @everyone role permissions (guild {guild_id or '?'})")
base = parse_permissions(everyone_role.get("permissions"))
for rid in member_role_ids:
role = roles.get(rid)
if role is not None and rid != guild_id:
base |= parse_permissions(role.get("permissions"))
if base & ADMINISTRATOR:
return ALL_PERMISSIONS, {}
tracer.step(ALL_PERMISSIONS, base,
"not granted by @everyone or any of the bot's roles (Server Settings → Roles)")
perms = base
everyone_ow = None
role_ows: list[Mapping[str, Any]] = []
member_ow = None
for ow in channel.get("permission_overwrites") or []:
oid = _sid(ow.get("id"))
kind = _overwrite_kind(ow)
if oid == guild_id and kind != "member":
everyone_ow = ow
elif kind == "member" or (kind is None and oid == member_id):
if oid == member_id:
member_ow = ow
elif oid in member_role_ids:
role_ows.append(ow)
# 3. @everyone overwrite.
if everyone_ow is not None:
deny = parse_permissions(everyone_ow.get("deny"))
allow = parse_permissions(everyone_ow.get("allow"))
perms = tracer.step(perms, (perms & ~deny) | allow,
f"denied by {chan}'s override for @everyone")
# 4. Role overwrites: all denies, then all allows.
if role_ows:
deny = allow = 0
deniers: dict[int, list[str]] = {}
for ow in sorted(role_ows, key=lambda o: _sid(o.get("id"))):
d = parse_permissions(ow.get("deny"))
deny |= d
allow |= parse_permissions(ow.get("allow"))
oid = _sid(ow.get("id"))
for bit, _ in TRACKED:
if d & bit:
deniers.setdefault(bit, []).append(_role_label(roles.get(oid), oid))
after = (perms & ~deny) | allow
for bit, _ in TRACKED:
if perms & bit and not after & bit:
tracer.reasons[bit] = (
f"denied by {chan}'s override for {', '.join(deniers.get(bit, []))}")
elif after & bit:
tracer.reasons.pop(bit, None)
perms = after
# 5. The member's own overwrite.
if member_ow is not None:
deny = parse_permissions(member_ow.get("deny"))
allow = parse_permissions(member_ow.get("allow"))
perms = tracer.step(perms, (perms & ~deny) | allow,
f"denied by {chan}'s override for the bot itself")
# 6. Implicit denials.
if not perms & VIEW_CHANNEL:
perms = tracer.step(perms, 0, f"the bot can't see {chan} without View Channel")
elif channel.get("type") == GUILD_VOICE and not perms & CONNECT:
perms = tracer.step(perms, perms & ~VOICE_IMPLIED_BY_CONNECT,
f"Discord removes it in {chan} without Connect")
# 7. Timeout.
if _timed_out(member, now):
perms = tracer.step(perms, perms & TIMEOUT_MASK, "the bot is timed out in this server")
return perms, tracer.result(perms)
def check_voice(
bits: int,
channel: Mapping[str, Any],
trace: Optional[Mapping[str, str]] = None,
) -> dict[str, Any]:
"""Can a bot with ``bits`` in ``channel`` stream into it?
Only a normal voice channel (type 2) counts; anything else is
``ok=False, is_voice=False`` with nothing ``missing``. ``user_limit`` is
advisory: REST can't see who is connected, so the edge's ``channel_full``
stays the enforcement. ``limit_bypass`` is Move Members, which Discord
lets past a full channel. ``reasons`` carries the trace sentence for each
missing label when ``trace`` is given.
"""
is_voice = channel.get("type") == GUILD_VOICE
try:
user_limit = int(channel.get("user_limit") or 0)
except (TypeError, ValueError):
user_limit = 0
missing = [label for bit, label in REQUIRED_VOICE if not bits & bit] if is_voice else []
reasons = {label: trace[label] for label in missing if trace and label in trace}
return {
"ok": is_voice and not missing,
"is_voice": is_voice,
"missing": missing,
"reasons": reasons,
"user_limit": user_limit,
"limit_bypass": bool(bits & MOVE_MEMBERS),
}
+288
View File
@@ -0,0 +1,288 @@
"""
Discord channel permission math (server-26#225, slice S1).
Payloads are shaped like Discord's REST JSON: permission integers are
strings, overwrite ``type`` is 0 (role) / 1 (member).
"""
from datetime import datetime, timedelta, timezone
import pytest
from app.internal import discord_leases as leases
from app.internal import discord_perms as perms
from app.internal.discord_perms import (
ADMINISTRATOR, ALL_PERMISSIONS, CONNECT, MOVE_MEMBERS, READ_MESSAGE_HISTORY,
SPEAK, VIEW_CHANNEL, check_voice, effective_permissions, parse_permissions,
)
GUILD = "100"
BOT = "900"
OWNER = "1"
MUTED = "201"
BOTS = "202"
DJ = "203"
NOW = datetime(2026, 10, 5, 12, 0, tzinfo=timezone.utc)
VOICE = VIEW_CHANNEL | CONNECT | SPEAK
def guild(everyone=VOICE, extra_roles=()):
roles = [{"id": GUILD, "name": "@everyone", "permissions": str(everyone)}]
roles += [{"id": rid, "name": name, "permissions": str(p)} for rid, name, p in extra_roles]
return {"id": GUILD, "owner_id": OWNER, "roles": roles}
def member(roles=(), uid=BOT, timeout=None):
return {"user": {"id": uid}, "roles": list(roles), "communication_disabled_until": timeout}
def ow(oid, kind, allow=0, deny=0):
return {"id": oid, "type": kind, "allow": str(allow), "deny": str(deny)}
def channel(overwrites=(), type_=2, user_limit=0, name="dispatch"):
return {"id": "500", "name": name, "type": type_, "user_limit": user_limit,
"permission_overwrites": list(overwrites)}
def eff(g, m, c):
return effective_permissions(g, m, c, now=NOW)
def test_required_bits_match_radio_bot_invite():
assert VIEW_CHANNEL | CONNECT | SPEAK == leases.RADIO_BOT_PERMISSIONS
def test_plain_everyone_grant_is_ok():
bits, trace = eff(guild(), member(), channel())
assert bits == VOICE and trace["Move Members"]
r = check_voice(bits, channel(), trace)
assert r == {"ok": True, "is_voice": True, "missing": [], "reasons": {},
"user_limit": 0, "limit_bypass": False}
def test_owner_gets_everything_despite_denies():
c = channel([ow(GUILD, 0, deny=VOICE), ow(OWNER, 1, deny=VOICE)])
bits, trace = eff(guild(everyone=0), member(uid=OWNER), c)
assert bits == ALL_PERMISSIONS and trace == {}
def test_administrator_role_bypasses_overwrites():
g = guild(everyone=0, extra_roles=[(BOTS, "Bots", ADMINISTRATOR)])
c = channel([ow(GUILD, 0, deny=VOICE), ow(BOT, 1, deny=VOICE)])
bits, _ = eff(g, member([BOTS]), c)
assert bits == ALL_PERMISSIONS
assert check_voice(bits, c)["ok"] and check_voice(bits, c)["limit_bypass"]
def test_admin_on_everyone_counts_too():
bits, _ = eff(guild(everyone=ADMINISTRATOR), member(), channel())
assert bits == ALL_PERMISSIONS
def test_base_is_everyone_or_member_roles_only():
# A role the bot doesn't hold grants nothing.
g = guild(everyone=VIEW_CHANNEL, extra_roles=[(DJ, "DJ", CONNECT | SPEAK), (BOTS, "Bots", 0)])
bits, trace = eff(g, member([BOTS]), channel())
assert bits == VIEW_CHANNEL
assert check_voice(bits, channel(), trace)["missing"] == ["Connect", "Speak"]
assert "not granted by @everyone or any of the bot's roles" in trace["Connect"]
bits, _ = eff(g, member([DJ]), channel())
assert bits == VOICE
def test_everyone_overwrite_deny_then_allow():
c = channel([ow(GUILD, 0, deny=SPEAK)])
bits, trace = eff(guild(), member(), c)
assert check_voice(bits, c, trace)["missing"] == ["Speak"]
assert trace["Speak"] == "denied by #dispatch's override for @everyone"
# An allow in the same overwrite restores a bit the base lacked.
c2 = channel([ow(GUILD, 0, allow=SPEAK)])
bits, _ = eff(guild(everyone=VIEW_CHANNEL | CONNECT), member(), c2)
assert bits == VOICE
def test_role_overwrite_denial_is_named():
g = guild(extra_roles=[(MUTED, "Muted", 0)])
c = channel([ow(MUTED, 0, deny=SPEAK)])
bits, trace = eff(g, member([MUTED]), c)
r = check_voice(bits, c, trace)
assert not r["ok"] and r["missing"] == ["Speak"]
assert r["reasons"]["Speak"] == "denied by #dispatch's override for @Muted"
def test_role_overwrite_for_role_bot_lacks_is_ignored():
g = guild(extra_roles=[(MUTED, "Muted", 0)])
c = channel([ow(MUTED, 0, deny=SPEAK)])
bits, _ = eff(g, member([]), c)
assert bits == VOICE
def test_role_allow_beats_role_deny_across_roles():
g = guild(extra_roles=[(MUTED, "Muted", 0), (DJ, "DJ", 0)])
c = channel([ow(MUTED, 0, deny=SPEAK), ow(DJ, 0, allow=SPEAK)])
bits, trace = eff(g, member([MUTED, DJ]), c)
assert bits & SPEAK and "Speak" not in trace
def test_role_overwrites_override_everyone_overwrite():
g = guild(extra_roles=[(DJ, "DJ", 0)])
c = channel([ow(GUILD, 0, deny=CONNECT | SPEAK), ow(DJ, 0, allow=CONNECT | SPEAK)])
bits, trace = eff(g, member([DJ]), c)
assert bits == VOICE and trace.get("Connect") is None
def test_several_denying_roles_are_all_named():
g = guild(extra_roles=[(MUTED, "Muted", 0), (BOTS, "Bots", 0)])
c = channel([ow(MUTED, 0, deny=CONNECT), ow(BOTS, 0, deny=CONNECT)])
_, trace = eff(g, member([MUTED, BOTS]), c)
assert trace["Connect"] == "denied by #dispatch's override for @Muted, @Bots"
def test_member_overwrite_applies_last():
g = guild(extra_roles=[(DJ, "DJ", 0)])
c = channel([ow(DJ, 0, allow=SPEAK), ow(BOT, 1, deny=SPEAK)])
bits, trace = eff(g, member([DJ]), c)
assert not bits & SPEAK
assert trace["Speak"] == "denied by #dispatch's override for the bot itself"
# ...and its allow beats a role deny.
c2 = channel([ow(GUILD, 0, deny=SPEAK), ow(BOT, 1, allow=SPEAK)])
bits, _ = eff(guild(), member(), c2)
assert bits == VOICE
def test_member_overwrite_for_another_user_is_ignored():
c = channel([ow("777", 1, deny=VOICE)])
bits, _ = eff(guild(), member(), c)
assert bits == VOICE
def test_no_view_channel_means_nothing():
c = channel([ow(GUILD, 0, deny=VIEW_CHANNEL)])
bits, trace = eff(guild(everyone=VOICE | MOVE_MEMBERS), member(), c)
assert bits == 0
r = check_voice(bits, c, trace)
assert r["missing"] == ["View Channel", "Connect", "Speak"] and not r["limit_bypass"]
assert trace["View Channel"] == "denied by #dispatch's override for @everyone"
assert trace["Connect"] == "the bot can't see #dispatch without View Channel"
def test_no_connect_takes_speak_and_move_in_voice():
c = channel([ow(GUILD, 0, deny=CONNECT)])
bits, trace = eff(guild(everyone=VOICE | MOVE_MEMBERS), member(), c)
assert bits == VIEW_CHANNEL
assert check_voice(bits, c, trace)["missing"] == ["Connect", "Speak"]
assert trace["Connect"] == "denied by #dispatch's override for @everyone"
assert trace["Speak"] == "Discord removes it in #dispatch without Connect"
def test_earlier_cause_survives_later_implicit_denial():
# Speak denied by a role AND later wiped by missing View: the trace keeps
# the last step that removed it, which is the View wipe only when Speak
# was still present at that point.
g = guild(extra_roles=[(MUTED, "Muted", 0)])
c = channel([ow(MUTED, 0, deny=SPEAK), ow(BOT, 1, deny=VIEW_CHANNEL)])
_, trace = eff(g, member([MUTED]), c)
assert trace["Speak"] == "denied by #dispatch's override for @Muted"
assert trace["Connect"] == "the bot can't see #dispatch without View Channel"
def test_timed_out_keeps_only_view_and_history():
future = (NOW + timedelta(hours=1)).isoformat()
g = guild(everyone=VOICE | READ_MESSAGE_HISTORY | MOVE_MEMBERS)
bits, trace = eff(g, member(timeout=future), channel())
assert bits == VIEW_CHANNEL | READ_MESSAGE_HISTORY
assert trace["Connect"] == "the bot is timed out in this server"
past = (NOW - timedelta(hours=1)).isoformat().replace("+00:00", "Z")
bits, _ = eff(g, member(timeout=past), channel())
assert bits & VOICE == VOICE
def test_unparseable_timeout_is_ignored():
bits, _ = eff(guild(), member(timeout="not-a-date"), channel())
assert bits == VOICE
def test_permission_strings_above_2_53():
big = (1 << 60) | (1 << 53) | 1
assert parse_permissions(str(big)) == big
assert parse_permissions(str((1 << 53) + 1)) == (1 << 53) + 1
# Float rounding would turn this into 2**53 and drop the low bit.
assert parse_permissions("9007199254740993") == 9007199254740993
g = guild(everyone=VOICE | (1 << 60))
c = channel([ow(GUILD, 0, allow=1 << 55, deny=1 << 60)])
bits, _ = eff(g, member(), c)
assert bits == VOICE | (1 << 55)
@pytest.mark.parametrize("value", ["abc", "-5", "1.5", True, 3.0, -1, [], {}])
def test_malformed_permission_values_raise(value):
with pytest.raises(ValueError):
parse_permissions(value)
def test_missing_permission_values_are_zero():
assert parse_permissions(None) == 0
assert parse_permissions("") == 0
assert parse_permissions(7) == 7
def test_malformed_payload_raises_not_silently_denies():
g = guild()
g["roles"][0]["permissions"] = "garbage"
with pytest.raises(ValueError):
eff(g, member(), channel())
def test_legacy_string_overwrite_types():
g = guild(extra_roles=[(MUTED, "Muted", 0)])
c = channel([ow(MUTED, "role", deny=SPEAK), ow(BOT, "member", deny=CONNECT)])
_, trace = eff(g, member([MUTED]), c)
assert set(trace) >= {"Connect", "Speak"}
def test_unknown_member_role_ids_are_ignored():
bits, _ = eff(guild(), member(["999999"]), channel())
assert bits == VOICE
@pytest.mark.parametrize("g", [
{"id": GUILD, "owner_id": OWNER, "roles": []},
{"id": GUILD, "owner_id": OWNER},
{"id": GUILD, "owner_id": OWNER, "roles": [{"id": GUILD, "name": "@everyone"}]},
{"owner_id": OWNER, "roles": [{"id": GUILD, "name": "@everyone", "permissions": "0"}]},
])
def test_partial_guild_payload_raises_not_silently_denies(g):
# A guild without its @everyone role (or its permissions) is a partial
# payload: "unknown" (ValueError, start goes ahead), never a definite deny.
with pytest.raises(ValueError):
eff(g, member(), channel())
def test_everyone_role_granting_nothing_is_a_real_deny():
bits, trace = eff(guild(everyone=0), member(), channel())
assert bits == 0 and "not granted" in trace["View Channel"]
def test_check_voice_only_counts_type_2():
for t in (0, 13, 4):
r = check_voice(ALL_PERMISSIONS, channel(type_=t))
assert r["ok"] is False and r["is_voice"] is False and r["missing"] == []
def test_check_voice_user_limit_and_bypass():
c = channel(user_limit=5)
assert check_voice(VOICE, c)["user_limit"] == 5
assert check_voice(VOICE, c)["limit_bypass"] is False
assert check_voice(VOICE | MOVE_MEMBERS, c)["limit_bypass"] is True
assert check_voice(VOICE, channel(user_limit=None))["user_limit"] == 0
def test_labels_match_edge_join_refusal_strings():
# node-26#38 discord_radio._join_refusal emits exactly these labels.
assert [label for _, label in perms.REQUIRED_VOICE] == ["View Channel", "Connect", "Speak"]
def test_now_defaults_to_current_time():
future = (datetime.now(timezone.utc) + timedelta(days=1)).isoformat()
bits, _ = effective_permissions(guild(), member(timeout=future), channel())
assert bits == VIEW_CHANNEL