signal-bot/signal_bot/selftest.py

# Copyright (c) 2026 Krzysztof Rudnicki. MIT licence.
"""Let the bot exercise its own commands without a second Signal account.

Signal never delivers an account its own sends, so the bot cannot type
`!papież` at itself, and every command shipped so far was verified either by
asking a human to type it or by reconstructing the object graph in a scratch
script. Both are weaker than they look: the scratch script is not the wiring
that runs in production, and a human typing it proves nothing about the path
between the websocket and the handler.

This injects a real envelope frame into the *live* dispatcher, so everything
downstream of the websocket read is the code that actually serves the group:
parsing, the scoping gate, the tally, trigger resolution, the cooldown, the
handler and the real transport. Only the socket read itself is skipped.

Fail-closed: SELFTEST_GROUPS names the groups this may post into, and is
empty by default. Without it the endpoint refuses, so a loopback caller
cannot make the bot post into the real group.
"""

from __future__ import annotations

import json
import os
import time
from typing import TYPE_CHECKING, Any, Final

if TYPE_CHECKING:
    from collections.abc import Mapping

    import httpx

    from signal_bot.bot import Bot
    from signal_bot.config import Config

GROUPS_ENV: Final = "SELFTEST_GROUPS"


class SelfTestError(RuntimeError):
    """Raised when a self-test cannot be run as asked."""


def allowed_groups(env: Mapping[str, str]) -> frozenset[str]:
    """Return the group names the self-test may post into.

    Empty unless SELFTEST_GROUPS says otherwise. A bot that would replay
    arbitrary text into the production group on a loopback POST is a worse
    problem than the one this module solves.
    """
    raw = env.get(GROUPS_ENV, "")
    return frozenset(name.strip() for name in raw.split(",") if name.strip())


async def resolve_group(client: httpx.AsyncClient, config: Config, name: str) -> tuple[str, str]:
    """Return (internal group id, an author to attribute the message to).

    The author must be somebody Signal can resolve, because the reply quotes
    them: a made-up uuid comes back as "user ... is not registered" and the
    send fails. Any real member other than the bot will do.
    """
    response = await client.get(config.groups_url)
    response.raise_for_status()
    for entry in response.json():
        if entry.get("name") != name:
            continue
        group_id = str(entry.get("internal_id") or entry.get("id") or "")
        members = [m for m in entry.get("members", []) if m != config.phone_number]
        if not group_id or not members:
            msg = f"group {name!r} has no usable id or no other member"
            raise SelfTestError(msg)
        return group_id, str(members[0])
    msg = f"no group named {name!r}"
    raise SelfTestError(msg)


def build_frame(group_id: str, author: str, text: str) -> str:
    """Build the websocket frame signal-cli would emit for this message."""
    stamp = int(time.time() * 1000)
    payload: dict[str, Any] = {
        "timestamp": stamp,
        "message": text,
        "groupInfo": {"groupId": group_id, "type": "DELIVER"},
    }
    envelope: dict[str, Any] = {
        "source": author,
        "sourceNumber": author,
        "sourceUuid": author,
        "sourceName": "self-test",
        "sourceDevice": 1,
        "timestamp": stamp,
        "dataMessage": payload,
    }
    return json.dumps({"envelope": envelope, "account": author})


async def run(
    bot: Bot,
    client: httpx.AsyncClient,
    group: str,
    text: str,
    env: Mapping[str, str] | None = None,
) -> str:
    """Feed one synthetic message through the live dispatcher."""
    if group not in allowed_groups(os.environ if env is None else env):
        msg = f"{group!r} is not in {GROUPS_ENV}; refusing to post there"
        raise SelfTestError(msg)
    group_id, author = await resolve_group(client, bot.config, group)
    await bot.dispatcher.handle(build_frame(group_id, author, text))
    return group_id