griches--PS5_DevTools/tools/ps5devtools.py

#!/usr/bin/env python3
# PS5 DevTools - the host side: load the payload, read crash reports with
# function names, and serve the dashboard with symbols.
# Copyright (C) 2026 Gary Riches
# SPDX-License-Identifier: GPL-3.0-or-later
"""Crash reports and memory graphs from a jailbroken PS5.

  ps5devtools.py load                 send the payload to the ELF loader (once a boot)
  ps5devtools.py status               the titles it has seen
  ps5devtools.py crashes              the crash reports on the console
  ps5devtools.py report [ID|FILE]     one report (the newest by default, or a
                                      .json someone sent), with function names
                                      from the build's ELF
  ps5devtools.py errors               the console's own error history
  ps5devtools.py serve [--port 8113]  the dashboard on this machine, with symbols
  ps5devtools.py test-crash [PID]     make an attached title fault, to test it all
  ps5devtools.py quit                 stop the payload

The console's address comes from PS5_HOST, or PS5_HOST= in .env beside this
tool's folder (or PS5_DevLoop's). ELFs are found by title id: every folder
beside PS5_DevTools whose sce_sys/param.json names the title, then its build/
*.elf files; the one whose code matches the report's hash is used. A
devtools.json beside this tool's folder can name more:
  {"elfs": {"PPSA99312": ["/path/to/eboot.elf"]}}
"""

import argparse
import base64
import glob
import http.server
import json
import os
import shlex
import shutil
import socket
import struct
import subprocess
import sys
import time
import urllib.error
import urllib.request
from pathlib import Path

ROOT = Path(__file__).resolve().parent.parent
WORKSPACE = ROOT.parent
PORT = 9113
LOADER_PORT = int(os.environ.get("PS5_LOADER_PORT", "9021"))


def console_host():
    if os.environ.get("PS5_HOST"):
        return os.environ["PS5_HOST"]
    for env in (ROOT / ".env", WORKSPACE / "PS5_DevLoop/.env"):
        if env.is_file():
            for line in env.read_text().splitlines():
                if line.startswith("PS5_HOST="):
                    return line.split("=", 1)[1].strip()
    raise SystemExit("set PS5_HOST, or PS5_HOST= in .env")


def get(path, timeout=10, method="GET"):
    request = urllib.request.Request(f"http://{console_host()}:{PORT}{path}", method=method)
    try:
        with urllib.request.urlopen(request, timeout=timeout) as reply:
            return reply.read(), reply.headers.get("Content-Type", "")
    except urllib.error.HTTPError as error:
        raise SystemExit(f"{path}: HTTP {error.code}")
    except OSError as error:
        raise SystemExit(f"the payload does not answer on port {PORT} ({error}); run `load` first")


def get_json(path, **kw):
    return json.loads(get(path, **kw)[0])


# ------------------------------------------------------------- symbols

def fnv64(data):
    h = 1469598103934665603
    for b in data:
        h = ((h ^ b) * 1099511628211) & 0xFFFFFFFFFFFFFFFF
    return h


_hash_cache = {}


def elf_text_hash(path):
    """FNV-1a 64 of the first 64 KiB of the segment loaded at address 0, as
    the payload hashes the title's code in memory."""
    st = os.stat(path)
    key = (path, st.st_mtime, st.st_size)
    if key in _hash_cache:
        return _hash_cache[key]
    result = None
    with open(path, "rb") as f:
        head = f.read(64)
        if head[:4] == b"\x7fELF" and head[4] == 2:
            phoff, = struct.unpack_from("<Q", head, 32)
            phentsize, phnum = struct.unpack_from("<HH", head, 54)
            f.seek(phoff)
            table = f.read(phentsize * phnum)
            for i in range(phnum):
                p_type, _, p_offset, p_vaddr = struct.unpack_from("<IIQQ", table, i * phentsize)
                if p_type == 1 and p_vaddr == 0:
                    f.seek(p_offset)
                    result = "%016x" % fnv64(f.read(65536))
                    break
    _hash_cache[key] = result
    return result


