Go pclntab function recovery for IDA for Golang 1.2, 1.16, 1.18 and 1.20. Wrote with Codex (GPT)

gopclntab_recovery.py

# Minimal Go pclntab function recovery for IDA.
#
# Run from IDA with File > Script file..., or copy to IDA/plugins and invoke
# Edit > Plugins > Go pclntab recover.

import re
import struct

import ida_auto
import ida_bytes
import ida_funcs
import ida_ida
import ida_idaapi
import ida_name
import ida_segment
import idautils

try:
    import ida_dirtree
except Exception:
    ida_dirtree = None


PCLN_MAGICS = {
    0xFFFFFFFB: "v12",
    0xFFFFFFFA: "v116",
    0xFFFFFFF0: "v118",
    0xFFFFFFF1: "v120",
}
BUILDINFO_MAGIC = b"\xff Go buildinf:"
MAX_FUNCS = 2_000_000
MAX_SEG_SCAN = 128 * 1024 * 1024


class PclnHeader:
    def __init__(self, base, version, ptr_size):
        self.base = base
        self.version = version
        self.ptr_size = ptr_size
        self.quantum = read_u8(base + 6) or 1
        self.nfunc = 0
        self.nfiles = 0
        self.text_start = 0
        self.funcname_offset = 0
        self.cu_offset = 0
        self.filetab_offset = 0
        self.pctab_offset = 0
        self.pcln_offset = 0


class GoFunc:
    def __init__(self, ea, full_name, package, receiver, short_name):
        self.ea = ea
        self.full_name = full_name
        self.package = package
        self.receiver = receiver
        self.short_name = short_name
        self.name = sanitize_go_name(full_name)


def msg(s):
    print("[go-recover] %s" % s)


def read_bytes(ea, n):
    if ea == ida_idaapi.BADADDR or n <= 0:
        return None
    return ida_bytes.get_bytes(ea, n)


def read_u8(ea):
    b = read_bytes(ea, 1)
    return b[0] if b else None


def read_u32(ea):
    b = read_bytes(ea, 4)
    return struct.unpack("<I", b)[0] if b and len(b) == 4 else None


def read_u64(ea):
    b = read_bytes(ea, 8)
    return struct.unpack("<Q", b)[0] if b and len(b) == 8 else None


def read_uintp(ea, ptr_size):
    if ptr_size == 4:
        return read_u32(ea)
    if ptr_size == 8:
        return read_u64(ea)
    return None


def read_cstr(ea, max_len=512):
    b = read_bytes(ea, max_len)
    if not b:
        return ""
    end = b.find(b"\x00")
    if end >= 0:
        b = b[:end]
    try:
        return b.decode("utf-8")
    except UnicodeDecodeError:
        return ""


def read_uleb128(ea):
    value = 0
    shift = 0
    off = 0
    while off < 16:
        b = read_u8(ea + off)
        if b is None:
            return None
        off += 1
        value |= (b & 0x7F) << shift
        if (b & 0x80) == 0:
            return value, off
        shift += 7
    return None


def read_varint_bytes(ea):
    got = read_uleb128(ea)
    if not got:
        return None
    size, used = got
    if size > 1024 * 1024:
        return None
    b = read_bytes(ea + used, size)
    if b is None:
        return None
    return b, used + size


def is_valid_ea(ea):
    return ida_segment.getseg(ea) is not None and read_bytes(ea, 1) is not None


def best_text_start():
    entry = ida_ida.inf_get_start_ea()
    if is_valid_ea(entry):
        seg = ida_segment.getseg(entry)
        if seg:
            return seg.start_ea
    for start in idautils.Segments():
        seg = ida_segment.getseg(start)
        if seg and (seg.perm & ida_segment.SEGPERM_EXEC):
            return seg.start_ea
    return entry


def iter_segments():
    for start in idautils.Segments():
        seg = ida_segment.getseg(start)
        if seg:
            yield seg


def seg_name(seg):
    try:
        return ida_segment.get_segm_name(seg) or ""
    except Exception:
        return ""


