diff --git a/tools/extractor/binary/pdb.py b/tools/extractor/binary/pdb.py index ebff28f..50c1bf8 100644 --- a/tools/extractor/binary/pdb.py +++ b/tools/extractor/binary/pdb.py @@ -48,7 +48,11 @@ def _find_pdbutil() -> str | None: # addr = 0001:00001080 _ADDR_RE = re.compile(r'addr\s*=\s*([0-9a-fA-F]+)\s*:\s*([0-9a-fA-F]+)', re.IGNORECASE) -_PROC_KINDS = frozenset({"S_GPROC32", "S_LPROC32", "S_GPROC32_ID", "S_LPROC32_ID"}) +_PROC_KINDS = frozenset({ + "S_GPROC32", "S_LPROC32", "S_GPROC32_ID", "S_LPROC32_ID", + "S_GPROC32_DPC", "S_LPROC32_DPC", + "S_GPROC32_DPC_ID", "S_LPROC32_DPC_ID", +}) _REC_START_RE = re.compile(r'^\s*(0x[0-9A-Fa-f]+)\s*\|\s*LF_(\w+)\b(.*)') _NAME_RE = re.compile(r'`([^`]+)`') @@ -57,10 +61,41 @@ def _find_pdbutil() -> str | None: _MEMBER_RE = re.compile(r'-\s*LF_MEMBER\s*\[name = `(.+?)`,.*?\boffset = (-?\d+)') _BCLASS_RE = re.compile(r'type\s*=\s*(0x[0-9A-Fa-f]+),\s*offset\s*=\s*(\d+)') _INDEX_RE = re.compile(r'-\s*LF_INDEX\s+continuation\s*=\s*(0x[0-9A-Fa-f]+)') +_SYM_TYPE_RE = re.compile(r'\btype\s*=\s*`?(0x[0-9A-Fa-f]+)') +_SIMPLE_TYPE_RE = { + 0x0003: ("void", "void"), + 0x0010: ("i8", "char"), + 0x0011: ("i16", "short"), + 0x0012: ("i32", "long"), + 0x0013: ("i64", "long long"), + 0x0020: ("u8", "unsigned char"), + 0x0021: ("u16", "unsigned short"), + 0x0022: ("u32", "unsigned long"), + 0x0023: ("u64", "unsigned long long"), + 0x0030: ("u8", "bool"), + 0x0040: ("f32", "float"), + 0x0041: ("f64", "double"), + 0x0068: ("i8", "int8_t"), + 0x0069: ("u8", "uint8_t"), + 0x0070: ("i8", "char"), + 0x0071: ("i16", "wchar_t"), + 0x0072: ("i16", "short"), + 0x0073: ("u16", "unsigned short"), + 0x0074: ("i32", "int"), + 0x0075: ("u32", "unsigned int"), + 0x0076: ("i64", "long long"), + 0x0077: ("u64", "unsigned long long"), + 0x0080: ("f32", "float"), + 0x0081: ("f64", "double"), +} _CLASS_KINDS = frozenset({"CLASS", "STRUCTURE", "UNION"}) +def _normalize_type_id(type_id: str) -> str: + return "0x" + type_id[2:].lower() + + def _resolve_fieldlist(fid, fieldlists): members: Dict[str, int] = {} bases = [] @@ -219,6 +254,7 @@ def _parse_pdbutil_int(s: str) -> int: pdbutil prints address fields as either: - plain decimal ("17573936") — section-relative offsets, code sizes - hex without 0x prefix ("010C2830", "1000") — when hex chars are present + - zero-padded 8-digit hex ("00000100") - hex with 0x prefix ("0x1000") — rare, but handle it Detect by presence of a–f; only base-10 chars are ambiguous. """ @@ -229,6 +265,9 @@ def _parse_pdbutil_int(s: str) -> int: if any(c in s for c in "abcdefABCDEF"): return int(s, 16) + if len(s) == 8 and s.startswith("0"): + return int(s, 16) + return int(s) def _resolve_rva(seg_str: str, off_str: str, section_map: Dict[int, int]) -> Optional[int]: @@ -249,8 +288,8 @@ def _resolve_rva(seg_str: str, off_str: str, section_map: Dict[int, int]) -> Opt def _parse_function_rvas(text: str, section_map: Dict[int, int]) -> Dict[str, int]: """ Walk the symbols section of a pdbutil dump and return - {mangled_name: rva_int} for every S_GPROC32 / S_LPROC32 record - that has a resolvable non-zero address. + {symbol_name: rva_int} for every supported procedure record with a + resolvable non-zero address. """ result: Dict[str, int] = {} current_name: Optional[str] = None @@ -279,6 +318,249 @@ def _parse_function_rvas(text: str, section_map: Dict[int, int]) -> Dict[str, in return result + +def _parse_type_records(lines) -> Dict[str, Dict[str, str]]: + records: Dict[str, Dict[str, str]] = {} + current_id: Optional[str] = None + for line in lines: + match = _REC_START_RE.match(line) + if match: + current_id, kind, rest = match.groups() + current_id = _normalize_type_id(current_id) + records[current_id] = {"kind": kind, "text": rest} + elif current_id is not None: + records[current_id]["text"] += "\n" + line + return records + + +def _dump_type_records(binary_path: str, ids: bool = False) -> Dict[str, Dict[str, str]]: + args = [_PDBUTIL, "dump", "-ids" if ids else "-types", binary_path] + proc = subprocess.Popen( + args, + stdout=subprocess.PIPE, + stderr=subprocess.DEVNULL, + text=True, + encoding="utf-8", + errors="ignore", + ) + try: + records = _parse_type_records(proc.stdout) + finally: + proc.stdout.close() + returncode = proc.wait() + if returncode != 0: + raise subprocess.CalledProcessError(returncode, args) + return records + + +def _record_ref(record: Optional[Dict[str, str]], label: str) -> Optional[str]: + if record is None: + return None + match = re.search( + rf'\b{re.escape(label)}\s*=\s*`?(0x[0-9A-Fa-f]+)', record["text"] + ) + return _normalize_type_id(match.group(1)) if match else None + + +def _record_name(record: Dict[str, str]) -> Optional[str]: + match = _NAME_RE.search(record["text"]) + return match.group(1) if match else None + + +def _resolve_codeview_type( + type_id: str, + type_records: Dict[str, Dict[str, str]], + seen: Optional[set[str]] = None, +) -> Dict[str, Optional[str]]: + """Resolve a CodeView type into the extractor's argument type shape.""" + type_id = _normalize_type_id(type_id) + seen = set() if seen is None else seen + if type_id in seen: + return {"kind": "ptr", "name": None, "view": None} + seen.add(type_id) + + try: + numeric_id = int(type_id, 16) + except ValueError: + return {"kind": "ptr", "name": None, "view": None} + + simple_id = numeric_id & 0xFF + pointer_mode = numeric_id & 0xF00 + if numeric_id < 0x1000 and pointer_mode and simple_id in _SIMPLE_TYPE_RE: + primitive_kind, primitive_name = _SIMPLE_TYPE_RE[simple_id] + view = primitive_name if primitive_name != "void" else None + return {"kind": "ptr", "name": primitive_name, "view": view} + primitive = _SIMPLE_TYPE_RE.get(numeric_id) + if primitive: + return {"kind": primitive[0], "name": primitive[1], "view": None} + + record = type_records.get(type_id) + if record is None: + return {"kind": "ptr", "name": None, "view": None} + + kind = record["kind"] + if kind == "POINTER": + referent = _record_ref(record, "referent") + if referent: + pointee = _resolve_codeview_type(referent, type_records, seen) + name = pointee.get("name") + return { + "kind": "ptr", + "name": name, + "view": name if name and name != "void" else None, + } + elif kind in {"MODIFIER", "ALIAS"}: + target = _record_ref( + record, + "modified type" if kind == "MODIFIER" else "underlying type", + ) + target = target or (_record_ref(record, "type") if kind == "ALIAS" else None) + if target: + return _resolve_codeview_type(target, type_records, seen) + elif kind in {"CLASS", "STRUCTURE", "UNION"}: + name = _record_name(record) + if name: + return {"kind": f"struct:{name}", "name": name, "view": name} + elif kind == "ENUM": + target = _record_ref(record, "underlying type") + if target: + return _resolve_codeview_type(target, type_records, seen) + + return {"kind": "ptr", "name": None, "view": None} + + +def _parse_arglist(record: Optional[Dict[str, str]]) -> list[str]: + if record is None or record["kind"] != "ARGLIST": + return [] + args = [] + for line in record["text"].splitlines(): + match = re.match(r'\s*(0x[0-9A-Fa-f]+)\s*:', line) + if match: + args.append(_normalize_type_id(match.group(1))) + return args + + +def _resolve_procedure( + type_id: str, + type_records: Dict[str, Dict[str, str]], + id_records: Dict[str, Dict[str, str]], +) -> Optional[Dict]: + type_id = _normalize_type_id(type_id) + seen: set[str] = set() + is_member_id = False + class_id = None + while type_id not in seen: + seen.add(type_id) + record = type_records.get(type_id) + if record is not None and record["kind"] in {"PROCEDURE", "MFUNCTION"}: + break + id_record = id_records.get(type_id) + if id_record is None or id_record["kind"] not in {"FUNC_ID", "MFUNC_ID"}: + return None + if id_record["kind"] == "MFUNC_ID": + is_member_id = True + class_id = _record_ref(id_record, "class type") + type_id = _record_ref(id_record, "function type") or "" + if not type_id: + return None + else: + return None + + record = type_records.get(type_id) + if record is None or record["kind"] not in {"PROCEDURE", "MFUNCTION"}: + return None + + return_id = _record_ref(record, "return type") + arglist_id = _record_ref(record, "param list") + if return_id is None: + return None + + args = [] + for index, arg_type_id in enumerate( + _parse_arglist(type_records.get(arglist_id or "")) + ): + arg_type = _resolve_codeview_type(arg_type_id, type_records) + arg = {"name": f"arg{index}", "kind": arg_type["kind"]} + if arg_type["view"]: + arg["view"] = arg_type["view"] + args.append(arg) + + return_type = _resolve_codeview_type(return_id, type_records) + ret = return_type["kind"] + if ret == "ptr" and return_type["name"]: + ret = f'{return_type["name"]}*' + + member = False + self_view = None + if record["kind"] == "MFUNCTION" or is_member_id: + class_id = class_id or _record_ref(record, "class type") + this_id = _record_ref(record, "this type") + class_type = _resolve_codeview_type(class_id, type_records) if class_id else None + class_name = class_type.get("name") if class_type else None + if is_member_id and class_name: + member = True + self_view = class_name + elif class_name and this_id: + this_type = _resolve_codeview_type(this_id, type_records) + if this_type.get("name") == class_name: + member = True + self_view = class_name + + return { + "ret": ret, + "args": args, + "member": member, + "self_view": self_view, + } + + +def _parse_function_signatures( + symbol_text: str, + section_map: Dict[int, int], + type_records: Dict[str, Dict[str, str]], + id_records: Optional[Dict[str, Dict[str, str]]] = None, +) -> Dict[str, Dict]: + """Return the selected signature for each symbol name with its resolved RVA.""" + id_records = id_records or {} + result: Dict[str, Dict] = {} + current: Optional[Dict[str, str]] = None + + def save() -> None: + if not current or not current.get("name") or not current.get("type"): + return + rva = current.get("rva") + signature = _resolve_procedure(current["type"], type_records, id_records) + if rva is None or signature is None: + return + entry = {**signature, "rva": rva} + name = current["name"] + if name not in result or rva > result[name]["rva"]: + result[name] = entry + + for line in symbol_text.splitlines(): + match = _SYM_REC_RE.match(line) + if match: + save() + current = None + if match.group(1) in _PROC_KINDS: + raw_name = match.group(2) + name = raw_name.strip(" `\r\n") if raw_name else "" + if name: + current = {"name": name} + continue + if current is None: + continue + type_match = _SYM_TYPE_RE.search(line) + if type_match: + current["type"] = _normalize_type_id(type_match.group(1)) + addr_match = _ADDR_RE.search(line) + if addr_match: + rva = _resolve_rva(addr_match.group(1), addr_match.group(2), section_map) + if rva is not None: + current["rva"] = rva + save() + return result + class PdbBackend(DebugInfoBackend): name = "pdb" @@ -304,6 +586,42 @@ def extract_rvas(self, binary_path: str) -> Dict[str, int]: section_map = _parse_section_headers(text) return _parse_function_rvas(text, section_map) + def extract_function_signatures(self, binary_path: str) -> Dict[str, Dict]: + if not _PDBUTIL: + sys.exit( + "pdb backend: llvm-pdbutil not found on PATH; " + "cannot extract PDB function signatures." + ) + + symbols = subprocess.run( + [_PDBUTIL, "dump", "-symbols", "-section-headers", binary_path], + capture_output=True, + text=True, + encoding="utf-8", + errors="ignore", + check=True, + ).stdout + section_map = _parse_section_headers(symbols) + if not any( + re.search(r'\|\s*S_[GL]PROC32(?:_DPC)?(?:_ID)?\b', line) + for line in symbols.splitlines() + ): + return {} + + type_records = _dump_type_records(binary_path) + + has_id_symbols = any( + re.search(r'\|\s*S_[GL]PROC32(?:_DPC)?_ID\b', line) + for line in symbols.splitlines() + ) + id_records = {} + if has_id_symbols: + id_records = _dump_type_records(binary_path, ids=True) + + return _parse_function_signatures( + symbols, section_map, type_records, id_records + ) + def image_identity(self, binary_path: str) -> Dict[str, int] | None: if not _PDBUTIL: return None diff --git a/tools/extractor/extract.py b/tools/extractor/extract.py index 97a00e5..b5c33f4 100644 --- a/tools/extractor/extract.py +++ b/tools/extractor/extract.py @@ -27,6 +27,8 @@ --ast-manifest build/ast_manifest.json \\ --binary build/game.elf \\ --output build/augment.bin + +# Add --pdb-signatures --debug-format pdb to extract signatures from a PDB. """ from __future__ import annotations @@ -310,6 +312,13 @@ def _collect_struct_views(manifest: Dict) -> set[str]: self_view = (fn.get("self_view") or "").strip() if self_view: views.add(self_view) + ret = (fn.get("ret") or "").strip() + if ret.startswith("struct:"): + views.add(ret[len("struct:"):].strip()) + elif ret.endswith("*"): + pointee = ret[:-1].strip().removeprefix("const ").strip() + if pointee: + views.add(pointee) for arg in fn.get("args") or []: view = (arg.get("view") or "").strip() if view: @@ -362,7 +371,8 @@ def phase2( verbose: bool = False, exclude_prefixes: tuple[str, ...] = (), exclude_substrs: tuple[str, ...] = (), - exclude_paths: tuple[str, ...] = () + exclude_paths: tuple[str, ...] = (), + pdb_signatures: bool = False, ) -> Dict: """ Load the ast_manifest.json produced by phase1, extract RVAs from the @@ -385,12 +395,22 @@ def phase2( log.info("extracting RVAs from %s", binary_path) backend = _select_backend(binary_path, debug_format) + signatures = None + if pdb_signatures: + if backend.name != "pdb": + sys.exit( + "[extract] --pdb-signatures requires --debug-format pdb " + "(or a .pdb input when the format is inferred)" + ) + signatures = backend.extract_function_signatures(binary_path) + log.info("got %d PDB function signatures", len(signatures)) rva_map = backend.extract_rvas(binary_path) log.info("got %d RVA entries", len(rva_map)) from extractor.merge import merge manifest = merge(ast, rva_map, exclude_prefixes=exclude_prefixes, - exclude_substrs=exclude_substrs) + exclude_substrs=exclude_substrs, + pdb_signatures=signatures) struct_layouts = backend.extract_struct_layouts(binary_path) if struct_layouts: @@ -501,6 +521,8 @@ def _build_parser() -> argparse.ArgumentParser: p2.add_argument("--debug-format", dest="debug_format", choices=["dwarf", "pdb"], default=None, help="Force debug-info format (inferred from extension if omitted)") # noqa # fmt: skip + p2.add_argument("--pdb-signatures", action="store_true", default=False, + help="Extract function signatures from the PDB (requires PDB debug info)") _add_common_args(p2) return root @@ -534,7 +556,8 @@ def main(argv: list | None = None) -> None: verbose=args.verbose, exclude_prefixes=tuple(args.exclude_prefixes), exclude_substrs=tuple(args.exclude_substrs), - exclude_paths=tuple(args.exclude_paths) + exclude_paths=tuple(args.exclude_paths), + pdb_signatures=args.pdb_signatures, ) diff --git a/tools/extractor/merge.py b/tools/extractor/merge.py index 3a91b84..ea8addd 100644 --- a/tools/extractor/merge.py +++ b/tools/extractor/merge.py @@ -113,6 +113,7 @@ def merge( rva_map: Dict[str, int], exclude_prefixes: tuple[str, ...] = (), exclude_substrs: tuple[str, ...] = (), + pdb_signatures: Optional[Dict[str, Dict]] = None, ) -> Dict[str, List[dict]]: rva_keys = list(rva_map.keys()) ast_fns = ast.get("functions", []) @@ -122,7 +123,8 @@ def merge( global_dm_cache = _demangle_batch(all_mangled_strings) functions = _merge_functions(ast_fns, rva_map, global_dm_cache, - exclude_prefixes, exclude_substrs) + exclude_prefixes, exclude_substrs, + pdb_signatures) _apply_flat_names(functions, global_dm_cache, exclude_prefixes, exclude_substrs) structs = sorted(ast.get("structs", []), key=lambda s: s["name"]) @@ -235,6 +237,7 @@ def _merge_functions( dm_cache: Dict[str, str], exclude_prefixes: tuple, exclude_substrs: tuple, + pdb_signatures: Optional[Dict[str, Dict]] = None, ) -> List[dict]: out: List[dict] = [] consumed_rva_keys: set = set() @@ -261,6 +264,17 @@ def _merge_functions( except Exception: pass + signatures_by_rva = { + signature["rva"]: signature + for signature in (pdb_signatures or {}).values() + if isinstance(signature.get("rva"), int) + } + + def signature_for(symbol: Optional[str], rva: Optional[int]) -> Optional[Dict]: + if pdb_signatures and symbol in pdb_signatures: + return pdb_signatures[symbol] + return signatures_by_rva.get(rva) + for fn in ast_fns: if _is_generated_artifact(fn): continue @@ -312,6 +326,9 @@ def _merge_functions( record["args"] = args record["rva"] = _rva_hex(rva_int) if rva_int is not None else None + _apply_pdb_signature( + record, signature_for(used_key or mangled, rva_int), preserve_ast=True + ) if used_key: if _is_mangled(used_key): @@ -361,13 +378,37 @@ def _merge_functions( break if not chosen: continue - out.append(_opaque_stub(chosen, rva_int)) + record = _opaque_stub(chosen, rva_int) + _apply_pdb_signature( + record, signature_for(chosen, rva_int), preserve_ast=False + ) + out.append(record) except Exception: pass return out +def _apply_pdb_signature(record: dict, signature: Optional[Dict], + preserve_ast: bool) -> None: + if not signature: + return + pdb_ret = signature.get("ret") + if record.get("ret") in (None, "", "ptr") and pdb_ret and pdb_ret != "ptr": + record["ret"] = pdb_ret + pdb_args = signature.get("args") + if not record.get("args") and pdb_args: + record["args"] = [dict(arg) for arg in pdb_args] + pdb_self_view = signature.get("self_view") + if preserve_ast: + if (record.get("member") or "member" not in record) \ + and not record.get("self_view") and pdb_self_view: + record["self_view"] = pdb_self_view + else: + record["member"] = bool(signature.get("member")) + record["self_view"] = pdb_self_view + + def _opaque_stub(mangled: str, rva_int: int) -> dict: return { "flat": mangled, diff --git a/tools/extractor/model.py b/tools/extractor/model.py index 1d7d873..c058301 100644 --- a/tools/extractor/model.py +++ b/tools/extractor/model.py @@ -85,7 +85,7 @@ class Function: flat: str # demangled, namespace-flattened name mangled: str # raw mangled symbol member: bool # is a non-static member function - ret: str # ffi kind string for return type + ret: str # ffi kind, or a resolved pointer spelling such as "Entity*" args: List[Arg] = field(default_factory=list) self_view: Optional[str] = None # owning struct name when member == True rva: Optional[str] = None # hex string e.g. "0x1a2b3c", or None