rea/bridge/hopper_bridge.py

"""Authenticated REA adapter executed on Hopper's dedicated Python thread.

The bootstrap injects ``REA_SOCKET`` and a random ``REA_TOKEN`` before executing
this file with Hopper's supported ``--python`` launcher option. Keep all Hopper
API access on this thread: moving dispatch to a worker can deadlock Hopper.
"""

import json
import hmac
import os
import socket
import struct
from typing import Any, Optional, Protocol, Sequence

BAD_ADDRESSES = (-1, 0xFFFFFFFFFFFFFFFF, None)
# The native label setter truncates NSString names at this UTF-16 extent.
HOPPER_SYMBOL_NAME_UTF16_UNITS = 1024
_rea_session_document = None
_search_inventory_cache = {}
_pseudocode_cache = {}
_hopper_api = None


class HopperDocument(Protocol):
    """Typed minimum used by the provider-neutral Hopper API facade."""

    def getDocumentName(self) -> str: ...

    def getExecutableFilePath(self) -> Optional[str]: ...

    def getDatabaseFilePath(self) -> Optional[str]: ...

    def backgroundProcessActive(self) -> bool: ...


class HopperDocumentProvider(Protocol):
    """Hopper's injected global document provider."""

    @classmethod
    def getAllDocuments(cls) -> Sequence[HopperDocument]: ...

    @classmethod
    def getCurrentDocument(cls) -> Optional[HopperDocument]: ...


class HopperApiFacade:
    """Small injected boundary around Hopper globals for standalone tests."""

    def __init__(self, document_provider: HopperDocumentProvider):
        self._document_provider = document_provider

    def documents(self) -> Sequence[HopperDocument]:
        return self._document_provider.getAllDocuments()

    def current_document(self) -> Optional[HopperDocument]:
        return self._document_provider.getCurrentDocument()

    def require_analysis_complete(
        self, document: HopperDocument, operation: str
    ) -> None:
        if document.backgroundProcessActive():
            raise CapabilityUnavailableError(
                "%s requires completed Hopper background analysis" % operation
            )


def _configure_hopper_api(document_provider):
    """Inject Hopper's document provider or a standalone test facade."""
    global _hopper_api, _rea_session_document
    _hopper_api = HopperApiFacade(document_provider)
    _rea_session_document = None


def _api():
    """Resolve Hopper globals lazily so importing this module needs no Hopper."""
    global _hopper_api
    if _hopper_api is not None:
        return _hopper_api
    document_provider = globals().get("Document")
    if document_provider is None:
        raise CapabilityUnavailableError(
            "Hopper Document API is unavailable outside the Hopper runtime"
        )
    _hopper_api = HopperApiFacade(document_provider)
    return _hopper_api


def _session_document():
    """Return the exact document bound to this authenticated REA session."""
    global _rea_session_document
    documents = _api().documents()
    if _rea_session_document is not None:
        # Hopper recreates Python wrappers; equality compares native handles.
        # Retain our bound wrapper so cache identity stays stable across calls.
        return (
            _rea_session_document
            if any(document == _rea_session_document for document in documents)
            else None
        )
    return _bind_session_document(documents)


def _bind_session_document(documents=None):
    """Bind this bridge to Hopper's current document for its target."""
    global _rea_session_document
    api = _api()
    if documents is None:
        documents = api.documents()
    target = os.path.realpath(REA_TARGET_PATH)

    def matches_target(document):
        paths = (document.getExecutableFilePath(), document.getDatabaseFilePath())
        return any(path and os.path.realpath(path) == target for path in paths)

    current = api.current_document()
    if current is not None and matches_target(current):
        _rea_session_document = current
        return current
    matching = [document for document in documents if matches_target(document)]
    if len(matching) == 1:
        _rea_session_document = matching[0]
        return matching[0]
    if len(matching) > 1:
        raise CapabilityUnavailableError(
            "The REA session document is ambiguous; Hopper did not identify the launched document"
        )
    return None


class CapabilityUnavailableError(Exception):
    """The active Hopper build lacks one admitted public API operation."""


class InvalidRequestError(Exception):
    """Caller input failed bridge request or operation validation."""


EXHAUSTIVE_ANALYSIS_METHODS = frozenset(
    (
        "analyze_function",
        "list_names",
        "list_procedures",
        "list_strings",
        "procedure_address",
        "procedure_assembly",
        "procedure_callees",
        "procedure_callers",
        "procedure_info",
        "procedure_pseudo_code",
        "procedure_references",
        "read_function_instructions",
        "resolve_containing_procedure",
        "search_procedures",
        "search_strings",
    )
)


def _hex(value):
    if not isinstance(value, int) or isinstance(value, bool):
        try:
            value = int(str(value), 16)
        except ValueError as error:
            raise ValueError("Hopper returned a non-hexadecimal address") from error
    return "0x%x" % value


def _json_safe(value):
    """Project Hopper-specific Python values into the JSON protocol boundary."""
    if value is None or isinstance(value, (bool, int, float, str)):
        return value
    if isinstance(value, (list, tuple)):
        return [_json_safe(item) for item in value]
    if isinstance(value, dict):
        return {str(key): _json_safe(item) for key, item in value.items()}
    return str(value)


def _document(name=None):
    """Keep observations and mutations bound to the active REA target."""
    documents = _api().documents()
    session_document = _session_document()
    if session_document is None:
        raise CapabilityUnavailableError("The active REA target's Hopper document is no longer open; reopen the target with open_binary")
    if name is not None:
        # A document name is display metadata; scope it to the bound native
        # handle even when another open target happens to have the same name.
        if session_document.getDocumentName() == name:
            return session_document
        if any(candidate.getDocumentName() == name for candidate in documents):
            raise InvalidRequestError("Hopper document belongs to another target; use open_binary to select that target")
        raise InvalidRequestError("Unknown Hopper document")
    return session_document


