SalastilandClaude Opus 5.5
3b4786ff4d
Ghidra 12.1.2 headless decompilation of Steam client version 1788652215: - decomp/linux64_steamclient.so, ubuntu12_32_steamclient.so, ubuntu12_32_steam: per-class pseudo-C (RTTI class recovery + Class::Method string naming), functions.tsv and strings.tsv - decomp/protos: 107 .proto files and descriptors.pb extracted from steamclient.so - decomp/steamui: prettified Library UI JS; steamclient_js_api.txt lists the SteamClient.* backend API it calls - tools/: decompile/recover/export scripts and proto extractor GCC exception-handler analysis is disabled: it truncated ~47% of function bodies on these binaries. Co-Authored-By: Claude Opus 5.5 <[email protected]>
285 lines
10 KiB
Python
Executable File
285 lines
10 KiB
Python
Executable File
#!/usr/bin/env python3
|
|
"""Extract embedded protobuf FileDescriptorProtos from a binary.
|
|
|
|
Writes:
|
|
<out>/descriptors.pb FileDescriptorSet (usable directly by prost-build's compile_fds)
|
|
<out>/*.proto human-readable reconstructions
|
|
|
|
Usage: extract_protos.py <binary> [<binary>...] <out_dir>
|
|
"""
|
|
import re
|
|
import sys
|
|
from pathlib import Path
|
|
|
|
from google.protobuf import descriptor_pb2 as dpb
|
|
from google.protobuf.unknown_fields import UnknownFieldSet
|
|
|
|
FDP = dpb.FieldDescriptorProto
|
|
# Wire types of FileDescriptorProto fields; anything else ends the message.
|
|
FILE_FIELDS = {1: 2, 2: 2, 3: 2, 4: 2, 5: 2, 6: 2, 7: 2, 8: 2, 9: 2, 10: 0, 11: 0, 12: 2, 14: 0}
|
|
|
|
|
|
def varint(buf, i):
|
|
shift = val = 0
|
|
while True:
|
|
if i >= len(buf) or shift > 63:
|
|
raise ValueError
|
|
b = buf[i]
|
|
i += 1
|
|
val |= (b & 0x7F) << shift
|
|
if not b & 0x80:
|
|
return val, i
|
|
shift += 7
|
|
|
|
|
|
def scan_message_end(buf, start):
|
|
"""Walk wire-format fields from start, returning offset where FileDescriptorProto fields stop."""
|
|
i = start
|
|
end = start
|
|
while i < len(buf):
|
|
try:
|
|
tag, j = varint(buf, i)
|
|
field, wt = tag >> 3, tag & 7
|
|
if FILE_FIELDS.get(field) != wt:
|
|
break
|
|
if wt == 0:
|
|
_, j = varint(buf, j)
|
|
else:
|
|
ln, j = varint(buf, j)
|
|
j += ln
|
|
if j > len(buf):
|
|
break
|
|
except ValueError:
|
|
break
|
|
i = end = j
|
|
return end
|
|
|
|
|
|
def extract(data):
|
|
found = {}
|
|
for m in re.finditer(rb"\x0a([\x01-\x7f])([\w/.\-]+\.proto)", data):
|
|
if m.group(1)[0] != len(m.group(2)):
|
|
continue
|
|
start = m.start()
|
|
end = scan_message_end(data, start)
|
|
fd = dpb.FileDescriptorProto()
|
|
try:
|
|
fd.ParseFromString(data[start:end])
|
|
except Exception:
|
|
continue
|
|
if fd.name and (fd.message_type or fd.enum_type or fd.service or fd.extension):
|
|
prev = found.get(fd.name)
|
|
if prev is None or prev.ByteSize() < fd.ByteSize():
|
|
found[fd.name] = fd
|
|
return found
|
|
|
|
|
|
LABEL = {FDP.LABEL_OPTIONAL: "optional", FDP.LABEL_REQUIRED: "required", FDP.LABEL_REPEATED: "repeated"}
|
|
SCALAR = {v: k[5:].lower() for k, v in FDP.Type.items() if k not in ("TYPE_MESSAGE", "TYPE_ENUM", "TYPE_GROUP")}
|
|
|
|
|
|
EXTENSIONS = {} # (options message name, field number) -> (qualified name, field, enum values, file)
|
|
NEEDED_IMPORTS = set()
|
|
|
|
|
|
def index_extensions(files):
|
|
enums = {}
|
|
|
|
def walk_enums(prefix, msgs, es):
|
|
for e in es:
|
|
enums[prefix + "." + e.name] = {v.number: v.name for v in e.value}
|
|
for m in msgs:
|
|
walk_enums(prefix + "." + m.name, m.nested_type, m.enum_type)
|
|
|
|
for fd in files:
|
|
walk_enums("." + fd.package if fd.package else "", fd.message_type, fd.enum_type)
|
|
for fd in files:
|
|
pkg = "." + fd.package if fd.package else ""
|
|
for x in fd.extension:
|
|
EXTENSIONS[(x.extendee.rsplit(".", 1)[-1], x.number)] = (
|
|
(pkg + "." + x.name).lstrip("."), x, enums.get(x.type_name, {}), fd.name)
|
|
|
|
|
|
def type_name(f):
|
|
return f.type_name if f.type_name else SCALAR[f.type]
|
|
|
|
|
|
def fmt_options(opts):
|
|
"""Render set option fields (including extensions, shown as raw [field_number])."""
|
|
parts = []
|
|
for fd, val in opts.ListFields():
|
|
if fd.is_repeated:
|
|
continue
|
|
if isinstance(val, str):
|
|
val = '"%s"' % val.replace("\\", "\\\\").replace('"', '\\"')
|
|
elif isinstance(val, bool):
|
|
val = str(val).lower()
|
|
elif fd.enum_type is not None:
|
|
val = fd.enum_type.values_by_number[val].name
|
|
parts.append(("(%s)" % fd.full_name if fd.is_extension else fd.name, val))
|
|
# Steam's custom options are extensions defined in the extracted files themselves;
|
|
# resolve them by field number via EXTENSIONS (filled in main) and note the import.
|
|
for u in UnknownFieldSet(opts):
|
|
val = u.data
|
|
ext = EXTENSIONS.get((type(opts).__name__, u.field_number))
|
|
if ext:
|
|
name, f, enum_vals, src = ext
|
|
NEEDED_IMPORTS.add(src)
|
|
if f.type == FDP.TYPE_BOOL:
|
|
val = str(bool(val)).lower()
|
|
elif f.type == FDP.TYPE_ENUM and val in enum_vals:
|
|
val = enum_vals[val]
|
|
elif isinstance(val, bytes) and f.type == FDP.TYPE_STRING:
|
|
val = '"%s"' % val.decode("utf-8").replace("\\", "\\\\").replace('"', '\\"').replace("\n", "\\n")
|
|
parts.append(("(%s)" % name, val))
|
|
continue
|
|
if isinstance(val, bytes):
|
|
try:
|
|
val = '"%s"' % val.decode("utf-8").replace("\\", "\\\\").replace('"', '\\"').replace("\n", "\\n")
|
|
except UnicodeDecodeError:
|
|
val = repr(val)
|
|
parts.append(("(ext_%d)" % u.field_number, val))
|
|
return parts
|
|
|
|
|
|
def field_line(f, ind, syntax3=False):
|
|
label = "" if (syntax3 and f.label == FDP.LABEL_OPTIONAL and not f.proto3_optional) else LABEL[f.label] + " "
|
|
opts = []
|
|
if f.HasField("default_value"):
|
|
dv = f.default_value
|
|
if f.type in (FDP.TYPE_STRING,):
|
|
dv = '"%s"' % dv.replace("\\", "\\\\").replace('"', '\\"')
|
|
elif f.type == FDP.TYPE_BYTES:
|
|
dv = '"%s"' % dv
|
|
opts.append(("default", dv))
|
|
if f.HasField("options"):
|
|
opts += fmt_options(f.options)
|
|
o = " [%s]" % ", ".join("%s = %s" % kv for kv in opts) if opts else ""
|
|
return "%s%s%s %s = %d%s;" % (ind, label, type_name(f), f.name, f.number, o)
|
|
|
|
|
|
def emit_enum(e, ind, out):
|
|
out.append("%senum %s {" % (ind, e.name))
|
|
for k, v in fmt_options(e.options) if e.HasField("options") else []:
|
|
out.append("%s\toption %s = %s;" % (ind, k, v))
|
|
for v in e.value:
|
|
o = fmt_options(v.options) if v.HasField("options") else []
|
|
o = " [%s]" % ", ".join("%s = %s" % kv for kv in o) if o else ""
|
|
out.append("%s\t%s = %d%s;" % (ind, v.name, v.number, o))
|
|
out.append("%s}" % ind)
|
|
|
|
|
|
def emit_message(m, ind, out, syntax3):
|
|
out.append("%smessage %s {" % (ind, m.name))
|
|
i2 = ind + "\t"
|
|
for k, v in fmt_options(m.options) if m.HasField("options") else []:
|
|
out.append("%soption %s = %s;" % (i2, k, v))
|
|
for e in m.enum_type:
|
|
emit_enum(e, i2, out)
|
|
for n in m.nested_type:
|
|
if n.options.map_entry:
|
|
continue
|
|
emit_message(n, i2, out, syntax3)
|
|
oneofs = {}
|
|
for f in m.field:
|
|
if f.HasField("oneof_index") and not f.proto3_optional:
|
|
oneofs.setdefault(f.oneof_index, []).append(f)
|
|
done = set()
|
|
for f in m.field:
|
|
if f.HasField("oneof_index") and not f.proto3_optional:
|
|
idx = f.oneof_index
|
|
if idx in done:
|
|
continue
|
|
done.add(idx)
|
|
out.append("%soneof %s {" % (i2, m.oneof_decl[idx].name))
|
|
for of in oneofs[idx]:
|
|
out.append(field_line(of, i2 + "\t", syntax3).replace("optional ", "", 1))
|
|
out.append("%s}" % i2)
|
|
else:
|
|
out.append(field_line(f, i2, syntax3))
|
|
for r in m.extension_range:
|
|
out.append("%sextensions %d to %s;" % (i2, r.start, "max" if r.end >= 536870912 else r.end - 1))
|
|
if m.extension:
|
|
emit_extensions(m.extension, i2, out)
|
|
out.append("%s}" % ind)
|
|
|
|
|
|
def emit_extensions(exts, ind, out):
|
|
by = {}
|
|
for x in exts:
|
|
by.setdefault(x.extendee, []).append(x)
|
|
for ext, fs in by.items():
|
|
out.append("%sextend %s {" % (ind, ext))
|
|
for f in fs:
|
|
out.append(field_line(f, ind + "\t"))
|
|
out.append("%s}" % ind)
|
|
|
|
|
|
def render(fd):
|
|
NEEDED_IMPORTS.clear()
|
|
body = render_body(fd)
|
|
missing = sorted(NEEDED_IMPORTS - set(fd.dependency) - {fd.name})
|
|
if missing:
|
|
body = body.replace("\n\n", "\n\n" + "".join('import "%s";\n' % m for m in missing) + "\n", 1)
|
|
return body
|
|
|
|
|
|
def render_body(fd):
|
|
syntax3 = fd.syntax == "proto3"
|
|
out = ['syntax = "%s";' % (fd.syntax or "proto2"), ""]
|
|
if fd.package:
|
|
out += ["package %s;" % fd.package, ""]
|
|
for d in fd.dependency:
|
|
out.append('import "%s";' % d)
|
|
if fd.dependency:
|
|
out.append("")
|
|
for k, v in fmt_options(fd.options) if fd.HasField("options") else []:
|
|
out.append("option %s = %s;" % (k, v))
|
|
for e in fd.enum_type:
|
|
out.append("")
|
|
emit_enum(e, "", out)
|
|
for m in fd.message_type:
|
|
out.append("")
|
|
emit_message(m, "", out, syntax3)
|
|
if fd.extension:
|
|
out.append("")
|
|
emit_extensions(fd.extension, "", out)
|
|
for s in fd.service:
|
|
out.append("")
|
|
out.append("service %s {" % s.name)
|
|
for k, v in fmt_options(s.options) if s.HasField("options") else []:
|
|
out.append("\toption %s = %s;" % (k, v))
|
|
for mt in s.method:
|
|
o = fmt_options(mt.options) if mt.HasField("options") else []
|
|
body = " {\n%s\t}" % "".join("\t\toption %s = %s;\n" % kv for kv in o) if o else ";"
|
|
out.append("\trpc %s (%s%s) returns (%s%s)%s" % (
|
|
mt.name, "stream " if mt.client_streaming else "", mt.input_type,
|
|
"stream " if mt.server_streaming else "", mt.output_type, body))
|
|
out.append("}")
|
|
return "\n".join(out) + "\n"
|
|
|
|
|
|
def main():
|
|
if len(sys.argv) < 3:
|
|
sys.exit(__doc__)
|
|
out = Path(sys.argv[-1])
|
|
out.mkdir(parents=True, exist_ok=True)
|
|
found = {}
|
|
for b in sys.argv[1:-1]:
|
|
for name, fd in extract(Path(b).read_bytes()).items():
|
|
if name not in found or found[name].ByteSize() < fd.ByteSize():
|
|
found[name] = fd
|
|
index_extensions(found.values())
|
|
fds = dpb.FileDescriptorSet()
|
|
for name in sorted(found):
|
|
fds.file.append(found[name])
|
|
p = out / name
|
|
p.parent.mkdir(parents=True, exist_ok=True)
|
|
p.write_text(render(found[name]))
|
|
(out / "descriptors.pb").write_bytes(fds.SerializeToString())
|
|
print("extracted %d proto files -> %s" % (len(found), out))
|
|
|
|
|
|
if __name__ == "__main__":
|
|
main()
|