src/kneiff_dev/privacy_audit.py

"""Reject identity metadata and content before publishing Kneiff."""

from __future__ import annotations

import argparse
from collections.abc import Iterable, Sequence
from dataclasses import dataclass
from pathlib import Path
import re
import subprocess


EXPECTED_NAME = "kneiff"
EXPECTED_EMAIL = "kneiff@users.noreply.huggingface.co"
EMAIL_PATTERN = re.compile(rb"[A-Za-z0-9._%+-]+@[A-Za-z0-9.-]+\.[A-Za-z]{2,}")
TAGGER_PATTERN = re.compile(
    rb"^tagger (.+) <([^>]+)> [0-9]+ ([+-][0-9]{4})$",
    re.MULTILINE,
)


def forbidden_fragments() -> tuple[bytes, ...]:
    """Build private-workspace fingerprints without embedding them verbatim.

    Keeping the parts separate lets this module audit its own tracked blob.

    :return: Lowercase byte fragments rejected in paths and object content.
    """

    parts = (
        ("mark", "ur4"),
        ("mart", "i"),
        ("graph", "igs"),
        ("plota", "stic"),
        ("ned", "data"),
        ("ha", "iu"),
        ("build", "ben"),
        ("mongodb", "api"),
    )
    return tuple("".join(fragment_parts).encode() for fragment_parts in parts)


@dataclass(frozen=True, order=True)
class PrivacyViolation:
    """One publication blocker found by the repository audit."""

    surface: str
    detail: str


def _run_git(repository: Path, *args: str) -> str:
    """Run one read-only Git query.

    :param repository: Repository root used by Git.
    :param args: Git arguments after ``git -C <repository>``.
    :return: Standard output decoded as UTF-8.
    """

    result = subprocess.run(
        ["git", "-C", str(repository), *args],
        check=True,
        capture_output=True,
        text=True,
    )
    return result.stdout


def _allowed_email(email: bytes) -> bool:
    """Return whether one email belongs to the anonymous publication surface.

    :param email: Address extracted from tracked content or Git metadata.
    :return: Whether the address is allowed.
    """

    lowered = email.lower()
    return lowered in {b"git@hf.co", EXPECTED_EMAIL.encode()}


def _scan_bytes(surface: str, payload: bytes) -> list[PrivacyViolation]:
    """Inspect one path or object payload for publication blockers.

    :param surface: Human-readable object or file identifier.
    :param payload: Raw bytes to inspect.
    :return: Detected violations.
    """

    lowered = payload.lower()
    violations = [
        PrivacyViolation(
            surface,
            f"forbidden fingerprint: {fragment.decode()}",
        )
        for fragment in forbidden_fragments()
        if fragment in lowered
    ]
    for email in sorted(set(EMAIL_PATTERN.findall(payload))):
        if not _allowed_email(email):
            violations.append(
                PrivacyViolation(
                    surface,
                    f"unexpected email: {email.decode(errors='replace')}",
                )
            )
    return violations


def _publishable_paths(repository: Path) -> tuple[Path, ...]:
    """List versioned and untracked paths that Git would publish.

    :param repository: Repository root used by Git.
    :return: Non-ignored paths relative to the repository.
    """

    output = subprocess.run(
        [
            "git",
            "-C",
            str(repository),
            "ls-files",
            "--cached",
            "--others",
            "--exclude-standard",
            "-z",
        ],
        check=True,
        capture_output=True,
    ).stdout
    return tuple(Path(part.decode()) for part in output.split(b"\0") if part)


def audit_tracked_tree(repository: Path) -> tuple[PrivacyViolation, ...]:
    """Inspect the current versioned tree.

    :param repository: Repository root to inspect.
    :return: Sorted publication blockers.
    """

    violations: list[PrivacyViolation] = []
    for relative_path in _publishable_paths(repository):
        surface = f"tree:{relative_path.as_posix()}"
        violations.extend(_scan_bytes(surface, relative_path.as_posix().encode()))
        path = repository / relative_path
        if path.is_symlink():
            violations.extend(_scan_bytes(surface, path.readlink().as_posix().encode()))
        elif path.is_file():
            violations.extend(_scan_bytes(surface, path.read_bytes()))
        else:
            violations.append(PrivacyViolation(surface, "tracked path is missing"))
    return tuple(sorted(set(violations)))


def _reachable_objects(repository: Path) -> tuple[tuple[str, str], ...]:
    """List unique objects reachable from local branches and tags.

    :param repository: Repository root used by Git.
    :return: Object IDs paired with their first known path.
    """

    objects: dict[str, str] = {}
    for line in _run_git(
        repository,
        "rev-list",
        "--objects",
        "--branches",
        "--tags",
    ).splitlines():
        object_id, _, path = line.partition(" ")
        objects.setdefault(object_id, path)
    for line in _run_git(
        repository,
        "for-each-ref",
        "--format=%(objectname) %(refname)",
        "refs/tags",
    ).splitlines():
        object_id, _, ref_name = line.partition(" ")
        objects.setdefault(object_id, ref_name)
    return tuple(objects.items())