def _api_text(value, field):
    """Validate text before passing it to Hopper's native UTF-8 string API."""
    if not isinstance(value, str):
        raise InvalidRequestError("%s must be a string" % field)
    if "\0" in value:
        raise InvalidRequestError("%s contains a NUL character, which Hopper's Python API cannot represent" % field)
    try:
        value.encode("utf-8")
    except UnicodeEncodeError as error:
        raise InvalidRequestError("%s contains an unpaired Unicode surrogate, which Hopper's Python API cannot represent" % field) from error
    return value


def _validate_name_targets(document, names):
    """Keep Hopper's unique-label moves within the explicitly selected addresses."""
    targets = {address for _, address, _ in names}
    if len(targets) != len(names):
        raise InvalidRequestError("Batch rename selects the same address more than once; use one canonical address per target")
    labels = set()
    for _, address, name in names:
        _segment(document, address)
        if not name:
            continue
        if name in labels:
            raise InvalidRequestError("Batch rename assigns the same name more than once: %s" % name)
        labels.add(name)
        owner = document.getAddressForName(name)
        if owner not in BAD_ADDRESSES and owner != address and owner not in targets:
            raise InvalidRequestError("Name %s is already assigned to %s; include that address in the batch with a replacement name" % (name, _hex(owner)))


def _symbol_name(value, field):
    """Reject native label truncation without altering the caller's name."""
    value = _api_text(value, field)
    if len(value.encode("utf-16-le")) // 2 > HOPPER_SYMBOL_NAME_UTF16_UNITS:
        raise InvalidRequestError("%s exceeds Hopper's %s UTF-16 code-unit symbol-name limit" % (field, HOPPER_SYMBOL_NAME_UTF16_UNITS))
    return value


def _name_matches(document, address, name):
    observed = _segment(document, address).getNameAtAddress(address)
    return observed == name or (not name and observed is None)


def _address(document, value=None):
    """Keep explicit hexadecimal coordinates distinct from symbol names."""
    if value is None:
        return document.getCurrentAddress()
    _api_text(value, "Address")
    if value.lower().startswith("0x"):
        try:
            address = int(value, 16)
        except ValueError as error:
            raise InvalidRequestError("Invalid hexadecimal Hopper address: %s" % value) from error
    else:
        # Names such as `add` and `face` are valid hexadecimal strings too.
        # Prefer an observed symbol, retaining bare-hex compatibility only
        # when Hopper has no symbol with the supplied name.
        address = document.getAddressForName(value)
        if address in BAD_ADDRESSES:
            try:
                address = int(value, 16)
            except ValueError as error:
                raise InvalidRequestError("Unknown Hopper address or name: %s" % value) from error
    if not 0 <= address < 0xFFFFFFFFFFFFFFFF:
        raise InvalidRequestError("Address is outside Hopper's usable unsigned 64-bit range: %s" % value)
    return address


def _segment(document, address):
    result = document.getSegmentAtAddress(address)
    if result is None:
        raise InvalidRequestError("Address is outside every segment")
    return result


def _procedure(document, value=None):
    address = document.getCurrentAddress() if value is None else _address(document, value)
    result = _segment(document, address).getProcedureAtAddress(address)
    if result is None:
        raise InvalidRequestError("No procedure exists at the requested address")
    return result


def _original_file_offset(document, offset):
    """Translate Hopper's image-relative mapping using the observed loaded header."""
    if offset > 9007199254740991:
        raise CapabilityUnavailableError("Provider file offset exceeds the exact JSON integer range")
    path = globals().get("REA_ORIGINAL_EXECUTABLE_PATH") or document.getExecutableFilePath()
    if not path:
        raise CapabilityUnavailableError("Hopper did not identify the original executable for file-offset verification")
    try:
        with open(path, "rb") as source:
            size = os.fstat(source.fileno()).st_size
            header = source.read(8)
            formats = {
                b"\xca\xfe\xba\xbe": (">", False),
                b"\xbe\xba\xfe\xca": ("<", False),
                b"\xca\xfe\xba\xbf": (">", True),
                b"\xbf\xba\xfe\xca": ("<", True),
            }
            fat = formats.get(header[:4])
            if fat is None:
                if offset >= size:
                    raise InvalidRequestError("Mapped offset is outside the original executable file at %s" % path)
                return offset, path, 0
            if len(header) != 8:
                raise CapabilityUnavailableError("Original FAT executable has a truncated header at %s" % path)
            byte_order, wide = fat
            count = struct.unpack(byte_order + "I", header[4:])[0]
            stride = 32 if wide else 20
            table_end = 8 + count * stride
            if count == 0 or table_end > size:
                raise CapabilityUnavailableError("Original FAT executable has an invalid architecture table at %s" % path)
            header_address = document.getAddressFromFileOffset(0)
            reader = getattr(document, "readBytes", None)
            if header_address in BAD_ADDRESSES or document.getSegmentAtAddress(header_address) is None or not callable(reader):
                raise CapabilityUnavailableError("Hopper did not expose a readable loaded Mach-O header to identify its FAT slice")
            loaded = reader(header_address, 12)
            if not isinstance(loaded, (bytes, bytearray)) or len(loaded) != 12:
                raise CapabilityUnavailableError("Hopper did not expose the loaded Mach-O header to identify its FAT slice")
            loaded_order = {b"\xce\xfa\xed\xfe": "<", b"\xcf\xfa\xed\xfe": "<", b"\xfe\xed\xfa\xce": ">", b"\xfe\xed\xfa\xcf": ">"}.get(bytes(loaded[:4]))
            if loaded_order is None:
                raise CapabilityUnavailableError("Hopper's loaded header is not a supported Mach-O image")
            identity = struct.unpack(loaded_order + "II", loaded[4:12])
            selected = None
            for index in range(count):
                source.seek(8 + index * stride)
                entry = source.read(stride)
                if len(entry) != stride:
                    raise CapabilityUnavailableError("Original FAT executable architecture table changed during verification")
                cpu, subtype, start, length, alignment = struct.unpack(byte_order + ("IIQQI" if wide else "IIIII"), entry[:28] if wide else entry)
                if length < 12 or start < table_end or start > size or length > size - start or alignment > 52 or start % (1 << alignment):
                    raise CapabilityUnavailableError("Original FAT executable has an invalid slice range at index %s" % index)
                if (cpu, subtype) != identity:
                    continue
                source.seek(start)
                if source.read(12) != bytes(loaded):
                    raise CapabilityUnavailableError("FAT slice declaration disagrees with Hopper's loaded header")
                if selected is not None:
                    raise CapabilityUnavailableError("Multiple FAT slices match Hopper's loaded header; original-file mapping is ambiguous")
                selected = (start, length)
            if selected is None:
                raise CapabilityUnavailableError("No original FAT slice matches Hopper's loaded header")
            start, length = selected
            if offset >= length:
                raise InvalidRequestError("Mapped offset is outside Hopper's loaded FAT slice")
            if start + offset > 9007199254740991:
                raise CapabilityUnavailableError("Original file offset exceeds the exact JSON integer range")
            return start + offset, path, start
    except OSError as error:
        raise CapabilityUnavailableError("Cannot read the original executable for file-offset verification at %s: %s" % (path, error)) from error