def candidate_elfs(title_id):
    found = []
    config = ROOT / "devtools.json"
    if config.is_file():
        found += json.loads(config.read_text()).get("elfs", {}).get(title_id, [])
    for param in glob.glob(str(WORKSPACE / "*/ps5/sce_sys/param.json")) + glob.glob(str(WORKSPACE / "*/sce_sys/param.json")):
        try:
            if json.loads(Path(param).read_text()).get("titleId") != title_id:
                continue
        except (OSError, ValueError):
            continue
        project = Path(param).parents[2] if "/ps5/sce_sys/" in param else Path(param).parents[1]
        for pattern in ("build/last-title.elf", "build/*.elf", "build/ps5/eboot.elf"):
            found += sorted(glob.glob(str(project / pattern)), key=os.path.getmtime, reverse=True)
    seen, unique = set(), []
    for path in found:
        if path not in seen and os.path.isfile(path):
            seen.add(path)
            unique.append(path)
    return unique


def pick_elf(report, override=None):
    """(path, matched): the build whose code matches the report, else the newest."""
    if override:
        return override, None
    want = next((m.get("text_fnv64") for m in report["modules"] if m.get("text_fnv64")), None)
    elfs = candidate_elfs(report["title_id"])
    for path in elfs:
        if want and elf_text_hash(path) == want:
            return path, True
    return (elfs[0], False) if elfs else (None, None)


def llvm(name):
    """the command for an LLVM tool (llvm-symbolizer, llvm-dwarfdump, llvm-nm): here, or in the build VM"""
    for candidate in (name, name + "-18"):
        if shutil.which(candidate):
            return [shutil.which(candidate)]
    for path in glob.glob(f"/opt/homebrew/opt/llvm*/bin/{name}") + glob.glob(f"/usr/local/opt/llvm*/bin/{name}"):
        return [path]
    if shutil.which("limactl"):
        # the workspace is mounted at the same path in the build VM
        return ["limactl", "shell", os.environ.get("PS5LOOP_VM", "ps5"), "--", name]
    return None


def run_each(tool, arguments):
    """one tool's output for each argument list, in one go when the tool is in the VM"""
    if tool[0].endswith("limactl"):
        marker = "@@devtools@@"
        script = "; ".join(f"echo {marker}; " + " ".join(shlex.quote(a) for a in [tool[-1]] + args) for args in arguments)
        out = subprocess.run(tool[:-1] + ["bash", "-c", script], capture_output=True, text=True, timeout=300).stdout
        return out.split(marker + "\n")[1:]
    return [subprocess.run(tool + args, capture_output=True, text=True, timeout=60).stdout for args in arguments]


def live_functions(path, offsets):
    """The offsets that DWARF puts inside a function the link kept.

    The linker leaves the debug information of the functions it discarded at
    address 0, so an address near the start of the code (the SDK's _start, which
    has none of its own) can land in one: an inlined piece of a removed function
    with no function around it, or a function whose low_pc is 0."""
    tool = llvm("llvm-dwarfdump")
    if not tool:
        return set(offsets)
    live = set()
    for offset, out in zip(offsets, run_each(tool, [[f"--lookup=0x{o:x}", path] for o in offsets])):
        in_function = False
        for line in out.splitlines():
            line = line.strip()
            if line.startswith("0x") and "DW_TAG_" in line:
                in_function = line.endswith("DW_TAG_subprogram")
            elif in_function and line.startswith("DW_AT_ranges"):
                live.add(offset)
            elif in_function and line.startswith("DW_AT_low_pc"):
                if int(line.split("(")[1].rstrip(")"), 16) != 0:
                    live.add(offset)
    return live


_symbols_cache = {}


def elf_symbols(path):
    """[(start, end, name)] of the ELF's functions, from its symbol table"""
    key = (path, os.path.getmtime(path))
    if key not in _symbols_cache:
        tool = llvm("llvm-nm")
        symbols = []
        if tool:
            out = subprocess.run(tool + ["-n", "-S", "-C", "--defined-only", path], capture_output=True, text=True, timeout=120).stdout
            for line in out.splitlines():
                parts = line.split(" ", 3)
                # "address size type name"; symbols with no size have no size column
                if len(parts) == 4 and len(parts[1]) == 16 and parts[2] in ("t", "T", "w", "W"):
                    start, size = int(parts[0], 16), int(parts[1], 16)
                    symbols.append((start, start + max(size, 1), parts[3]))
        _symbols_cache[key] = symbols
    return _symbols_cache[key]


def symbol_at(path, offset):
    best = None
    for start, end, name in elf_symbols(path):
        if start > offset:
            break
        if offset < end:
            best = name
    return best


