import contextlib import io import json import os import stat import struct import subprocess import sys import tempfile import unittest from pathlib import Path from unittest.mock import patch sys.path.insert(0, str(Path(__file__).resolve().parents[1])) import import_audit import nid_names GLOBAL_FUNCTION = 0x12 LOCAL_FUNCTION = 0x02 STUB_SOURCE = "int APS5_VABI {name}(int value) {{ NotImplemented_nid_no_patch(__func__); return 0; }}\n" DONE_SOURCE = "int APS5_VABI {name}(int value) {{ return value; }}\n" MESSAGE_NID = "wTpfglkmv34" MESSAGE_NAME = "sceMsgDialogProgressBarSetValue" def build_elf(symbols, needed=()): strings = b"\0" entries = [struct.pack(" 2: record["weak"] = entry[2] records.append(record) path.write_text(json.dumps(records), encoding=encoding) class Workspace: def __init__(self): self._directory = tempfile.TemporaryDirectory() self.root = Path(self._directory.name) self.libs = self.root / "libs" self.modules = self.root / "modules" self.source = self.root / "prx" for path in (self.libs, self.modules, self.source): path.mkdir() def close(self): self._directory.cleanup() def run(self, registry, *extra, registries=()): out, err = io.StringIO(), io.StringIO() arguments = [str(registry), *map(str, registries), "--libs", str(self.libs), "--modules", str(self.modules), "--source", str(self.source), *extra] with contextlib.redirect_stdout(out), contextlib.redirect_stderr(err): code = import_audit.main(arguments) return code, out.getvalue(), err.getvalue() class ImportAuditTests(unittest.TestCase): def setUp(self): self.work = Workspace() self.addCleanup(self.work.close) def test_failed_json_write_preserves_existing_report_and_leaves_no_partial_report(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 / "audit.json" previous = b'{"previous": "complete report"}\n' def disk_full(path, content, **kwargs): with path.open("w", encoding="utf-8") as handle: handle.write(content[:24]) raise OSError(28, "No space left on device", str(path)) for exists in (True, False): with self.subTest(existing_report=exists): if exists: output.write_bytes(previous) else: output.unlink(missing_ok=True) files = set(work.root.iterdir()) with patch.object(Path, "write_text", disk_full): code, out, err = work.run(registry, "--json", str(output)) self.assertEqual((code, out), (2, "")) self.assertIn("No space left on device", err) self.assertEqual(set(work.root.iterdir()), files) if exists: self.assertEqual(output.read_bytes(), previous) else: self.assertFalse(output.exists()) def test_failed_json_replacement_preserves_report_and_cleans_staging(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 / "audit.json" previous = b'{"previous": "complete report"}\n' output.write_bytes(previous) files = set(work.root.iterdir()) with patch.object(Path, "replace", side_effect=PermissionError(13, "Permission denied", str(output))): code, out, err = work.run(registry, "--json", str(output)) self.assertEqual((code, out), (2, "")) self.assertIn("Permission denied", err) self.assertEqual(output.read_bytes(), previous) self.assertEqual(set(work.root.iterdir()), files) def test_successful_json_write_replaces_complete_report_without_staging_files(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 / "audit.json" output.write_text("old report", encoding="utf-8") files = set(work.root.iterdir()) code, _, err = work.run(registry, "--json", str(output)) self.assertEqual((code, err), (0, "")) self.assertEqual(json.loads(output.read_text(encoding="utf-8"))["unique_by_class"]["implemented"], 1) self.assertEqual(set(work.root.iterdir()), files) @unittest.skipUnless(sys.platform == "win32", "Windows read-only replacement behavior") def test_read_only_json_report_stays_intact_without_a_read_only_temporary(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 / "audit.json" previous = b'{"previous": "complete report"}\n' output.write_bytes(previous) files = set(work.root.iterdir()) output.chmod(0o444) try: code, out, err = work.run(registry, "--json", str(output)) self.assertEqual((code, out), (2, "")) self.assertIn("FAIL:", err) self.assertEqual(output.read_bytes(), previous) self.assertEqual(set(work.root.iterdir()), files) finally: output.chmod(0o666) def test_json_report_symlink_keeps_its_target(self): work = self.work (work.libs / "libSceA.prx").write_bytes(exporting("AAAAAAAAAAA")) registry = work.root / "game.json" write_registry(registry, [("AAAAAAAAAAA", "libSceA.prx")]) target = work.root / "actual-report.json" target.write_text("old report", encoding="utf-8") output = work.root / "audit.json" try: output.symlink_to(target.name) except (OSError, NotImplementedError) as error: self.skipTest(str(error)) files = set(work.root.iterdir()) code, _, err = work.run(registry, "--json", str(output)) self.assertEqual((code, err), (0, "")) self.assertTrue(output.is_symlink()) self.assertEqual(json.loads(target.read_text(encoding="utf-8"))["unique_imports"], 1) self.assertEqual(set(work.root.iterdir()), files) def test_json_directory_error_preserves_the_supplied_path(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.modules / ".." / work.modules.name self.assertNotEqual(str(output), str(output.resolve())) files = set(work.root.iterdir()) code, out, err = work.run(registry, "--json", str(output)) self.assertEqual((code, out), (2, "")) self.assertEqual(err, f"FAIL: {output}: JSON output is a directory\n") self.assertEqual(set(work.root.iterdir()), files) @unittest.skipUnless(os.name == "posix", "POSIX report permissions") def test_new_json_report_permissions_follow_the_process_umask(self): script = "\n".join([ "import os, sys", "from pathlib import Path", "sys.path.insert(0, sys.argv[1])", "import import_audit", "os.umask(int(sys.argv[4], 8))", "Path(sys.argv[3]).write_text('{}\\n', encoding='utf-8')", "import_audit.write_json(Path(sys.argv[2]), {'complete': True})", ]) for mask, mode in ((0o022, 0o644), (0o002, 0o664), (0o077, 0o600)): with self.subTest(umask=oct(mask)): output = self.work.root / f"audit-{mask:o}.json" reference = self.work.root / f"plain-{mask:o}.json" files = set(self.work.root.iterdir()) completed = subprocess.run([sys.executable, "-c", script, str(import_audit.ROOT / "tools"), str(output), str(reference), oct(mask)], capture_output=True, text=True) self.assertEqual(completed.returncode, 0, completed.stderr) self.assertEqual(stat.S_IMODE(output.stat().st_mode), mode) self.assertEqual(stat.S_IMODE(output.stat().st_mode), stat.S_IMODE(reference.stat().st_mode)) self.assertEqual(json.loads(output.read_text(encoding="utf-8")), {"complete": True}) self.assertEqual(set(self.work.root.iterdir()), files | {output, reference}) @unittest.skipUnless(os.name == "posix", "POSIX report permissions") def test_existing_json_report_permissions_are_preserved(self): output = self.work.root / "audit.json" for mode in (0o600, 0o640, 0o664): with self.subTest(mode=oct(mode)): output.write_text("old report", encoding="utf-8") output.chmod(mode) files = set(self.work.root.iterdir()) import_audit.write_json(output, {"complete": True}) self.assertEqual(stat.S_IMODE(output.stat().st_mode), mode) self.assertEqual(json.loads(output.read_text(encoding="utf-8")), {"complete": True}) self.assertEqual(set(self.work.root.iterdir()), files) def test_classification_accounts_for_every_reference(self): work = self.work (work.libs / "libSceMsgDialog.prx").write_bytes(build_elf([ (MESSAGE_NID, 1, GLOBAL_FUNCTION), ("AAAAAAAAAAA", 1, GLOBAL_FUNCTION), ("BBBBBBBBBBB", 1, GLOBAL_FUNCTION), ("UNDEFINED11", 0, GLOBAL_FUNCTION), ("LOCALLOCAL1", 1, LOCAL_FUNCTION)])) (work.libs / "libc.prx").write_bytes(exporting("CCCCCCCCCCC")) (work.modules / "libSceNpCppWebApi.prx").write_bytes(exporting()) write_source(work.source, "libSceMsgDialog", STUB_SOURCE.format(name=MESSAGE_NAME) + 'APS5_EXPORT("BBBBBBBBBBB", sceFooUnknown_B);\n' + STUB_SOURCE.format(name="sceFooUnknown_B")) registry = work.root / "game.registry.json" write_registry(registry, [ (MESSAGE_NID + "#A#B", "libSceMsgDialog.prx"), (MESSAGE_NID, "libSceMsgDialog.prx"), ("AAAAAAAAAAA", "libSceMsgDialog.prx"), ("BBBBBBBBBBB", "libSceMsgDialog.prx"), ("DDDDDDDDDDD", "libSceMsgDialog.prx"), ("EEEEEEEEEEE", "libSceNpCppWebApi.prx"), ("CCCCCCCCCCC", "libSceFoo.prx"), ("AAAAAAAAAAA", "libSceOther.prx"), ("UNDEFINED11", "libSceMsgDialog.prx"), ("LOCALLOCAL1", "libSceMsgDialog.prx")]) result = work.root / "audit.json" code, out, err = work.run(registry, "--json", str(result)) self.assertEqual((code, err), (1, "")) data = json.loads(result.read_text()) self.assertEqual(data["references"], 10) self.assertEqual(data["unique_imports"], 9) self.assertEqual(data["unique_by_class"], {"implemented": 3, "stub": 2, "absent": 3, "module": 1}) self.assertEqual(data["references_by_class"], {"implemented": 3, "stub": 3, "absent": 3, "module": 1}) self.assertEqual(data["missing_libraries"], {"libSceFoo.prx": 1, "libSceOther.prx": 1}) self.assertEqual(data["library_mismatch"], 1) kinds = {(record["nid"], record["library"]): record["class"] for record in data["imports"]} self.assertEqual(kinds[(MESSAGE_NID, "libSceMsgDialog.prx")], "stub") self.assertEqual(kinds[("BBBBBBBBBBB", "libSceMsgDialog.prx")], "stub") self.assertEqual(kinds[("UNDEFINED11", "libSceMsgDialog.prx")], "absent") self.assertEqual(kinds[("LOCALLOCAL1", "libSceMsgDialog.prx")], "absent") self.assertEqual(kinds[("EEEEEEEEEEE", "libSceNpCppWebApi.prx")], "module") self.assertIn(f"{MESSAGE_NID} {MESSAGE_NAME}", out) self.assertIn("DDDDDDDDDDD", out) def test_module_imports_are_audited(self): work = self.work stub = nid_names.compute_nid("sceModuleStub") (work.libs / "libSceA.prx").write_bytes(exporting("AAAAAAAAAAA")) (work.libs / "libSceB.prx").write_bytes(exporting(stub)) write_source(work.source, "libSceB", STUB_SOURCE.format(name="sceModuleStub")) (work.modules / "sibling.prx").write_bytes(exporting("BBBBBBBBBBB")) (work.modules / "main.prx").write_bytes(build_elf([ ("AAAAAAAAAAA", 0, GLOBAL_FUNCTION), ("BBBBBBBBBBB", 0, GLOBAL_FUNCTION), (stub, 0, GLOBAL_FUNCTION), ("EEEEEEEEEEE", 0, GLOBAL_FUNCTION)], ["libSceA.prx", "sibling.prx"])) registry = work.root / "game.json" write_registry(registry, []) result = work.root / "audit.json" code, out, err = work.run(registry, "--json", str(result)) self.assertEqual((code, err), (1, "")) data = json.loads(result.read_text()) self.assertEqual(data["references"], 4) self.assertEqual(data["unique_by_class"], {"implemented": 2, "stub": 1, "absent": 1, "module": 0}) self.assertEqual(data["missing_libraries"], {}) records = {(record["nid"], record["module"]): record for record in data["imports"]} self.assertEqual(records[("AAAAAAAAAAA", "main.prx")]["class"], "implemented") self.assertEqual(records[("AAAAAAAAAAA", "main.prx")]["providers"], ["libSceA.prx"]) self.assertEqual(records[("BBBBBBBBBBB", "main.prx")]["class"], "implemented") self.assertEqual(records[("BBBBBBBBBBB", "main.prx")]["providers"], ["sibling.prx"]) self.assertEqual(records[(stub, "main.prx")]["class"], "stub") self.assertEqual(records[(stub, "main.prx")]["name"], "sceModuleStub") self.assertEqual(records[("EEEEEEEEEEE", "main.prx")]["class"], "absent") self.assertIn("[library not named]", out) self.assertIn("EEEEEEEEEEE (imported by main.prx)", out) def test_module_sidecars_do_not_change_the_audit(self): work = self.work (work.libs / "libSceA.prx").write_bytes(exporting("AAAAAAAAAAA")) (work.modules / "provider.prx").write_bytes(exporting("BBBBBBBBBBB")) (work.modules / "consumer.prx").write_bytes(build_elf([ ("AAAAAAAAAAA", 0, GLOBAL_FUNCTION), ("BBBBBBBBBBB", 0, GLOBAL_FUNCTION)], ["libSceA.prx", "provider.prx"])) registry, result = work.root / "game.json", work.root / "audit.json" write_registry(registry, [("BBBBBBBBBBB", "provider.prx")]) code, before_out, err = work.run(registry, "--json", str(result)) self.assertEqual((code, err), (0, "")) before = json.loads(result.read_text()) for name, data in (("README", b"module notes\n"), ("notes.txt", b"EEEEEEEEEEE\n"), ("provider.registry.json", b'[{"nid":"EEEEEEEEEEE","library":"missing.prx"}]'), (".DS_Store", b"\x00\xffmetadata"), ("empty.txt", b"")): (work.modules / name).write_bytes(data) code, out, err = work.run(registry, "--json", str(result)) self.assertEqual((code, err), (0, "")) self.assertEqual(out, before_out) self.assertEqual(json.loads(result.read_text()), before) write_registry(registry, [("EEEEEEEEEEE", "notes.txt")]) code, _, err = work.run(registry, "--json", str(result)) self.assertEqual((code, err), (1, "")) data = json.loads(result.read_text()) record = next(record for record in data["imports"] if record["nid"] == "EEEEEEEEEEE") self.assertEqual((record["class"], record["providers"]), ("absent", [])) self.assertEqual(data["missing_libraries"], {"notes.txt": 1}) def test_elf_modules_are_audited_regardless_of_filename(self): work = self.work (work.libs / "libSceA.prx").write_bytes(exporting("AAAAAAAAAAA")) (work.modules / "provider.data").write_bytes(build_elf([ ("AAAAAAAAAAA", 0, GLOBAL_FUNCTION), ("BBBBBBBBBBB", 1, GLOBAL_FUNCTION)], ["libSceA.prx"])) (work.modules / "consumer.prx").write_bytes(build_elf([ ("BBBBBBBBBBB", 0, GLOBAL_FUNCTION)], ["provider.data"])) registry, result = work.root / "game.json", work.root / "audit.json" write_registry(registry, [("BBBBBBBBBBB", "provider.data")]) code, _, err = work.run(registry, "--json", str(result)) self.assertEqual((code, err), (0, "")) data = json.loads(result.read_text()) self.assertEqual(data["references"], 3) self.assertEqual(data["unique_by_class"], {"implemented": 2, "stub": 0, "absent": 0, "module": 1}) self.assertEqual(data["missing_libraries"], {}) records = {(record["nid"], record.get("module")): record for record in data["imports"]} self.assertEqual(records[("AAAAAAAAAAA", "provider.data")]["providers"], ["libSceA.prx"]) self.assertEqual(records[("BBBBBBBBBBB", "consumer.prx")]["providers"], ["provider.data"]) def test_bad_named_modules_and_malformed_elf_are_not_ignored(self): work = self.work (work.libs / "libc.prx").write_bytes(exporting()) registry, result = work.root / "game.json", work.root / "audit.json" write_registry(registry, []) previous = b'{"previous":"report"}\n' invalid_class = bytearray(exporting()) invalid_class[4] = 1 candidates = [("guest" + suffix, b"module placeholder\n") for suffix in (".prx", ".sprx", ".suprx", ".debug_prx", ".elf", ".PRX")] candidates += [("truncated.data", b"\x7fELF"), ("unsupported.data", invalid_class)] for name, image in candidates: module = work.modules / name module.write_bytes(image) for exists in (False, True): with self.subTest(name=name, existing_report=exists): if exists: result.write_bytes(previous) else: result.unlink(missing_ok=True) code, out, err = work.run(registry, "--json", str(result)) self.assertEqual((code, out), (2, "")) self.assertIn(name, err) self.assertIn("ELF", err) if exists: self.assertEqual(result.read_bytes(), previous) else: self.assertFalse(result.exists()) module.unlink() def test_pe_and_self_modules_are_not_ignored_as_sidecars(self): work = self.work (work.libs / "libc.prx").write_bytes(exporting()) registry, result = work.root / "game.json", work.root / "audit.json" write_registry(registry, []) previous = b'{"previous":"report"}\n' for image in (build_pe(["alpha"]), b"\x4f\x15\x3d\x1d" + bytes(64), b"\x54\x14\xf5\xee" + bytes(64)): for name in ("guest.prx", "metadata.txt"): module = work.modules / name module.write_bytes(image) with self.subTest(name=name, magic=bytes(image[:4])): result.write_bytes(previous) code, out, err = work.run(registry, "--json", str(result)) self.assertEqual((code, out), (2, "")) self.assertIn(name, err) self.assertIn("ELF", err) self.assertEqual(result.read_bytes(), previous) module.unlink() def test_module_needed_library_without_a_file_is_missing(self): work = self.work (work.libs / "libc.prx").write_bytes(exporting()) (work.modules / "main.prx").write_bytes(build_elf([], ["libSceMissing.prx"])) registry = work.root / "game.json" write_registry(registry, []) result = work.root / "audit.json" code, out, err = work.run(registry, "--json", str(result)) self.assertEqual((code, err), (1, "")) data = json.loads(result.read_text()) self.assertEqual((data["references"], data["imports"]), (0, [])) self.assertEqual(data["missing_libraries"], {"libSceMissing.prx": 1}) self.assertIn("libSceMissing.prx", out) def test_conflicting_module_files_preserve_or_leave_no_json_report(self): work = self.work first, second = work.modules, work.root / "other-modules" second.mkdir() (work.libs / "libc.prx").write_bytes(exporting()) (first / "main.prx").write_bytes(build_elf([ ("AAAAAAAAAAA", 0, GLOBAL_FUNCTION), ("BBBBBBBBBBB", 1, GLOBAL_FUNCTION)], ["libSceMissing.prx"])) registry, result = work.root / "game.json", work.root / "audit.json" write_registry(registry, []) code, _, err = work.run(registry, "--json", str(result)) self.assertEqual((code, err), (1, "")) data = json.loads(result.read_text()) self.assertEqual(data["unique_by_class"]["absent"], 1) self.assertEqual(data["missing_libraries"], {"libSceMissing.prx": 1}) (second / "main.prx").write_bytes(exporting("CCCCCCCCCCC")) previous = b'{"previous": "complete report"}\n' for directories in ((first, second), (second, first)): for exists in (False, True): with self.subTest(directories=directories, existing_report=exists): if exists: result.write_bytes(previous) else: result.unlink(missing_ok=True) with patch.object(work, "modules", directories[0]): code, out, err = work.run(registry, "--modules", str(directories[1]), "--json", str(result)) self.assertEqual((code, out), (2, "")) self.assertIn("ambiguous guest module main.prx", err) self.assertIn(str(first / "main.prx"), err) self.assertIn(str(second / "main.prx"), err) if exists: self.assertEqual(result.read_bytes(), previous) else: self.assertFalse(result.exists()) def test_identical_module_files_are_audited_once(self): work = self.work first, second = work.modules, work.root / "other-modules" second.mkdir() (work.libs / "libc.prx").write_bytes(exporting()) image = build_elf([ ("AAAAAAAAAAA", 0, GLOBAL_FUNCTION), ("BBBBBBBBBBB", 1, GLOBAL_FUNCTION)], ["libSceMissing.prx"]) for directory in (first, second): (directory / "main.prx").write_bytes(image) (first / "consumer.prx").write_bytes(build_elf([("BBBBBBBBBBB", 0, GLOBAL_FUNCTION)])) registry, result = work.root / "game.json", work.root / "audit.json" write_registry(registry, []) for directories in ((first, second, first), (second, first, second)): with self.subTest(directories=directories), patch.object(work, "modules", directories[0]): extra = [argument for directory in directories[1:] for argument in ("--modules", str(directory))] code, _, err = work.run(registry, *extra, "--json", str(result)) self.assertEqual((code, err), (1, "")) data = json.loads(result.read_text()) self.assertEqual((data["references"], data["unique_imports"]), (2, 2)) self.assertEqual(data["unique_by_class"], {"implemented": 1, "stub": 0, "absent": 1, "module": 0}) self.assertEqual(data["references_by_class"], data["unique_by_class"]) self.assertEqual(data["missing_libraries"], {"libSceMissing.prx": 1}) records = {record["nid"]: record for record in data["imports"]} self.assertEqual((records["AAAAAAAAAAA"]["class"], records["AAAAAAAAAAA"]["references"]), ("absent", 1)) self.assertEqual((records["BBBBBBBBBBB"]["class"], records["BBBBBBBBBBB"]["providers"]), ("implemented", ["main.prx"])) def test_module_files_must_be_elf(self): work = self.work (work.libs / "libc.prx").write_bytes(exporting()) (work.modules / "guest.prx").write_bytes(build_pe(["alpha"])) registry = work.root / "game.json" write_registry(registry, []) code, out, err = work.run(registry) self.assertEqual((code, out), (2, "")) self.assertIn("guest.prx", err) self.assertIn("ELF", err) def test_sectionless_module_imports_and_exports_are_audited(self): work = self.work (work.libs / "libSceA.prx").write_bytes(exporting("AAAAAAAAAAA")) (work.modules / "stripped.prx").write_bytes(build_sectionless_elf([ ("AAAAAAAAAAA", 0, GLOBAL_FUNCTION), ("BBBBBBBBBBB", 1, GLOBAL_FUNCTION)], ["libSceA.prx"])) (work.modules / "main.prx").write_bytes(build_elf([("BBBBBBBBBBB", 0, GLOBAL_FUNCTION)], [])) registry = work.root / "game.json" write_registry(registry, []) result = work.root / "audit.json" code, out, err = work.run(registry, "--json", str(result)) self.assertEqual((code, err), (0, "")) data = json.loads(result.read_text()) self.assertEqual(data["references"], 2) records = {(record["nid"], record["module"]): record for record in data["imports"]} self.assertEqual(records[("AAAAAAAAAAA", "stripped.prx")]["class"], "implemented") self.assertEqual(records[("BBBBBBBBBBB", "main.prx")]["class"], "implemented") def test_elf_with_out_of_file_section_table_uses_program_headers(self): path = Path("module.prx") image = build_sectionless_elf([ ("AAAAAAAAAAA", 0, GLOBAL_FUNCTION), ("BBBBBBBBBBB", 1, GLOBAL_FUNCTION)], ["libSceA.prx"]) expected_imports = import_audit.elf_imports(path, image) expected_exports = import_audit.elf_exports(path, image) self.assertEqual(expected_imports, ({"AAAAAAAAAAA"}, {"libSceA.prx"})) self.assertEqual(expected_exports, {"BBBBBBBBBBB"}) for section_offset in (len(image), len(image) + 64): for section_count in (1, 0): modified = bytearray(image) struct.pack_into("