tools: Migrate to external repo

This commit is contained in:
MonsterDruide1
2021-08-07 21:59:16 +02:00
parent 802be64d8e
commit 18152c5909
42 changed files with 38 additions and 3634 deletions
View File
-223
View File
@@ -1,223 +0,0 @@
import struct
from collections import defaultdict
from typing import Set, DefaultDict, Dict, Optional, Tuple
import capstone as cs
from util import dsym, elf, utils
_store_instructions = ("str", "strb", "strh", "stur", "sturb", "sturh")
class FunctionChecker:
def __init__(self, log_mismatch_cause: bool = False):
self.md = cs.Cs(cs.CS_ARCH_ARM64, cs.CS_MODE_ARM)
self.md.detail = True
self.my_symtab = elf.build_name_to_symbol_table(elf.my_symtab)
self.dsymtab = dsym.DataSymbolContainer()
self.decompiled_fns: Dict[int, str] = dict()
self._log_mismatch_cause = log_mismatch_cause
self._mismatch_addr1 = -1
self._mismatch_addr2 = -1
self._mismatch_cause = ""
self._base_got_section = elf.base_elf.get_section_by_name(".got")
self._decomp_glob_data_table = elf.build_glob_data_table(elf.my_elf)
self._got_data_symbol_check_cache: Dict[Tuple[int, int], bool] = dict()
self.load_data_for_project()
def _reset_mismatch(self) -> None:
self._mismatch_addr1 = -1
self._mismatch_addr2 = -1
self._mismatch_cause = ""
def get_data_symtab(self) -> dsym.DataSymbolContainer:
return self.dsymtab
def get_mismatch(self) -> (int, int, str):
return self._mismatch_addr1, self._mismatch_addr2, self._mismatch_cause
def load_data_for_project(self) -> None:
self.decompiled_fns = {func.addr: func.decomp_name for func in utils.get_functions() if func.decomp_name}
self.get_data_symtab().load_from_csv(utils.get_repo_root() / "data" / "data_symbols.csv")
def check(self, base_fn: elf.Function, my_fn: elf.Function) -> bool:
self._reset_mismatch()
gprs1: DefaultDict[int, int] = defaultdict(int)
gprs2: DefaultDict[int, int] = defaultdict(int)
adrp_pair_registers: Set[int] = set()
size = len(base_fn)
if len(base_fn) != len(my_fn):
if self._log_mismatch_cause:
self._set_mismatch_cause(None, None, "different function length")
return False
def forget_modified_registers(insn):
_, regs_write = insn.regs_access()
for reg in regs_write:
adrp_pair_registers.discard(reg)
for i1, i2 in zip(self.md.disasm(base_fn.data, base_fn.addr), self.md.disasm(my_fn.data, my_fn.addr)):
if i1.bytes == i2.bytes:
if i1.mnemonic == 'adrp':
gprs1[i1.operands[0].reg] = i1.operands[1].imm
gprs2[i2.operands[0].reg] = i2.operands[1].imm
adrp_pair_registers.add(i1.operands[0].reg)
elif i1.mnemonic == 'b':
branch_target = i1.operands[0].imm
if not (base_fn.addr <= branch_target < base_fn.addr + size):
if not self._check_function_call(i1, i2, branch_target, i2.operands[0].imm):
return False
else:
forget_modified_registers(i1)
continue
if i1.mnemonic != i2.mnemonic:
if self._log_mismatch_cause:
self._set_mismatch_cause(i1, i2, "mnemonics are different")
return False
# Ignore some address differences until a fully matching executable can be generated.
if i1.mnemonic == 'bl':
if not self._check_function_call(i1, i2, i1.operands[0].imm, i2.operands[0].imm):
return False
continue
if i1.mnemonic == 'b':
branch_target = i1.operands[0].imm
# If we are branching outside the function, this is likely a tail call.
# Treat this as a function call.
if not (base_fn.addr <= branch_target < base_fn.addr + size):
if not self._check_function_call(i1, i2, branch_target, i2.operands[0].imm):
return False
continue
# Otherwise, it's a mismatch.
return False
if i1.mnemonic == 'adrp':
if i1.operands[0].reg != i2.operands[0].reg:
return False
reg = i1.operands[0].reg
gprs1[reg] = i1.operands[1].imm
gprs2[reg] = i2.operands[1].imm
adrp_pair_registers.add(reg)
continue
if i1.mnemonic == 'ldp' or i1.mnemonic == 'ldpsw' or i1.mnemonic == 'stp':
if i1.operands[0].reg != i2.operands[0].reg:
return False
if i1.operands[1].reg != i2.operands[1].reg:
return False
if i1.operands[2].value.mem.base != i2.operands[2].value.mem.base:
return False
reg = i1.operands[2].value.mem.base
if reg not in adrp_pair_registers:
return False
gprs1[reg] += i1.operands[2].value.mem.disp
gprs2[reg] += i2.operands[2].value.mem.disp
if not self._check_data_symbol_load(i1, i2, gprs1[reg], gprs2[reg]):
return False
forget_modified_registers(i1)
continue
if i1.mnemonic.startswith('ld') or i1.mnemonic in _store_instructions:
if i1.operands[0].reg != i2.operands[0].reg:
return False
if i1.operands[1].value.mem.base != i2.operands[1].value.mem.base:
return False
reg = i1.operands[1].value.mem.base
if reg not in adrp_pair_registers:
return False
gprs1[reg] += i1.operands[1].value.mem.disp
gprs2[reg] += i2.operands[1].value.mem.disp
if not self._check_data_symbol_load(i1, i2, gprs1[reg], gprs2[reg]):
return False
forget_modified_registers(i1)
continue
if i1.mnemonic == 'add':
if i1.operands[0].reg != i2.operands[0].reg:
return False
if i1.operands[1].reg != i2.operands[1].reg:
return False
reg = i1.operands[1].reg
if reg not in adrp_pair_registers:
return False
gprs1[reg] += i1.operands[2].imm
gprs2[reg] += i2.operands[2].imm
if not self._check_data_symbol(i1, i2, gprs1[reg], gprs2[reg]):
return False
forget_modified_registers(i1)
continue
return False
return True
def _set_mismatch_cause(self, i1: Optional[any], i2: Optional[any], description: str) -> None:
self._mismatch_addr1 = i1.address if i1 else -1
self._mismatch_addr2 = i2.address if i2 else -1
self._mismatch_cause = description
def _check_data_symbol(self, i1, i2, orig_addr: int, decomp_addr: int) -> bool:
symbol = self.dsymtab.get_symbol(orig_addr)
if symbol is None:
return True
decomp_symbol = self.my_symtab[symbol.name]
if decomp_symbol.addr == decomp_addr:
return True
if self._log_mismatch_cause:
self._set_mismatch_cause(i1, i2, f"data symbol mismatch: {symbol.name} (original address: {orig_addr:#x}, "
f"expected: {decomp_symbol.addr:#x}, "
f"actual: {decomp_addr:#x})")
return False
def _check_data_symbol_load(self, i1, i2, orig_addr: int, decomp_addr: int) -> bool:
cached_result = self._got_data_symbol_check_cache.get((orig_addr, decomp_addr), None)
if cached_result is not None:
return cached_result
if not elf.is_in_section(self._base_got_section, orig_addr, 8):
return True
ptr1, = struct.unpack("<Q", elf.read_from_elf(elf.base_elf, orig_addr, 8))
if self.dsymtab.get_symbol(ptr1) is None:
return True
ptr2 = self._decomp_glob_data_table[decomp_addr]
result = self._check_data_symbol(i1, i2, ptr1, ptr2)
self._got_data_symbol_check_cache[(orig_addr, decomp_addr)] = result
return result
def _check_function_call(self, i1, i2, orig_addr: int, decomp_addr: int) -> bool:
name = self.decompiled_fns.get(orig_addr, None)
if name is None:
self.on_unknown_fn_call(orig_addr, decomp_addr)
return True
decomp_symbol = self.my_symtab[name]
if decomp_symbol.addr == decomp_addr:
return True
if self._log_mismatch_cause:
self._set_mismatch_cause(i1, i2, f"function call mismatch: {name}")
return False
def on_unknown_fn_call(self, orig_addr: int, decomp_addr: int) -> None:
pass
-62
View File
@@ -1,62 +0,0 @@
import csv
from pathlib import Path
import typing as tp
import util.elf
class DataSymbol(tp.NamedTuple):
addr: int # without the 0x7100000000 base
name: str
size: int
_IDA_BASE = 0x7100000000
class DataSymbolContainer:
def __init__(self) -> None:
self.symbols: tp.List[DataSymbol] = []
def load_from_csv(self, path: Path):
symtab = util.elf.build_name_to_symbol_table(util.elf.my_symtab)
with path.open("r") as f:
for i, line in enumerate(csv.reader(f)):
if len(line) != 2:
raise RuntimeError(f"Invalid line format at line {i}")
addr = int(line[0], 16) - _IDA_BASE
name = line[1]
if name not in symtab:
continue
size = symtab[name].size
self.symbols.append(DataSymbol(addr, name, size))
# Sort the list, just in case the entries were not sorted in the CSV.
self.symbols.sort(key=lambda sym: sym.addr)
def get_symbol(self, addr: int) -> tp.Optional[DataSymbol]:
"""If addr is part of a known data symbol, this function returns the corresponding symbol."""
# Perform a binary search on self.symbols.
a = 0
b = len(self.symbols) - 1
while a <= b:
m = (a + b) // 2
symbol: DataSymbol = self.symbols[m]
addr_begin = symbol.addr
addr_end = addr_begin + symbol.size
if addr_begin <= addr < addr_end:
return symbol
if addr <= addr_begin:
b = m - 1
elif addr >= addr_end:
a = m + 1
else:
return None
return None
-169
View File
@@ -1,169 +0,0 @@
import io
import struct
from typing import Any, Dict, NamedTuple, Tuple
from elftools.elf.elffile import ELFFile
from elftools.elf.relocation import RelocationSection
from elftools.elf.sections import Section
import diff_settings
from util import utils
_config: Dict[str, Any] = {}
diff_settings.apply(_config, {})
_root = utils.get_repo_root()
base_elf_data = io.BytesIO((_root / _config["baseimg"]).read_bytes())
my_elf_data = io.BytesIO((_root / _config["myimg"]).read_bytes())
base_elf = ELFFile(base_elf_data)
my_elf = ELFFile(my_elf_data)
my_symtab = my_elf.get_section_by_name(".symtab")
if not my_symtab:
utils.fail(f'{_config["myimg"]} has no symbol table')
class Symbol(NamedTuple):
addr: int
name: str
size: int
class Function(NamedTuple):
data: bytes
addr: int
_ElfSymFormat = struct.Struct("<IBBHQQ")
class _ElfSym(NamedTuple):
st_name: int
info: int
other: int
shndx: int
st_value: int
st_size: int
@staticmethod
def parse(d: bytes):
return _ElfSym._make(_ElfSymFormat.unpack(d))
def get_file_offset(elf, addr: int) -> int:
for seg in elf.iter_segments():
if seg.header["p_type"] != "PT_LOAD":
continue
if seg["p_vaddr"] <= addr < seg["p_vaddr"] + seg["p_filesz"]:
return addr - seg["p_vaddr"] + seg["p_offset"]
raise KeyError(f"No segment found for {addr:#x}")
def is_in_section(section: Section, addr: int, size: int) -> bool:
begin = section["sh_addr"]
end = begin + section["sh_size"]
return begin <= addr < end and begin <= addr + size < end
_TableCache = dict()
def make_table_cached(symtab):
table = _TableCache.get(id(symtab))
if table is None:
table = build_name_to_symbol_table(symtab)
_TableCache[id(symtab)] = table
return table
def get_symbol(symtab, name: str) -> Symbol:
table = make_table_cached(symtab)
return table[name]
def get_symbol_file_offset_and_size(elf, table, name: str) -> (int, int):
sym = get_symbol(table, name)
return get_file_offset(elf, sym.addr), sym.size
def iter_symbols(symtab):
offset = symtab["sh_offset"]
entsize = symtab["sh_entsize"]
for i in range(symtab.num_symbols()):
symtab.stream.seek(offset + i * entsize)
entry = _ElfSym.parse(symtab.stream.read(_ElfSymFormat.size))
name = symtab.stringtable.get_string(entry.st_name)
yield Symbol(entry.st_value, name, entry.st_size)
def build_addr_to_symbol_table(symtab) -> Dict[int, str]:
table = dict()
for sym in iter_symbols(symtab):
addr = sym.addr
existing_value = table.get(addr, None)
if existing_value is None or not existing_value.startswith("_Z"):
table[addr] = sym.name
return table
def build_name_to_symbol_table(symtab) -> Dict[str, Symbol]:
return {sym.name: sym for sym in iter_symbols(symtab)}
def read_from_elf(elf: ELFFile, addr: int, size: int) -> bytes:
addr &= ~0x7100000000
offset: int = get_file_offset(elf, addr)
elf.stream.seek(offset)
return elf.stream.read(size)
def get_fn_from_base_elf(addr: int, size: int) -> Function:
return Function(read_from_elf(base_elf, addr, size), addr)
def get_fn_from_my_elf(name: str) -> Function:
sym = get_symbol(my_symtab, name)
return Function(read_from_elf(my_elf, sym.addr, sym.size), sym.addr)
R_AARCH64_GLOB_DAT = 1025
R_AARCH64_RELATIVE = 1027
def build_glob_data_table(elf: ELFFile) -> Dict[int, int]:
table: Dict[int, int] = dict()
section = elf.get_section_by_name(".rela.dyn")
assert isinstance(section, RelocationSection)
symtab = elf.get_section(section["sh_link"])
offset = symtab["sh_offset"]
entsize = symtab["sh_entsize"]
for reloc in section.iter_relocations():
symtab.stream.seek(offset + reloc["r_info_sym"] * entsize)
sym_value = _ElfSym.parse(symtab.stream.read(_ElfSymFormat.size)).st_value
info_type = reloc["r_info_type"]
if info_type == R_AARCH64_GLOB_DAT:
table[reloc["r_offset"]] = sym_value + reloc["r_addend"]
elif info_type == R_AARCH64_RELATIVE:
# FIXME: this should be Delta(S) + A
table[reloc["r_offset"]] = sym_value + reloc["r_addend"]
return table
def unpack_vtable_fns(vtable_bytes: bytes, num_entries: int) -> Tuple[int, ...]:
return struct.unpack(f"<{num_entries}Q", vtable_bytes[:num_entries * 8])
def get_vtable_fns_from_base_elf(vtable_addr: int, num_entries: int) -> Tuple[int, ...]:
vtable_bytes = read_from_elf(base_elf, vtable_addr, num_entries * 8)
return unpack_vtable_fns(vtable_bytes, num_entries)
def get_vtable_fns_from_my_elf(vtable_name: str, num_entries: int) -> Tuple[int, ...]:
offset, size = get_symbol_file_offset_and_size(my_elf, my_symtab, vtable_name)
my_elf.stream.seek(offset + 0x10)
vtable_bytes = my_elf.stream.read(size - 0x10)
return unpack_vtable_fns(vtable_bytes, num_entries)
-61
View File
@@ -1,61 +0,0 @@
from collections import defaultdict
_Visiting = 0
_Visited = 1
class Graph:
def __init__(self):
self.nodes = defaultdict(set)
def add_edge(self, a, b):
self.nodes[a].add(b)
def find_connected_components(self):
nodes = defaultdict(list)
for u in self.nodes:
for v in self.nodes[u]:
nodes[u].append(v)
nodes[v].append(u)
cc = []
visited = set()
def dfs(start):
result = []
to_visit = [start]
while to_visit:
x = to_visit.pop()
result.append(x)
visited.add(x)
for y in nodes[x]:
if y not in visited:
to_visit.append(y)
return result
for u in nodes.keys():
if u in visited:
continue
cc.append(dfs(u))
return cc
def topological_sort(self) -> list:
result = []
statuses = dict()
def dfs(node):
if statuses.get(node) == _Visiting:
raise RuntimeError("Graph is not acyclic")
if statuses.get(node) == _Visited:
return
statuses[node] = _Visiting
for y in self.nodes.get(node, set()):
dfs(y)
statuses[node] = _Visited
result.insert(0, node)
for x in self.nodes:
dfs(x)
return result
-129
View File
@@ -1,129 +0,0 @@
import io
from colorama import Fore, Style
import csv
import warnings
import enum
from pathlib import Path
import sys
import typing as tp
try:
import cxxfilt
except:
# cxxfilt cannot be used on Windows.
warnings.warn("cxxfilt could not be imported; demangling functions will fail")
class FunctionStatus(enum.Enum):
Matching = 0
Equivalent = 1 # semantically equivalent but not perfectly matching
NonMatching = 2
Wip = 3
NotDecompiled = 4
class FunctionInfo(tp.NamedTuple):
addr: int # without the 0x7100000000 base
name: str
size: int
decomp_name: str
library: bool
status: FunctionStatus
raw_row: tp.List[str]
_markers = {
"O": FunctionStatus.Matching,
"m": FunctionStatus.Equivalent,
"M": FunctionStatus.NonMatching,
"W": FunctionStatus.Wip,
"U": FunctionStatus.NotDecompiled,
"L": FunctionStatus.NotDecompiled,
}
def parse_function_csv_entry(row) -> FunctionInfo:
ea, stat, size, name = row
status = _markers.get(stat, FunctionStatus.NotDecompiled)
decomp_name = ""
if status != FunctionStatus.NotDecompiled:
decomp_name = name
addr = int(ea, 16) - 0x7100000000
return FunctionInfo(addr, name, int(size), decomp_name, stat == "L", status, row)
def get_functions_csv_path() -> Path:
return get_repo_root() / "data" / "uking_functions.csv"
def get_functions(path: tp.Optional[Path] = None) -> tp.Iterable[FunctionInfo]:
if path is None:
path = get_functions_csv_path()
with path.open() as f:
reader = csv.reader(f)
# Skip headers
next(reader)
for row in reader:
try:
entry = parse_function_csv_entry(row)
# excluded library function
if entry.library:
continue
yield entry
except ValueError as e:
raise Exception(f"Failed to parse line {reader.line_num}") from e
def add_decompiled_functions(new_matches: tp.Dict[int, str],
new_orig_names: tp.Optional[tp.Dict[int, str]] = None) -> None:
buffer = io.StringIO()
writer = csv.writer(buffer, lineterminator="\n")
for func in get_functions():
if new_orig_names is not None and func.status == FunctionStatus.NotDecompiled and func.addr in new_orig_names:
func.raw_row[3] = new_orig_names[func.addr]
if func.status == FunctionStatus.NotDecompiled and func.addr in new_matches:
func.raw_row[3] = new_matches[func.addr]
writer.writerow(func.raw_row)
get_functions_csv_path().write_text(buffer.getvalue())
def format_symbol_name(name: str) -> str:
try:
return f"{cxxfilt.demangle(name)} {Style.DIM}({name}){Style.RESET_ALL}"
except:
return name
def format_symbol_name_for_msg(name: str) -> str:
try:
return f"{Fore.BLUE}{cxxfilt.demangle(name)}{Fore.RESET} {Style.DIM}({name}){Style.RESET_ALL}{Style.BRIGHT}"
except:
return name
def are_demangled_names_equal(name1: str, name2: str):
return cxxfilt.demangle(name1) == cxxfilt.demangle(name2)
def print_note(msg: str, prefix: str = ""):
sys.stderr.write(f"{Style.BRIGHT}{prefix}{Fore.CYAN}note:{Fore.RESET} {msg}{Style.RESET_ALL}\n")
def warn(msg: str, prefix: str = ""):
sys.stderr.write(f"{Style.BRIGHT}{prefix}{Fore.MAGENTA}warning:{Fore.RESET} {msg}{Style.RESET_ALL}\n")
def print_error(msg: str, prefix: str = ""):
sys.stderr.write(f"{Style.BRIGHT}{prefix}{Fore.RED}error:{Fore.RESET} {msg}{Style.RESET_ALL}\n")
def fail(msg: str, prefix: str = ""):
print_error(msg, prefix)
sys.exit(1)
def get_repo_root() -> Path:
return Path(__file__).parent.parent.parent