def tidy(name):
    """libc++'s spellings, as people write them"""
    name = name.replace("std::__1::basic_string<char, std::__1::char_traits<char>, std::__1::allocator<char>>", "std::string")
    return name.replace("std::__1::", "std::").replace("[abi:ne180100]", "")


def symbolicate(report, elf=None):
    """{addr: {"function", "location"}} for every frame in the main executable."""
    path, matched = pick_elf(report, elf)
    tool = llvm("llvm-symbolizer")
    if not path or not tool:
        return {}, path, matched
    frames = list(report["backtrace"])
    for thread in report.get("threads", []):
        frames += thread["backtrace"]
    main = next((m["path"] for m in report["modules"] if m.get("main") or m.get("text_fnv64")), None)
    wanted = {}
    for f in frames:
        module = f["module"].rsplit("/", 1)[-1]
        if f["module"] == main or module in ("eboot.bin", "eboot.elf"):
            offset = int(f["offset"], 16)
            # a return address's call is the instruction before it
            wanted[f["addr"]] = offset if f["how"] == "pc" else offset - 1
    if not wanted:
        return {}, path, matched
    offsets = sorted(set(wanted.values()))
    out = subprocess.run(tool + ["--output-style=JSON", "--demangle", "--relativenames", f"--obj={path}"] +
                         ["0x%x" % o for o in offsets], capture_output=True, text=True, timeout=300)
    by_offset = {}
    entries = []
    for line in out.stdout.splitlines():
        try:
            parsed = json.loads(line)
        except ValueError:
            continue
        # one array for addresses given as arguments, one object a line from stdin
        entries += parsed if isinstance(parsed, list) else [parsed]
    for entry in entries:
        symbol = entry.get("Symbol") or [{}]
        inner = symbol[0]
        names = [tidy(s.get("FunctionName", "??")) for s in symbol]
        location = "%s:%s" % (inner.get("FileName", "?").replace(str(WORKSPACE) + "/", ""), inner.get("Line", 0))
        # inlined code: the function it was inlined into is the frame, the
        # innermost one is where the line is
        by_offset[int(entry["Address"], 16)] = {
            "function": names[-1] + (f" [inlined: {names[0].split('(')[0]}]" if len(names) > 1 else ""),
            "location": location if inner.get("FileName") else "",
        }
    # where the DWARF has no function of the build's own, the symbol table's name is the truth
    live = live_functions(path, offsets)
    for offset in offsets:
        if offset not in live:
            name = symbol_at(path, offset)
            by_offset[offset] = {"function": tidy(name) if name else "??", "location": ""}
    return {addr: by_offset.get(o, {"function": "??", "location": ""}) for addr, o in wanted.items()}, path, matched


# ------------------------------------------------------------ commands

def cmd_load(args):
    elf = Path(args.elf) if args.elf else ROOT / "build/ps5devtools.elf"
    if not elf.is_file():
        raise SystemExit(f"missing {elf}: run tools/build.sh")
    with socket.create_connection((console_host(), LOADER_PORT), timeout=10) as connection:
        connection.sendall(elf.read_bytes())
    print(f"sent {elf.name} ({elf.stat().st_size} bytes)")
    for _ in range(20):
        time.sleep(0.5)
        try:
            status = json.loads(urllib.request.urlopen(f"http://{console_host()}:{PORT}/api/status", timeout=3).read())
            print(f"PS5 DevTools {status['devtools']} is running: open http://<console>:{PORT}/ or run `serve`")
            return
        except OSError:
            continue
    raise SystemExit("sent, but the payload does not answer: read /data/devtools/devtools.log over FTP")


def cmd_status(args):
    status = get_json("/api/status")
    print(f"PS5 DevTools {status['devtools']}, up {status['now'] - status['started']} s")
    for t in status["titles"]:
        state = "running" if t["live"] else "ended"
        watch = ("attached" if t["traced"] else f"attach failed (errno {t['trace_error']})" if t["trace_error"]
                 else "not attached") if t["watch"] and t["live"] else "" if t["watch"] else "sampled only"
        print(f"  {t['title_id']}  {t['name']:<32} pid {t['pid']:<5} {state:<8} {watch + ', ' if watch else ''}{t['reports']} report(s)")


