signal-bot/tool/e2e_media.py

#!/usr/bin/env python3
# Copyright (c) 2026 Krzysztof Rudnicki. MIT licence.
"""Seed the fake's attachment store, and check what the album ended up storing.

The samples themselves are built by tool/e2e_fixtures.py; this half moves them
in and reads the result back out.

Two modes:

    --upload <base-url>     put the media into the fake's attachment store
    --verify-pack <dir>     assert every stored file came out clean

The second is the half that matters. It re-reads the album with exiftool and
fails on any surviving non-structural tag, which is the same check
``media_tags.is_allowed`` makes -- run here against files that went through
the whole live path rather than through a unit test's fake stripper.
"""

from __future__ import annotations

import argparse
import base64
import sys
from pathlib import Path
from typing import Final

import httpx
from e2e_fixtures import EXIFTOOL, PROBE_CREATOR, PROBE_MAKE, build, run_tool

from signal_bot.media_tags import is_allowed

# The suffixes exiftool 13.55 cannot write, and which are therefore stored
# exactly as they arrived. Excluded from the verification sweep because they
# are *expected* to still carry everything.
UNSTRIPPED: Final = frozenset({".svg", ".webm"})


def upload(base_url: str) -> int:
    """Put every attachment into the fake container's store."""
    for attachment_id, payload in build().items():
        response = httpx.post(
            f"{base_url}/_control/attachment",
            json={"id": attachment_id, "base64": base64.b64encode(payload).decode("ascii")},
            timeout=10,
        )
        response.raise_for_status()
        sys.stdout.write(f"   uploaded {attachment_id} ({len(payload)} bytes)\n")
    return 0


def _survivors(path: Path) -> list[str]:
    """Every non-structural tag one stored file still reports."""
    report = run_tool([EXIFTOOL, "-G0", "-s", "-all", str(path)])
    return [name for name, allowed in _tags(report) if not allowed]


def _tags(report: str) -> list[tuple[str, bool]]:
    """Each reported tag as "Group:Tag", and whether it may remain."""
    return [_tag(line) for line in report.splitlines() if line.startswith("[")]


def _tag(line: str) -> tuple[str, bool]:
    """One "[Group] Tag : value" line as its name and whether it may remain."""
    group, _, rest = line[1:].partition("]")
    tag, _, value = rest.strip().partition(":")
    return f"{group}:{tag.strip()}", is_allowed(group, tag.strip(), value.strip())


def _stored(directory: Path) -> list[Path]:
    """Every media file the album holds: in the album, and waiting to be."""
    folders = (directory / "media", directory / "pending")
    return sorted(
        path
        for folder in folders
        if folder.is_dir()
        for path in folder.iterdir()
        # The queue keeps a JSON sidecar beside each file; it is not media.
        if path.suffix != ".json"
    )


def verify_pack(directory: Path) -> int:
    """Fail unless every strippable file, stored or queued, came out clean."""
    failures: list[str] = []
    checked = 0
    for path in _stored(directory):
        if path.suffix.lower() in UNSTRIPPED:
            sys.stdout.write(f"   {path.name}: stored unstripped by decision, not checked\n")
            continue
        raw = path.read_bytes()
        # The raw byte search is not redundant with the tag sweep: exiftool
        # reports what it can parse, and a string left in a chunk it does not
        # model would be invisible to the sweep and present in the file.
        failures.extend(
            f"{path.name} still contains {probe!r}"
            for probe in (PROBE_MAKE.encode(), PROBE_CREATOR.encode())
            if probe in raw
        )
        leaks = _survivors(path)
        if leaks:
            failures.append(f"{path.name} still reports {leaks}")
        checked += 1
    for line in failures:
        sys.stderr.write(f"Error: {line}\n")
    if failures:
        return 1
    sys.stdout.write(f"   {checked} stored file(s) carry no metadata beyond the container\n")
    return 0


def main() -> int:
    """Generate, upload or verify, depending on the mode asked for."""
    parser = argparse.ArgumentParser(description=__doc__)
    group = parser.add_mutually_exclusive_group(required=True)
    group.add_argument("--upload", metavar="BASE_URL", help="seed the fake's attachment store")
    group.add_argument("--verify-pack", metavar="DIR", type=Path, help="check a written album")
    arguments = parser.parse_args()
    if arguments.upload:
        return upload(arguments.upload)
    return verify_pack(arguments.verify_pack)


if __name__ == "__main__":
    raise SystemExit(main())