Compare binary files in a git working tree against HEAD

bindiff.py

#!/usr/bin/env python3
"""bindiff - compare binary files in a git working tree against HEAD.

Shows a side-by-side hex+ASCII diff with change locations and ASCII column.
Per-byte old->new details are available via -v/--verbose.

Zero third-party dependencies (stdlib only).
"""

from __future__ import annotations

import argparse
import subprocess
import sys
from dataclasses import dataclass, field
from difflib import SequenceMatcher
from pathlib import Path
from typing import Iterable

__version__ = "0.1.0"

MAX_BYTES_DEFAULT = 50 * 1024 * 1024


# =============================================================================
# Errors
# =============================================================================


class BinDiffError(Exception):
    """Base exception for bindiff."""


class GitNotFound(BinDiffError):
    """The git executable could not be found on PATH."""


class NotTracked(BinDiffError):
    """A given path is not tracked by git."""


class OutsideRepo(BinDiffError):
    """A given path resolves outside the repository root."""


class FileTooLarge(BinDiffError):
    """A file exceeds the safety threshold; --force overrides."""


# =============================================================================
# Git integration
# =============================================================================


def _git(args: list[str], *, text: bool = False) -> subprocess.CompletedProcess:
    try:
        return subprocess.run(
            ["git", *args],
            capture_output=True,
            text=text,
            check=False,
        )
    except FileNotFoundError as exc:
        raise GitNotFound("`git` executable not found on PATH") from exc


def find_repo_for(path: Path) -> Path | None:
    """Locate the git working tree that contains `path`.

    Walks up from `path` (and its parents) and returns the toplevel reported
    by `git -C <dir> rev-parse --show-toplevel` for the first enclosing
    working tree. Returns None if no repo is found.

    Submodules return their own toplevel (each submodule is independent).
    """
    try:
        abs_path = path if path.is_absolute() else (Path.cwd() / path).resolve()
    except OSError:
        return None

    candidates = [abs_path, *abs_path.parents]
    seen: set[Path] = set()
    for d in candidates:
        try:
            key = d.resolve()
        except OSError:
            continue
        if key in seen:
            continue
        seen.add(key)
        r = _git(["-C", str(d), "rev-parse", "--show-toplevel"], text=True)
        if r.returncode == 0 and r.stdout.strip():
            return Path(r.stdout.strip())
    return None


def is_tracked(rel_path: str, root: Path | None = None) -> bool:
    """Check whether a repo-relative POSIX path is tracked.

    `root`, when given, scopes the git command to that repository so the
    check works even when the current working directory is elsewhere.
    """
    args = ["ls-files", "--error-unmatch", "--", rel_path]
    if root is not None:
        args = ["-C", str(root), *args]
    r = _git(args)
    return r.returncode == 0


def head_blob(rel_path: str, root: Path | None = None) -> bytes | None:
    """Return the contents of the file at HEAD, or None if absent at HEAD."""
    args = ["show", f"HEAD:{rel_path}"]
    if root is not None:
        args = ["-C", str(root), *args]
    r = _git(args)
    if r.returncode != 0:
        return None
    return r.stdout


def to_repo_relative(path: Path, root: Path) -> str:
    """Convert an absolute or cwd-relative path to a repo-relative POSIX string."""
    abs_path = path if path.is_absolute() else (Path.cwd() / path).resolve()
    try:
        rel = abs_path.relative_to(root)
    except ValueError as exc:
        raise OutsideRepo(f"{path} is outside the repository root") from exc
    return rel.as_posix()


# =============================================================================
# Diff
# =============================================================================


@dataclass
class ChangeRegion:
    old_offset: int
    new_offset: int
    old_len: int
    new_len: int
    kind: str  # 'replace' | 'delete' | 'insert'


@dataclass
class Hunk:
    old_start: int
    old_end: int
    new_start: int
    new_end: int
    changes: list[ChangeRegion] = field(default_factory=list)
    old_bytes: bytes = b""
    new_bytes: bytes = b""


def find_regions(old: bytes, new: bytes) -> list[ChangeRegion]:
    if old == new:
        return []
    sm = SequenceMatcher(a=old, b=new, autojunk=False)
    out: list[ChangeRegion] = []
    for tag, i1, i2, j1, j2 in sm.get_opcodes():
        if tag == "equal":
            continue
        out.append(ChangeRegion(
            old_offset=i1, new_offset=j1,
            old_len=i2 - i1, new_len=j2 - j1,
            kind=tag,
        ))
    return out