def _procedure_name(procedure):
    entry = procedure.getEntryPoint()
    return procedure.getSegment().getNameAtAddress(entry) or _hex(entry)


def _procedure_identity(procedure):
    return {
        "address": _hex(procedure.getEntryPoint()),
        "name": _procedure_name(procedure),
        "classification": None,
        "body": {
            "available": False,
            "reason": "Hopper's public Python API does not expose complete function body ranges",
        },
    }


def _procedure_locals(procedure):
    """Preserve public local-variable names and observed stack displacements."""
    result = []
    for local in procedure.getLocalVariableList():
        name = local.name()
        displacement = local.displacement()
        result.append({
            "description": "%s (stack displacement %s)" % (name if name is not None else "Unnamed local", displacement),
            "name": name,
            "stack_displacement": str(displacement) if displacement is not None else None,
            "provenance": "hopper-public-python-api",
        })
    return result


def _containing_procedure(document, address):
    segment = document.getSegmentAtAddress(address)
    if segment is None:
        return None, "outside_segments"
    procedure = segment.getProcedureAtAddress(address)
    return (procedure, None) if procedure is not None else (None, "not_in_procedure")


def _basic_block_end(procedure, block):
    """Normalize native block endpoints to an exclusive byte address."""
    end = block.getEndingAddress()
    # Hopper 6.1 returns the final instruction address, despite the Python
    # documentation describing an exclusive end. Membership distinguishes
    # that representation from an actual exclusive endpoint in other builds.
    if procedure.getBasicBlockAtAddress(end) == block:
        instruction = procedure.getSegment().getInstructionAtAddress(end)
        if instruction is not None and instruction.getInstructionLength() > 0:
            return end + instruction.getInstructionLength()
    return end


def _instruction_addresses(procedure):
    result = []
    seen = set()
    segment = procedure.getSegment()
    for block in procedure.basicBlockIterator():
        address = block.getStartingAddress()
        end = _basic_block_end(procedure, block)
        while address < end and address not in seen:
            seen.add(address)
            instruction = segment.getInstructionAtAddress(address)
            if instruction is None:
                break
            result.append(address)
            length = instruction.getInstructionLength()
            if length <= 0:
                break
            address += length
    return result


def _native_calls(procedure, direction):
    references = procedure.getAllCallees() if direction == "outgoing" else procedure.getAllCallers()
    kinds = {0: "none", 1: "unknown", 2: "direct", 3: "objective_c"}
    calls = {}
    unresolved = []
    for reference in references:
        source, target, kind = reference.fromAddress(), reference.toAddress(), reference.type()
        if source in BAD_ADDRESSES:
            continue
        if target in BAD_ADDRESSES:
            unresolved.append({"address": _hex(source), "reason": "Native CallReference has no resolved target (type %s)" % kind})
            continue
        calls[(source, target)] = {"classification": kinds.get(kind, "unknown"), "provider_type": kind, "provenance": "hopper-public-python-api:CallReference"}
    return calls, sorted(unresolved, key=lambda item: int(item["address"], 16))


def _procedure_references(document, params):
    procedure = _procedure(document, params.get("procedure"))
    direction = params.get("direction", "outgoing")
    if direction not in ("incoming", "outgoing"):
        raise InvalidRequestError("direction must be incoming or outgoing")
    addresses = _instruction_addresses(procedure)
    edges = set()
    for address in addresses:
        segment = _segment(document, address)
        references = segment.getReferencesFromAddress(address) if direction == "outgoing" else segment.getReferencesOfAddress(address)
        for reference in references:
            edges.add((address, reference) if direction == "outgoing" else (reference, address))
    calls, unresolved = _native_calls(procedure, direction)
    edges.update(calls)
    ordered = sorted(edges)
    items = []
    for source, target in ordered:
        source_procedure, _ = _containing_procedure(document, source)
        target_procedure, _ = _containing_procedure(document, target)
        items.append({
            "source_address": _hex(source),
            "target_address": _hex(target),
            "source_procedure": _procedure_identity(source_procedure) if source_procedure is not None else None,
            "target_procedure": _procedure_identity(target_procedure) if target_procedure is not None else None,
            "kind": _unavailable("Hopper exposes partial CallReference classification, but not detailed reference flags"),
            **({"call": calls[(source, target)]} if (source, target) in calls else {}),
        })
    return {
        "procedure": _procedure_identity(procedure), "direction": direction,
        "reference_kinds_available": False,
        # These are native CallReference objects, not proof that every unresolved
        # dispatch site has been enumerated by Hopper.
        "unresolved_calls": unresolved,
        "references": items,
    }


