Add macho_diff package with CLI and helpers

macho_diff/__init__.py

@@ -0,0 +1 @@
+"""Mach-O diff tool."""

macho_diff/cli.py

@@ -0,0 +1,142 @@
+"""
+CLI module for the macho-diff tool.
+"""
+import argparse
+import os
+import sys
+
+import lief
+
+from macho_diff.helpers import (
+    parse_functions,
+    compare_instructions,
+    generate_batch_patches,
+)
+
+lief.logging.disable()
+
+
+def parse_args():
+    """Parse command line arguments."""
+    parser = argparse.ArgumentParser(description="Mach-O binary visual diff tool")
+    parser.add_argument("src", help="Path to the original/source Mach-O binary")
+    parser.add_argument("patched", help="Path to the patched Mach-O binary")
+    parser.add_argument(
+        "-b",
+        "--batch",
+        action="store_true",
+        help="Generate a batch patch file representing the differences.",
+    )
+    return parser.parse_args()
+
+
+def load_binary(filepath):
+    """Load a Mach-O binary from the given file path."""
+    if not os.path.isfile(filepath):
+        print(f"Error: File not found: '{filepath}'", file=sys.stderr)
+        return None
+    try:
+        fat = lief.MachO.parse(filepath)
+        return fat.at(0) if fat else None
+    except Exception as e:  # pylint: disable=broad-exception-caught
+        print(f"Error parsing binary '{filepath}': {e}", file=sys.stderr)
+        return None
+
+
+def read_file_raw(filepath):
+    """Read the raw bytes of a file."""
+    try:
+        with open(filepath, "rb") as f:
+            return f.read()
+    except OSError as e:
+        print(f"Error reading file '{filepath}': {e}", file=sys.stderr)
+        return None
+
+
+def process_diffs(args, binary_src, src_raw, functions_src, functions_patched):
+    """Process and print the differences between functions."""
+    diff_count = 0
+    all_batch_patches = []
+
+    for addr, src_data in functions_src.items():
+        patched_data = functions_patched.get(addr)
+        if not patched_data:
+            continue
+
+        if src_data["checksum"] != patched_data["checksum"]:
+            if args.batch:
+                patches = generate_batch_patches(
+                    binary_src, src_raw, addr, src_data["data"], patched_data["data"]
+                )
+                if patches:
+                    all_batch_patches.append((src_data["name"], patches))
+            else:
+                print(
+                    f"\n[DIFF] {src_data['name']} @ {hex(addr)} size={src_data['size']}"
+                )
+                compare_instructions(
+                    binary_src, addr, src_data["data"], patched_data["data"]
+                )
+            diff_count += 1
+
+    return diff_count, all_batch_patches
+
+
+def print_batch_patches(binary_src, all_batch_patches):
+    """Print the batch patches."""
+    cputype_name = binary_src.header.cpu_type.name
+    print("\nBatch Patches:" + "\n" + "=" * 40)
+    print(f"{cputype_name}:\n")
+
+    for _, patches in all_batch_patches:
+        for find_str, replace_str in patches:
+            print(find_str)
+            print("to")
+            print(replace_str)
+            print()
+    print("=" * 40)
+
+
+def main():
+    """Main entry point for the CLI."""
+    args = parse_args()
+
+    print(f"Loading '{args.src}'...")
+    binary_src = load_binary(args.src)
+    if not binary_src:
+        return
+
+    print(f"Loading '{args.patched}'...")
+    binary_patched = load_binary(args.patched)
+    if not binary_patched:
+        return
+
+    print("Reading file contents...")
+    src_raw = read_file_raw(args.src)
+    if not src_raw:
+        return
+    patched_raw = read_file_raw(args.patched)
+    if not patched_raw:
+        return
+
+    print("Parsing functions (this might take a moment)...")
+    functions_src = parse_functions(binary_src, src_raw)
+    functions_patched = parse_functions(binary_patched, patched_raw)
+
+    print(
+        f"Found {len(functions_src)} functions in source binary "
+        f"and {len(functions_patched)} in patched binary."
+    )
+
+    diff_count, all_batch_patches = process_diffs(
+        args, binary_src, src_raw, functions_src, functions_patched
+    )
+
+    if args.batch:
+        print_batch_patches(binary_src, all_batch_patches)
+
+    print(f"\nTotal diffs: {diff_count}")
+
+
+if __name__ == "__main__":
+    main()

macho_diff/helpers.py