def detect_pclntab():
    for seg in iter_segments():
        if ".gopclntab" in seg_name(seg):
            hit = try_pclntab_at(seg.start_ea)
            if hit:
                return hit

    likely = (".rodata", ".rdata", ".text", ".data.rel.ro", "__rodata", "__DATA_CONST")
    ordered = sorted(
        iter_segments(),
        key=lambda s: 0 if any(x in seg_name(s) for x in likely) else 1,
    )
    for seg in ordered:
        size = min(seg.end_ea - seg.start_ea, MAX_SEG_SCAN)
        ea = seg.start_ea
        end = seg.start_ea + size
        while ea + 8 <= end:
            magic = read_u32(ea)
            if magic in PCLN_MAGICS:
                hit = try_pclntab_at(ea)
                if hit:
                    return hit
            ea += 4
    return None


def try_pclntab_at(ea):
    magic = read_u32(ea)
    version = PCLN_MAGICS.get(magic)
    ptr_size = read_u8(ea + 7)
    if not version or ptr_size not in (4, 8):
        return None

    h = PclnHeader(ea, version, ptr_size)
    if version in ("v12", "v116"):
        h.nfunc = read_uintp(ea + 8, ptr_size) or 0
        if version == "v116":
            func_table = 8 + 2 * ptr_size
            after_table = func_table + h.nfunc * 2 * ptr_size
            h.funcname_offset = read_uintp(ea + after_table, ptr_size) or 0
    else:
        h.nfunc = read_u64(ea + 8) or 0
        h.nfiles = read_u64(ea + 16) or 0
        h.text_start = read_u64(ea + 24) or 0
        h.funcname_offset = read_u64(ea + 32) or 0
        h.cu_offset = read_u64(ea + 40) or 0
        h.filetab_offset = read_u64(ea + 48) or 0
        h.pctab_offset = read_u64(ea + 56) or 0
        h.pcln_offset = read_u64(ea + 64) or 0
        if not is_valid_ea(h.text_start):
            h.text_start = best_text_start()

    if h.nfunc == 0 or h.nfunc > MAX_FUNCS:
        return None
    return h


def parse_functions(h):
    if h.version == "v12":
        return parse_v12(h)
    if h.version == "v116":
        return parse_v116(h)
    return parse_v118(h)


def parse_v12(h):
    out = []
    p = h.ptr_size
    table = h.base + 8 + p
    for i in range(h.nfunc):
        ent = table + i * 2 * p
        entry_pc = read_uintp(ent, p)
        rec_off = read_uintp(ent + p, p)
        if entry_pc is None or rec_off is None or not is_valid_ea(entry_pc):
            continue
        name_off = read_u32(h.base + rec_off + p)
        if name_off is None:
            continue
        add_go_func(out, entry_pc, read_cstr(h.base + name_off))
    return out


def parse_v116(h):
    out = []
    p = h.ptr_size
    table = h.base + 8 + 2 * p
    names = h.base + h.funcname_offset
    for i in range(h.nfunc):
        ent = table + i * 2 * p
        entry_pc = read_uintp(ent, p)
        rec_off = read_uintp(ent + p, p)
        if entry_pc is None or rec_off is None or not is_valid_ea(entry_pc):
            continue
        name_off = read_u32(h.base + rec_off + p)
        if name_off is None:
            continue
        add_go_func(out, entry_pc, read_cstr(names + name_off))
    return out


def parse_v118(h):
    out = []
    table = h.base + h.pcln_offset
    names = h.base + h.funcname_offset
    bad = 0
    for i in range(h.nfunc):
        ent = table + i * 8
        pc_off = read_u32(ent)
        rec_off = read_u32(ent + 4)
        if pc_off is None or rec_off is None:
            continue
        entry_pc = h.text_start + pc_off
        if not is_valid_ea(entry_pc):
            bad += 1
            continue
        rec = table + rec_off
        name_off = read_u32(rec + 4)
        if name_off is None:
            continue
        add_go_func(out, entry_pc, read_cstr(names + name_off))
    if bad:
        msg("skipped %d funcs with invalid entry PCs" % bad)
    return out


def add_go_func(out, ea, full_name):
    if not full_name or not looks_like_go_name(full_name):
        return
    package, receiver, short = parse_go_func_name(full_name)
    out.append(GoFunc(ea, full_name, package, receiver, short))


def looks_like_go_name(s):
    if len(s) > 500 or "\n" in s or "\r" in s:
        return False
    return "." in s or s in ("init", "main")