def build_hunks(
    old: bytes,
    new: bytes,
    regions: list[ChangeRegion],
    context: int,
) -> list[Hunk]:
    if not regions:
        return []

    old_len, new_len, ctx = len(old), len(new), max(0, context)
    hunks: list[Hunk] = []

    cur_old_lo = max(0, regions[0].old_offset - ctx)
    cur_new_lo = max(0, regions[0].new_offset - ctx)
    cur_old_hi = regions[0].old_offset + regions[0].old_len
    cur_new_hi = regions[0].new_offset + regions[0].new_len
    cur_changes: list[ChangeRegion] = [regions[0]]

    def flush() -> None:
        hi_old = min(old_len, cur_old_hi + ctx)
        hi_new = min(new_len, cur_new_hi + ctx)
        hunks.append(Hunk(
            old_start=cur_old_lo, old_end=hi_old,
            new_start=cur_new_lo, new_end=hi_new,
            changes=list(cur_changes),
            old_bytes=old[cur_old_lo:hi_old],
            new_bytes=new[cur_new_lo:hi_new],
        ))

    for r in regions[1:]:
        r_old_hi = r.old_offset + r.old_len
        r_new_hi = r.new_offset + r.new_len
        gap_old = r.old_offset - cur_old_hi
        gap_new = r.new_offset - cur_new_hi
        if gap_old <= ctx and gap_new <= ctx:
            cur_old_hi = max(cur_old_hi, r_old_hi)
            cur_new_hi = max(cur_new_hi, r_new_hi)
            cur_changes.append(r)
        else:
            flush()
            cur_old_lo = max(0, r.old_offset - ctx)
            cur_new_lo = max(0, r.new_offset - ctx)
            cur_old_hi = r_old_hi
            cur_new_hi = r_new_hi
            cur_changes = [r]

    flush()
    return hunks


def total_changed_bytes(regions: list[ChangeRegion]) -> int:
    return sum(max(r.old_len, r.new_len) for r in regions)


# =============================================================================
# Formatting
# =============================================================================


class _Color:
    """Minimal ANSI color helper. Auto-disables when stdout isn't a TTY."""

    def __init__(self) -> None:
        self.enabled: bool = bool(getattr(sys.stdout, "isatty", lambda: False)())

    def _wrap(self, code: str, text: str) -> str:
        if not self.enabled or not text:
            return text
        return f"\x1b[{code}m{text}\x1b[0m"

    def green(self, t: str) -> str: return self._wrap("32", t)
    def red(self, t: str) -> str: return self._wrap("31", t)
    def cyan(self, t: str) -> str: return self._wrap("36", t)
    def dim(self, t: str) -> str: return self._wrap("2", t)
    def bold(self, t: str) -> str: return self._wrap("1", t)
    def old_diff(self, t: str) -> str: return self._wrap("91", t)   # bright red
    def new_diff(self, t: str) -> str: return self._wrap("92", t)   # bright green


Color = _Color()


def _ascii(b: int) -> str:
    return chr(b) if 0x20 <= b < 0x7F else "."


def _render_hex_cell(
    data: bytes, base_offset: int, changed: set[int],
    width: int, colorize: str | None,
) -> str:
    out: list[str] = []
    for i in range(width):
        if i < len(data):
            b = data[i]
            token = f"{b:02x}"
            if base_offset + i in changed:
                if colorize == "old":
                    out.append(Color.old_diff(token))
                elif colorize == "new":
                    out.append(Color.new_diff(token))
                else:
                    out.append(Color.dim(token))
            else:
                out.append(token)
        else:
            out.append("  ")
        if i < width - 1:
            out.append(" ")
    return "".join(out)


def _render_ascii_cell(
    data: bytes, base_offset: int, changed: set[int],
    width: int, colorize: str | None,
) -> str:
    out: list[str] = []
    for i in range(width):
        if i < len(data):
            b = data[i]
            ch = _ascii(b)
            if base_offset + i in changed:
                if colorize == "old":
                    out.append(Color.old_diff(ch))
                elif colorize == "new":
                    out.append(Color.new_diff(ch))
                else:
                    out.append(Color.dim(ch))
            else:
                out.append(ch)
        else:
            out.append(" ")
    return "".join(out)


def _caret_row(
    data: bytes, base_offset: int, changed: set[int],
    width: int,
) -> str:
    out: list[str] = []
    saw = False
    for i in range(width):
        if i < len(data) and (base_offset + i) in changed:
            out.append("^^")
            saw = True
        else:
            out.append("  ")
        if i < width - 1:
            out.append(" ")
    return "".join(out) if saw else ""