def _procedure_map(document):
    result = {}
    for segment in document.getSegmentsList():
        for index in range(segment.getProcedureCount()):
            procedure = segment.getProcedureAtIndex(index)
            result[_hex(procedure.getEntryPoint())] = _procedure_name(procedure)
    return result


def _strings(document):
    return {address: record["value"] for address, record in _string_inventory(document).items()}


def _string_record(segment, address, display):
    """Read one native typed object, retaining its shortened display separately."""
    length = segment.getObjectLength(address)
    end = segment.getStartingAddress() + segment.getLength()
    if isinstance(length, bool) or not isinstance(length, int) or length <= 0 or length > end - address:
        raise CapabilityUnavailableError("Hopper reported an invalid string object extent at %s" % _hex(address))
    raw = segment.readBytes(address, length)
    if not isinstance(raw, (bytes, bytearray)) or len(raw) != length:
        raise CapabilityUnavailableError("Hopper could not read the complete string object at %s" % _hex(address))
    kind = segment.getTypeAtAddress(address)
    encodings = ("utf-8", "latin-1") if kind == segment.TYPE_ASCII else ("utf-16-le", "utf-16-be") if kind == segment.TYPE_UNICODE else ()
    # Native display escaping is not invertible (a literal backslash-n and a
    # newline can render alike). Decode byte strings directly; for UTF-16,
    # compare rendered candidates to establish an unambiguous byte order.
    candidates = []
    for encoding in encodings:
        terminator = b"\x00\x00" if encoding.startswith("utf-16") else b"\x00"
        terminated = raw.endswith(terminator)
        payload = raw[:-len(terminator)] if terminated else raw
        try:
            value = payload.decode(encoding)
        except UnicodeDecodeError:
            continue
        record = {"value": value, "provider_value": display, "string": {"encoding": encoding, "encoding_status": "inferred", "termination": "present_or_not_required" if terminated else "missing", "byte_length": length}}
        if kind == segment.TYPE_ASCII:
            return record
        rendered = value.replace('"', '\\"').replace("\n", "\\n").replace("\r", "\\r").replace("\t", "\\t")
        if rendered == display or (display.endswith("\u2026") and display[:-1] and rendered.startswith(display[:-1])):
            candidates.append(record)
    if candidates and len({candidate["value"] for candidate in candidates}) == 1:
        return candidates[0]
    raise CapabilityUnavailableError("Hopper's string display cannot be reconciled with its typed bytes at %s; inspect the bytes and encoding explicitly" % _hex(address))


def _string_inventory(document, addresses=None):
    key = (id(document), "string_objects")
    objects = _search_inventory_cache.get(key)
    if objects is None:
        objects = {address: (segment, display) for segment in document.getSegmentsList() for display, address in segment.getStringsList()}
        _search_inventory_cache[key] = objects
    key = (id(document), "string_records")
    records = _search_inventory_cache.setdefault(key, {})
    selected = sorted(objects if addresses is None else objects.keys() & addresses)
    for address in selected:
        name = _hex(address)
        if name not in records:
            segment, display = objects[address]
            try:
                records[name] = _string_record(segment, address, display)
            except CapabilityUnavailableError as error:
                records[name] = {"value": display, "provider_value": display, "decoding": _unavailable(str(error))}
    return {_hex(address): records[_hex(address)] for address in selected}


def _invalidate_search_inventory(document):
    """Discard derived names after analysis metadata changes."""
    document_id = id(document)
    for key in list(_search_inventory_cache):
        if key[0] == document_id:
            del _search_inventory_cache[key]


def _invalidate_pseudocode(document):
    """Discard cached decompilation after an analysis annotation changes."""
    document_id = id(document)
    for key in list(_pseudocode_cache):
        if key[0] == document_id:
            del _pseudocode_cache[key]


def _pseudocode(document, procedure):
    """Reuse one provider decompilation for repeated bounded projections."""
    key = (id(document), procedure.getEntryPoint())
    if key not in _pseudocode_cache:
        _pseudocode_cache[key] = procedure.decompile()
    return _pseudocode_cache[key]


def _search_inventory(document, kind):
    """Cache an immutable, address-sorted inventory for an unchanged document."""
    key = (id(document), kind)
    inventory = _search_inventory_cache.get(key)
    if inventory is None:
        if kind == "procedure":
            values = _procedure_map(document)
        elif kind == "string":
            values = _strings(document)
        elif kind == "name":
            values = {_hex(address): value for address, value in _name_map(document).items()}
        else:
            raise ValueError("Unknown inventory kind")
        inventory = tuple(sorted(values.items(), key=lambda item: int(item[0], 16)))
        _search_inventory_cache[key] = inventory
    return inventory