def parse_go_func_name(full):
    slash = full.rfind("/") + 1
    dot = full.find(".", slash)
    if dot < 0:
        return "", None, full
    package = full[:dot]
    rest = full[dot + 1 :]
    if rest.startswith("("):
        end = rest.find(").")
        if end >= 0:
            return package, rest[: end + 1], rest[end + 2 :]
    return package, None, rest


def clean_component(s):
    s = re.sub(r"[^0-9A-Za-z_]", "_", s)
    s = re.sub(r"_+", "_", s).strip("_")
    if not s:
        return "go"
    if s[0].isdigit():
        return "_" + s
    return s


def sanitize_go_name(full):
    package, receiver, short = parse_go_func_name(full)
    if "." in package:
        package = package.rsplit("/", 1)[-1]
    parts = []
    if package:
        parts.append(clean_component(package))
    if receiver:
        parts.append(clean_component(receiver.replace("*", "").strip("()")))
    if short:
        parts.append(clean_component(short))
    return "_".join(parts) or "go_func"


def unique_name(base, ea, used):
    name = base
    if name in used:
        name = "%s_%x" % (base, ea)
    i = 1
    while name in used:
        name = "%s_%x_%d" % (base, ea, i)
        i += 1
    used.add(name)
    return name


def find_build_info(ptr_size):
    for seg in iter_segments():
        size = min(seg.end_ea - seg.start_ea, MAX_SEG_SCAN)
        off = 0
        while off + len(BUILDINFO_MAGIC) <= size:
            ea = seg.start_ea + off
            if read_bytes(ea, len(BUILDINFO_MAGIC)) == BUILDINFO_MAGIC:
                info = parse_build_info(ea, ptr_size)
                if info:
                    return info
            off += 16
    return {}


def parse_build_info(ea, fallback_ptr_size):
    hdr_ptr_size = read_u8(ea + 14)
    flags = read_u8(ea + 15)
    ptr_size = hdr_ptr_size if hdr_ptr_size in (4, 8) else fallback_ptr_size
    if flags is None:
        return {}

    if flags & 2:
        start = ea + 16 + 2 * ptr_size
        ver = read_varint_bytes(start)
        if not ver:
            return {}
        mod = read_varint_bytes(start + ver[1])
        if not mod:
            return {}
        raw = mod[0]
        if len(raw) > 32:
            raw = raw[16:-16]
        modinfo = raw.decode("utf-8", "ignore").rstrip("\x00")
    else:
        mod_ptr = read_uintp(ea + 16 + 2 * ptr_size, ptr_size)
        mod_len = read_uintp(ea + 16 + 3 * ptr_size, ptr_size)
        if not mod_ptr or not mod_len or mod_len > 1024 * 1024:
            return {}
        raw = read_bytes(mod_ptr, mod_len)
        if not raw:
            return {}
        modinfo = raw.decode("utf-8", "ignore").rstrip("\x00")

    return parse_modinfo(modinfo)


def parse_modinfo(text):
    info = {}
    for line in text.splitlines():
        fields = line.split("\t")
        if len(fields) >= 2 and fields[0] == "path":
            info["path"] = fields[1]
        elif len(fields) >= 2 and fields[0] == "mod":
            info["module"] = fields[1]
    return info


def root_package(path):
    cut = len(path)
    for ch in "[({":
        pos = path.find(ch)
        if pos >= 0:
            cut = min(cut, pos)
    s = path[:cut].rstrip(". ")
    if ":" in s:
        return s.split(":", 1)[0]
    if "/" not in s and "." in s:
        return s.split(".", 1)[0]
    return s


def classify_package(pkg, module_root):
    first = pkg.split("/", 1)[0]
    if "." not in first:
        return "stdlib"
    if module_root and pkg.startswith(module_root):
        return "user"
    return "thirdparty"


def folder_for_package(pkg, module_root):
    pkg = root_package(pkg)
    if not pkg:
        return "/unpackaged"
    if pkg == "main":
        return "/main"
    kind = classify_package(pkg, module_root)
    if kind == "stdlib":
        return "/_stdlib/" + clean_folder_path(pkg)
    if kind == "thirdparty":
        return "/_thirdparty/" + clean_folder_path(pkg.replace(".", "_"))
    return "/" + clean_folder_path(pkg)


def clean_folder_path(path):
    return "/".join(clean_component(p) for p in path.split("/") if p)