def _cat_objects(
    repository: Path,
    objects: Sequence[tuple[str, str]],
) -> Iterable[tuple[str, str, str, bytes]]:
    """Read reachable Git objects through one batch process.

    :param repository: Repository root used by Git.
    :param objects: Object IDs and display paths to read.
    :return: Iterator of object ID, display path, type, and raw payload.
    """

    if not objects:
        return
    request = b"".join(f"{object_id}\n".encode() for object_id, _path in objects)
    process = subprocess.run(
        ["git", "-C", str(repository), "cat-file", "--batch"],
        input=request,
        check=True,
        capture_output=True,
    )
    output = process.stdout
    offset = 0
    for expected_id, path in objects:
        header_end = output.index(b"\n", offset)
        header = output[offset:header_end].decode()
        object_id, object_type, size_text = header.split()
        if object_id != expected_id:
            raise RuntimeError(f"Git returned {object_id} while reading {expected_id}")
        size = int(size_text)
        payload_start = header_end + 1
        payload_end = payload_start + size
        payload = output[payload_start:payload_end]
        offset = payload_end + 1
        yield object_id, path, object_type, payload


def _audit_commit_metadata(repository: Path) -> list[PrivacyViolation]:
    """Verify raw author, committer, date, and signature metadata.

    :param repository: Repository root used by Git.
    :return: Detected metadata violations.
    """

    output = _run_git(
        repository,
        "log",
        "--branches",
        "--tags",
        "--format=%H%x09%aN%x09%aE%x09%cN%x09%cE%x09%aI%x09%cI",
    )
    violations: list[PrivacyViolation] = []
    for line in output.splitlines():
        (
            commit_id,
            author_name,
            author_email,
            committer_name,
            committer_email,
            author_date,
            committer_date,
        ) = line.split("\t")
        surface = f"commit:{commit_id}"
        if (author_name, author_email) != (EXPECTED_NAME, EXPECTED_EMAIL):
            violations.append(
                PrivacyViolation(
                    surface,
                    f"unexpected author: {author_name} <{author_email}>",
                )
            )
        if (committer_name, committer_email) != (EXPECTED_NAME, EXPECTED_EMAIL):
            violations.append(
                PrivacyViolation(
                    surface,
                    f"unexpected committer: {committer_name} <{committer_email}>",
                )
            )
        if not author_date.endswith(("Z", "+00:00")):
            violations.append(
                PrivacyViolation(surface, f"author date is not UTC: {author_date}")
            )
        if not committer_date.endswith(("Z", "+00:00")):
            violations.append(
                PrivacyViolation(
                    surface,
                    f"committer date is not UTC: {committer_date}",
                )
            )
    return violations


def audit_reachable_history(repository: Path) -> tuple[PrivacyViolation, ...]:
    """Inspect local branch and tag history, including raw object content.

    :param repository: Repository root to inspect.
    :return: Sorted publication blockers.
    """

    violations = _audit_commit_metadata(repository)
    objects = _reachable_objects(repository)
    for object_id, path, object_type, payload in _cat_objects(repository, objects):
        if object_type not in {"blob", "commit", "tag"}:
            continue
        surface = f"{object_type}:{object_id}"
        if path:
            surface = f"{surface}:{path}"
        violations.extend(_scan_bytes(surface, payload))
        signature_markers = (
            b"\ngpgsig ",
            b"-----BEGIN PGP SIGNATURE-----",
            b"-----BEGIN SSH SIGNATURE-----",
        )
        if object_type in {"commit", "tag"} and any(
            marker in b"\n" + payload for marker in signature_markers
        ):
            violations.append(PrivacyViolation(surface, "signed Git object"))
        if object_type == "tag":
            tagger = TAGGER_PATTERN.search(payload)
            if tagger is None:
                violations.append(PrivacyViolation(surface, "missing tagger metadata"))
            else:
                name, email, timezone = tagger.groups()
                if (name.decode(), email.decode()) != (
                    EXPECTED_NAME,
                    EXPECTED_EMAIL,
                ):
                    violations.append(
                        PrivacyViolation(
                            surface,
                            f"unexpected tagger: {name.decode()} <{email.decode()}>",
                        )
                    )
                if timezone != b"+0000":
                    violations.append(
                        PrivacyViolation(
                            surface,
                            f"tagger date is not UTC: {timezone.decode()}",
                        )
                    )
    return tuple(sorted(set(violations)))


def audit_repository(
    repository: Path,
    *,
    include_history: bool = True,
) -> tuple[PrivacyViolation, ...]:
    """Audit the versioned tree and optionally its reachable Git history.

    :param repository: Repository root to inspect.
    :param include_history: Whether to inspect local branches and tags.
    :return: Sorted publication blockers.
    """

    violations = list(audit_tracked_tree(repository))
    if include_history:
        violations.extend(audit_reachable_history(repository))
    return tuple(sorted(set(violations)))


def _parser() -> argparse.ArgumentParser:
    """Build the command-line parser.

    :return: Configured parser.
    """

    parser = argparse.ArgumentParser(
        description="Reject identity metadata and content before publication."
    )
    parser.add_argument(
        "--repository",
        type=Path,
        default=Path.cwd(),
        help="Repository root. Defaults to the current directory.",
    )
    parser.add_argument(
        "--tree-only",
        action="store_true",
        help="Inspect tracked files without checking existing Git history.",
    )
    return parser


def main(argv: Sequence[str] | None = None) -> int:
    """Run the publication audit.

    :param argv: Optional command-line arguments.
    :return: Process exit code.
    """

    args = _parser().parse_args(argv)
    repository = args.repository.resolve()
    violations = audit_repository(
        repository,
        include_history=not args.tree_only,
    )
    if violations:
        for violation in violations:
            print(f"{violation.surface}: {violation.detail}")
        return 1
    print("Privacy audit passed.")
    return 0


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