def _search_results(document, kind, params):
    pattern = params.get("pattern")
    if not isinstance(pattern, str) or not pattern:
        raise InvalidRequestError("pattern must be a non-empty string")
    mode = params.get("mode", "literal")
    if mode not in ("literal", "regex"):
        raise InvalidRequestError("mode must be literal or regex")
    case_sensitive = params.get("case_sensitive", False)
    if not isinstance(case_sensitive, bool):
        raise InvalidRequestError("case_sensitive must be a boolean")

    if mode == "literal":
        needle = pattern if case_sensitive else pattern.casefold()
        matches = lambda value: needle in (value if case_sensitive else value.casefold())
    else:
        raise CapabilityUnavailableError("Regex matching requires REA's supervised worker; the native bridge only executes literal searches")

    selected = []
    for item in _search_inventory(document, kind):
        if not matches(item[1]):
            continue
        selected.append(item)
    return [{"address": address, **_string_inventory(document)[address]} for address, _ in selected] if kind == "string" else [
            {
                "address": address,
                "value": value,
            }
            for address, value in selected
        ]


def _unavailable(reason):
    """Describe evidence the public Hopper API cannot truthfully provide."""
    return {"available": False, "reason": reason}


def _name_map(document):
    result = {}
    for segment in document.getSegmentsList():
        for address in segment.getNamedAddresses():
            name = segment.getNameAtAddress(address)
            if name is not None:
                result[address] = name
    return result


def _render_instruction(segment, address):
    instruction = segment.getInstructionAtAddress(address)
    if instruction is None:
        return None
    arguments = [
        instruction.getFormattedArgument(index)
        for index in range(instruction.getArgumentCount())
    ]
    suffix = ", ".join(value for value in arguments if value is not None)
    return "%s: %s%s" % (
        _hex(address),
        instruction.getInstructionString(),
        (" " + suffix) if suffix else "",
    )


def _assembly(procedure):
    """Render assembly while guarding against malformed instruction cycles."""
    segment = procedure.getSegment()
    return "\n".join(
        line for address in _instruction_addresses(procedure)
        if (line := _render_instruction(segment, address)) is not None
    )


def _read_function_instructions(document, params):
    """Read all raw instructions without invoking decompilation."""
    procedure = _procedure(document, params.get("procedure"))
    addresses = _instruction_addresses(procedure)
    segment = procedure.getSegment()
    items = []
    for address in addresses:
        line = _render_instruction(segment, address)
        if line is not None:
            items.append(line)
    return {
        "procedure": _procedure_identity(procedure),
        "instructions": items,
        "limitations": [
            "Instruction text and ordering are Hopper-specific representations.",
            "The fast path does not decompile the procedure or scan whole-program names and strings.",
        ],
    }


def _analyze_function(document, params):
    """Collect the complete function dossier for agent callers."""
    procedure = _procedure(document, params.get("procedure"))
    addresses = _instruction_addresses(procedure)
    blocks = []
    all_blocks = list(procedure.basicBlockIterator())
    for block in all_blocks:
        successors = []
        for index in range(block.getSuccessorCount()):
            successor = block.getSuccessorAddressAtIndex(index)
            if successor not in BAD_ADDRESSES:
                successors.append(_hex(successor))
        blocks.append({
            "start": _hex(block.getStartingAddress()),
            "end": _hex(_basic_block_end(procedure, block)),
            "successors": sorted(set(successors), key=lambda value: int(value, 16)),
        })
    pseudo = _pseudocode(document, procedure) or ""
    assembly_lines = _assembly(procedure).splitlines()
    callers = sorted((_procedure_identity(item) for item in procedure.getAllCallerProcedures()), key=lambda item: int(item["address"], 16))
    callees = sorted((_procedure_identity(item) for item in procedure.getAllCalleeProcedures()), key=lambda item: int(item["address"], 16))
    comments = []
    edges = set()
    for address in addresses:
        segment = _segment(document, address)
        comment = segment.getCommentAtAddress(address)
        inline_comment = segment.getInlineCommentAtAddress(address)
        if comment:
            comments.append({"address": _hex(address), "kind": "comment", "text": comment})
        if inline_comment:
            comments.append({"address": _hex(address), "kind": "inline", "text": inline_comment})
        for target in segment.getReferencesFromAddress(address):
            edges.add((address, target))
        for source in segment.getReferencesOfAddress(address):
            edges.add((source, address))
    incoming = []
    outgoing = []
    procedure_addresses = set(addresses)
    outgoing_calls, unresolved_calls = _native_calls(procedure, "outgoing")
    incoming_calls, _ = _native_calls(procedure, "incoming")
    calls = {**incoming_calls, **outgoing_calls}
    edges.update(calls)
    for source, target in sorted(edges):
        source_procedure, _ = _containing_procedure(document, source)
        target_procedure, _ = _containing_procedure(document, target)
        item = {
            "source_address": _hex(source),
            "target_address": _hex(target),
            "source_procedure": _procedure_identity(source_procedure) if source_procedure is not None else None,
            "target_procedure": _procedure_identity(target_procedure) if target_procedure is not None else None,
            "kind": _unavailable("Hopper exposes partial CallReference classification, but not detailed reference flags"),
            **({"call": calls[(source, target)]} if (source, target) in calls else {}),
        }
        if target in procedure_addresses and source not in procedure_addresses:
            incoming.append(item)
        if source in procedure_addresses:
            outgoing.append(item)
    string_map = {int(address, 16): record for address, record in _string_inventory(document, {int(edge["target_address"], 16) for edge in outgoing}).items()}
    name_map = {
        int(address, 16): value
        for address, value in _search_inventory(document, "name")
    }
    referenced_strings = []
    referenced_names = []
    for edge in outgoing:
        target = int(edge["target_address"], 16)
        if target in string_map:
            referenced_strings.append({"address": edge["target_address"], **string_map[target], "source_address": edge["source_address"]})
        if target in name_map:
            referenced_names.append({"address": edge["target_address"], "value": name_map[target], "source_address": edge["source_address"]})
    comments.sort(key=lambda item: (int(item["address"], 16), item["kind"]))
    referenced_strings.sort(key=lambda item: (int(item["address"], 16), int(item["source_address"], 16)))
    referenced_names.sort(key=lambda item: (int(item["address"], 16), int(item["source_address"], 16)))
    return {
        "procedure": {
            **_procedure_identity(procedure),
            "signature": procedure.signatureString(),
            "locals": _procedure_locals(procedure),
        },
        "pseudocode": pseudo,
        "assembly": assembly_lines,
        "comments": comments,
        "callers": callers, "callees": callees,
        "incoming_references": incoming,
        "outgoing_references": outgoing,
        "unresolved_calls": unresolved_calls,
        "referenced_strings": referenced_strings,
        "referenced_names": referenced_names,
        "basic_blocks": blocks,
        "limitations": [
            "Native CallReference classifications are observed where available; detailed reference flags remain unknown.",
            "Hopper's public Python API does not expose equivalent external or thunk classification in this dossier.",
            "Native calls without resolved target addresses are not represented as edges; enumeration beyond reported CallReference objects is unknown.",
            "Pseudocode and assembly are provider-specific representations, not original source.",
        ],
    }


