From a9f6875abe64f6cae05fbd8e2608e2a6bff0edbe Mon Sep 17 00:00:00 2001 From: Krzysztof kuhy Rudnicki Date: Sat, 29 Aug 2026 19:04:39 +0200 Subject: [PATCH] fix: never let the trust gate un-watch a group it already adopted _permitted feeds both _join_pending and _adopt, and _adopt treats anything missing from its result as a removal: it drops the group, purges its tally with store.forget_group and logs "no longer a member". So one listing whose membership no longer vouches -- the same group has been seen spelling a member as +E.164 in one row and as a bare uuid in another, and a row can come back with no members field at all -- would silently un-watch a live group and delete its counts, all three claims false. That is exactly the narrowing discovery.py's own docstring forbids ("a failed refresh never narrows the set"), reintroduced by the gate one commit earlier. Adoption is itself the vouch, so an already-watched id short- circuits the check; only groups the bot has never adopted are tested. Co-Authored-By: Claude Opus 5 (1M context) Claude-Session: https://claude.ai/code/session_01KbjfWdAyHbCubac7wJKJZW --- signal_bot/discovery.py | 14 ++++++++------ tests/test_trust.py | 31 +++++++++++++++++++++++++++++++ 2 files changed, 39 insertions(+), 6 deletions(-) diff --git a/signal_bot/discovery.py b/signal_bot/discovery.py index 8d4bb9c..ac55fef 100644 --- a/signal_bot/discovery.py +++ b/signal_bot/discovery.py @@ -191,13 +191,15 @@ class GroupDirectory: group_id = canonical_group_id(raw) if not self._config.allows(group_id): return None - # A pinned id was named by the operator in .env, which is already the - # strongest possible vouch; running it past the trust gate as well - # would make verify_allowlist fail with a message about joins that - # says nothing about trust. - if self._config.is_pinned or self._trust.allows(group_id, entry): + # Two things that are already vouched for. A pinned id was named by + # the operator in .env; a group already being watched was vouched + # when it was adopted. Re-testing an adopted group every sweep would + # let one listing that spells a member as a uuid -- or omits members + # entirely -- un-watch it and drop its tally, which is the narrowing + # this module exists to prevent. + if self._config.is_pinned or group_id in self._groups: return group_id - return None + return group_id if self._trust.allows(group_id, entry) else None async def _join_pending(self, entries: list[dict[str, Any]]) -> bool: """Try to accept every invitation worth retrying. diff --git a/tests/test_trust.py b/tests/test_trust.py index 6c891a4..1ba9938 100644 --- a/tests/test_trust.py +++ b/tests/test_trust.py @@ -7,6 +7,7 @@ import logging from dataclasses import replace from typing import TYPE_CHECKING, Any +import httpx import pytest from signal_bot.discovery import GroupDirectory @@ -175,3 +176,33 @@ def test_ordinary_frames_vouch_for_nothing(config: Config, store: Store) -> None assert message is not None found.note_group_update(message) assert trust.allows(WATCHED_GROUP, row(WATCHED_GROUP, members=[STRANGER_NUMBER])) is False + + +async def test_an_adopted_group_is_never_un_watched_by_the_gate( + config: Config, store: Store +) -> None: + """Adoption is itself the vouch; a later listing must not purge a tally. + + The listing spells a member as a phone number for a known contact and as + a bare uuid for everybody else, and the same group has been seen both + ways. If that flip could refuse an adopted group, _adopt would treat it + as a removal, drop its counts and log that the bot was removed from it -- + all three of which would be false. + """ + listings = iter( + [ + [row(WATCHED_GROUP, members=[TRUSTED_NUMBER])], + [row(WATCHED_GROUP, members=[STRANGER_NUMBER])], + ] + ) + client = httpx.AsyncClient( + transport=httpx.MockTransport(lambda request: httpx.Response(200, json=next(listings))) + ) + found = GroupDirectory(config, client, store, InviteTrust((TRUSTED_NUMBER,))) + await found.refresh() + assert found.groups == (WATCHED_GROUP,) + store.increment(WATCHED_GROUP, "uuid-alice", "Alice") + + await found.refresh() + assert found.groups == (WATCHED_GROUP,) + assert [(t.display_name, t.count) for t in store.tallies(WATCHED_GROUP)] == [("Alice", 1)]