signal-bot/tests/test_trust.py

# Copyright (c) 2026 Krzysztof Rudnicki. MIT licence.
"""Who may add this bot to a group, and what happens to everybody else."""

from __future__ import annotations

import logging
from dataclasses import replace
from typing import TYPE_CHECKING, Any

import pytest

from signal_bot.discovery import GroupDirectory
from signal_bot.groups import ConfigError
from signal_bot.models import parse_envelope
from signal_bot.trust import InviteTrust, parse_trusted_inviters
from tests.conftest import (
    SECOND_GROUP,
    STRANGER_NUMBER,
    TRUSTED_NUMBER,
    WATCHED_GROUP,
    WATCHED_INTERNAL,
    envelope,
    group_entry,
    listing_client,
    update_envelope,
)

if TYPE_CHECKING:
    from signal_bot.config import Config
    from signal_bot.store import Store

OPERATOR_UUID = "b1c2d3e4-0000-4a00-8a00-000000000001"


def row(group_id: str, *, members: list[str], pending: list[str] | None = None) -> dict[str, Any]:
    """A /v1/groups row with the membership fields the gate reads."""
    entry = group_entry(group_id)
    entry["members"] = members
    entry["pending_invites"] = pending if pending is not None else []
    return entry


def test_parsing_accepts_numbers_uuids_and_nothing() -> None:
    assert parse_trusted_inviters({}) == ()
    assert parse_trusted_inviters({"TRUSTED_INVITERS": "   "}) == ()
    assert parse_trusted_inviters({"TRUSTED_INVITERS": f" {TRUSTED_NUMBER} , ,"}) == (
        TRUSTED_NUMBER,
    )
    assert parse_trusted_inviters({"TRUSTED_INVITERS": f"{TRUSTED_NUMBER},{OPERATOR_UUID}"}) == (
        TRUSTED_NUMBER,
        OPERATOR_UUID,
    )


def test_parsing_de_duplicates_and_rejects_junk() -> None:
    same = f"{TRUSTED_NUMBER},{TRUSTED_NUMBER}"
    assert parse_trusted_inviters({"TRUSTED_INVITERS": same}) == (TRUSTED_NUMBER,)
    with pytest.raises(ConfigError, match="must be"):
        parse_trusted_inviters({"TRUSTED_INVITERS": "kuhy"})


def test_an_empty_allowlist_trusts_everyone_but_says_so(
    caplog: pytest.LogCaptureFixture,
) -> None:
    with caplog.at_level(logging.WARNING):
        trust = InviteTrust(())
    assert "will be watched" in caplog.text
    assert trust.allows(WATCHED_GROUP, row(WATCHED_GROUP, members=[STRANGER_NUMBER])) is True


def test_a_trusted_member_vouches_for_the_group() -> None:
    trust = InviteTrust((TRUSTED_NUMBER,))
    assert trust.allows(WATCHED_GROUP, row(WATCHED_GROUP, members=[TRUSTED_NUMBER])) is True


def test_a_trusted_uuid_vouches_too() -> None:
    """The listing spells an unknown contact as a uuid, not a number."""
    trust = InviteTrust((OPERATOR_UUID,))
    assert trust.allows(WATCHED_GROUP, row(WATCHED_GROUP, members=[OPERATOR_UUID])) is True


def test_a_trusted_number_that_is_only_invited_does_not_vouch() -> None:
    """Bait: a stranger adds the operator, who has not accepted."""
    trust = InviteTrust((TRUSTED_NUMBER,))
    entry = row(WATCHED_GROUP, members=[STRANGER_NUMBER, TRUSTED_NUMBER], pending=[TRUSTED_NUMBER])
    assert trust.allows(WATCHED_GROUP, entry) is False


def test_a_refusal_is_logged_once_and_names_the_members(
    caplog: pytest.LogCaptureFixture,
) -> None:
    trust = InviteTrust((TRUSTED_NUMBER,))
    entry = row(WATCHED_GROUP, members=[STRANGER_NUMBER])
    with caplog.at_level(logging.WARNING):
        assert trust.allows(WATCHED_GROUP, entry) is False
        assert trust.allows(WATCHED_GROUP, entry) is False
    assert caplog.text.count("refusing group") == 1
    assert STRANGER_NUMBER in caplog.text


