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.

A message may carry attachments, which is what lets ``!add_photo`` be
verified this way at all: the ids name files already in the container's own
store, so the handler fetches them back over the real
GET /v1/attachments/{id} rather than being handed bytes. Signal never
delivers an account its own sends -- measured 2026-08-31, and a single-device
account gets no sync copy either -- so this is the only way the bot can put a
photo in front of itself.

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 mimetypes
import os
import time
from typing import TYPE_CHECKING, Any, Final

if TYPE_CHECKING:
    from collections.abc import Mapping, Sequence

    import httpx

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

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 attachment_entry(attachment_id: str) -> dict[str, Any]:
    """One attachments[] entry, shaped the way signal-cli reports it.

    The type is guessed from the id's suffix because that is all signal-cli
    itself has to go on for a stored file -- and it is not load-bearing:
    !add_photo sniffs the magic bytes and ignores what the envelope claims.
    """
    guessed, _ = mimetypes.guess_type(attachment_id)
    return {"contentType": guessed or "application/octet-stream", "id": attachment_id}


def build_frame(group_id: str, author: str, text: str, attachments: Sequence[str] = ()) -> str:
    """Build the websocket frame signal-cli would emit for this message.

    ``attachments`` names files that are **already in the container's own
    attachment store**, exactly as a real inbound envelope does. Nothing here
    invents bytes: the handler fetches them back over the real
    GET /v1/attachments/{id}, so the ingest path under test is the live one.
    """
    stamp = int(time.time() * 1000)
    payload: dict[str, Any] = {
        "timestamp": stamp,
        "message": text,
        "groupInfo": {"groupId": group_id, "type": "DELIVER"},
    }
    if attachments:
        # Top level, never under "quote": a reply carries the *quoted*
        # message's media there, and !add_photo must never take that.
        payload["attachments"] = [attachment_entry(a) for a in attachments]
    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,
    attachments: Sequence[str] = (),
) -> tuple[str, Served | None]:
    """Feed one synthetic message, optionally carrying media, to the dispatcher.

    Returns the group posted into and, when the text was a command, what
    it cost -- through ``Bot.serve``, so the frame queues for a slot the way
    a real one does and the timings are the bot's own.
    """
    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)
    served = await bot.serve(build_frame(group_id, author, text, attachments))
    return group_id, served