signal-bot/scripts/personal_data_sources.py

#!/usr/bin/env python3
# Copyright (c) 2026 Krzysztof Rudnicki. MIT licence.
"""Where the forbidden words come from -- all of it outside this repository.

Split out of check_personal_data.py at the 250-line cap, and the seam is the
honest one: this half knows the private data, that half knows nothing but how
to search. A list of real names kept *in* the repository to check against
would be the leak the check exists to prevent, so nothing here is ever
written down -- it is read at run time from the photo pack, the message
database and the local .env, and on a machine with none of those it simply
comes back empty.
"""

from __future__ import annotations

import base64
import json
import os
import sqlite3
from pathlib import Path
from typing import Final

REPO: Final = Path(__file__).resolve().parent.parent

# Three, because real first names are three letters long and one of this
# group's is. Two would match inside half the words in the repository; three
# needs a handful of allowlist entries for the given names that are also
# popes, which is a cost worth paying to catch a real one.
MIN_NAME: Final = 3


def _pack_names(directory: Path) -> set[str]:
    """Every trigger word in a photo pack manifest -- i.e. every name."""
    manifest = directory / "people.json"
    if not manifest.is_file():
        return set()
    try:
        document = json.loads(manifest.read_text(encoding="utf-8"))
    except OSError, json.JSONDecodeError:
        return set()
    entries = [*document.get("people", []), *document.get("combos", [])]
    return {
        str(trigger).lstrip("!").strip()
        for entry in entries
        for trigger in entry.get("triggers", [])
    }


def _database_names(path: Path) -> set[str]:
    """Display names of everybody the bot has ever seen speak."""
    if not path.is_file():
        return set()
    try:
        connection = sqlite3.connect(f"file:{path}?mode=ro", uri=True)
    except sqlite3.Error:
        return set()
    # Literal statements, not a table name interpolated into one: a
    # formatted query is a SQL-injection shape even when every input is a
    # constant, and this repository takes no lint suppressions.
    queries = (
        "SELECT DISTINCT display_name FROM message_counts",
        "SELECT DISTINCT display_name FROM group_people",
        "SELECT DISTINCT display_name FROM message_events",
    )
    found: set[str] = set()
    with connection:
        for query in queries:
            try:
                rows = connection.execute(query)
            except sqlite3.Error:
                continue
            found.update(str(row[0]) for row in rows if row[0])
    connection.close()
    return found


def _database_groups(path: Path) -> set[str]:
    """Every real group id the bot has seen, in both spellings.

    Both, because they are the same group written two ways: /v1/groups
    reports "group.<base64>" and an inbound envelope carries the id that
    base64 decodes to. A check that knew only the first would have missed the
    one that was sitting in tests/test_config.py.
    """
    if not path.is_file():
        return set()
    try:
        connection = sqlite3.connect(f"file:{path}?mode=ro", uri=True)
    except sqlite3.Error:
        return set()
    queries = (
        "SELECT DISTINCT group_id FROM group_meta",
        "SELECT DISTINCT group_id FROM message_events",
    )
    found: set[str] = set()
    with connection:
        for query in queries:
            try:
                rows = connection.execute(query)
            except sqlite3.Error:
                continue
            found.update(str(row[0]) for row in rows if row[0])
    connection.close()
    inner: set[str] = set()
    for group in found:
        payload = group.removeprefix("group.")
        try:
            inner.add(base64.b64decode(payload, validate=True).decode("utf-8"))
        except ValueError, UnicodeDecodeError:
            continue
    return found | inner


def _env_values(path: Path) -> set[str]:
    """Phone numbers, group ids and inviters from the deployment's .env."""
    if not path.is_file():
        return set()
    interesting = {"PHONE_NUMBER", "GROUP_IDS", "GROUP_ID", "TRUSTED_INVITERS"}
    found: set[str] = set()
    for line in path.read_text(encoding="utf-8").splitlines():
        key, _, value = line.partition("=")
        if key.strip() in interesting:
            found.update(part.strip() for part in value.split(",") if part.strip())
    return found


def forbidden(permitted: set[str]) -> tuple[set[str], list[str]]:
    """Every literal that must not be committed, and where they came from.

    The sources are all outside the repository on purpose. A list of real
    names kept *in* the repository to check against would be the leak it is
    meant to prevent.
    """
    pack = os.environ.get("PHOTO_PACK_DIR", "").strip()
    directory = (
        Path(pack)
        if pack
        else Path(os.environ.get("XDG_DATA_HOME", Path.home() / ".local" / "share"))
        / "signal-bot"
        / "pack"
    )
    sources: list[str] = []
    words: set[str] = set()
    for label, found in (
        (f"pack manifest ({directory})", _pack_names(directory)),
        ("message database", _database_names(REPO / "signal_bot.sqlite3")),
        ("group ids", _database_groups(REPO / "signal_bot.sqlite3")),
        (".env", _env_values(REPO / ".env")),
    ):
        if found:
            sources.append(label)
            words |= found
    return {
        word.lower() for word in words if len(word) >= MIN_NAME and word.lower() not in permitted
    }, sources