diff options
| -rw-r--r-- | idatui/codemode_client.py | 6 | ||||
| -rw-r--r-- | idatui/remote_tools.py | 89 | ||||
| -rw-r--r-- | tests/test_scenarios.py | 48 |
3 files changed, 143 insertions, 0 deletions
diff --git a/idatui/codemode_client.py b/idatui/codemode_client.py index d0d1937..2361e08 100644 --- a/idatui/codemode_client.py +++ b/idatui/codemode_client.py @@ -1205,6 +1205,12 @@ _OPERATIONS["pc_num_format"] = _remote_op( # would reformat) and had no digest/expect support (so every page was re-sent # after any edit), and whose span walk was the per-character loop our own # version had already been rewritten to avoid. +#: Row count + seek anchors for a whole segment, in ONE call. See +#: remote_tools.segment_index: the alternative is fetching every row. +_OPERATIONS["segment_index"] = _remote_op( + 'segment_index(addr=a["addr"], end=a.get("end", ""),' + ' page_rows=int(a.get("page_rows", 500)))') + _HEADS = _remote_op( 'heads(addr=a["addr"], count=int(a.get("count", 200)),' ' offset=int(a.get("offset", 0)), end=a.get("end", ""),' diff --git a/idatui/remote_tools.py b/idatui/remote_tools.py index d2cb02e..614fba4 100644 --- a/idatui/remote_tools.py +++ b/idatui/remote_tools.py @@ -510,6 +510,95 @@ def _idatui_func_footer_rows(ea, func): ] +def segment_index( + addr: Annotated[str, "Any address in the segment to index"], + end: Annotated[str, "Optional exclusive end address; default = segment end"] = "", + page_rows: Annotated[int, "Rows between anchors (default 500)"] = 500, +) -> dict: + """How many listing rows a segment has, and where to seek into it. + + The listing needs a total row count to size its scrollbar, and the only way + to get one used to be to fetch every row: 455 round trips and 227k rows for + a 1.2MB bash, none of which is looked at. This walks the same items and + counts what ``heads`` WOULD emit, without building or rendering any of them. + + Returns ``{rows, heads, anchors}`` where anchors is ``[[row, ea], ...]`` + every ``page_rows`` logical rows -- enough to turn "show me row N" into a + ``heads(addr=anchor)`` call, so pages can be fetched on demand instead of + streamed in order. + + **The count must match what heads() actually emits, exactly**, or the + scrollbar lies and a jump lands on the wrong row. It therefore mirrors + _rows_for's arithmetic rather than approximating it: 3 banner rows at a + function start, a label row for a named code head that is not one, the head + row itself, struct member rows for data, 2 footer rows at a function end, + and an undefined run counted as its byte length (the client presents one + collapsed row as that many logical rows). Verified equal to summing the real + pages, head for head, over a whole segment. + """ + + count = max(int(page_rows), 1) + try: + start = parse_address(addr) + except Exception as e: + return {"addr": str(addr), "error": str(e), "rows": 0, "anchors": []} + import ida_segment + seg = ida_segment.getseg(start) + if not seg: + return {"addr": str(addr), "error": "no segment", "rows": 0, "anchors": []} + lo, hi = seg.start_ea, seg.end_ea + if end: + try: + hi = min(hi, parse_address(end)) + except Exception: + pass + + get_flags = ida_bytes.get_flags + get_item_end = ida_bytes.get_item_end + next_head = ida_bytes.next_head + get_ea_name = ida_name.get_ea_name + get_func = idaapi.get_func + BAD = idaapi.BADADDR + + rows = 0 + n_heads = 0 + anchors = [] + fn = None + ea = ida_bytes.get_item_head(lo) + while ea != BAD and ea < hi: + if rows // count >= len(anchors): + anchors.append([rows, hex(ea)]) + f = get_flags(ea) + cls = f & _MS_CLS + if cls != _FF_CODE and cls != _FF_DATA: + # An undefined run is ONE emitted row that PRESENTS as one logical + # row per byte (see _idatui_unknown_row and the client's _span). + nh = next_head(ea, hi) + stop = nh if (nh != BAD and ea < nh <= hi) else hi + run = stop - ea + rows += run if run > 1 else 1 + n_heads += 1 + ea = stop + continue + if fn is None or not (fn.start_ea <= ea < fn.end_ea): + fn = get_func(ea) + n = 1 + if fn is not None and fn.start_ea == ea: + n += 3 # blank, banner, `proc` + elif cls == _FF_CODE and get_ea_name(ea): + n += 1 # loc_XXX label on its own row + if cls == _FF_DATA: + n += len(_idatui_struct_member_rows(ea)) + item_end = get_item_end(ea) + if fn is not None and item_end >= fn.end_ea: + n += 2 # `endp` + separator + rows += n + n_heads += 1 + ea = item_end if item_end > ea else ea + 1 + return {"addr": hex(lo), "end": hex(hi), "rows": rows, + "heads": n_heads, "anchors": anchors} + + def heads( addr: Annotated[str, "Start address or name to walk from"], count: Annotated[int, "Max heads to return (default 200, max 2000)"] = 200, diff --git a/tests/test_scenarios.py b/tests/test_scenarios.py index 780459a..b256b80 100644 --- a/tests/test_scenarios.py +++ b/tests/test_scenarios.py @@ -412,6 +412,54 @@ async def s_auto_land(c: Ctx): c.check("auto-land is idempotent (guarded)", app._cur is prev) +@scenario("segment_index") +async def s_segment_index(c: Ctx): + """segment_index must count EXACTLY what streaming the pages produces. + + It exists so the listing can know its row total without fetching every row + (1 call and ~0.5s instead of 458 calls and ~1.8s on bash). That is only + usable if the number is exact: the total sizes the scrollbar, and the + anchors are what a future "seek to row N" would jump through, so being off + by a handful of rows means the bar lies and a jump lands in the wrong place. + + Approximating it is the tempting mistake, which is why this compares against + the real thing rather than a tolerance. + """ + app = c.app + await c.open_biggest("listing") + lv = app.query_one(ListingView) + model = lv.model + if model is None: + c.check("listing model exists", False) + return + await c.wait(lambda: model.complete, 30) + if not model.complete: + c.check("segment streamed for comparison", False) + return + + idx = app.program.client.invoke("segment_index", addr=hex(model.seg_start)) + c.check("segment_index counts exactly what streaming produced", + idx.get("rows") == len(model), + f"index={idx.get('rows')} streamed={len(model)}") + c.check("it reports the same segment", + int(str(idx.get("addr")), 16) == model.seg_start, + f"{idx.get('addr')} vs {model.seg_start:#x}") + anchors = idx.get("anchors") or [] + c.check("anchors cover the segment", + len(anchors) >= max(1, len(model) // 500), + f"{len(anchors)} anchors for {len(model)} rows") + + # Every anchor must name the address of the row it claims, or seeking to it + # would land somewhere else entirely. + bad = [] + for row, ea in anchors: + h = model.get(row) + if h is None or h.ea != int(str(ea), 16): + bad.append((row, ea, hex(h.ea) if h else None)) + c.check("every anchor points at the row it claims", not bad, + f"{len(bad)} wrong, first={bad[:2]}") + + @scenario("skeleton_pages") async def s_skeleton_pages(c: Ctx): """The background grower loads text-less pages; reading one must fill it in. |
