path: root/tests/test_blob_ui.py

#!/usr/bin/env python3
"""A blob that analyses to NOTHING must still be usable.

Zero functions is the normal outcome for a raw image described wrongly — and it
used to leave two empty panes and a status that said "functions still loading…",
which was a lie: loading had finished, there was simply nothing to land on.

Needs IDA (spawns a real worker). ~40s.
"""

#: spawns a real worker on a raw blob and drives the pilot.
#: Read by tests/run.py (--fast skips every NEEDS_IDA file).
NEEDS_IDA = True
import asyncio
import os
import sys
import tempfile

sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
sys.path.insert(0, os.path.dirname(os.path.abspath(__file__)))

from textual.widgets import Input, Static  # noqa: E402

from idatui.app import ConfirmScreen, IdaTui, ListingView  # noqa: E402
from _fixtures import staged, synthetic  # noqa: E402
from idatui._sync import settle  # noqa: E402

PASS = FAIL = 0


def check(name, ok, detail=""):
    global PASS, FAIL
    if ok:
        PASS += 1
        print(f"  ok   {name}")
    else:
        FAIL += 1
        print(f"  FAIL {name}   {detail}")


async def wait(pred, pilot, t=240.0):
    for _ in range(int(t / 0.05)):
        await pilot.pause(0.05)
        try:
            if pred():
                return True
        except Exception:  # noqa: BLE001
            pass
    return False


#: File offset of the planted instruction run -> ea 0x4000 + PLANTED.
PLANTED = 0x40


def _blob_bytes() -> bytes:
    """A DETERMINISTIC pseudo-random blob with real AArch64 instructions planted.

    Seeded, not os.urandom: the bytes must be identical every run or the
    pristine-database cache can never apply (this suite used to pay ~40s of
    auto-analysis per run because the content, and the path, changed each time).
    Determinism also removes a genuine flake -- whether 64KB of chance bytes
    contains something IDA reads as a function is luck, and "and really has no
    functions" is asserted below.
    """
    import random
    data = bytearray(random.Random(0xB10BCAFE).randbytes(64 * 1024))
        # -parm puts IDA in AArch64 mode, so these are A64 encodings; the ARM32
        # spelling of a nop (0xE1A00000) is NOT decodable there and made this
        # test fail for a reason that had nothing to do with what it checks.
    for k, insn in enumerate((0xD503201F,   # nop
                              0xD503201F,   # nop
                              0xD65F03C0)):  # ret  <- the run must stop here
        data[PLANTED + k * 4:PLANTED + k * 4 + 4] = insn.to_bytes(4, "little")
    return bytes(data)


#: -parm puts IDA in AArch64 mode (the ARM32 spelling of a nop is not decodable
#: there); -b400 sets the image base. The cached database must be built with the
#: SAME switches, so both go through one factory.
BLOB_ARGS = "-parm -b400"


def _blob_app(path):
    return IdaTui(open_path=path, keepalive=False, load_args=BLOB_ARGS)


def head_at(lst, ea):
    """The listing row for ``ea`` off the LIVE model, or None.

    Always re-reads ``lst.model``: an edit may rebuild the model, and holding
    the old object shows pre-edit rows -- which looks exactly like the edit
    silently failing.
    """
    m = lst.model
    if m is None:
        return None
    i = m.index_of_ea(ea)
    return m.get(i) if i is not None and i >= 0 else None


