signal-bot/signal_bot/media_strip.py

# Copyright (c) 2026 Krzysztof Rudnicki. MIT licence.
"""Removing metadata from an upload, and *proving* it is gone.

The gate is a positive check on the output, never an exit code, because an
exit code is exactly what lied the first time. ``ffmpeg -map_metadata -1 -c
copy`` was tested against a JPEG carrying a real APP1 Exif segment and
produced a byte-identical file -- 292 bytes in, 292 bytes out, GPS marker
still there -- and exited 0. ``-c copy`` hands the compressed frame through
as an opaque packet, and a JPEG's Exif lives inside it. Do not reintroduce
ffmpeg here.

So: strip with exiftool, then read the result back and require every
surviving tag to be structural (media_tags). A metadata atom nobody
anticipated shows up as a tag that is not on the allowlist, and the upload is
refused rather than stored.

Temp files go in /tmp, which the unit makes per-service with PrivateTmp=true.
Staging beside ``media/`` would let a crash mid-strip leave an unstripped
original inside the album permanently, where nothing in pack_load would ever
flag it.

The ``Stripper`` protocol is the seam: unit tests inject a fake so they never
shell out, and ``scripts/e2e.sh`` runs the real one against generated media.
"""

from __future__ import annotations

import logging
import os
import re
import shutil
import tempfile
from pathlib import Path
from typing import TYPE_CHECKING, Final, Protocol

from signal_bot.external import run
from signal_bot.media_tags import is_allowed

if TYPE_CHECKING:
    from signal_bot.media_sniff import MediaKind

logger = logging.getLogger(__name__)

EXIFTOOL: Final = "exiftool"

# Where to look when PATH does not have it. Arch's perl-image-exiftool
# installs to /usr/bin/vendor_perl, which is on an interactive shell's PATH
# (perl's profile.d adds it) and *not* on a systemd unit's, whose default is
# "/usr/local/bin:/usr/bin". Found the hard way on 2026-08-31: the stripper
# worked in every test and in the shell, and the first live run under the
# unit refused every upload with "exiftool is not installed". Looking in the
# known locations is the fix; setting PATH in the unit would only be a note
# somebody has to remember on the next machine.
FALLBACK_PATHS: Final = (
    "/usr/bin/vendor_perl/exiftool",
    "/usr/local/bin/exiftool",
    "/usr/bin/exiftool",
    "/opt/homebrew/bin/exiftool",
)

# How long either exiftool call may take. A phone video is a few tens of
# megabytes and exiftool rewrites the container, not the frames.
TIMEOUT_SECONDS: Final = 60

# "[Group]    TagName    : value", which is what -G0 -s prints.
_TAG_LINE: Final = re.compile(r"^\[(\w+)\]\s+(\S+)\s*:\s?(.*)$")


def locate() -> str | None:
    """The exiftool binary's absolute path, or None if it is not installed.

    PATH first, then the known install locations -- see FALLBACK_PATHS for
    why the second half is not redundant.
    """
    found = shutil.which(EXIFTOOL)
    if found is not None:
        return found
    return next((path for path in FALLBACK_PATHS if Path(path).is_file()), None)


class Stripper(Protocol):
    """Removes metadata from one upload, or refuses it."""

    def strip(self, payload: bytes, kind: MediaKind) -> bytes | None:
        """Return the cleaned file, or None when it cannot be proven clean."""


class ExiftoolStripper:
    """The real stripper: exiftool 13.55, plus a positive post-check."""

    def __init__(self, executable: str | None = None) -> None:
        """Locate exiftool once, by absolute path.

        Args:
            executable: Override for the binary, used by the e2e script.
        """
        self._exiftool = executable or locate()

    def strip(self, payload: bytes, kind: MediaKind) -> bytes | None:
        """Strip one upload and verify the result, or refuse it.

        Args:
            payload: The file exactly as it arrived.
            kind: What the sniffer decided it is.

        Returns:
            The cleaned bytes, or None when the strip failed or the output
            still carries a tag that is not structural.
        """
        if not kind.strippable:
            # kuhy's decision of 2026-08-31: .webm and .svg stay on the
            # allowlist even though exiftool 13.55 can write neither, so they
            # are stored exactly as they arrived, metadata included. Flagged
            # at the time and recorded in DOCS-photo-pack.md.
            logger.warning("storing %s unstripped: exiftool cannot write it", kind.suffix)
            return payload
        if self._exiftool is None:
            logger.error(
                "exiftool not found on PATH (%s) or in %s; refusing to store %s",
                os.environ.get("PATH", ""),
                FALLBACK_PATHS,
                kind.suffix,
            )
            return None
        with tempfile.TemporaryDirectory(prefix="signal-bot-strip-") as workspace:
            return self._run(Path(workspace), payload, kind)

    def _run(self, workspace: Path, payload: bytes, kind: MediaKind) -> bytes | None:
        """Do the strip inside a temp directory that is removed either way."""
        source = workspace / f"in{kind.suffix}"
        destination = workspace / f"out{kind.suffix}"
        source.write_bytes(payload)
        # Never --overwrite_original: writing in place would leave a partial
        # file where the original was if exiftool died halfway.
        if not self._exiftool_ok(["-all=", "-o", str(destination), str(source)]):
            return None
        if not destination.is_file():
            logger.warning("exiftool produced no output for %s", kind.suffix)
            return None
        report = self._exiftool_output(["-G0", "-s", "-all", str(destination)])
        if report is None or not self._verify(report, kind):
            return None
        return destination.read_bytes()

    def _verify(self, report: str, kind: MediaKind) -> bool:
        """True when every tag the stripped file still reports is structural."""
        for line in report.splitlines():
            match = _TAG_LINE.match(line)
            if match is None:
                continue
            group, tag, value = match.group(1), match.group(2), match.group(3)
            if not is_allowed(group, tag, value):
                # The tag name, never its value: the value is the metadata.
                logger.warning("refusing %s: %s:%s survived the strip", kind.suffix, group, tag)
                return False
        return True

    def _exiftool_ok(self, arguments: list[str]) -> bool:
        """Run exiftool for its effect, reporting whether it succeeded."""
        return self._exiftool_output(arguments) is not None

    def _exiftool_output(self, arguments: list[str]) -> str | None:
        """Run exiftool and return its stdout, or None if it failed at all.

        Every subprocess in this repository goes through signal_bot.external;
        see that module for why there is exactly one.
        """
        result = run([str(self._exiftool), *arguments], timeout=TIMEOUT_SECONDS)
        return result.stdout if result.ok else None