def cmd_crashes(args):
    crashes = get_json("/api/crashes")
    if not crashes:
        print("no crash reports")
    for c in crashes:
        print(f"{c['id']}  {c['signal_name']:<8} {c['where']:<36} {c['outcome']}")


def format_frames(frames, syms):
    lines = []
    for i, f in enumerate(frames):
        s = syms.get(f["addr"])
        module = f["module"].rsplit("/", 1)[-1]
        text = f"  #{i:<2} {f['addr']:>14}  {module}+{f['offset']}"
        if s:
            text += f"  {s['function']}" + (f"  ({s['location']})" if s["location"] else "")
        lines.append(text + ("" if f["how"] in ("pc", "fp") else f"  [{f['how']}]"))
    return lines


def cmd_report(args):
    report_id = args.id
    if report_id and report_id.endswith(".json") and os.path.isfile(report_id):
        # a report someone sent, downloaded from the dashboard or copied over FTP
        report = json.loads(Path(report_id).read_text())
        report_id = report["id"]
    else:
        if not report_id:
            crashes = get_json("/api/crashes")
            if not crashes:
                raise SystemExit("no crash reports")
            report_id = crashes[0]["id"]
        report = get_json("/api/crash/" + report_id)
    syms, elf, matched = symbolicate(report, args.elf)
    out = [f"{report['title_name']} ({report['title_id']} {report['title_version']}) {report['signal_name']} at {report['where']}",
           f"  {report['time']}, pid {report['pid']}, up {report['uptime_s']} s, outcome: {report['outcome']}",
           f"  fault address {report['fault_addr']}, signal {report['signal']} code {report['si_code']}, "
           f"thread {report['thread']['name']} ({report['thread']['id']})",
           f"  symbols: {os.path.relpath(elf, WORKSPACE) if elf else 'no ELF found'}" + ("" if matched is None else " (matches)" if matched else
                                                   " (DOES NOT MATCH the running build: names may be wrong)"),
           ]
    mine = next((f for f in report["backtrace"] if report.get("title_frame") and
                 f["module"].endswith("eboot.bin") and f["offset"] == report["title_frame"].split("+")[1]), None)
    if mine:
        s = syms.get(mine["addr"])
        out.append(f"  in the title's code: {report['title_frame']}" +
                   (f"  {s['function']}" + (f"  ({s['location']})" if s["location"] else "") if s else ""))
    out += ["", "backtrace:"] + format_frames(report["backtrace"], syms)
    out += ["", "registers:"]
    regs = list(report["registers"].items())
    for i in range(0, len(regs), 4):
        out.append("  " + "  ".join(f"{k:>6} {v:>18}" for k, v in regs[i:i + 4]))
    out += ["", f"code at pc: {report['code_at_pc']}"]
    for thread in report["threads"]:
        if thread["crashed"] and not args.all_threads:
            continue
        if args.all_threads:
            out += ["", f"thread {thread['id']} {thread['name']}" + (" (crashed)" if thread["crashed"] else "")]
            out += format_frames(thread["backtrace"], syms)
    memory = report.get("memory") or []
    if memory:
        out += ["", f"resident memory before the crash: {memory[0][1] // 1024} MiB -> {memory[-1][1] // 1024} MiB "
                    f"over {memory[-1][0] - memory[0][0]} s"]
    text = "\n".join(out)
    print(text)
    captures = ROOT / "captures"
    captures.mkdir(exist_ok=True)
    (captures / f"{report_id}.json").write_text(json.dumps(report, indent=1))
    (captures / f"{report_id}.txt").write_text(text + "\n")


def cmd_errors(args):
    entries = [e["entry"] for e in get_json("/api/errors")]
    entries.sort(key=lambda e: e.get("tick", 0))
    for e in entries[-args.last:]:
        when = time.strftime("%Y-%m-%d %H:%M:%S", time.localtime(e["tick"] / 1e6 - 62135596800))
        print(f"{when}  {e['shortError']:<12} {e['titleId']:<10} {e['messageBuffer'].splitlines()[0] if e['messageBuffer'] else ''}")


def cmd_fault(args):
    """make an attached title take a real page fault, to test the reporter"""
    pid = args.pid
    if not pid:
        live = [t for t in get_json("/api/status")["titles"] if t["live"] and t["traced"]]
        if not live:
            raise SystemExit("no attached title is running")
        pid = live[0]["pid"]
    print(get(f"/api/test/fault?pid={pid}" + ("&untraced=1" if args.untraced else ""), method="POST")[0].decode())


