1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
|
"""Ghidra bridge: symbols in, hit annotations out.
Two flows:
*out of Ghidra*: export function names/addresses/sizes to a green-eye JSON
symbol database (from ``names.tsv``, ``decompiled_index.tsv``, exports or ELF);
*hits*: turn a recorded session into ``greeneye_hits.tsv``
(``addr<TAB>hits<TAB>symbol``) that ``MarkGreenEyeHits.java`` can annotate in
the Ghidra project.
This is the static/dynamic join: Ghidra owns the address space and
decompilation, green-eye owns what actually executed.
"""
from __future__ import annotations
import os
from collections import Counter
from .record import load_events
from .symbols import SymbolDB
def build_symbol_db(
*,
names_tsv: str | None = None,
index_tsv: str | None = None,
exports_tsv: str | None = None,
callgraph_tsv: str | None = None,
elf: str | None = None,
bias: int = 0,
) -> SymbolDB:
db = SymbolDB()
if elf:
db.load_elf(elf, bias=0) # ELF symbol values are already runtime addresses
if names_tsv:
db.load_names_tsv(names_tsv, bias=bias)
if index_tsv:
db.load_index_tsv(index_tsv, bias=bias)
if exports_tsv:
db.load_index_tsv(exports_tsv, bias=bias)
if callgraph_tsv:
db.load_callgraph(callgraph_tsv)
return db
def export_hits(session_dir: str, out_tsv: str, symbols: SymbolDB | None = None) -> str:
events = load_events(session_dir)
hits: Counter = Counter()
for event in events:
if event.get("type") == "bp_hit":
hits[event.get("pc")] += 1
elif event.get("type") == "watch_hit":
hits[event.get("addr")] += 1
with open(out_tsv, "w", encoding="utf-8") as fh:
fh.write("addr\thits\tsymbol\tannotation\n")
for addr, count in sorted(hits.items(), key=lambda kv: -kv[1]):
if not isinstance(addr, int):
continue
sym = symbols.resolve(addr) if symbols else None
name = sym.name if sym else ""
annotation = symbols.annotate(addr) if symbols else f"{addr:#x}"
fh.write(f"{addr:#x}\t{count}\t{name}\t{annotation}\n")
return os.path.abspath(out_tsv)
def corpus_location(symbol, corpus_root: str) -> str | None:
"""Return ``<bucket>:<line>`` for a symbol with Ghidra index extras."""
bucket = symbol.extras.get("bucket")
line = symbol.extras.get("line")
if not bucket:
return None
return os.path.join(corpus_root, bucket) + (f":{line}" if line else "")
def render_ghidra_command(project_dir: str, project_name: str, program: str, hits_tsv: str) -> str:
return (
"analyzeHeadless {proj} {name} -process {program} -noanalysis "
"-scriptPath <open-iov>/debug/risc-v-gsp-rm-vm/ghidra "
"-postScript MarkGreenEyeHits.java {hits}"
).format(proj=project_dir, name=project_name, program=program, hits=hits_tsv)
|