def _byte_diff_line(ch: ChangeRegion, old: bytes, new: bytes) -> Iterable[str]:
    o = old[ch.old_offset:ch.old_offset + ch.old_len]
    n = new[ch.new_offset:ch.new_offset + ch.new_len]
    if ch.kind == "delete":
        yield f"  {ch.old_offset:#010x}: delete {ch.old_len} B {o!r}"
        return
    if ch.kind == "insert":
        yield f"  {ch.new_offset:#010x}: insert {ch.old_len}->{ch.new_len} B {n!r}"
        return
    common = min(ch.old_len, ch.new_len)
    for k in range(common):
        if o[k] != n[k]:
            yield (
                f"  {ch.old_offset + k:#010x} (new={ch.new_offset + k:#010x}): "
                f"{o[k]:#04x} {_ascii(o[k])} -> {n[k]:#04x} {_ascii(n[k])}"
            )
    if ch.old_len > ch.new_len:
        for k in range(common, ch.old_len):
            yield f"  {ch.old_offset + k:#010x}: delete {o[k]:#04x} {_ascii(o[k])}"
    elif ch.new_len > ch.old_len:
        for k in range(common, ch.new_len):
            yield f"  {ch.new_offset + k:#010x}: insert {n[k]:#04x} {_ascii(n[k])}"


def format_file_header(
    display_path: str,
    old: bytes | None,
    new: bytes | None,
    regions: list[ChangeRegion],
    n_hunks: int,
) -> str:
    old_size = len(old) if old is not None else None
    new_size = len(new) if new is not None else None

    lines: list[str] = []
    lines.append(Color.bold(f"== {display_path} ") + Color.dim("=" * 40))
    if old_size is None and new_size is not None:
        lines.append(Color.green(
            f"  added in working tree ({new_size} B); no diff"
        ))
        return "\n".join(lines)
    if new_size is None and old_size is not None:
        lines.append(Color.red(
            f"  deleted from working tree; was {old_size} B at HEAD; no diff"
        ))
        return "\n".join(lines)
    if not regions:
        lines.append(Color.dim(
            f"  old: {old_size} B | new: {new_size} B | no changes"
        ))
        return "\n".join(lines)

    delta = (new_size or 0) - (old_size or 0)
    delta_s = f"+{delta}" if delta >= 0 else str(delta)
    changed = total_changed_bytes(regions)
    lines.append(
        f"  old: {old_size} B (HEAD)   new: {new_size} B (worktree)   "
        f"d {delta_s} B   "
        f"{changed} B in {len(regions)} change(s), {n_hunks} hunk(s)"
    )
    return "\n".join(lines)


def format_hunk(
    hunk: Hunk,
    index: int,
    total: int,
    width: int,
    context: int,
    old: bytes,
    new: bytes,
    verbose: bool = False,
) -> str:
    lines: list[str] = []

    header = (
        f"  hunk #{index}/{total} | "
        f"old [{hunk.old_start:#x}..{hunk.old_end:#x}) "
        f"({hunk.old_end - hunk.old_start} B) | "
        f"new [{hunk.new_start:#x}..{hunk.new_end:#x}) "
        f"({hunk.new_end - hunk.new_start} B) | ctx={context}"
    )
    lines.append(Color.cyan(header))

    changed_old: set[int] = set()
    changed_new: set[int] = set()
    for ch in hunk.changes:
        for i in range(ch.old_len):
            changed_old.add(ch.old_offset + i)
        for i in range(ch.new_len):
            changed_new.add(ch.new_offset + i)

    hex_w = width * 3 - 1
    lines.append(Color.dim(
        f"  {'OFFSET':<10}  {'OLD':<{hex_w}} | {'NEW':<{hex_w}}  "
        f"{'OLD ASCII':<{width}} | {'NEW ASCII':<{width}}"
    ))

    max_len = max(len(hunk.old_bytes), len(hunk.new_bytes))
    n_rows = (max_len + width - 1) // width if max_len else 0

    for row in range(n_rows):
        old_lo = row * width
        old_hi = min(old_lo + width, len(hunk.old_bytes))
        new_lo = row * width
        new_hi = min(new_lo + width, len(hunk.new_bytes))

        old_row = hunk.old_bytes[old_lo:old_hi]
        new_row = hunk.new_bytes[new_lo:new_hi]

        offset = hunk.old_start + old_lo
        offset_str = f"{offset:#010x}"

        old_hex = _render_hex_cell(old_row, hunk.old_start + old_lo,
                                   changed_old, width, "old")
        new_hex = _render_hex_cell(new_row, hunk.new_start + new_lo,
                                   changed_new, width, "new")
        old_asc = _render_ascii_cell(old_row, hunk.old_start + old_lo,
                                     changed_old, width, "old")
        new_asc = _render_ascii_cell(new_row, hunk.new_start + new_lo,
                                     changed_new, width, "new")

        lines.append(
            f"  {offset_str}  {old_hex:<{hex_w}} | {new_hex:<{hex_w}}  "
            f"{old_asc:<{width}} | {new_asc:<{width}}"
        )

        old_caret = _caret_row(old_row, hunk.old_start + old_lo,
                               changed_old, width)
        new_caret = _caret_row(new_row, hunk.new_start + new_lo,
                               changed_new, width)
        if old_caret or new_caret:
            lines.append(
                f"  {'':<10}  {old_caret:<{hex_w}} | {new_caret:<{hex_w}}"
            )

    if verbose:
        for ch in hunk.changes:
            for bl in _byte_diff_line(ch, old, new):
                lines.append(Color.dim(bl))

    return "\n".join(lines)


