learningcircuit--local-deep-research/.pre-commit-hooks/check-author-identity.py

#!/usr/bin/env python3
"""Author identity consistency check.

Ensures every commit attributed to a project author declared in
``pyproject.toml`` (the ``authors`` list) uses an acceptable email identity:
either a GitHub ``@users.noreply.github.com`` address (always allowed — these
are privacy-preserving), or that author's own declared address. This keeps
contributor attribution consistent and prevents a declared author's commits
from going out under an unintended personal address.

It inspects commit *metadata* (author, committer, and ``Co-authored-by``
trailers) rather than file contents, so it runs once per invocation
(``always_run: true`` / ``pass_filenames: false``):

- In CI on a pull request, it checks every commit the PR adds (``merge-base..head``),
  and reads the allow-list from the *base* ref so a PR cannot authorize an
  address by editing ``pyproject.toml`` in the same change.
- Locally at the pre-commit stage, it checks the commit about to be made.

The allow-list is read from ``pyproject.toml`` at runtime — nothing is
hard-coded here. A mismatching address is never printed (so it can't end up in
logs); messages name only the declared author. Range-resolution failures fail
*closed* (the check fails rather than silently passing).
"""

from __future__ import annotations

import json
import os
import re
import subprocess
import sys
from pathlib import Path

NOREPLY_SUFFIX = "@users.noreply.github.com"


def _git(*args: str) -> tuple[int, str]:
    """Run git; return (returncode, stdout). stderr is captured and discarded."""
    proc = subprocess.run(["git", *args], capture_output=True, text=True)
    return proc.returncode, proc.stdout


def parse_identities(text: str) -> dict[str, set[str]]:
    """Map declared author name (lower-cased) -> set of declared emails (lower).

    Tolerant of key order and surrounding whitespace; anchored to a top-level
    ``authors = [`` line so an unrelated ``*authors`` key can't hijack it.
    """
    identities: dict[str, set[str]] = {}
    block = re.search(r"(?m)^authors\s*=\s*\[(.*?)\]", text, re.S)
    if not block:
        return identities
    for entry in re.findall(r"\{([^}]*)\}", block.group(1)):
        name = re.search(r'name\s*=\s*"([^"]+)"', entry)
        email = re.search(r'email\s*=\s*"([^"]+)"', entry)
        if name and email:
            identities.setdefault(name.group(1).strip().lower(), set()).add(
                email.group(1).strip().lower()
            )
    return identities


def _is_noreply(email: str) -> bool:
    return email.endswith(NOREPLY_SUFFIX)


def _mismatch(kind: str, name: str, email: str, declared: dict[str, set[str]]):
    """Return a message if (name, email) is a disallowed identity, else None.

    The offending email is intentionally NOT included in the message.
    """
    raw_name = (name or "").strip()
    name_lc = raw_name.lower()
    email = (email or "").strip().lower()
    allowed = declared.get(name_lc)
    if not allowed:
        return None  # not a declared author -> not enforced
    if _is_noreply(email):
        return None  # any GitHub noreply is privacy-safe and allowed
    if email in allowed:
        return None  # the author's own declared address
    return (
        f'{kind} "{raw_name}" is a declared author but is using a non-noreply '
        f"address that is not its declared identity"
    )


_CO_AUTHOR = re.compile(
    r"^\s*Co-authored-by:\s*(?P<name>.*?)\s*<(?P<email>[^>]+)>\s*$", re.I | re.M
)


def _check(record, declared) -> list[str]:
    sha, an, ae, cn, ce, body = record
    out = []
    for kind, name, email in (("author", an, ae), ("committer", cn, ce)):
        msg = _mismatch(kind, name, email, declared)
        if msg:
            out.append(f"  {sha[:9]}: {msg}")
    for m in _CO_AUTHOR.finditer(body or ""):
        msg = _mismatch(
            "Co-authored-by", m.group("name"), m.group("email"), declared
        )
        if msg:
            out.append(f"  {sha[:9]}: {msg}")
    return out


def _parse_log(raw: str) -> list[tuple]:
    records = []
    for chunk in raw.split("\x1e"):
        chunk = chunk.strip("\n")
        if not chunk:
            continue
        fields = chunk.split("\x00")
        if len(fields) < 6:
            # A control char (\x1e / \x00) injected into the author/committer
            # name or body corrupts the framing. Never silently skip a commit
            # -> fail closed.
            raise RuntimeError("malformed commit record in git log output")
        # body may itself contain NULs -> rejoin the tail
        records.append(
            (
                fields[0],
                fields[1],
                fields[2],
                fields[3],
                fields[4],
                "\x00".join(fields[5:]),
            )
        )
    return records


