Merge pull request #3666 from MrMha/fix/export-audit-current

fix(import-audit): validate ELF and PE exports before reporting
This commit is contained in:
Adrià 2026-10-11 23:46:56 +02:00 committed by GitHub
commit cda695db72
No known key found for this signature in database
GPG Key ID: B5690EEEBB952194
2 changed files with 256 additions and 24 deletions

View File

@ -1,4 +1,5 @@
import argparse
import hashlib
import json
import re
import struct
@ -33,14 +34,14 @@ def unpack(layout, data, offset, path, what):
return struct.unpack(layout, take(data, offset, struct.calcsize(layout), path, what))
def cstring(data, offset, limit, path, what):
def cstring(data, offset, limit, path, what, encoding="utf-8", errors="surrogateescape"):
end = data.find(b"\0", offset, limit)
if offset < 0 or offset >= limit or end < 0:
raise AuditError(f"{path}: {what} at 0x{offset:x} is not a NUL-terminated string inside 0x{limit:x} bytes")
try:
return data[offset:end].decode("utf-8")
return data[offset:end].decode(encoding, errors=errors)
except UnicodeDecodeError as error:
raise AuditError(f"{path}: {what} at 0x{offset:x} is not UTF-8: {error}") from error
raise AuditError(f"{path}: {what} at 0x{offset:x} is not {encoding}: {error}") from error
def elf_exports(path, data):
@ -48,9 +49,14 @@ def elf_exports(path, data):
ident, section_offset, section_size, section_count = header[0], header[6], header[11], header[12]
if ident[4] != 2 or ident[5] != 1 or section_size != 64:
raise AuditError(f"{path}: only little-endian ELF64 with 64-byte section headers is supported (class {ident[4]}, data {ident[5]}, shentsize {section_size})")
if not section_offset:
raise AuditError(f"{path}: ELF exports without a section header table are not supported")
if not section_count:
section_count = unpack("<IIQQQQIIQQ", data, section_offset, path, "extended section count")[5]
take(data, section_offset, section_count * 64, path, "section header table")
sections = [unpack("<IIQQQQIIQQ", data, section_offset + index * 64, path, f"section header {index}") for index in range(section_count)]
names = set()
for section in sections:
for section_index, section in enumerate(sections):
kind, offset, size, link, entry = section[1], section[4], section[5], section[6], section[9]
if kind != 11:
continue
@ -58,10 +64,21 @@ def elf_exports(path, data):
raise AuditError(f"{path}: invalid dynamic symbol table (entry size {entry}, size {size}, string table section {link})")
strings_offset, strings_size = sections[link][4], sections[link][5]
take(data, strings_offset, strings_size, path, "dynamic string table")
indexes = [candidate for candidate in sections if candidate[1] == 18 and candidate[6] == section_index]
if len(indexes) > 1:
raise AuditError(f"{path}: multiple extended section index tables for section {section_index}")
for position in range(0, size, 24):
name, info, other, index, _, _ = unpack("<IBBHQQ", data, offset + position, path, "dynamic symbol")
if index == 0xffff:
if not indexes or indexes[0][9] != 4 or indexes[0][5] != size // 24 * 4:
raise AuditError(f"{path}: invalid or missing extended section index table for section {section_index}")
index = unpack("<I", data, indexes[0][4] + position // 24 * 4, path, "extended symbol section index")[0]
if index >= len(sections):
raise AuditError(f"{path}: extended symbol section index {index} lies outside the section table")
elif 0 < index < 0xff00 and index >= len(sections):
raise AuditError(f"{path}: symbol section index {index} lies outside the section table")
binding, visibility = info >> 4, other & 3
if index == 0 or binding not in (1, 2, 10) or visibility not in (0, 3):
if name == 0 or index == 0 or binding not in (1, 2, 10) or visibility not in (0, 3):
continue
names.add(cstring(data, strings_offset + name, strings_offset + strings_size, path, "export name"))
return names
@ -79,12 +96,19 @@ def pe_exports(path, data):
directory = PE_DIRECTORY_OFFSET[magic]
if optional_size < directory + 8:
raise AuditError(f"{path}: PE optional header of {optional_size} bytes has no export directory")
take(data, optional, optional_size, path, "PE optional header")
if unpack("<I", data, optional + directory - 4, path, "data directory count")[0] == 0:
raise AuditError(f"{path}: PE file declares no data directories")
headers_size = unpack("<I", data, optional + 60, path, "header size")[0]
rva, size = unpack("<II", data, optional + directory, path, "export data directory")
if not rva or size < 40:
raise AuditError(f"{path}: PE file has no export table")
sections = [unpack("<8sIIIIIIHHI", data, optional + optional_size + index * 40, path, f"PE section {index}") for index in range(section_count)]
def to_offset(address, length):
if address < headers_size and length <= headers_size - address:
take(data, address, length, path, f"header data at RVA 0x{address:x}")
return address, min(headers_size, len(data))
for section in sections:
virtual, raw_size, raw_offset = section[2], section[3], section[4]
if address < virtual:
@ -93,24 +117,43 @@ def pe_exports(path, data):
if relative >= raw_size or length > raw_size - relative:
continue
take(data, raw_offset + relative, length, path, f"data at RVA 0x{address:x}")
return raw_offset + relative
return raw_offset + relative, min(raw_offset + raw_size, len(data))
raise AuditError(f"{path}: RVA 0x{address:x} is not inside any section")
table = unpack("<IIHHIIIIIII", data, to_offset(rva, 40), path, "export directory")
name_count, names_rva = table[7], table[9]
table = unpack("<IIHHIIIIIII", data, to_offset(rva, 40)[0], path, "export directory")
function_count, name_count, functions_rva, names_rva, ordinals_rva = table[6:11]
names = set()
if not name_count:
return names
names_offset = to_offset(names_rva, name_count * 4)
if not functions_rva or not names_rva or not ordinals_rva:
raise AuditError(f"{path}: named PE exports require address, name and ordinal tables")
functions_offset = to_offset(functions_rva, function_count * 4)[0]
names_offset = to_offset(names_rva, name_count * 4)[0]
ordinals_offset = to_offset(ordinals_rva, name_count * 2)[0]
previous_name = None
for index in range(name_count):
name_rva = unpack("<I", data, names_offset + index * 4, path, "export name RVA")[0]
offset = to_offset(name_rva, 1)
names.add(cstring(data, offset, len(data), path, "export name"))
offset, limit = to_offset(name_rva, 1)
name = cstring(data, offset, limit, path, "export name", "ASCII", "strict")
if previous_name is not None and name < previous_name:
raise AuditError(f"{path}: PE export name pointer table is not sorted")
previous_name = name
ordinal = unpack("<H", data, ordinals_offset + index * 2, path, "export ordinal")[0]
if ordinal >= function_count:
raise AuditError(f"{path}: export ordinal {ordinal} lies outside the address table ({function_count} entries)")
function_rva = unpack("<I", data, functions_offset + ordinal * 4, path, "export address")[0]
if not function_rva:
continue
if rva <= function_rva < rva + size:
offset, limit = to_offset(function_rva, 1)
cstring(data, offset, min(limit, offset + rva + size - function_rva), path, "export forwarder", "ASCII", "strict")
names.add(name)
return names
def read_exports(path):
data = path.read_bytes()
def read_exports(path, data=None):
if data is None:
data = path.read_bytes()
if data[:4] == b"\x7fELF":
return elf_exports(path, data)
if data[:2] == b"MZ":
@ -128,13 +171,22 @@ def directory_files(directory, pattern):
def built_libraries(directories):
exports = defaultdict(set)
libraries = set()
images = {}
for directory in directories:
files = directory_files(directory, "*.prx")
if not files:
raise AuditError(f"{directory}: no .prx files; pass the directory the libs target builds (build/core/libs/libs)")
for file in files:
data = file.read_bytes()
digest = hashlib.sha256(data).digest()
if file.name in images:
previous, fingerprint = images[file.name]
if digest != fingerprint:
raise AuditError(f"ambiguous built library {file.name}: {previous} and {file} have different contents")
continue
images[file.name] = (file, digest)
libraries.add(file.name)
for name in read_exports(file):
for name in read_exports(file, data):
exports[name].add(file.name)
return {name: sorted(files) for name, files in exports.items()}, libraries
@ -236,10 +288,10 @@ def attach_names(records, names):
record["name_verified"] = nid_names.compute_nid(name) == record["nid"]
def git_stamp():
def git_stamp(source=ROOT):
try:
commit = subprocess.run(["git", "-C", str(ROOT), "rev-parse", "HEAD"], capture_output=True, text=True, check=True).stdout.strip()
dirty = bool(subprocess.run(["git", "-C", str(ROOT), "status", "--porcelain"], capture_output=True, text=True, check=True).stdout.strip())
commit = subprocess.run(["git", "-C", str(source), "rev-parse", "HEAD"], capture_output=True, text=True, check=True).stdout.strip()
dirty = bool(subprocess.run(["git", "-C", str(source), "status", "--porcelain"], capture_output=True, text=True, check=True).stdout.strip())
except (OSError, subprocess.CalledProcessError):
return None, None
return commit, dirty
@ -300,6 +352,14 @@ def main(argv=None):
parser.add_argument("--json", type=Path, help="write the full result to this file")
args = parser.parse_args(argv)
try:
if args.json:
inputs = args.registries + ([Path(args.names)] if args.names else [])
inputs += [file for directory in args.libs for file in directory_files(directory, "*.prx")]
inputs += [file for directory in args.modules for file in directory_files(directory, "*")]
inputs += list(Path(args.source).rglob("*.cpp"))
output = args.json.resolve()
if any(output == path.resolve() or (args.json.exists() and args.json.samefile(path)) for path in inputs):
raise AuditError(f"{args.json}: JSON output would overwrite an input file")
imports = [entry for registry in args.registries for entry in read_registry(registry)]
exports, libraries = built_libraries(args.libs)
modules = module_files(args.modules)
@ -308,7 +368,7 @@ def main(argv=None):
attach_names(records, load_names(args.names))
summary = summarize(records, len(imports), missing)
if args.json:
commit, dirty = git_stamp()
commit, dirty = git_stamp(Path(args.source))
result = {"source_commit": commit, "source_dirty": dirty, "registries": [str(path) for path in args.registries], **summary, "imports": records}
args.json.write_text(json.dumps(result, indent=2) + "\n", encoding="utf-8")
except AuditError as error:

View File

@ -47,21 +47,26 @@ def exporting(*names):
def build_pe(names):
section_rva, raw_offset = 0x1000, 0x200
strings, rvas = b"", []
start = 40 + 4 * len(names)
start = 40 + 10 * len(names)
for name in names:
rvas.append(section_rva + start + len(strings))
strings += name.encode() + b"\0"
raw = struct.pack("<IIHHIIIIIII", 0, 0, 0, 0, 0, 1, len(names), len(names), 0, section_rva + 40, 0)
raw += b"".join(struct.pack("<I", rva) for rva in rvas) + strings
raw = struct.pack("<IIHHIIIIIII", 0, 0, 0, 0, 0, 1, len(names), len(names), section_rva + 40, section_rva + 40 + 4 * len(names), section_rva + 40 + 8 * len(names))
raw += b"".join(struct.pack("<I", 0x2000 + 16 * index) for index in range(len(names)))
raw += b"".join(struct.pack("<I", rva) for rva in rvas)
raw += b"".join(struct.pack("<H", index) for index in range(len(names))) + strings
image = bytearray(raw_offset)
image[0:2] = b"MZ"
image[0x3c:0x40] = struct.pack("<I", 0x40)
image[0x40:0x44] = b"PE\0\0"
image[0x44:0x58] = struct.pack("<HHIIIHH", 0x8664, 1, 0, 0, 0, 240, 0x2022)
image[0x44:0x58] = struct.pack("<HHIIIHH", 0x8664, 2, 0, 0, 0, 240, 0x2022)
image[0x58:0x5a] = struct.pack("<H", 0x20b)
image[0x58 + 60:0x58 + 64] = struct.pack("<I", raw_offset)
image[0x58 + 108:0x58 + 112] = struct.pack("<I", 16)
image[0x58 + 112:0x58 + 120] = struct.pack("<II", section_rva, len(raw))
image[0x58 + 240:0x58 + 280] = struct.pack("<8sIIIIIIHHI", b".edata", len(raw), section_rva, len(raw), raw_offset, 0, 0, 0, 0, 0x40000040)
return bytes(image) + raw
image[0x58 + 280:0x58 + 320] = struct.pack("<8sIIIIIIHHI", b".text", 16 * max(1, len(names)), 0x2000, 16 * max(1, len(names)), raw_offset + len(raw), 0, 0, 0, 0, 0x60000020)
return bytes(image) + raw + b"\xc3" * (16 * max(1, len(names)))
def write_source(root, library, text):
@ -189,6 +194,9 @@ class ImportAuditTests(unittest.TestCase):
@unittest.skipUnless((import_audit.ROOT / ".git").exists(), "needs a git checkout")
def test_json_is_stamped_with_the_source_commit(self):
work = self.work
import_audit.subprocess.run(["git", "-C", str(work.source), "init", "-q"], check=True)
import_audit.subprocess.run(["git", "-C", str(work.source), "-c", "user.name=Audit test", "-c", "user.email=audit@example.invalid", "commit", "--allow-empty", "-qm", "Source stamp fixture"], check=True)
commit, dirty = import_audit.git_stamp(work.source)
(work.libs / "libSceA.prx").write_bytes(exporting("AAAAAAAAAAA"))
registry = work.root / "game.json"
write_registry(registry, [("AAAAAAAAAAA", "libSceA.prx")])
@ -196,7 +204,7 @@ class ImportAuditTests(unittest.TestCase):
work.run(registry, "--json", str(result))
data = json.loads(result.read_text())
self.assertRegex(data["source_commit"], r"^[0-9a-f]{40}$")
self.assertIsInstance(data["source_dirty"], bool)
self.assertEqual((data["source_commit"], data["source_dirty"]), (commit, dirty))
self.assertEqual(data["registries"], [str(registry)])
def test_nid_anchor_matches_the_known_pair(self):
@ -220,6 +228,170 @@ class ImportAuditTests(unittest.TestCase):
with self.assertRaisesRegex(import_audit.AuditError, "lib.prx"):
import_audit.pe_exports(path, image[:cut])
def test_pe_ignores_bytes_after_the_declared_directory_count(self):
image = bytearray(build_pe(["alpha"]))
struct.pack_into("<I", image, 0x58 + 108, 0)
with self.assertRaises(import_audit.AuditError):
import_audit.pe_exports(Path("lib.prx"), image)
def test_pe_null_export_address_is_not_an_export(self):
image = bytearray(build_pe(["alpha"]))
struct.pack_into("<I", image, 0x200 + 40, 0)
self.assertEqual(import_audit.pe_exports(Path("lib.prx"), image), set())
def test_pe_export_ordinal_must_index_the_address_table(self):
image = bytearray(build_pe(["alpha"]))
struct.pack_into("<H", image, 0x200 + 48, 1)
with self.assertRaises(import_audit.AuditError):
import_audit.pe_exports(Path("lib.prx"), image)
def test_pe_export_name_cannot_terminate_in_unmapped_overlay(self):
image = bytearray(build_pe(["alpha"]))
image[0x200 + 55] = ord("x")
image[0x238:] = b"y" * (len(image) - 0x238)
image += b"\0"
with self.assertRaises(import_audit.AuditError):
import_audit.pe_exports(Path("lib.prx"), image)
def test_pe_exports_can_reside_in_headers(self):
image = bytearray(build_pe(["alpha"]))
raw = bytes(image[0x200:0x238])
image[0x200:0x200] = bytes(0x200)
image[0x200:0x238] = raw
struct.pack_into("<I", image, 0x58 + 60, 0x400)
struct.pack_into("<I", image, 0x58 + 112, 0x200)
struct.pack_into("<III", image, 0x200 + 28, 0x228, 0x22c, 0x230)
struct.pack_into("<I", image, 0x22c, 0x232)
struct.pack_into("<I", image, 0x58 + 240 + 20, 0x400)
struct.pack_into("<I", image, 0x58 + 280 + 20, 0x438)
self.assertEqual(import_audit.pe_exports(Path("lib.prx"), image), {"alpha"})
def test_elf_extended_section_count_keeps_exports(self):
image = bytearray(exporting("alpha"))
section_offset = struct.unpack_from("<Q", image, 40)[0]
image.extend(bytes((0xff00 - 3) * 64))
struct.pack_into("<H", image, 60, 0)
struct.pack_into("<Q", image, section_offset + 32, 0xff00)
self.assertEqual(import_audit.elf_exports(Path("lib.prx"), image), {"alpha"})
def test_elf_extended_undefined_index_is_not_an_export(self):
image = bytearray(exporting("alpha"))
section_offset = struct.unpack_from("<Q", image, 40)[0]
symbol_offset = struct.unpack_from("<Q", image, section_offset + 64 + 24)[0]
index_offset = len(image) + 64
image.extend(struct.pack("<IIQQQQIIQQ", 0, 18, 0, 0, index_offset, 8, 1, 0, 4, 4))
image.extend(struct.pack("<II", 0, 0))
struct.pack_into("<H", image, 60, 4)
struct.pack_into("<H", image, symbol_offset + 24 + 6, 0xffff)
self.assertEqual(import_audit.elf_exports(Path("lib.prx"), image), set())
struct.pack_into("<I", image, index_offset + 4, 1)
self.assertEqual(import_audit.elf_exports(Path("lib.prx"), image), {"alpha"})
def test_pe_export_aliases_share_an_address(self):
image = bytearray(build_pe(["alpha", "beta"]))
struct.pack_into("<H", image, 0x200 + 56 + 2, 0)
self.assertEqual(import_audit.pe_exports(Path("lib.prx"), image), {"alpha", "beta"})
def test_pe32_uses_its_own_directory_layout(self):
image = bytearray(build_pe(["alpha"]))
struct.pack_into("<H", image, 0x58, 0x10b)
struct.pack_into("<I", image, 0x58 + 92, 16)
struct.pack_into("<II", image, 0x58 + 96, 0x1000, 56)
self.assertEqual(import_audit.pe_exports(Path("lib.prx"), image), {"alpha"})
def test_elf_unnamed_defined_symbol_is_not_an_export(self):
image = bytearray(exporting("alpha"))
section_offset = struct.unpack_from("<Q", image, 40)[0]
symbol_offset = struct.unpack_from("<Q", image, section_offset + 64 + 24)[0]
struct.pack_into("<I", image, symbol_offset + 24, 0)
self.assertEqual(import_audit.elf_exports(Path("lib.prx"), image), set())
def test_elf_preserves_non_utf8_symbol_bytes(self):
image = bytearray(exporting("alpha"))
image[65] = 0xff
self.assertEqual(import_audit.elf_exports(Path("lib.prx"), image), {"\udcfflpha"})
def test_elf_rejects_missing_regular_symbol_section(self):
image = build_elf([("alpha", 3, GLOBAL_FUNCTION)])
with self.assertRaisesRegex(import_audit.AuditError, "section index"):
import_audit.elf_exports(Path("lib.prx"), image)
self.assertEqual(import_audit.elf_exports(Path("lib.prx"), build_elf([("absolute", 0xfff1, GLOBAL_FUNCTION)])), {"absolute"})
def test_pe_forwarder_cannot_end_outside_export_directory(self):
image = bytearray(build_pe(["alpha"]))
struct.pack_into("<I", image, 0x200 + 40, 0x1032)
struct.pack_into("<II", image, 0x58 + 112, 0x1000, 55)
with self.assertRaisesRegex(import_audit.AuditError, "forwarder"):
import_audit.pe_exports(Path("lib.prx"), image)
def test_pe_forwarder_inside_export_directory_keeps_public_name(self):
image = bytearray(build_pe(["alpha"]))
target = b"NTDLL.#27\0"
image[0x238:0x238] = target
struct.pack_into("<I", image, 0x200 + 40, 0x1038)
struct.pack_into("<II", image, 0x58 + 112, 0x1000, 56 + len(target))
struct.pack_into("<I", image, 0x58 + 240 + 8, 56 + len(target))
struct.pack_into("<I", image, 0x58 + 240 + 16, 56 + len(target))
struct.pack_into("<I", image, 0x58 + 280 + 20, 0x238 + len(target))
self.assertEqual(import_audit.pe_exports(Path("lib.prx"), image), {"alpha"})
def test_pe_export_name_pointer_table_must_be_sorted(self):
image = build_pe(["beta", "alpha"])
with self.assertRaisesRegex(import_audit.AuditError, "sorted"):
import_audit.pe_exports(Path("lib.prx"), image)
def test_pe_export_name_must_be_ascii(self):
image = bytearray(build_pe(["alpha"]))
image[0x200 + 50:0x200 + 52] = b"\xc3\xa9"
with self.assertRaisesRegex(import_audit.AuditError, "ASCII"):
import_audit.pe_exports(Path("lib.prx"), image)
def test_libraries_with_conflicting_basenames_are_not_merged(self):
first, second = self.work.libs, self.work.root / "other-libs"
second.mkdir()
(first / "libSceA.prx").write_bytes(exporting("AAAAAAAAAAA"))
(second / "libSceA.prx").write_bytes(exporting("BBBBBBBBBBB"))
with self.assertRaisesRegex(import_audit.AuditError, "ambiguous"):
import_audit.built_libraries([first, second])
(second / "libSceA.prx").write_bytes(exporting("AAAAAAAAAAA"))
self.assertEqual(import_audit.built_libraries([first, second, first]), ({"AAAAAAAAAAA": ["libSceA.prx"]}, {"libSceA.prx"}))
def test_json_output_cannot_overwrite_registry_or_names(self):
work = self.work
(work.libs / "libSceA.prx").write_bytes(exporting("AAAAAAAAAAA"))
registry = work.root / "game.json"
write_registry(registry, [("AAAAAAAAAAA", "libSceA.prx")])
before = registry.read_bytes()
code, out, err = work.run(registry, "--json", str(registry))
self.assertEqual((code, out), (2, ""))
self.assertIn("input", err)
self.assertEqual(registry.read_bytes(), before)
names = work.root / "names.csv"
names.write_text("AAAAAAAAAAA name\n")
code, out, err = work.run(registry, "--json", str(names), "--names", str(names))
self.assertEqual((code, out), (2, ""))
self.assertEqual(names.read_text(), "AAAAAAAAAAA name\n")
def test_json_output_cannot_overwrite_hard_linked_input(self):
work = self.work
(work.libs / "libSceA.prx").write_bytes(exporting("AAAAAAAAAAA"))
registry = work.root / "game.json"
write_registry(registry, [("AAAAAAAAAAA", "libSceA.prx")])
output = work.root / "hard-link.json"
output.hardlink_to(registry)
before = registry.read_bytes()
code, out, err = work.run(registry, "--json", str(output))
self.assertEqual((code, out), (2, ""))
self.assertIn("input", err)
self.assertEqual(registry.read_bytes(), before)
def test_git_stamp_uses_selected_source_tree(self):
completed = type("Completed", (), {"stdout": "abcdef\n"})()
with patch.object(import_audit.subprocess, "run", return_value=completed) as run:
self.assertEqual(import_audit.git_stamp(self.work.source), ("abcdef", True))
self.assertEqual([call.args[0][2] for call in run.call_args_list], [str(self.work.source), str(self.work.source)])
def test_library_directories(self):
work = self.work
(work.libs / "libSceA.prx").write_bytes(build_pe(["AAAAAAAAAAA"]))