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 TYPE_CHECKING, Any, Final

from signal_bot.groups import ConfigError, canonical_group_id

if TYPE_CHECKING:
    from collections.abc import Collection

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 is_trusted(trusted: Collection[str], number: str, uuid: str) -> bool:
    """Whether a sender is on the TRUSTED_INVITERS list, by either identity.

    Strict: an empty list trusts nobody here. What an *unset* variable means
    is the caller's policy -- InviteTrust reads it as "every group", the
    unmute command as "anybody" -- and each says so where it decides.
    """
    return number in trusted or uuid in trusted


def identities(entry: dict[str, Any], key: str) -> set[str]:
    """Return one listing field as a set of strings, tolerating anything.

    Public because the roster cache reads the same two fields off the same
    listing rows, and two spellings of "who is in this group" is how the
    nightly report would come to disagree with the trust gate.
    """
    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 not is_trusted(self._trusted, source_number, source_uuid):
            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