Improve standalone symbol extractor to not require a json crashlog

Author: tmolitor-stud-tu

tools/symbol_extractor/symbolicator.py

@@ -1,23 +1,45 @@
-import json
+#!/usr/bin/env python3
+
 import sqlite3
 import re
 import argparse
 from pathlib import Path
 
-def replace_redacted_in_crash_log(text_log: str, resolved_json: dict) -> str:
-    # Build symbol lookup table from resolved_json
-    addr_to_symbol = {}
-
-    threads = resolved_json.get("crash", {}).get("threads", [])
-    for thread in threads:
-        for frame in thread.get("backtrace", {}).get("contents", []):
-            addr = frame.get("instruction_addr")
-            symbol = frame.get("symbol_name")
-            lib = frame.get("object_name")
-            if addr is not None and symbol and symbol != "<redacted>" and lib is not None:
-                # print(f"{addr=}, {symbol=}, {lib=}")
-                addr_to_symbol[lib] = addr_to_symbol.get(lib, {})
-                addr_to_symbol[lib][addr] = symbol
+def extract_crash_metadata(crash: str) -> dict:
+    os_type, os_version, build_number, cpu_arch = None, None, None, None
+
+    for line in crash.splitlines():
+        line = line.strip()
+
+        # Example: "OS Version:          iOS 18.6.2 (22G100)"
+        if line.startswith("OS Version:"):
+            match = re.match(r"OS Version:\s+(\w+)\s+([\d\.]+)\s+\(([^)]+)\)", line)
+            if match:
+                os_type, os_version, build_number = match.groups()
+        # Example: "Code Type:           ARM-64 (Native)"
+        elif line.startswith("Code Type:"):
+            match = re.search(r"Code Type:\s+([A-Za-z0-9\-]+)", line)
+            if match:
+                arch = match.group(1).lower()
+                # Map Apple-style arch names to canonical strings
+                cpu_arch_map = {
+                    "arm-64": "arm64e",
+                    "arm64": "arm64",
+                    "x86-64": "x86_64"
+                }
+                cpu_arch = cpu_arch_map.get(arch, arch)
+
+    return {
+        "osType": os_type,
+        "osVersion": os_version,
+        "buildNumber": build_number,
+        "cpuArch": cpu_arch,
+    }
+
+def replace_redacted_in_crash_log(crash: str, sqlite_db_path: str) -> str:
+    metadata = extract_crash_metadata(crash)
+    conn = sqlite3.connect(sqlite_db_path)
+    cursor = conn.cursor()
 
     # Match stack frame lines, matches:
     # 7   Foundation                    	0x0000000199bc8500 0x199b11000 + 750848 (<redacted> + 212)
@@ -32,75 +54,44 @@ def replace_redacted_in_crash_log(text_log: str, resolved_json: dict) -> str:
     )
 
     def replacer(match):
+        offset = int(match.group("offset")) - int(match.group("delta"))
         lib = match.group("lib")
-        absaddr = int(match.group("absaddr"), 16)
-        symbol_name = addr_to_symbol.get(lib).get(absaddr)
-        if symbol_name:
-            return match.group(0).replace("<redacted>", symbol_name)
+        query = """
+            SELECT symbols.name
+            FROM symbols
+            JOIN files ON symbols.file_id = files.id
+            JOIN builds ON files.build_id = builds.id
+            WHERE symbols.address = ?
+                AND files.name = ?
+                AND builds.build = ?
+                AND builds.arch = ?
+            LIMIT 1;
+        """
+        # print("Searching for %s, %s, %s, %s" % (offset, lib, metadata["buildNumber"], metadata["cpuArch"]))
+        cursor.execute(query, (offset, lib, metadata["buildNumber"], metadata["cpuArch"]))
+        result = cursor.fetchone()
+        if result:
+            return match.group(0).replace("<redacted>", result[0])
         return match.group(0)
 
-    return frame_regex.sub(replacer, text_log)
-
-def resolve_redacted_symbols(data, sqlite_db_path):
-    # Connect to the SQLite database
-    conn = sqlite3.connect(sqlite_db_path)
-    cursor = conn.cursor()
-
-    os_version = data["system"]["os_version"]
-
-    for thread in data["crash"]["threads"]:
-        if "backtrace" not in thread:
-            continue
-        for frame in thread["backtrace"]["contents"]:
-            if frame.get("symbol_name") == "<redacted>":
-                symbol_addr = frame["symbol_addr"]
-                object_addr = frame["object_addr"]
-                object_name = frame["object_name"]
-
-                offset = symbol_addr - object_addr
-
-                query = """
-                    SELECT symbols.name
-                    FROM symbols
-                    JOIN files ON symbols.file_id = files.id
-                    JOIN builds ON files.build_id = builds.id
-                    WHERE symbols.address = ?
-                      AND files.name = ?
-                      AND builds.build = ?
-                    LIMIT 1;
-                """
-
-                cursor.execute(query, (offset, object_name, os_version))
-                result = cursor.fetchone()
-
-                if result:
-                    frame["symbol_name"] = result[0]
-                else:
-                    frame["symbol_name"] = "<unresolved>"
-
+    retval = frame_regex.sub(replacer, crash)
     conn.close()
-    return data
+    return retval
 
 
-parser = argparse.ArgumentParser(description="Replace '<redacted>' symbols in Apple crash logs using JSON crash data")
-parser.add_argument("json_file", help="Path to JSON crash report (*.json)")
+parser = argparse.ArgumentParser(description="Replace '<redacted>' symbols in Apple crash logs using symbols.db")
 parser.add_argument("crash_log", help="Path to Apple crash log (*.crash)")
 parser.add_argument("symbols_db", help="Path to symbols.db")
 args = parser.parse_args()
 
-json_path = Path(args.json_file)
 crash_log_path = Path(args.crash_log)
 symbols_db_path = Path(args.symbols_db)
-if not json_path.exists():
-    print(f"Error: JSON file not found: {json_path}", file=sys.stderr)
-    sys.exit(1)
 if not crash_log_path.exists():
     print(f"Error: Crash log file not found: {crash_log_path}", file=sys.stderr)
     sys.exit(1)
 if not symbols_db_path.exists():
     print(f"Warning: symbols.db file not found: {symbols_db_path}", file=sys.stderr)
 
-with open(json_path, 'r') as f:
-    data =resolve_redacted_symbols(json.load(f), symbols_db_path)
-    with open(crash_log_path, 'r') as t:
-        print(replace_redacted_in_crash_log(t.read(), data))
+with open(crash_log_path, 'r') as t:
+    symbolicated_crash = replace_redacted_in_crash_log(t.read(), symbols_db_path)
+    print(symbolicated_crash)