async def run() -> int:
    blob_src = synthetic("rnd.bin", _blob_bytes)
    # staged() analyses once ever and copies the result in on later runs.
    async with staged(blob_src, _blob_app) as blob:
        # Skip the dialog by answering up front; this test is about what
        # happens AFTER a described blob turns out to contain nothing.
        app = _blob_app(blob)
        async with app.run_test(size=(140, 44)) as pilot:
            ok = await wait(lambda: app._func_index is not None
                            and app._func_index.complete, pilot)
            check("a random blob still finishes loading", ok)
            check("and really has no functions", len(app._func_index) == 0,
                  f"n={len(app._func_index)}")

            landed = await wait(lambda: app._cur is not None, pilot, 60)
            check("it lands somewhere instead of leaving empty panes", landed,
                  f"cur={app._cur}")
            lst = app.query_one(ListingView)
            check("the listing actually has rows to show",
                  lst.total > 0, f"total={lst.total}")
            check("landed at the image base", app._cur is not None
                  and app._cur.ea == 0x4000, f"{app._cur.ea if app._cur else None:#x}")

            status = str(app.query_one("#status", Static).render())
            check("the status says there are no functions (not 'still loading')",
                  "no functions" in status and "still loading" not in status,
                  status[:90])
            check("and points at the likely cause",
                  "processor" in status and "Ctrl+L" in status, status[:90])

            # The hint is a property of the database, so it must survive moving
            # around — an earlier version wrote it once and the next status
            # write erased it.
            lst.focus()
            await pilot.press("down")
            await pilot.press("down")
            await pilot.pause(0.3)
            status2 = str(app.query_one("#status", Static).render())
            check("the hint survives navigating", "no functions" in status2,
                  status2[:90])

            # -- byte-granular carving ---------------------------------- #
            # An undefined run arrives as ONE head ("db N dup(?)"). It has to
            # PRESENT as one row per byte or there is no cursor position to
            # press `c` at, which is how IDA works and the only way to find an
            # instruction stream that doesn't start at the run's first byte.
            m = lst.model
            check("an undefined run presents one row per byte",
                  m.loaded() >= 4096 and len(m._heads) < 64,
                  f"rows={m.loaded()} physical heads={len(m._heads)}")
            rows = m.window(0, 4)
            check("each row is a single addressable byte",
                  [h.ea for h in rows] == [0x4000, 0x4001, 0x4002, 0x4003]
                  and all(h.size == 1 for h in rows),
                  f"{[(hex(h.ea), h.size) for h in rows]}")
            check("and shows its value, not a placeholder",
                  all(h.text.startswith("db ") and "dup" not in h.text
                      for h in rows), f"{[h.text for h in rows]}")

            # The point of all this: land on an arbitrary byte and convert it.
            i = m.index_of_ea(0x4021)
            check("an address inside the run resolves to its own row",
                  i >= 0 and m.get(i).ea == 0x4021,
                  f"row={i} ea={m.get(i).ea if i >= 0 else None}")

            target = 0x4000 + PLANTED        # a NOP we put there ourselves
            lst.cursor = m.index_of_ea(target)
            lst._scroll_cursor_into_view()
            await pilot.pause(0.1)
            check("the cursor sits on the byte we aimed at",
                  lst._cursor_ea() == target,
                  f"{lst._cursor_ea():#x} want {target:#x}")
            await pilot.press("c")
            # settle(), not a fixed sleep AND not a bare predicate: an edit can
            # look done for a moment and then be replaced when a queued listing
            # rebuild lands, so the gate has to be "the row is code AND the app
            # has stopped working". settle() is the same helper the app's own
            # RPC layer uses, so tests and driver agree on what "done" means.
            await settle(app, lambda: (lambda h: h is not None and h.kind == "code")(
                head_at(lst, target)), timeout=30)
            # Re-read the model: defining an item rebuilds it, and holding the
            # old object shows pre-edit rows -- which looks exactly like the
            # edit silently failing.
            m = lst.model
            h = head_at(lst, target)
            check("`c` on a chosen byte carves an instruction there",
                  h is not None and h.kind == "code",
                  f"kind={h.kind if h else None} text={h.text if h else None!r}")
            if h is not None and h.kind == "code":
                print(f"       carved {target:#x}: {h.text}")
                # `c` runs until something stops it, like IDA — one instruction
                # at a time means pressing it once per opcode for the length of
                # a routine. We planted nop/nop/nop/ret, so it must take all
                # four and stop AT the ret, not run on into the random bytes
                # after it.
                run = [m.get(m.index_of_ea(target + k * 4)) for k in range(3)]
                check("`c` keeps going until control flow ends",
                      all(x is not None and x.kind == "code" for x in run),
                      f"{[(hex(x.ea), x.kind) for x in run if x]}")
                check("and stops at the ret instead of running into junk",
                      m.get(m.index_of_ea(target + 12)).kind == "unknown",
                      f"{m.get(m.index_of_ea(target + 12)).text!r}")
                status = str(app.query_one("#status", Static).render())
                check("the status reports what the run did",
                      "3 instructions" in status and "control flow" in status,
                      status[:80])
                check("the carved row spans the instruction, not one byte",
                      h.size == 4, f"size={h.size}")
                check("bytes before it stay individually addressable",
                      m.get(m.index_of_ea(target - 1)).size == 1
                      and m.get(m.index_of_ea(target - 1)).ea == target - 1)

            # -- an edit must not move the view -------------------------- #
            # Every mutation rebuilds the model, and row indices don't survive
            # that: carving collapses four byte rows into one instruction row.
            # All the edit paths go through ViewAnchor, which remembers
            # ADDRESSES. This runs in its own app, so unlike the shared scenario
            # suite it can mutate the database freely.
            lst.cursor = m.index_of_ea(0x4200)
            lst._scroll_cursor_into_view()
            await pilot.pause(0.4)
            ctop = lst.model.get(round(lst.scroll_offset.y)).ea
            ccur = lst._cursor_ea()
            old = lst.model
            await pilot.press("semicolon")
            await wait(lambda: app.query_one("#comment", Input).display, pilot, 20)
            for ch in "note":
                await pilot.press(ch)
            await pilot.press("enter")
            # Wait for the COMMENT ITSELF to show up, not for the model object to
            # be replaced: a comment now re-renders the listing in place (the
            # walk is kept), so `model is not old` never becomes true and this
            # burned its full 30s timeout on every run -- after which the check
            # below passed vacuously, because nothing had happened at all.
            # The prompt closing plus quiescence is the real end of the edit.
            # (The listing re-renders its text lazily, so the comment is not
            # necessarily visible in model rows the moment the worker returns --
            # which is why this waits for the app, not for the text.)
            await settle(app, lambda: not app.query_one("#comment", Input).display,
                         timeout=30)
            check("commenting leaves the view where it was",
                  lst.model.get(round(lst.scroll_offset.y)).ea == ctop
                  and lst._cursor_ea() == ccur,
                  f"top {ctop:#x} -> "
                  f"{lst.model.get(round(lst.scroll_offset.y)).ea:#x}, "
                  f"cursor {ccur:#x} -> {lst._cursor_ea():#x}")

            # -- carving must not move the view -------------------------- #
            # Defining code collapses rows (four byte rows become one
            # instruction), so anything that remembers a row INDEX puts you
            # somewhere else afterwards. Scroll down far enough that there is a
            # viewport to lose, then check the top ADDRESS is unchanged.
            far = 0x4000 + 0x600
            lst.cursor = lst.model.index_of_ea(far)
            lst._scroll_cursor_into_view()
            await pilot.pause(0.4)
            top_before = lst.model.get(round(lst.scroll_offset.y)).ea
            cur_before = lst._cursor_ea()
            check("scrolled somewhere with rows above us",
                  round(lst.scroll_offset.y) > 0, f"top={lst.scroll_offset.y}")
            await pilot.press("c")
            # No predicate here on purpose: this spot is random data, so the
            # carve may legitimately produce nothing and "the row became code"
            # would never hold (it timed out for 30s and then passed anyway).
            # What is being checked is that the VIEW did not move, so the gate
            # is simply "the app has finished reacting".
            await settle(app, timeout=30)
            m2 = lst.model
            top_after = m2.get(round(lst.scroll_offset.y)).ea
            check("carving leaves the scroll position where it was",
                  top_after == top_before,
                  f"{top_before:#x} -> {top_after:#x}")
            check("and leaves the cursor on the same address",
                  lst._cursor_ea() == cur_before,
                  f"{cur_before:#x} -> {lst._cursor_ea():#x}")

            # -- `p` after carving: the rest of the app must notice ------ #
            # The "no functions" hint was latched at load and only cleared on a
            # reload, so it kept telling you the processor/base were wrong long
            # after you'd defined a function. The function index was never
            # rebuilt either, which meant Ctrl+N couldn't find what `p` made.
            check("no functions yet, and the hint says so",
                  len(app._func_index) == 0
                  and "no functions" in str(app.query_one("#status", Static).render()),
                  f"n={len(app._func_index)}")
            lst.cursor = lst.model.index_of_ea(target)
            lst._scroll_cursor_into_view()
            await pilot.pause(0.3)
            mp = lst.model
            await pilot.press("p")
            await wait(lambda: lst.model is not mp and lst.model is not None,
                       pilot, 40)
            await wait(lambda: app._func_index is not None
                       and len(app._func_index) > 0, pilot, 60)
            check("`p` creates a function the index can see",
                  len(app._func_index) == 1, f"n={len(app._func_index)}")
            status = str(app.query_one("#status", Static).render())
            check("and the stale 'no functions' hint is gone",
                  "no functions" not in status, status[:90])
            check("the status names the function it made",
                  "created function" in status, status[:90])

            await pilot.press("ctrl+l")
            opened = await wait(lambda: isinstance(app.screen, ConfirmScreen), pilot, 20)
            check("Ctrl+L offers to reload with different options", opened,
                  f"screen={type(app.screen).__name__}")
            if opened:
                note = str(app.screen.query_one("#confirm-note", Static).render())
                # We just made a function, so it must NOT claim nothing is lost —
                # reloading throws the database away and that is now a real cost.
                check("the confirmation counts what would be lost",
                      "1 function," in note and "nothing is lost" not in note,
                      note[:80])
                await pilot.press("escape")
                await pilot.pause(0.3)
                check("declining leaves the binary open",
                      not isinstance(app.screen, ConfirmScreen) and app._cur is not None)

    print(f"\n{PASS} passed, {FAIL} failed")
    return 1 if FAIL else 0


if __name__ == "__main__":
    raise SystemExit(asyncio.run(run()))