def _resolve_merge_base(base: str, head: str) -> str:
    """Return merge-base(base, head). Raise on failure.

    Resolve from locally-available history FIRST and fetch only as a fallback.
    This hook runs inside ``pre-commit run``, which treats any working-tree or
    git-state change a hook makes as a hook failure. The workflow checks out
    full history (``fetch-depth: 0``) precisely so the PR-CI path resolves the
    range with no fetch -- mutating nothing. The fetch fallback exists only for
    a shallow checkout and should not run in the PR-CI path.
    """
    rc, mb = _git("merge-base", base, head)
    if rc == 0 and mb.strip():
        return mb.strip()
    # Endpoints not present locally (shallow clone): fetch, then retry.
    subprocess.run(
        ["git", "fetch", "--quiet", "--depth=1000", "origin", head, base],
        check=False,
    )
    rc, mb = _git("merge-base", base, head)
    if rc != 0 or not mb.strip():
        # Diverged further than the shallow window — get full history, retry.
        subprocess.run(
            ["git", "fetch", "--quiet", "--unshallow", "origin"], check=False
        )
        rc, mb = _git("merge-base", base, head)
    if rc != 0 or not mb.strip():
        raise RuntimeError("could not resolve the PR commit range")
    return mb.strip()


def _pr_context():
    """In PR CI: return (records, base_pyproject_text). None if not PR CI.

    Raises RuntimeError on an unresolvable range (caller fails closed).
    """
    is_pr_event = os.environ.get("GITHUB_EVENT_NAME", "") in (
        "pull_request",
        "pull_request_target",
    )

    def give_up(reason: str):
        # On a genuine PR event an unusable payload must FAIL CLOSED, never drop
        # to the local no-op path. Off a PR event, this simply isn't PR CI.
        if is_pr_event:
            raise RuntimeError(reason)
        return

    event_path = os.environ.get("GITHUB_EVENT_PATH")
    if not (event_path and Path(event_path).exists()):
        return give_up("pull_request event but no event payload")
    try:
        with open(event_path, encoding="utf-8") as fh:
            payload = json.load(fh)
    except (OSError, ValueError):
        return give_up("could not read the PR event payload")
    pr = payload.get("pull_request") if isinstance(payload, dict) else None
    if not isinstance(pr, dict):
        return give_up("pull_request event missing its payload")
    base = (pr.get("base") or {}).get("sha")
    head = (pr.get("head") or {}).get("sha")
    if not (base and head):
        return give_up("pull_request event missing base/head sha")
    merge_base = _resolve_merge_base(base, head)  # may raise -> fail closed
    rc, raw = _git(
        "log",
        f"{merge_base}..{head}",
        "--format=%H%x00%an%x00%ae%x00%cn%x00%ce%x00%B%x1e",
    )
    if rc != 0:
        raise RuntimeError("git log failed for the PR commit range")
    rc, base_pyproject = _git("show", f"{base}:pyproject.toml")
    if rc != 0:
        raise RuntimeError("could not read pyproject.toml from the base ref")
    return _parse_log(raw), base_pyproject


def _pending_records() -> list[tuple]:
    """Identity of the commit about to be created (local pre-commit stage)."""

    def parse(ident: str):
        m = re.match(r"^(.*)<([^>]+)>", ident)
        return (m.group(1).strip(), m.group(2).strip()) if m else ("", "")

    an, ae = parse(_git("var", "GIT_AUTHOR_IDENT")[1])
    cn, ce = parse(_git("var", "GIT_COMMITTER_IDENT")[1])
    return [("pending00", an, ae, cn, ce, "")]


def main() -> int:
    try:
        ctx = _pr_context()
    except RuntimeError as exc:
        # Fail CLOSED: never silently pass a required check we couldn't run.
        print(f"author-identity: {exc}; failing closed.", file=sys.stderr)
        return 1

    if ctx is not None:
        records, pyproject_text = ctx
        declared = parse_identities(pyproject_text)
        if not declared and re.search(
            r"(?m)^\s*(authors\s*=|\[\[[^\]]*authors\s*\]\])", pyproject_text
        ):
            # The base pyproject declares authors but we parsed none (e.g. the
            # table format changed). Fail CLOSED rather than silently disabling
            # the check -- a loud red is the whole point of this guard.
            print(
                "author-identity: base pyproject.toml declares authors but none "
                "could be parsed (unrecognized format?); failing closed.",
                file=sys.stderr,
            )
            return 1
    else:
        try:
            declared = parse_identities(
                Path("pyproject.toml").read_text(encoding="utf-8")
            )
        except OSError:
            declared = {}
        records = _pending_records()

    if not declared:
        return 0

    errors = []
    for rec in records:
        errors.extend(_check(rec, declared))
    if errors:
        print("Author identity check failed:\n", file=sys.stderr)
        print("\n".join(errors), file=sys.stderr)
        print(
            "\nA commit is attributed to a declared author but uses a personal "
            "(non-noreply) address.\nUse your GitHub `@users.noreply.github.com` "
            "address, e.g. re-author with:\n"
            "  git commit --amend --reset-author        # latest commit\n"
            "and ensure your git user.email is your noreply address.",
            file=sys.stderr,
        )
        return 1
    return 0


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