@@ -0,0 +1,206 @@
+"""
+Helper functions for macho-diff tool.
+"""
+import hashlib
+import capstone
+
+
+def va_to_file_offset(binary, va):
+    """Convert a virtual address to a file offset."""
+    for segment in binary.segments:
+        start = segment.virtual_address
+        end = start + segment.virtual_size
+
+        if start <= va < end:
+            delta = va - start
+            return segment.file_offset + delta
+    return None
+
+
+def get_function_bytes(binary, raw_data, address, size):
+    """Retrieve the raw bytes for a function given its virtual address and size."""
+    if size <= 0:
+        return b""
+    file_offset = va_to_file_offset(binary, address)
+    if file_offset is None:
+        return b""
+    end = file_offset + size
+    if end > len(raw_data):
+        size = len(raw_data) - file_offset
+    if size <= 0:
+        return b""
+    return raw_data[file_offset : file_offset + size]
+
+
+def checksum(data):
+    """Calculate the SHA256 checksum of the given data."""
+    return hashlib.sha256(data).hexdigest()
+
+
+def parse_functions(binary, raw_data):
+    """Parse and extract functions from a Mach-O binary."""
+    funcs = sorted(binary.functions, key=lambda f: f.address)
+    out = {}
+    for i in range(len(funcs) - 1):
+        cur, nxt = funcs[i], funcs[i + 1]
+        addr, next_addr = cur.address, nxt.address
+        if next_addr <= addr:
+            continue
+        size = next_addr - addr
+        if size > 0x100000:
+            continue
+        name = cur.name or f"sub_{addr:x}"
+
+        data = get_function_bytes(binary, raw_data, addr, size)
+        if not data:
+            continue
+        out[addr] = {
+            "name": name,
+            "size": size,
+            "data": data,
+            "start": hex(addr),
+            "end": hex(next_addr - 4),
+            "checksum": checksum(data),
+        }
+    return out
+
+
+def find_nearest_function(functions, target_addr):
+    """Find the function that encompasses or immediately precedes the target address."""
+    closest_addr = None
+    for addr in sorted(functions.keys()):
+        if addr <= target_addr:
+            closest_addr = addr
+        else:
+            break
+
+    if closest_addr is not None:
+        return closest_addr, functions[closest_addr]
+    return None, None
+
+
+def get_capstone_context(binary):
+    """Initialize a capstone disassembler context for the binary architecture."""
+    cputype = binary.header.cpu_type
+    if cputype.name == "ARM64":
+        arch = capstone.CS_ARCH_ARM64
+        mode = capstone.CS_MODE_ARM
+    elif cputype.name == "ARM":
+        arch = capstone.CS_ARCH_ARM
+        mode = capstone.CS_MODE_ARM
+    elif cputype.name == "X86_64":
+        arch = capstone.CS_ARCH_X86
+        mode = capstone.CS_MODE_64
+    elif cputype.name == "X86":
+        arch = capstone.CS_ARCH_X86
+        mode = capstone.CS_MODE_32
+    else:
+        raise ValueError(f"Unsupported CPU type: {cputype.name}")
+
+    return capstone.Cs(arch, mode)
+
+
+def _get_diff_blocks(src_insns, patched_insns):
+    """Identify contiguous blocks of differing instructions."""
+    max_len = max(len(src_insns), len(patched_insns))
+    blocks = []
+    current_block = []
+
+    for i in range(max_len):
+        src_i = src_insns[i] if i < len(src_insns) else None
+        patch_i = patched_insns[i] if i < len(patched_insns) else None
+
+        src_str = f"{src_i.mnemonic} {src_i.op_str}" if src_i else "<none>"
+        patch_str = f"{patch_i.mnemonic} {patch_i.op_str}" if patch_i else "<none>"
+
+        if src_str != patch_str:
+            current_block.append(i)
+        else:
+            if current_block:
+                blocks.append((current_block[0], current_block[-1]))
+                current_block = []
+
+    if current_block:
+        blocks.append((current_block[0], current_block[-1]))
+
+    return blocks
+
+
+def _expand_and_format_patches(src_raw, src_insns, patched_insns, blocks):
+    """Expand each differing block until unique, and return formatted hex patches."""
+    # pylint: disable=too-many-locals
+    patches = []
+    for start_idx, end_idx in blocks:
+        curr_start = start_idx
+        curr_end = end_idx
+
+        while True:
+            find_bytes = b"".join(
+                bytes(src_insns[j].bytes)
+                for j in range(curr_start, curr_end + 1)
+                if j < len(src_insns)
+            )
+
+            if src_raw.count(find_bytes) <= 1:
+                break
+
+            can_expand_left = curr_start > 0
+            can_expand_right = (
+                curr_end < len(src_insns) - 1 and curr_end < len(patched_insns) - 1
+            )
+
+            if not can_expand_left and not can_expand_right:
+                break
+
+            if can_expand_left:
+                curr_start -= 1
+            elif can_expand_right:
+                curr_end += 1
+
+        find_hex = []
+        replace_hex = []
+        for j in range(curr_start, curr_end + 1):
+            src_i = src_insns[j] if j < len(src_insns) else None
+            patch_i = patched_insns[j] if j < len(patched_insns) else None
+            if src_i:
+                find_hex.extend(f"{b:02X}" for b in src_i.bytes)
+            if patch_i:
+                replace_hex.extend(f"{b:02X}" for b in patch_i.bytes)
+        patches.append((" ".join(find_hex), " ".join(replace_hex)))
+
+    return patches
+
+
+def generate_batch_patches(binary, src_raw, addr, src_bytes, patched_bytes):
+    """Generate batch patches comparing two sets of function bytes."""
+    md = get_capstone_context(binary)
+    src_insns = list(md.disasm(src_bytes, addr))
+    patched_insns = list(md.disasm(patched_bytes, addr))
+
+    blocks = _get_diff_blocks(src_insns, patched_insns)
+    return _expand_and_format_patches(src_raw, src_insns, patched_insns, blocks)
+
+
+def compare_instructions(binary, addr, src_bytes, patched_bytes):
+    """Print the differences between two instruction streams."""
+    md = get_capstone_context(binary)
+    src_insns = list(md.disasm(src_bytes, addr))
+    patched_insns = list(md.disasm(patched_bytes, addr))
+
+    max_len = max(len(src_insns), len(patched_insns))
+    last_diff_idx = None
+    for i in range(max_len):
+        src_i = src_insns[i] if i < len(src_insns) else None
+        patch_i = patched_insns[i] if i < len(patched_insns) else None
+
+        src_str = f"{src_i.mnemonic} {src_i.op_str}" if src_i else "<none>"
+        patch_str = f"{patch_i.mnemonic} {patch_i.op_str}" if patch_i else "<none>"
+
+        if src_str != patch_str:
+            if last_diff_idx is not None and i > last_diff_idx + 1:
+                print("  ...")
+            curr_addr = (
+                src_i.address if src_i else (patch_i.address if patch_i else addr)
+            )
+            print(f"  {hex(curr_addr)}: {src_str:<35} | patch: {patch_str}")
+            last_diff_idx = i