# =============================================================================
# CLI
# =============================================================================


def _build_parser() -> argparse.ArgumentParser:
    p = argparse.ArgumentParser(
        prog="bindiff",
        description=(
            "Compare specified binary files in the git working tree against HEAD, "
            "showing a side-by-side hex+ASCII diff with change locations."
        ),
    )
    p.add_argument(
        "files",
        nargs="+", metavar="FILE",
        help=(
            "Path(s) to binary files (absolute, or relative to CWD). "
            "The git repo is located by walking up from each path, so the "
            "current working directory does not need to be inside a repo."
        ),
    )
    p.add_argument("-c", "--context", type=int, default=8,
                   help="Bytes of equal context around each change (default: 8).")
    p.add_argument("-w", "--width", type=int, default=16,
                   help="Bytes per hex row (default: 16).")
    p.add_argument("-v", "--verbose", action="store_true",
                   help="Show per-byte old->new details under each hunk.")
    p.add_argument("--summary", action="store_true",
                   help="Only print per-file summaries, no hex dump.")
    p.add_argument("--no-color", action="store_true",
                   help="Disable ANSI color output.")
    p.add_argument("--color", action="store_true",
                   help="Force ANSI color output even when stdout is not a TTY.")
    p.add_argument("--force", action="store_true",
                   help="Skip the large-file safety check.")
    p.add_argument("--version", action="version",
                   version=f"bindiff {__version__}")
    return p


def _configure_color(args: argparse.Namespace) -> None:
    if args.no_color:
        Color.enabled = False
    elif args.color:
        Color.enabled = True


def _process_one(
    display: str,
    rel: str,
    abs_path: Path,
    root: Path,
    args: argparse.Namespace,
) -> int:
    old = head_blob(rel, root=root)
    new: bytes | None = None
    if abs_path.exists():
        new = abs_path.read_bytes()

    max_size = max(
        len(old) if old is not None else 0,
        len(new) if new is not None else 0,
    )
    if max_size > MAX_BYTES_DEFAULT and not args.force:
        raise FileTooLarge(
            f"{max_size} bytes exceeds the 50 MB safety limit; pass --force to override"
        )

    regions = [] if old is None or new is None else find_regions(old, new)
    hunks = ([] if args.summary or old is None or new is None
             else build_hunks(old, new, regions, args.context))

    print()
    print(format_file_header(display, old, new, regions, len(hunks)))

    if args.summary or old is None or new is None or not regions:
        return 0

    for i, h in enumerate(hunks, 1):
        print()
        print(format_hunk(h, i, len(hunks), args.width, args.context, old, new,
                          verbose=args.verbose))
    return 0


def main(argv: list[str] | None = None) -> int:
    parser = _build_parser()
    args = parser.parse_args(argv)
    _configure_color(args)

    rc = 0
    for f in args.files:
        try:
            p = Path(f)
            try:
                abs_path = p if p.is_absolute() else (Path.cwd() / p).resolve()
            except OSError as exc:
                print(f"bindiff: error: {f}: cannot resolve path ({exc})",
                      file=sys.stderr)
                rc = 1
                continue

            root = find_repo_for(abs_path)
            if root is None:
                print(f"bindiff: error: {f}: not inside any git working tree",
                      file=sys.stderr)
                rc = 1
                continue

            try:
                rel = to_repo_relative(abs_path, root)
            except OutsideRepo as exc:
                print(f"bindiff: error: {f}: {exc}", file=sys.stderr)
                rc = 1
                continue

            if not is_tracked(rel, root=root):
                print(f"bindiff: error: {f}: {rel} is not tracked by git",
                      file=sys.stderr)
                rc = 1
                continue

            try:
                rc |= _process_one(f, rel, abs_path, root, args)
            except BinDiffError as exc:
                print(f"bindiff: error: {f}: {exc}", file=sys.stderr)
                rc = 1
        except BinDiffError as exc:
            print(f"bindiff: error: {f}: {exc}", file=sys.stderr)
            rc = 1
    return rc


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