def _dispatch(method, params):
    """Dispatch only the closed operation set implemented by REA's public tools."""
    if method == "health":
        return {"name": "REA Hopper bridge", "version": "1.0.0", "run_id": REA_RUN_ID}
    if method in ("shutdown", "shutdown_document"):
        document = _session_document()
        if document is None:
            return {"shutdown": True, "analysis_stopped": True, "document_closed": True}
        if method == "shutdown" and not REA_OWNS_PROCESS_LIFETIME:
            return {
                "shutdown": True,
                "analysis_stopped": not document.backgroundProcessActive(),
                "document_closed": False,
                "document_retained": True,
            }
        if document.backgroundProcessActive():
            document.requestBackgroundProcessStop()
        if method == "shutdown" and REA_OWNS_PROCESS_LIFETIME:
            return {
                "shutdown": True,
                "analysis_stopped": not document.backgroundProcessActive(),
                "document_closed": False,
                "cleanup_required": True,
            }
        if document.backgroundProcessActive():
            document.waitForBackgroundProcessToEnd()
        document.closeDocument()
        analysis_stopped = not document.backgroundProcessActive()
        document_closed = _session_document() is None
        return {
            "shutdown": True,
            "analysis_stopped": analysis_stopped,
            "document_closed": document_closed,
        }
    if method == "list_documents":
        return [document.getDocumentName() for document in _api().documents()]
    if method == "current_document":
        return _document().getDocumentName()
    document = _document(params.get("document"))
    if method in EXHAUSTIVE_ANALYSIS_METHODS:
        _api().require_analysis_complete(document, method)

    if method == "analyze_function":
        return _analyze_function(document, params)
    if method == "read_function_instructions":
        return _read_function_instructions(document, params)
    if method == "read_bytes":
        length = params.get("length", 256)
        if isinstance(length, bool) or not isinstance(length, int):
            raise InvalidRequestError("Byte-read length must be an integer")
        if length < 1:
            raise InvalidRequestError("Byte-read length must be at least 1")
        reader = getattr(document, "readBytes", None)
        if not callable(reader):
            raise CapabilityUnavailableError(
                "Hopper's public Python API does not expose readBytes"
            )
        address = _address(document, params.get("address"))
        segment = _segment(document, address)
        # Document.readBytes delegates to one Segment and returns False when
        # the requested range crosses its end. Preserve the readable prefix.
        available = segment.getStartingAddress() + segment.getLength() - address
        value = reader(address, min(length, available))
        if value is False or value is None:
            raise InvalidRequestError("Hopper could not read the requested byte range at %s" % _hex(address))
        if not isinstance(value, (bytes, bytearray)):
            raise CapabilityUnavailableError(
                "Hopper readBytes returned %s for the requested %s-byte range at %s"
                % (type(value).__name__, length, _hex(address))
            )
        data = bytes(value)
        return {
            "address": _hex(address),
            "requested_bytes": length,
            "returned_bytes": len(data),
            "bytes_hex": data.hex(),
            "complete": len(data) == length,
        }
    if method == "address_to_file_offset":
        mapper = getattr(document, "getFileOffsetFromAddress", None)
        if not callable(mapper):
            raise CapabilityUnavailableError(
                "Hopper's public Python API does not expose getFileOffsetFromAddress"
            )
        address = _address(document, params.get("address"))
        offset = mapper(address)
        if (
            isinstance(offset, bool)
            or not isinstance(offset, int)
            or offset in BAD_ADDRESSES
            or offset < 0
        ):
            raise InvalidRequestError("Address has no authoritative file-offset mapping")
        inverse = getattr(document, "getAddressFromFileOffset", None)
        if not callable(inverse):
            raise CapabilityUnavailableError(
                "Hopper's public Python API cannot verify the file-offset mapping by reverse lookup"
            )
        # Synthetic memory can report offset zero, which belongs to the actual
        # file header. An integer alone does not prove source-file provenance.
        if inverse(offset) != address:
            raise InvalidRequestError("Address has no authoritative file-offset mapping: reverse lookup disagrees at %s" % _hex(address))
        original_offset, source_path, image_base = _original_file_offset(document, offset)
        return {"address": _hex(address), "file_offset": original_offset, "provider_file_offset": offset, "image_base_file_offset": image_base, "source_path": source_path}
    if method == "resolve_containing_procedure":
        address = _address(document, params.get("address"))
        procedure, reason = _containing_procedure(document, address)
        if procedure is None:
            return {"query_address": _hex(address), "found": False, "procedure": None, "reason": reason}
        return {"query_address": _hex(address), "found": True, "procedure": _procedure_identity(procedure)}
    if method == "procedure_references":
        return _procedure_references(document, params)

    if method == "current_address":
        return _hex(document.getCurrentAddress())
    if method == "current_procedure":
        return _procedure_name(_procedure(document))
    if method == "goto_address":
        address = _address(document, params.get("address"))
        segment = _segment(document, address)
        expected = segment.getInstructionStart(address)
        document.moveCursorAtAddress(address)
        observed = document.getCurrentAddress()
        if observed != (address if expected in BAD_ADDRESSES else expected):
            raise InvalidRequestError("Hopper cursor did not reach the requested object at %s; observed %s" % (_hex(address), _hex(observed)))
        return _hex(observed)
    if method in ("address_name", "comment", "inline_comment", "xrefs"):
        target = _address(document, params.get("address"))
        segment = _segment(document, target)
        if method == "address_name":
            return segment.getNameAtAddress(target)
        if method == "comment":
            return segment.getCommentAtAddress(target)
        if method == "inline_comment":
            return segment.getInlineCommentAtAddress(target)
        return [_hex(value) for value in segment.getReferencesOfAddress(target)]
    if method in ("next_address", "prev_address"):
        target = _address(document, params.get("address"))
        segment = _segment(document, target)
        if method == "next_address":
            start = segment.getInstructionStart(target)
            if start in BAD_ADDRESSES:
                raise InvalidRequestError("No analyzed object exists at the requested address")
            length = segment.getObjectLength(start)
            if length in BAD_ADDRESSES or length <= 0:
                raise InvalidRequestError("No next analyzed object exists at the requested address")
            result = start + length
        else:
            previous_segment = document.getSegmentAtAddress(target - 1) if target > 0 else None
            if previous_segment is None:
                raise InvalidRequestError("No previous address exists in mapped memory")
            result = previous_segment.getInstructionStart(target - 1)
        if result in BAD_ADDRESSES or document.getSegmentAtAddress(result) is None:
            raise InvalidRequestError("No adjacent analyzed object exists in mapped memory")
        return _hex(result)
    if method == "list_segments":
        result = []
        permission_limitation = _unavailable(
            "Hopper's public Python API does not expose segment or section permissions"
        )
        for segment in document.getSegmentsList():
            start = segment.getStartingAddress()
            sections = [{
                "name": section.getName(),
                "start": _hex(section.getStartingAddress()),
                "end": _hex(section.getStartingAddress() + section.getLength()),
                "readable": None,
                "writable": None,
                "executable": None,
                "permissions": permission_limitation,
                "provenance": "hopper-public-python-api",
            } for section in segment.getSectionsList()]
            result.append({
                "name": segment.getName(),
                "start": _hex(start),
                "end": _hex(start + segment.getLength()),
                "readable": None,
                "writable": None,
                "executable": None,
                "permissions": permission_limitation,
                "provenance": "hopper-public-python-api",
                "sections": sections,
            })
        return result
    if method == "list_procedures":
        return [{"address": address, "value": value} for address, value in _search_inventory(document, "procedure")]
    if method == "list_strings":
        requested = params.get("address")
        values = _string_inventory(document, None if requested is None else {_address(document, requested)})
        return [{"address": address, **record} for address, record in values.items()]
    if method == "list_names":
        result = dict(_search_inventory(document, "name"))
        requested = params.get("address")
        if requested is not None:
            key = _hex(_address(document, requested))
            result = {key: result[key]} if key in result else {}
        return [{"address": address, "value": value} for address, value in result.items()]
    if method in ("search_procedures", "search_strings"):
        kind = "procedure" if method == "search_procedures" else "string"
        return _search_results(document, kind, params)
    if method.startswith("procedure_"):
        procedure = _procedure(document, params.get("procedure"))
        if method == "procedure_address":
            return _hex(procedure.getEntryPoint())
        if method == "procedure_assembly":
            return _assembly(procedure)
        if method == "procedure_pseudo_code":
            return _pseudocode(document, procedure)
        if method == "procedure_callers":
            return sorted((_hex(item.getEntryPoint()) for item in procedure.getAllCallerProcedures()), key=lambda value: int(value, 16))
        if method == "procedure_callees":
            return sorted((_hex(item.getEntryPoint()) for item in procedure.getAllCalleeProcedures()), key=lambda value: int(value, 16))
        if method == "procedure_info":
            blocks = list(procedure.basicBlockIterator())
            length = sum(max(0, _basic_block_end(procedure, block) - block.getStartingAddress()) for block in blocks)
            return {
                "name": _procedure_name(procedure),
                "entrypoint": _hex(procedure.getEntryPoint()),
                "basicblock_count": procedure.getBasicBlockCount(),
                "length": length,
                "signature": procedure.signatureString(),
                "locals": _procedure_locals(procedure),
                "classification": None,
                "body": _procedure_identity(procedure)["body"],
            }
    if method == "set_address_name":
        address = _address(document, params.get("address"))
        name = _symbol_name(params["name"], "Name")
        _validate_name_targets(document, [(params.get("address"), address, name)])
        try:
            result = document.setNameAtAddress(address, name)
        finally:
            _invalidate_search_inventory(document)
            _invalidate_pseudocode(document)
        return bool(result) and _name_matches(document, address, name)
    if method == "set_addresses_names":
        # Validate all destinations before applying any annotation. A malformed
        # later address must not discard the result of an earlier mutation.
        names = [(key, _address(document, key), _symbol_name(value, "Name for %s" % key)) for key, value in params["names"].items()]
        _validate_name_targets(document, names)
        try:
            result = {key: document.setNameAtAddress(address, value) for key, address, value in names}
        finally:
            _invalidate_search_inventory(document)
            _invalidate_pseudocode(document)
        return {key: bool(result[key]) and _name_matches(document, address, value) for key, address, value in names}
    if method in ("set_comment", "set_inline_comment"):
        address = _address(document, params.get("address"))
        comment = _api_text(params["comment"], "Comment")
        segment = _segment(document, address)
        setter = segment.setCommentAtAddress if method == "set_comment" else segment.setInlineCommentAtAddress
        getter = segment.getCommentAtAddress if method == "set_comment" else segment.getInlineCommentAtAddress
        setter(address, comment)
        _invalidate_pseudocode(document)
        observed = getter(address)
        return observed == comment or (comment == "" and observed is None)
    if method == "list_bookmarks":
        return [{"address": _hex(item), "name": document.getBookmarkName(item)} for item in document.getBookmarks()]
    if method == "set_bookmark":
        address = _address(document, params.get("address"))
        _segment(document, address)
        name = params.get("name")
        if name is not None:
            _api_text(name, "Bookmark name")
        document.setBookmarkAtAddress(address, name)
        return document.hasBookmarkAtAddress(address)
    if method == "unset_bookmark":
        address = _address(document, params.get("address"))
        # Permit removing a legacy orphan bookmark, but do not report success
        # for an unmapped address with no bookmark to remove.
        if not document.hasBookmarkAtAddress(address):
            _segment(document, address)
        document.removeBookmarkAtAddress(address)
        return not document.hasBookmarkAtAddress(address)
    raise InvalidRequestError("Unknown bridge method")


