diff --git a/tools/symbolize_samples.py b/tools/symbolize_samples.py index 7a6f9ffead..c78ff4e18e 100644 --- a/tools/symbolize_samples.py +++ b/tools/symbolize_samples.py @@ -291,6 +291,31 @@ def main(argv=None): samples = load(args.samples, args.thread) if args.since: samples = window(samples, load(args.since, args.thread)) + if args.out and samples.stacks: + referenced_modules = {name.lower() for name, _ in samples.self_counts} + referenced_modules.update(name.lower() for name, _ in samples.inclusive) + for frames in samples.stacks: + for frame in frames.split(";"): + name, separator, _ = frame.partition("+0x") + if separator: + referenced_modules.add(name.lower()) + module_output_alias = False + for directory in args.libs: + for source in Path(directory).iterdir(): + if not source.is_file() or not (source.resolve() == output.resolve() or + (source.exists() and output.exists() and source.samefile(output))): + continue + if source.name.lower() in referenced_modules or source.suffix.lower() in (".prx", ".dll", ".exe"): + module_output_alias = True + continue + try: + with source.open("rb") as module: + if module.read(2) == b"MZ": + module_output_alias = True + except OSError: + module_output_alias = True + if module_output_alias: + parser.error("symbolized output must be separate from module inputs") symbolizer = Symbolizer(args.libs, samples.headers, args.demangle, warn=lambda text: print("warning: " + text, file=sys.stderr), unverified=args.unverified) for line in summarize(symbolizer, samples, args.top): print(line) diff --git a/tools/tests/test_symbolize_samples.py b/tools/tests/test_symbolize_samples.py index 17825f69a4..b9fba34664 100644 --- a/tools/tests/test_symbolize_samples.py +++ b/tools/tests/test_symbolize_samples.py @@ -109,6 +109,150 @@ class SymbolizeSamplesTests(unittest.TestCase): for path, content in original.items(): self.assertEqual(path.read_bytes(), content) + def test_output_cannot_alias_a_module_input(self): + samples = self.root / "worker.collapsed.txt" + samples.write_text( + f"# module libSceAgcDriver.prx {SIZE_OF_IMAGE:#x} {TIMESTAMP:#x}\n" + "libSceAgcDriver.prx+0x1011 1\n", + encoding="utf-8", + ) + module = self.libs / "libSceAgcDriver.prx" + original = module.read_bytes() + for alias in ("same", "hardlink"): + with self.subTest(alias=alias): + output = module if alias == "same" else self.root / "module-output" + if alias == "hardlink": + os.link(module, output) + result = subprocess.run([sys.executable, symbolize_samples.__file__, str(samples), + "--libs", str(self.libs), "--out", str(output)], + capture_output=True, text=True, timeout=5) + self.assertEqual(result.returncode, 2, result.stdout + result.stderr) + self.assertIn("must be separate from module inputs", result.stderr) + self.assertNotIn("Traceback", result.stderr) + self.assertEqual(module.read_bytes(), original) + if alias == "hardlink": + output.unlink() + + def test_output_cannot_alias_an_unreferenced_module(self): + samples = self.root / "worker.collapsed.txt" + samples.write_text( + f"# module libSceAgcDriver.prx {SIZE_OF_IMAGE:#x} {TIMESTAMP:#x}\n" + "libSceAgcDriver.prx+0x1011 1\n", + encoding="utf-8", + ) + module = self.libs / "other.prx" + build_pe(module) + original = module.read_bytes() + for alias in ("same", "hardlink"): + with self.subTest(alias=alias): + output = module if alias == "same" else self.root / "other-output" + if alias == "hardlink": + os.link(module, output) + result = subprocess.run([sys.executable, symbolize_samples.__file__, str(samples), + "--libs", str(self.libs), "--out", str(output)], + capture_output=True, text=True, timeout=5) + self.assertEqual(result.returncode, 2, result.stdout + result.stderr) + self.assertIn("must be separate from module inputs", result.stderr) + self.assertNotIn("Traceback", result.stderr) + self.assertEqual(module.read_bytes(), original) + if alias == "hardlink": + output.unlink() + + def test_write_only_aliased_module_candidate_is_protected(self): + if os.name != "posix" or os.geteuid() == 0: + self.skipTest("An unprivileged POSIX user is required") + samples = self.root / "worker.collapsed.txt" + samples.write_text( + f"# module libSceAgcDriver.prx {SIZE_OF_IMAGE:#x} {TIMESTAMP:#x}\n" + "libSceAgcDriver.prx+0x1011 1\n", + encoding="utf-8", + ) + output = self.libs / "opaque-module" + output.write_bytes(b"opaque") + original = output.read_bytes() + output.chmod(0o200) + try: + result = subprocess.run([sys.executable, symbolize_samples.__file__, str(samples), + "--libs", str(self.libs), "--out", str(output)], + capture_output=True, text=True, timeout=5) + finally: + output.chmod(0o600) + self.assertEqual(result.returncode, 2, result.stdout + result.stderr) + self.assertIn("module inputs", result.stderr) + self.assertEqual(output.read_bytes(), original) + + def test_unreferenced_malformed_or_unsupported_module_candidate_is_protected(self): + samples = self.root / "worker.collapsed.txt" + samples.write_text( + f"# module libSceAgcDriver.prx {SIZE_OF_IMAGE:#x} {TIMESTAMP:#x}\n" + "libSceAgcDriver.prx+0x1011 1\n", + encoding="utf-8", + ) + for kind in ("malformed-prx", "unsupported-pe"): + with self.subTest(kind=kind): + output = self.libs / ("malformed.prx" if kind == "malformed-prx" else "unsupported.bin") + if kind == "malformed-prx": + output.write_bytes(b"not a PE image") + else: + build_pe(output) + image = bytearray(output.read_bytes()) + struct.pack_into("