def apply_functions(funcs, module_root):
    used = set()
    renamed = 0
    created = 0
    moved = 0
    funcs = sorted({f.ea: f for f in funcs}.values(), key=lambda f: f.ea)
    addrs = [f.ea for f in funcs]

    tree = get_func_tree()
    for i, f in enumerate(funcs):
        end = addrs[i + 1] if i + 1 < len(addrs) else ida_idaapi.BADADDR
        pfn = ida_funcs.get_func(f.ea)
        if not pfn or pfn.start_ea != f.ea:
            if ida_funcs.add_func(f.ea, end):
                created += 1
                pfn = ida_funcs.get_func(f.ea)
        old_name = ida_funcs.get_func_name(f.ea) or ida_name.get_name(f.ea)
        new_name = unique_name(f.name, f.ea, used)
        if ida_name.set_name(f.ea, new_name, ida_name.SN_CHECK | ida_name.SN_NOWARN):
            renamed += 1
        else:
            new_name = unique_name("%s_%x" % (f.name, f.ea), f.ea, used)
            if ida_name.set_name(f.ea, new_name, ida_name.SN_CHECK | ida_name.SN_NOWARN):
                renamed += 1

        folder = folder_for_package(f.package, module_root)
        ida_bytes.set_cmt(
            f.ea,
            "Go: %s\nPackage: %s\nFolder: %s" % (f.full_name, f.package, folder),
            0,
        )
        if tree and move_function(tree, old_name, new_name, folder):
            moved += 1
        if pfn:
            ida_funcs.set_func_cmt(pfn, "Go package: %s" % (f.package or "unknown"), False)

    return created, renamed, moved


def get_func_tree():
    if ida_dirtree is None:
        return None
    try:
        tree = ida_dirtree.get_std_dirtree(ida_dirtree.DIRTREE_FUNCS)
        if tree:
            tree.load()
        return tree
    except Exception:
        return None


def mkdirs(tree, folder):
    cur = ""
    for part in folder.strip("/").split("/"):
        if not part:
            continue
        cur += "/" + part
        if not tree.isdir(cur):
            err = tree.mkdir(cur)
            if err not in (ida_dirtree.DTE_OK, ida_dirtree.DTE_ALREADY_EXISTS):
                return False
    return tree.isdir(folder)


def move_function(tree, old_name, new_name, folder):
    if not mkdirs(tree, folder):
        return False
    dest = folder.rstrip("/") + "/" + new_name
    for name in (new_name, old_name):
        if not name:
            continue
        src = find_tree_file(tree, name)
        if src and src != dest:
            return tree.rename(src, dest) == ida_dirtree.DTE_OK
    return False


def find_tree_file(tree, name):
    direct = "/" + name
    if tree.isfile(direct):
        return direct

    class Visitor(ida_dirtree.dirtree_visitor_t):
        def __init__(self):
            ida_dirtree.dirtree_visitor_t.__init__(self)
            self.path = None

        def visit(self, cursor, entry):
            if tree.isfile(entry) and tree.get_entry_name(entry) == name:
                self.path = tree.get_abspath(cursor)
                return 1
            return 0

    v = Visitor()
    try:
        tree.traverse(v)
    except Exception:
        return None
    return v.path


def run():
    ida_auto.auto_wait()
    h = detect_pclntab()
    if not h:
        msg("no Go pclntab found")
        return
    funcs = parse_functions(h)
    if not funcs:
        msg("pclntab found at %#x, but no functions parsed" % h.base)
        return
    build = find_build_info(h.ptr_size)
    module_root = build.get("module") or build.get("path")
    created, renamed, moved = apply_functions(funcs, module_root)
    msg(
        "pclntab=%#x %s funcs=%d created=%d renamed=%d moved=%d module=%s"
        % (h.base, h.version, len(funcs), created, renamed, moved, module_root or "")
    )


class GoRecoverPlugin(ida_idaapi.plugin_t):
    flags = ida_idaapi.PLUGIN_MOD | ida_idaapi.PLUGIN_UNL
    comment = "Recover Go functions from pclntab"
    help = "Recover Go functions from pclntab"
    wanted_name = "Go pclntab recover"
    wanted_hotkey = ""

    def init(self):
        return ida_idaapi.PLUGIN_OK

    def run(self, arg):
        run()

    def term(self):
        pass


def PLUGIN_ENTRY():
    return GoRecoverPlugin()


if __name__ == "__main__":
    run()