def _serve_connection(connection):
    """Serve one capability-authenticated NDJSON connection."""
    file = connection.makefile("rwb")
    while True:
        line = file.readline()
        if not line:
            break
        request_id = None
        authenticated = False
        should_stop = False
        try:
            try:
                request = json.loads(line.decode("utf-8"))
            except (UnicodeDecodeError, json.JSONDecodeError) as error:
                raise InvalidRequestError("Invalid bridge request JSON") from error
            if not isinstance(request, dict) or set(request) != {"id", "token", "method", "params"}:
                raise InvalidRequestError("Invalid bridge request shape")
            request_id = request["id"]
            if isinstance(request_id, bool) or not isinstance(request_id, int) or request_id < 0:
                raise InvalidRequestError("Invalid bridge request id")
            if (
                not isinstance(request["method"], str) or not request["method"]
                or not isinstance(request["params"], dict)
            ):
                raise InvalidRequestError("Invalid bridge method or parameters")
            if (
                not isinstance(request["token"], str) or not request["token"].isascii()
                or not hmac.compare_digest(request["token"], REA_TOKEN)
            ):
                raise PermissionError("Invalid bridge capability")
            authenticated = True
            _write_message(file, {"id": request_id, "event": {
                "type": "progress",
                "phase": "hopper_bridge",
                "completed": 0,
                "total": 1,
                "message": "Hopper bridge started request",
            }})
            result = _dispatch(request["method"], request["params"])
            should_stop = request["method"] == "shutdown_document" or (
                request["method"] == "shutdown" and not result.get("cleanup_required", False)
            )
            safe_result = _json_safe(result)
            _write_message(file, {"id": request_id, "event": {
                "type": "progress",
                "phase": "hopper_bridge",
                "completed": 1,
                "total": 1,
                "message": "Hopper bridge completed request",
                "terminal": True,
            }})
            response = {"id": request_id, "result": safe_result}
        except Exception as error:
            diagnostic = _safe_diagnostic(error)
            if authenticated:
                _write_message(file, {"id": request_id, "event": {
                    "type": "diagnostic",
                    "error": diagnostic,
                }})
            response_id = request_id if (
                isinstance(request_id, int)
                and not isinstance(request_id, bool)
                and request_id >= 0
            ) else 0
            response = {"id": response_id, "error": diagnostic}
        _write_message(file, response)
        if should_stop:
            break
    file.close()
    connection.close()