def cmd_quit(args):
    print(get("/api/quit", method="POST")[0].decode())


def cmd_serve(args):
    reports = {}

    class Handler(http.server.BaseHTTPRequestHandler):
        def log_message(self, *a):
            pass

        def send(self, code, body, kind="application/json"):
            self.send_response(code)
            self.send_header("Content-Type", kind)
            self.send_header("Content-Length", str(len(body)))
            self.send_header("Cache-Control", "no-store")
            self.end_headers()
            if self.command != "HEAD":
                self.wfile.write(body)

        def do_HEAD(self):
            self.do_GET()

        def do_GET(self):
            path = self.path
            if path in ("/", "/index.html"):
                return self.send(200, (ROOT / "web/dashboard.html").read_bytes(), "text/html; charset=utf-8")
            if path == "/api/symbolicate":
                return self.send(200, b"{}")
            if path.startswith("/api/symbolicate/"):
                report_id = path.rsplit("/", 1)[1]
                try:
                    if report_id not in reports:
                        report = json.loads(urllib.request.urlopen(
                            f"http://{console_host()}:{PORT}/api/crash/{report_id}", timeout=20).read())
                        syms, elf, matched = symbolicate(report)
                        reports[report_id] = {"frames": syms, "elf": elf, "matched": matched}
                    return self.send(200, json.dumps(reports[report_id]).encode())
                except Exception as error:  # noqa: BLE001 - shown in the dashboard as no symbols
                    return self.send(500, json.dumps({"error": str(error)}).encode())
            if path.startswith("/api/"):
                return self.proxy()
            self.send(404, b"{}")

        def do_POST(self):
            self.proxy()

        def do_DELETE(self):
            self.proxy()

        def proxy(self):
            request = urllib.request.Request(f"http://{console_host()}:{PORT}{self.path}", method=self.command)
            try:
                with urllib.request.urlopen(request, timeout=20) as reply:
                    self.send(reply.status, reply.read(), reply.headers.get("Content-Type", "application/json"))
            except urllib.error.HTTPError as error:
                self.send(error.code, error.read())
            except OSError as error:
                self.send(502, json.dumps({"error": str(error)}).encode())

    server = http.server.ThreadingHTTPServer(("127.0.0.1" if not args.lan else "0.0.0.0", args.port), Handler)
    print(f"dashboard: http://localhost:{args.port}/  (ctrl-c to stop)")
    try:
        server.serve_forever()
    except KeyboardInterrupt:
        pass


def main():
    parser = argparse.ArgumentParser(description=__doc__, formatter_class=argparse.RawDescriptionHelpFormatter)
    commands = parser.add_subparsers(dest="command", required=True)
    p = commands.add_parser("load", help="send the payload to the ELF loader")
    p.add_argument("elf", nargs="?")
    p.set_defaults(func=cmd_load)
    commands.add_parser("status", help="the titles the payload has seen").set_defaults(func=cmd_status)
    commands.add_parser("crashes", help="list the crash reports").set_defaults(func=cmd_crashes)
    p = commands.add_parser("report", help="one crash report, with function names")
    p.add_argument("id", nargs="?", help="a report id on the console, or a report's .json file")
    p.add_argument("--elf", help="the ELF to take names from, instead of finding it")
    p.add_argument("--all-threads", action="store_true")
    p.set_defaults(func=cmd_report)
    p = commands.add_parser("errors", help="the console's own error history")
    p.add_argument("--last", type=int, default=20)
    p.set_defaults(func=cmd_errors)
    p = commands.add_parser("serve", help="the dashboard here, with symbols")
    p.add_argument("--port", type=int, default=8113)
    p.add_argument("--lan", action="store_true", help="listen on every interface, not just this machine")
    p.set_defaults(func=cmd_serve)
    p = commands.add_parser("test-crash", help="make an attached title fault for real (it will close)")
    p.add_argument("pid", nargs="?", type=int)
    p.add_argument("--untraced", action="store_true", help="detach first: the crash as the system sees it with no debugger")
    p.set_defaults(func=cmd_fault)
    commands.add_parser("quit", help="stop the payload").set_defaults(func=cmd_quit)
    args = parser.parse_args()
    args.func(args)


if __name__ == "__main__":
    main()