mirror of
https://github.com/kuhyx/signal-bot.git
synced 2026-11-11 13:57:32 +01:00
Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
159 lines
5.9 KiB
Python
159 lines
5.9 KiB
Python
# Copyright (c) 2026 Krzysztof Rudnicki. MIT licence.
|
|
"""Group ids: the three spellings, and the one that wins.
|
|
|
|
Split out of config.py because it is the whole reason the bot can watch more
|
|
than one group without answering in the wrong one. The REST API reports a
|
|
group as ``group.<base64>``, an envelope carries what that decodes to, and
|
|
only the first form can be sent to -- so every id is normalised here, once,
|
|
before anything downstream compares or addresses one.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import base64
|
|
import binascii
|
|
from typing import TYPE_CHECKING, Final
|
|
|
|
if TYPE_CHECKING:
|
|
from collections.abc import Iterable
|
|
|
|
GROUP_ID_PREFIX: Final = "group."
|
|
|
|
|
|
class ConfigError(RuntimeError):
|
|
"""Raised when the environment does not describe a runnable bot."""
|
|
|
|
|
|
def group_id_aliases(group_id: str) -> frozenset[str]:
|
|
"""Every spelling of one group id that can arrive in an envelope.
|
|
|
|
The REST API reports group ids two different ways and never says so:
|
|
``/v1/groups`` returns ``group.<base64>`` (what the setup script writes
|
|
into .env), while an incoming envelope's ``groupInfo.groupId`` carries
|
|
the *internal* id -- which is exactly what that base64 decodes to. The
|
|
two never compare equal, so a bot configured from the API's own output
|
|
silently ignores every message in its own group. Verified live on
|
|
2026-08-29: ``group.WTI5dVptbG...`` decodes to ``Y29uZmlnLWdy/...``, the id
|
|
the same endpoint reports as ``internal_id``.
|
|
|
|
Args:
|
|
group_id: The configured group id, in either spelling.
|
|
|
|
Returns:
|
|
The configured id plus its decoded internal form, when it has one.
|
|
"""
|
|
aliases = {group_id}
|
|
if group_id.startswith(GROUP_ID_PREFIX):
|
|
try:
|
|
decoded = base64.b64decode(group_id.removeprefix(GROUP_ID_PREFIX), validate=True)
|
|
except binascii.Error, ValueError:
|
|
return frozenset(aliases)
|
|
aliases.add(decoded.decode("utf-8", errors="replace"))
|
|
else:
|
|
aliases.add(canonical_group_id(group_id))
|
|
return frozenset(aliases)
|
|
|
|
|
|
def canonical_group_id(group_id: str) -> str:
|
|
"""Return the ``group.<base64>`` spelling of a group id.
|
|
|
|
One spelling has to win, because the same string is used for three
|
|
different jobs: matching an incoming envelope, addressing a reply, and
|
|
keying per-group state. The prefixed form wins because it is the only
|
|
one ``/v2/send`` and the group REST paths accept -- the internal form
|
|
contains ``/`` and ``+``. The internal id is itself base64 text, so the
|
|
conversion is just the inverse of what group_id_aliases decodes.
|
|
|
|
Args:
|
|
group_id: A group id in either spelling.
|
|
|
|
Returns:
|
|
The id in ``group.<base64>`` form, unchanged if it already is one.
|
|
"""
|
|
if group_id.startswith(GROUP_ID_PREFIX):
|
|
return group_id
|
|
return GROUP_ID_PREFIX + base64.b64encode(group_id.encode("utf-8")).decode("ascii")
|
|
|
|
|
|
def build_alias_map(group_ids: Iterable[str]) -> dict[str, str]:
|
|
"""Map every spelling of every configured group to its canonical id.
|
|
|
|
Args:
|
|
group_ids: Configured group ids, already canonical.
|
|
|
|
Returns:
|
|
A lookup from any spelling an envelope might carry to the one
|
|
canonical id that scopes state and addresses replies.
|
|
"""
|
|
mapping: dict[str, str] = {}
|
|
for group_id in group_ids:
|
|
for alias in group_id_aliases(group_id):
|
|
mapping[alias] = group_id
|
|
return mapping
|
|
|
|
|
|
def parse_group_ids(env: dict[str, str]) -> tuple[str, ...]:
|
|
"""Return the optional allowlist of groups, canonical and de-duplicated.
|
|
|
|
Empty is the normal case and means "every group this account is a full
|
|
member of". The bot is its own registered Signal account, not a linked
|
|
device on a personal one, so the groups it can see are exactly the
|
|
groups somebody put it in -- there is nothing to scope it away from,
|
|
and requiring a list here is what made adding the bot a two-step
|
|
procedure with a restart in the middle.
|
|
|
|
Set it to pin the bot to specific groups anyway. ``GROUP_IDS`` is the
|
|
list form; ``GROUP_ID`` remains accepted as a one-element spelling.
|
|
Setting both is an error rather than a precedence rule, because a
|
|
precedence rule means silently ignoring one of two values the operator
|
|
clearly meant.
|
|
|
|
Args:
|
|
env: The environment mapping to read.
|
|
|
|
Returns:
|
|
Canonical group ids in the order listed, or empty for "all of them".
|
|
|
|
Raises:
|
|
ConfigError: Both keys are set, a removed key is set, or the value
|
|
is nothing but separators.
|
|
"""
|
|
if env.get("GROUP_ID_SEND", "").strip():
|
|
msg = (
|
|
"GROUP_ID_SEND has been removed. Replies and the daily report now "
|
|
"go back to the group the message came from, which a fixed send "
|
|
"target cannot express once the bot watches more than one group. "
|
|
"Delete the line from .env."
|
|
)
|
|
raise ConfigError(msg)
|
|
|
|
listed = env.get("GROUP_IDS", "").strip()
|
|
single = env.get("GROUP_ID", "").strip()
|
|
if listed and single:
|
|
msg = "set GROUP_IDS or GROUP_ID, not both; GROUP_ID is the one-group spelling"
|
|
raise ConfigError(msg)
|
|
|
|
raw = listed or single
|
|
if not raw:
|
|
return ()
|
|
|
|
group_ids: list[str] = []
|
|
seen: dict[str, str] = {}
|
|
for part in raw.split(","):
|
|
candidate = part.strip()
|
|
if not candidate:
|
|
continue
|
|
canonical = canonical_group_id(candidate)
|
|
aliases = group_id_aliases(canonical)
|
|
clash = next((seen[alias] for alias in aliases if alias in seen), None)
|
|
if clash is not None:
|
|
msg = f"group {candidate!r} is listed twice, once as {clash!r}"
|
|
raise ConfigError(msg)
|
|
seen.update(dict.fromkeys(aliases, canonical))
|
|
group_ids.append(canonical)
|
|
|
|
if not group_ids:
|
|
msg = f"GROUP_IDS lists no group ids, only separators: {raw!r}"
|
|
raise ConfigError(msg)
|
|
return tuple(group_ids)
|