def _write_message(file, message):
    """Write and flush one compact bridge message."""
    file.write((json.dumps(message, separators=(",", ":")) + "\n").encode("utf-8"))
    file.flush()


def _diagnostic_type(error):
    if isinstance(error, CapabilityUnavailableError):
        return "capability_unavailable"
    if isinstance(error, PermissionError):
        return "authorization"
    if isinstance(error, InvalidRequestError):
        return "invalid_request"
    return "bridge_exception"


def _safe_diagnostic(error):
    """Preserve bridge validation reasons while excluding transport credentials."""
    diagnostic_type = _diagnostic_type(error)
    if diagnostic_type == "capability_unavailable":
        message = str(error)
    elif diagnostic_type == "authorization":
        message = "Invalid bridge capability"
    elif diagnostic_type == "invalid_request":
        message = str(error)
    else:
        message = "%s: Hopper bridge operation failed" % type(error).__name__
    return {"code": -32000, "message": message, "type": diagnostic_type}


def _run():
    """Own a permission-restricted, single-client Unix socket for this bridge."""
    if os.path.exists(REA_SOCKET):
        os.unlink(REA_SOCKET)
    server = socket.socket(socket.AF_UNIX, socket.SOCK_STREAM)
    server.bind(REA_SOCKET)
    os.chmod(REA_SOCKET, 0o600)
    server.listen(1)
    try:
        connection, _ = server.accept()
        _serve_connection(connection)
    finally:
        server.close()
        if os.path.exists(REA_SOCKET):
            os.unlink(REA_SOCKET)


# Hopper's public objects are bound to its dedicated Python execution thread.
# Keep dispatch on that thread; moving calls to a worker can deadlock.
if __name__ == "__main__":
    _bind_session_document()
    _run()