594 lines
30 KiB
Python
594 lines
30 KiB
Python
import argparse
|
|
import hashlib
|
|
import json
|
|
import re
|
|
import struct
|
|
import subprocess
|
|
import sys
|
|
import tempfile
|
|
from collections import defaultdict
|
|
from pathlib import Path
|
|
|
|
import nid_names
|
|
import progress
|
|
|
|
ROOT = Path(__file__).resolve().parent.parent
|
|
PRX = ROOT / "core" / "libs" / "prx"
|
|
NID = re.compile(r"[A-Za-z0-9+\-]{11}")
|
|
NID_POSTFIX = "_nid_postfix"
|
|
MODULE_SUFFIX = "#"
|
|
CLASSES = ("implemented", "stub", "absent", "module")
|
|
PE_DIRECTORY_OFFSET = {0x20b: 112, 0x10b: 96}
|
|
|
|
|
|
class AuditError(Exception):
|
|
pass
|
|
|
|
|
|
def take(data, offset, size, path, what):
|
|
if offset < 0 or size < 0 or offset + size > len(data):
|
|
raise AuditError(f"{path}: {what} at 0x{offset:x} with size {size} lies outside the file ({len(data)} bytes)")
|
|
return data[offset:offset + size]
|
|
|
|
|
|
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, 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(encoding, errors=errors)
|
|
except UnicodeDecodeError as error:
|
|
raise AuditError(f"{path}: {what} at 0x{offset:x} is not {encoding}: {error}") from error
|
|
|
|
|
|
def elf_sections(path, data):
|
|
header = unpack("<16sHHIQQQIHHHHHH", data, 0, path, "ELF header")
|
|
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 file without a section header table is 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")
|
|
return [unpack("<IIQQQQIIQQ", data, section_offset + index * 64, path, f"section header {index}") for index in range(section_count)]
|
|
|
|
|
|
def elf_program_dynamic_symbols(path, data):
|
|
header = unpack("<16sHHIQQQIHHHHHH", data, 0, path, "ELF header")
|
|
ident, program_offset, program_size, program_count = header[0], header[5], header[9], header[10]
|
|
if ident[4] != 2 or ident[5] != 1 or program_size != 56 or not program_offset or not program_count:
|
|
raise AuditError(f"{path}: sectionless ELF needs little-endian ELF64 program headers")
|
|
take(data, program_offset, program_count * program_size, path, "program header table")
|
|
segments = [unpack("<IIQQQQQQ", data, program_offset + index * program_size, path, f"program header {index}") for index in range(program_count)]
|
|
loads = [segment for segment in segments if segment[0] == 1]
|
|
|
|
def file_offset(address, size, what):
|
|
for segment in loads:
|
|
_, _, offset, virtual, _, file_size, _, _ = segment
|
|
if address >= virtual and address - virtual <= file_size and size <= file_size - (address - virtual):
|
|
result = offset + address - virtual
|
|
take(data, result, size, path, what)
|
|
return result
|
|
raise AuditError(f"{path}: {what} address 0x{address:x} is outside file-backed load segments")
|
|
|
|
dynamic_segments = [segment for segment in segments if segment[0] == 2]
|
|
if len(dynamic_segments) != 1:
|
|
raise AuditError(f"{path}: expected one dynamic segment, found {len(dynamic_segments)}")
|
|
_, _, dynamic_offset, _, _, dynamic_size, _, _ = dynamic_segments[0]
|
|
if dynamic_size % 16:
|
|
raise AuditError(f"{path}: dynamic segment size {dynamic_size} is not a multiple of 16")
|
|
dynamic = {}
|
|
needed_offsets = []
|
|
for position in range(0, dynamic_size, 16):
|
|
tag, value = unpack("<qQ", data, dynamic_offset + position, path, "dynamic entry")
|
|
if tag == 0:
|
|
break
|
|
if tag == 1:
|
|
needed_offsets.append(value)
|
|
else:
|
|
dynamic[tag] = value
|
|
string_address = dynamic.get(5)
|
|
string_size = dynamic.get(10)
|
|
symbol_address = dynamic.get(6)
|
|
symbol_size = dynamic.get(11)
|
|
if None in (string_address, string_size, symbol_address, symbol_size) or symbol_size != 24 or string_size == 0:
|
|
raise AuditError(f"{path}: incomplete dynamic string or symbol table")
|
|
string_offset = file_offset(string_address, string_size, "dynamic string table")
|
|
|
|
symbol_count = None
|
|
if 4 in dynamic:
|
|
hash_offset = file_offset(dynamic[4], 8, "SysV hash table")
|
|
buckets, chains = unpack("<II", data, hash_offset, path, "SysV hash header")
|
|
take(data, hash_offset, 8 + 4 * (buckets + chains), path, "SysV hash table")
|
|
symbol_count = chains
|
|
elif 0x6ffffef5 in dynamic:
|
|
hash_offset = file_offset(dynamic[0x6ffffef5], 16, "GNU hash header")
|
|
buckets, first_symbol, bloom_count, _ = unpack("<IIII", data, hash_offset, path, "GNU hash header")
|
|
bucket_offset = hash_offset + 16 + bloom_count * 8
|
|
bucket_data = take(data, bucket_offset, buckets * 4, path, "GNU hash buckets")
|
|
chain_offset = bucket_offset + buckets * 4
|
|
maximum = first_symbol - 1
|
|
for index in struct.unpack(f"<{buckets}I", bucket_data) if buckets else ():
|
|
if index == 0:
|
|
continue
|
|
if index < first_symbol:
|
|
raise AuditError(f"{path}: GNU hash bucket index {index} precedes symbol offset {first_symbol}")
|
|
symbol_index = index
|
|
while True:
|
|
chain_address = dynamic[0x6ffffef5] + (chain_offset - hash_offset) + (symbol_index - first_symbol) * 4
|
|
chain_file_offset = file_offset(chain_address, 4, "GNU hash chain")
|
|
chain_value = unpack("<I", data, chain_file_offset, path, "GNU hash chain entry")[0]
|
|
maximum = max(maximum, symbol_index)
|
|
if chain_value & 1:
|
|
break
|
|
symbol_index += 1
|
|
symbol_count = maximum + 1
|
|
if symbol_count is None:
|
|
raise AuditError(f"{path}: dynamic symbol table has no supported hash table")
|
|
symbol_offset = file_offset(symbol_address, symbol_count * symbol_size, "dynamic symbol table")
|
|
imports = set()
|
|
exports = set()
|
|
for index in range(symbol_count):
|
|
name, info, other, section_index, _, _ = unpack("<IBBHQQ", data, symbol_offset + index * symbol_size, path, "dynamic symbol")
|
|
if name == 0:
|
|
continue
|
|
symbol = cstring(data, string_offset + name, string_offset + string_size, path, "dynamic symbol name").split(MODULE_SUFFIX, 1)[0]
|
|
if section_index == 0:
|
|
imports.add(symbol)
|
|
elif info >> 4 in (1, 2, 10) and other & 3 in (0, 3):
|
|
exports.add(symbol)
|
|
needed = {cstring(data, string_offset + value, string_offset + string_size, path, "needed library name") for value in needed_offsets}
|
|
return imports, exports, needed
|
|
|
|
|
|
def elf_exports(path, data):
|
|
header = unpack("<16sHHIQQQIHHHHHH", data, 0, path, "ELF header")
|
|
if not header[6]:
|
|
return elf_program_dynamic_symbols(path, data)[1]
|
|
sections = elf_sections(path, data)
|
|
names = set()
|
|
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
|
|
if entry != 24 or size % 24 or link >= len(sections) or sections[link][1] != 3:
|
|
raise AuditError(f"{path}: invalid dynamic symbol table (entry size {entry}, size {size}, string table section {link})")
|
|
take(data, offset, size, path, "dynamic symbol table")
|
|
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 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
|
|
|
|
|
|
def elf_imports(path, data):
|
|
header = unpack("<16sHHIQQQIHHHHHH", data, 0, path, "ELF header")
|
|
if not header[6]:
|
|
imports, _, needed = elf_program_dynamic_symbols(path, data)
|
|
return imports, needed
|
|
sections = elf_sections(path, data)
|
|
symbols = set()
|
|
needed = set()
|
|
for section in sections:
|
|
kind, offset, size, link, entry = section[1], section[4], section[5], section[6], section[9]
|
|
if kind == 11:
|
|
if entry != 24 or size % 24 or link >= len(sections) or sections[link][1] != 3:
|
|
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")
|
|
for position in range(0, size, 24):
|
|
name, _, _, index, _, _ = unpack("<IBBHQQ", data, offset + position, path, "dynamic symbol")
|
|
if name and index == 0:
|
|
symbol = cstring(data, strings_offset + name, strings_offset + strings_size, path, "import name")
|
|
symbols.add(symbol.split(MODULE_SUFFIX, 1)[0])
|
|
elif kind == 6:
|
|
if entry != 16 or size % 16 or link >= len(sections) or sections[link][1] != 3:
|
|
raise AuditError(f"{path}: invalid dynamic section (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")
|
|
for position in range(0, size, 16):
|
|
tag, value = unpack("<qQ", data, offset + position, path, "dynamic entry")
|
|
if tag == 1:
|
|
needed.add(cstring(data, strings_offset + value, strings_offset + strings_size, path, "needed library name"))
|
|
return symbols, needed
|
|
|
|
|
|
def pe_exports(path, data):
|
|
header = unpack("<I", data, 0x3c, path, "PE header offset")[0]
|
|
if unpack("<I", data, header, path, "PE signature")[0] != 0x4550:
|
|
raise AuditError(f"{path}: invalid PE signature at 0x{header:x}")
|
|
section_count, optional_size = unpack("<H", data, header + 6, path, "section count")[0], unpack("<H", data, header + 20, path, "optional header size")[0]
|
|
optional = header + 24
|
|
magic = unpack("<H", data, optional, path, "optional header magic")[0]
|
|
if magic not in PE_DIRECTORY_OFFSET:
|
|
raise AuditError(f"{path}: unsupported PE optional header magic 0x{magic:x}")
|
|
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:
|
|
continue
|
|
relative = address - virtual
|
|
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, 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)[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
|
|
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, 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=None):
|
|
if data is None:
|
|
data = path.read_bytes()
|
|
if data[:4] == b"\x7fELF":
|
|
return elf_exports(path, data)
|
|
if data[:2] == b"MZ":
|
|
return pe_exports(path, data)
|
|
raise AuditError(f"{path}: neither ELF nor PE (first bytes {data[:4]!r})")
|
|
|
|
|
|
def directory_files(directory, pattern):
|
|
path = Path(directory)
|
|
if not path.is_dir():
|
|
raise AuditError(f"{directory}: not a directory")
|
|
return sorted(item for item in path.glob(pattern) if item.is_file())
|
|
|
|
|
|
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, data):
|
|
exports[name].add(file.name)
|
|
return {name: sorted(files) for name, files in exports.items()}, libraries
|
|
|
|
|
|
def read_modules(directories):
|
|
imports = {}
|
|
exports = defaultdict(set)
|
|
for directory in directories:
|
|
for file in directory_files(directory, "*"):
|
|
data = file.read_bytes()
|
|
own = elf_imports(file, data)
|
|
symbols, needed = own
|
|
imports[file.name] = (sorted(symbols), sorted(needed))
|
|
for name in elf_exports(file, data):
|
|
exports[name].add(file.name)
|
|
return imports, {name: sorted(files) for name, files in exports.items()}
|
|
|
|
|
|
def nid_of(name, aliases):
|
|
if name in aliases:
|
|
return aliases[name]
|
|
return nid_names.compute_nid(name.removesuffix(NID_POSTFIX))
|
|
|
|
|
|
def source_stubs(source):
|
|
if not Path(source).is_dir():
|
|
raise AuditError(f"{source}: not a directory")
|
|
stubs = {}
|
|
shared = {}
|
|
for library in sorted(item for item in Path(source).iterdir() if item.is_dir()):
|
|
aliases = {}
|
|
for file in library.rglob("*.cpp"):
|
|
for nid, name in nid_names.source_aliases(file.read_text(errors="ignore")):
|
|
aliases[name] = nid
|
|
group = progress.scan_library(library)
|
|
if "shared_sources" in group:
|
|
shared[library.name] = group["shared_sources"]
|
|
done = {nid_of(name, aliases) for name in group["done_names"]}
|
|
for name in group["todo_names"]:
|
|
nid = nid_of(name, aliases)
|
|
if nid not in done:
|
|
stubs.setdefault((library.name + ".prx", nid), name)
|
|
for library, owner in shared.items():
|
|
chain = [library]
|
|
while owner in shared:
|
|
if owner in chain:
|
|
raise AuditError(f"{source}: shared source cycle: {' -> '.join(chain + [owner])}")
|
|
chain.append(owner)
|
|
owner = shared[owner]
|
|
for (provider, nid), name in list(stubs.items()):
|
|
if provider == owner + ".prx":
|
|
stubs[(library + ".prx", nid)] = name
|
|
return stubs
|
|
|
|
|
|
def read_registry(path):
|
|
try:
|
|
entries = json.loads(Path(path).read_text(encoding="utf-8-sig"))
|
|
except (OSError, UnicodeDecodeError, json.JSONDecodeError) as error:
|
|
raise AuditError(f"{path}: cannot read registry: {error}") from error
|
|
if not isinstance(entries, list):
|
|
raise AuditError(f"{path}: a registry is a JSON list, got {type(entries).__name__}")
|
|
imports = []
|
|
for index, entry in enumerate(entries):
|
|
if not isinstance(entry, dict) or not isinstance(entry.get("nid"), str) or not isinstance(entry.get("library"), str):
|
|
raise AuditError(f"{path}: entry {index} needs string fields 'nid' and 'library': {entry!r}")
|
|
nid = entry["nid"].split("#", 1)[0]
|
|
if not NID.fullmatch(nid):
|
|
raise AuditError(f"{path}: entry {index} has an invalid NID {entry['nid']!r}")
|
|
imports.append((nid, entry["library"]))
|
|
return imports
|
|
|
|
|
|
def check_conservation(references, records):
|
|
classified = sum(record["references"] for record in records)
|
|
unknown = sorted({record["class"] for record in records} - set(CLASSES))
|
|
if classified != references or unknown:
|
|
raise AuditError(f"conservation failed: {references} references in, {classified} classified, unknown classes {unknown}")
|
|
|
|
|
|
def classify(nid, library, providers, stubs, modules):
|
|
if library and library in modules:
|
|
return "module"
|
|
if not providers:
|
|
return "absent"
|
|
if all((provider, nid) in stubs for provider in providers):
|
|
return "stub"
|
|
return "implemented"
|
|
|
|
|
|
def audit(imports, exports, stubs, libraries, modules, module_imports=None, module_exports=None):
|
|
records = {}
|
|
for nid, library in imports:
|
|
record = records.setdefault((nid, library), {"nid": nid, "library": library, "references": 0})
|
|
record["references"] += 1
|
|
for module, (symbols, needed) in (module_imports or {}).items():
|
|
for nid in symbols:
|
|
record = records.setdefault((nid, None, module), {"nid": nid, "library": None, "module": module, "references": 0})
|
|
record["references"] += 1
|
|
for record in records.values():
|
|
nid, library = record["nid"], record["library"]
|
|
providers = exports.get(nid, [])
|
|
if library is None:
|
|
providers = sorted(set(providers) | set((module_exports or {}).get(nid, [])))
|
|
kind = classify(nid, library, providers, stubs, modules)
|
|
record["class"] = kind
|
|
record["providers"] = providers
|
|
record["library_mismatch"] = kind in ("implemented", "stub") and bool(library) and library not in providers and "libc.prx" not in providers
|
|
record["name"] = next((stubs[(provider, nid)] for provider in providers if (provider, nid) in stubs), None) if kind == "stub" else None
|
|
ordered = sorted(records.values(), key=lambda record: (record["class"], record["library"] or "", record.get("module") or "", record["nid"]))
|
|
check_conservation(len(imports) + sum(len(symbols) for symbols, _ in (module_imports or {}).values()), ordered)
|
|
missing = defaultdict(int)
|
|
for record in ordered:
|
|
library = record["library"]
|
|
if library and library not in libraries and library not in modules:
|
|
missing[library] += 1
|
|
for _, needed in (module_imports or {}).values():
|
|
for library in needed:
|
|
if library not in libraries and library not in modules:
|
|
missing[library] += 1
|
|
return ordered, dict(sorted(missing.items()))
|
|
|
|
|
|
def current_stop(imports, records):
|
|
"""The first import the loader cannot resolve, in registry order."""
|
|
classes = {(record["nid"], record["library"]): record for record in records}
|
|
for nid, library in imports:
|
|
record = classes.get((nid, library))
|
|
if record is not None and record["class"] == "absent":
|
|
return record
|
|
return None
|
|
|
|
|
|
def load_names(path):
|
|
if not Path(path).is_file():
|
|
raise AuditError(f"{path}: names file not found; the audit never downloads one (see tools/nid_names.py for the source)")
|
|
return nid_names.load_db(Path(path))
|
|
|
|
|
|
def attach_names(records, names):
|
|
for record in records:
|
|
if record["name"] is None and record["class"] == "absent" and record["nid"] in names:
|
|
name = names[record["nid"]]
|
|
record["name"] = name
|
|
record["name_verified"] = nid_names.compute_nid(name) == record["nid"]
|
|
|
|
|
|
def write_json(path, result):
|
|
if path.is_dir():
|
|
raise AuditError(f"{path}: JSON output is a directory")
|
|
destination = path.resolve()
|
|
with tempfile.TemporaryDirectory(dir=destination.parent, prefix=destination.name + ".", suffix=".tmp") as directory:
|
|
temporary = Path(directory) / destination.name
|
|
try:
|
|
temporary.write_text(json.dumps(result, indent=2) + "\n", encoding="utf-8")
|
|
if destination.exists():
|
|
temporary.chmod(destination.stat().st_mode & 0o7777)
|
|
temporary.replace(destination)
|
|
finally:
|
|
if temporary.exists():
|
|
temporary.chmod(0o600)
|
|
temporary.unlink(missing_ok=True)
|
|
|
|
|
|
def git_stamp(source=ROOT):
|
|
try:
|
|
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
|
|
|
|
|
|
def summarize(records, references, missing):
|
|
counts = {kind: sum(1 for record in records if record["class"] == kind) for kind in CLASSES}
|
|
by_references = {kind: sum(record["references"] for record in records if record["class"] == kind) for kind in CLASSES}
|
|
return {
|
|
"references": references,
|
|
"unique_imports": len(records),
|
|
"unique_by_class": counts,
|
|
"references_by_class": by_references,
|
|
"missing_libraries": missing,
|
|
"library_mismatch": sum(1 for record in records if record["library_mismatch"]),
|
|
}
|
|
|
|
|
|
def label(record):
|
|
if record["name"] is None:
|
|
text = record["nid"]
|
|
else:
|
|
mark = "" if record.get("name_verified", True) else " (name does not hash to this NID)"
|
|
text = f"{record['nid']} {record['name']}{mark}"
|
|
return f"{text} (imported by {record['module']})" if record.get("module") else text
|
|
|
|
|
|
def render(summary, records, missing, stops):
|
|
lines = [f"{summary['references']} references, {summary['unique_imports']} unique imports"]
|
|
for kind in CLASSES:
|
|
lines.append(f" {kind:<12}{summary['unique_by_class'][kind]:>6} imports {summary['references_by_class'][kind]:>6} references")
|
|
for title, kind in (("Absent: no built library exports these, so the loader fails", "absent"), ("Stub: exported but throws when called", "stub")):
|
|
selected = [record for record in records if record["class"] == kind]
|
|
if selected:
|
|
lines.append("")
|
|
lines.append(f"{title} ({len(selected)})")
|
|
current = False
|
|
for record in selected:
|
|
if record["library"] != current:
|
|
current = record["library"]
|
|
lines.append(f" [{current or 'library not named'}]")
|
|
lines.append(f" {label(record)}")
|
|
if stops:
|
|
lines.append("")
|
|
lines.append("Current stop (first unresolved import, in registry order)")
|
|
for entry in stops:
|
|
stop = entry["stop"]
|
|
lines.append(f" {entry['registry']}: none" if stop is None else f" {entry['registry']}: {label(stop)} [{stop['library']}]")
|
|
if missing:
|
|
lines.append("")
|
|
lines.append(f"Needed libraries with no file in --libs or --modules ({len(missing)})")
|
|
lines.extend(f" {library}: {count} imports" for library, count in missing.items())
|
|
if summary["library_mismatch"]:
|
|
lines.append("")
|
|
lines.append(f"{summary['library_mismatch']} imports are exported only by a library other than the one they name: they resolve on Linux and can fail on Windows")
|
|
return "\n".join(lines)
|
|
|
|
|
|
def main(argv=None):
|
|
parser = argparse.ArgumentParser(description="classify the imports of a converted game against the built system libraries")
|
|
parser.add_argument("registries", nargs="+", type=Path, help="JSON written by 'relinker --registry'")
|
|
parser.add_argument("--libs", action="append", required=True, help="directory of built .prx files; repeat for several")
|
|
parser.add_argument("--modules", action="append", default=[], help="directory of the title's own modules; each module's own imports are audited")
|
|
parser.add_argument("--source", default=str(PRX), help="core/libs/prx tree used to find throwing stubs (default: this repository's)")
|
|
parser.add_argument("--names", help="aerolib.csv style file (NID, name per line) used to name absent imports")
|
|
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")
|
|
registries = [(registry, read_registry(registry)) for registry in args.registries]
|
|
imports = [entry for _, entries in registries for entry in entries]
|
|
exports, libraries = built_libraries(args.libs)
|
|
module_imports, module_exports = read_modules(args.modules)
|
|
records, missing = audit(imports, exports, source_stubs(args.source), libraries, set(module_imports), module_imports, module_exports)
|
|
if args.names:
|
|
attach_names(records, load_names(args.names))
|
|
summary = summarize(records, sum(record["references"] for record in records), missing)
|
|
stops = [{"registry": str(path), "stop": current_stop(entries, records)} for path, entries in registries]
|
|
if args.json:
|
|
commit, dirty = git_stamp(Path(args.source))
|
|
json_stops = [{"registry": entry["registry"],
|
|
"nid": entry["stop"]["nid"] if entry["stop"] else None,
|
|
"library": entry["stop"]["library"] if entry["stop"] else None,
|
|
"name": entry["stop"]["name"] if entry["stop"] else None} for entry in stops]
|
|
result = {"source_commit": commit, "source_dirty": dirty, "registries": [str(path) for path in args.registries], **summary, "stops": json_stops, "imports": records}
|
|
write_json(args.json, result)
|
|
except AuditError as error:
|
|
print(f"FAIL: {error}", file=sys.stderr)
|
|
return 2
|
|
except OSError as error:
|
|
print(f"FAIL: {error.strerror}: {error.filename}" if error.filename else f"FAIL: {error}", file=sys.stderr)
|
|
return 2
|
|
print(render(summary, records, missing, stops))
|
|
return 1 if summary["unique_by_class"]["absent"] or missing else 0
|
|
|
|
|
|
if __name__ == "__main__":
|
|
if hasattr(sys.stdout, "reconfigure"):
|
|
sys.stdout.reconfigure(encoding="utf-8", errors="replace")
|
|
raise SystemExit(main())
|