Logan CusanoandClaude Opus 5.5
9e850fedf0
A guild without its @everyone role (or that role's permissions) was read as base 0, a definite deny that would block a start. It now raises ValueError, which S3 treats as unknown and lets the start go ahead. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
291 lines
11 KiB
Python
291 lines
11 KiB
Python
"""
|
|
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),
|
|
}
|