def test_a_malformed_membership_field_refuses_rather_than_crashes() -> None:
    trust = InviteTrust((TRUSTED_NUMBER,))
    entry = group_entry(WATCHED_GROUP)
    entry["members"] = "not-a-list"
    assert trust.allows(WATCHED_GROUP, entry) is False


def test_an_update_from_a_trusted_number_vouches_in_either_spelling(
    caplog: pytest.LogCaptureFixture,
) -> None:
    """The envelope carries the internal id; the listing carries group.<b64>."""
    trust = InviteTrust((TRUSTED_NUMBER,))
    with caplog.at_level(logging.INFO):
        trust.note_update(WATCHED_INTERNAL, TRUSTED_NUMBER, "uuid-operator")
        trust.note_update(WATCHED_INTERNAL, TRUSTED_NUMBER, "uuid-operator")
    assert caplog.text.count("trusted inviter added the bot") == 1
    assert trust.allows(WATCHED_GROUP, row(WATCHED_GROUP, members=[STRANGER_NUMBER])) is True


def test_an_update_from_a_stranger_vouches_for_nothing() -> None:
    trust = InviteTrust((TRUSTED_NUMBER,))
    trust.note_update(WATCHED_INTERNAL, STRANGER_NUMBER, "uuid-stranger")
    assert trust.allows(WATCHED_GROUP, row(WATCHED_GROUP, members=[STRANGER_NUMBER])) is False


def test_an_update_by_a_trusted_uuid_vouches() -> None:
    trust = InviteTrust((OPERATOR_UUID,))
    trust.note_update(WATCHED_INTERNAL, "", OPERATOR_UUID)
    assert trust.allows(WATCHED_GROUP, row(WATCHED_GROUP, members=[STRANGER_NUMBER])) is True


async def test_the_directory_adopts_only_the_trusted_group(config: Config, store: Store) -> None:
    client = listing_client(
        [
            row(WATCHED_GROUP, members=[TRUSTED_NUMBER]),
            row(SECOND_GROUP, members=[STRANGER_NUMBER]),
        ]
    )
    found = GroupDirectory(config, client, store, InviteTrust((TRUSTED_NUMBER,)))
    await found.refresh()
    assert found.groups == (WATCHED_GROUP,)


async def test_a_pinned_group_bypasses_the_trust_gate(config: Config, store: Store) -> None:
    """GROUP_IDS is the operator naming a group by hand: vouch enough."""
    pinned = replace(config, group_ids=(WATCHED_GROUP,))
    client = listing_client([row(WATCHED_GROUP, members=[STRANGER_NUMBER])])
    found = GroupDirectory(pinned, client, store, InviteTrust((TRUSTED_NUMBER,)))
    await found.refresh()
    assert found.groups == (WATCHED_GROUP,)


async def test_an_update_frame_lets_the_next_refresh_adopt_the_group(
    config: Config, store: Store
) -> None:
    """The whole point: being added by the operator is still the whole setup."""
    client = listing_client([row(WATCHED_GROUP, members=[STRANGER_NUMBER])])
    found = GroupDirectory(config, client, store, InviteTrust((TRUSTED_NUMBER,)))
    await found.refresh()
    assert found.groups == ()

    message = parse_envelope(update_envelope(group_id=WATCHED_INTERNAL))
    assert message is not None
    found.note_group_update(message)
    await found.refresh()
    assert found.groups == (WATCHED_GROUP,)


def test_ordinary_frames_vouch_for_nothing(config: Config, store: Store) -> None:
    """Only an UPDATE names an inviter; a chat message and a DM must not."""
    trust = InviteTrust((TRUSTED_NUMBER,))
    found = GroupDirectory(config, listing_client([]), store, trust)
    for frame in (envelope(text="!cat"), envelope(text="hi", group_id=None)):
        message = parse_envelope(frame)
        assert message is not None
        found.note_group_update(message)
    assert trust.allows(WATCHED_GROUP, row(WATCHED_GROUP, members=[STRANGER_NUMBER])) is False