signal-bot/signal_bot/trust.py

# Copyright (c) 2026 Krzysztof Rudnicki. MIT licence.
"""Who is allowed to put this bot in a group.

The bot adopts every group its account is a member of, which is what makes
"add the bot" the whole setup. The cost is that *anybody* who knows the
number can pull it into a group and have it answer there. TRUSTED_INVITERS
puts a name on the door without taking the convenience away: a group still
needs no config edit and no restart, it just has to involve somebody the
operator listed.

Two independent pieces of evidence, because neither covers the other:

- the **group-update envelope**, which is the only frame that names who
  changed the membership. It is authoritative, and it arrives before the
  first message in a new group -- but only if the bot was running.
- the **listing row**, used when the bot was down when the group appeared.
  ``/v1/groups`` never says who invited anyone, so the proxy is "a trusted
  identity is a full member of this group". Deliberately a proxy: it stops
  a stranger's private group, and it does not stop a stranger's group that
  a trusted person genuinely joined.

Identities are matched as the listing spells them. A row lists a phone
number for contacts the account knows and a bare UUID for everybody else,
so TRUSTED_INVITERS accepts either form and a refusal logs the row's own
members, which is what tells the operator which string to add.
"""

from __future__ import annotations

import logging
import re
from typing import Any, Final

from signal_bot.groups import ConfigError, canonical_group_id

logger = logging.getLogger(__name__)

TRUSTED_INVITERS_KEY: Final = "TRUSTED_INVITERS"
# +E.164, or the UUID the listing uses for a member the account has never
# exchanged messages with.
_E164: Final = re.compile(r"^\+[1-9]\d{6,14}$")
_UUID: Final = re.compile(r"^[0-9a-fA-F]{8}(-[0-9a-fA-F]{4}){3}-[0-9a-fA-F]{12}$")


def parse_trusted_inviters(env: dict[str, str]) -> tuple[str, ...]:
    """Return the identities allowed to add this bot to a group.

    Args:
        env: The environment mapping to read.

    Returns:
        The listed identities, de-duplicated and in order. Empty means the
        gate is off and any group the account is in will be watched.

    Raises:
        ConfigError: An entry is neither +E.164 nor a UUID.
    """
    raw = env.get(TRUSTED_INVITERS_KEY, "").strip()
    if not raw:
        return ()
    trusted: list[str] = []
    for item in raw.split(","):
        identity = item.strip()
        if not identity:
            continue
        if not _E164.match(identity) and not _UUID.match(identity):
            msg = (
                f"{TRUSTED_INVITERS_KEY} entries must be +E.164 numbers or member "
                f"UUIDs, got {identity!r}"
            )
            raise ConfigError(msg)
        if identity not in trusted:
            trusted.append(identity)
    return tuple(trusted)


def _identities(entry: dict[str, Any], key: str) -> set[str]:
    """Return one listing field as a set of strings, tolerating anything."""
    values = entry.get(key)
    if not isinstance(values, list):
        return set()
    return {value for value in values if isinstance(value, str)}


class InviteTrust:
    """Decides whether a group in the listing is one the operator asked for."""

    def __init__(self, trusted: tuple[str, ...]) -> None:
        """Wire the gate to its allowlist.

        Args:
            trusted: Identities permitted to add the bot to a group. Empty
                turns the gate off, which is said out loud rather than
                assumed, because the difference is who can make this bot
                talk in their group.
        """
        self._trusted = frozenset(trusted)
        self._vouched: set[str] = set()
        self._refused: set[str] = set()
        if not self._trusted:
            logger.warning(
                "%s is unset: every group this account is added to will be watched, "
                "whoever added it",
                TRUSTED_INVITERS_KEY,
            )

    def note_update(self, group_id: str, source_number: str, source_uuid: str) -> None:
        """Record a membership change, vouching for the group when it is trusted.

        Args:
            group_id: The group id as the envelope spelled it.
            source_number: The +E.164 number that made the change, if known.
            source_uuid: The uuid that made the change.
        """
        if source_number not in self._trusted and source_uuid not in self._trusted:
            return
        canonical = canonical_group_id(group_id)
        if canonical not in self._vouched:
            self._vouched.add(canonical)
            self._refused.discard(canonical)
            logger.info("a trusted inviter added the bot to %s", canonical)

    def allows(self, group_id: str, entry: dict[str, Any]) -> bool:
        """True when this group may be watched.

        Args:
            group_id: The canonical group id.
            entry: The group's row from ``/v1/groups``.

        Returns:
            True when the gate is off, the group was vouched for by an
            update envelope, or a trusted identity is a full member.
        """
        if not self._trusted or group_id in self._vouched:
            return True
        members = _identities(entry, "members") - _identities(entry, "pending_invites")
        if self._trusted & members:
            return True
        if group_id not in self._refused:
            self._refused.add(group_id)
            logger.warning(
                "refusing group %r (%s): no trusted inviter. Its members are %s; "
                "add one of them to %s if this group is yours.",
                entry.get("name"),
                group_id,
                sorted(members),
                TRUSTED_INVITERS_KEY,
            )
        return False