diff --git a/core/relinker/CMakeLists.txt b/core/relinker/CMakeLists.txt index 0b57570996..f8fd8039fe 100644 --- a/core/relinker/CMakeLists.txt +++ b/core/relinker/CMakeLists.txt @@ -549,6 +549,10 @@ if(BUILD_TESTING) ${CMAKE_CURRENT_SOURCE_DIR}/relinker/tests/test_input_magic.py $) add_test(NAME input_preservation COMMAND ${Python3_EXECUTABLE} ${CMAKE_CURRENT_SOURCE_DIR}/relinker/tests/test_input_preservation.py $) + add_test(NAME output_collisions COMMAND ${Python3_EXECUTABLE} + ${CMAKE_CURRENT_SOURCE_DIR}/relinker/tests/test_output_collisions.py $) + add_test(NAME secondary_output_collisions COMMAND ${Python3_EXECUTABLE} + ${CMAKE_CURRENT_SOURCE_DIR}/relinker/tests/test_secondary_output_collisions.py $) add_test(NAME input_file COMMAND ${Python3_EXECUTABLE} ${CMAKE_CURRENT_SOURCE_DIR}/relinker/tests/test_input_file.py $) add_test(NAME null_symbol_relocations COMMAND ${Python3_EXECUTABLE} diff --git a/core/relinker/main.cpp b/core/relinker/main.cpp index 557840d16a..07f448aefc 100644 --- a/core/relinker/main.cpp +++ b/core/relinker/main.cpp @@ -14,6 +14,7 @@ #include #include #include +#include #include #include @@ -69,6 +70,100 @@ ConvertedExecutable Convert(const Cli::Args& args, const std::string& inputPath, return {inputPath, absPath, std::move(converted.Analysis), std::move(converted.Modules), std::move(converted.Executable), std::move(converted.WindowsDiagnostics)}; } +#ifdef _WIN32 + +class OutputIdentityDirectory { +public: + OutputIdentityDirectory(const OutputIdentityDirectory&) = delete; + OutputIdentityDirectory& operator=(const OutputIdentityDirectory&) = delete; + + explicit OutputIdentityDirectory(const std::filesystem::path& parent) { + std::random_device random; + for (unsigned attempt = 0; attempt < 64; ++attempt) { + const auto candidate = parent / (".anyps5-output-identity-" + std::to_string(random()) + "-" + std::to_string(random())); + std::error_code error; + if (std::filesystem::create_directory(candidate, error)) { + directory = candidate; + return; + } + if (error && error != std::errc::file_exists) + throw std::filesystem::filesystem_error("Cannot create output identity directory", candidate, error); + } + throw Domain::RelinkerException("Cannot reserve an output identity directory in " + Io::Utf8Path(parent)); + } + + ~OutputIdentityDirectory() { + if (directory.empty()) return; + try { + Remove(); + } catch (const std::exception& error) { + std::cerr << "FAIL: Output identity cleanup: " << error.what() << '\n'; + } + } + + std::filesystem::path GetPath() const { return directory / "file"; } + const std::filesystem::path& GetDirectory() const { return directory; } + + void Remove() { + std::filesystem::remove(GetPath()); + std::filesystem::remove(directory); + directory.clear(); + } + +private: + std::filesystem::path directory; +}; + +#endif + +bool SameOutputFile(const std::filesystem::path& first, const std::filesystem::path& second) { + if (std::filesystem::equivalent(first, second)) return true; +#ifdef _WIN32 + if (std::filesystem::status(first).permissions() != std::filesystem::status(second).permissions() && + std::filesystem::hard_link_count(first) > 1 && std::filesystem::hard_link_count(second) > 1) { + for (const auto& path : {first, second}) { + if ((std::filesystem::status(path).permissions() & std::filesystem::perms::owner_write) == std::filesystem::perms::none) + throw Domain::RelinkerException("Cannot compare read-only output identities: " + Io::Utf8Path(path)); + } + OutputIdentityDirectory firstDirectory(first.parent_path()); + OutputIdentityDirectory secondDirectory(second.parent_path()); + std::filesystem::create_hard_link(first, firstDirectory.GetPath()); + std::filesystem::create_hard_link(second, secondDirectory.GetPath()); + const bool same = std::filesystem::equivalent(firstDirectory.GetPath(), secondDirectory.GetPath()); + secondDirectory.Remove(); + firstDirectory.Remove(); + return same; + } +#endif + return false; +} + +bool SameOutputDestination(const std::filesystem::path& first, const std::filesystem::path& second) { + const auto firstPath = std::filesystem::weakly_canonical(first); + const auto secondPath = std::filesystem::weakly_canonical(second); + if (firstPath == secondPath) return true; + if (std::filesystem::exists(first) && std::filesystem::exists(second)) return SameOutputFile(first, second); +#ifdef _WIN32 + auto firstName = Io::Utf8Path(firstPath.filename()); + auto secondName = Io::Utf8Path(secondPath.filename()); + for (auto* name : {&firstName, &secondName}) + for (auto& character : *name) + if (character >= 'A' && character <= 'Z') character = static_cast(character + ('a' - 'A')); + if (firstName == secondName && std::filesystem::exists(firstPath.parent_path()) && + std::filesystem::exists(secondPath.parent_path()) && std::filesystem::equivalent(firstPath.parent_path(), secondPath.parent_path())) { + OutputIdentityDirectory probe(firstPath.parent_path()); + auto alternateName = Io::Utf8Path(probe.GetDirectory().filename()); + for (auto& character : alternateName) + if (character >= 'a' && character <= 'z') character = static_cast(character - ('a' - 'A')); + const auto alternate = firstPath.parent_path() / Io::NativePath(alternateName); + const bool same = std::filesystem::exists(alternate) && std::filesystem::equivalent(probe.GetDirectory(), alternate); + probe.Remove(); + return same; + } +#endif + return false; +} + int Run(const int argc, char* argv[]) { Cli::Args args; try { @@ -184,6 +279,24 @@ int Run(const int argc, char* argv[]) { checkOutputInputs(Io::NativePath(absPath), "Executable"); for (const auto& artifact : guestArtifacts) checkOutputInputs(artifact.Path, "Guest module"); + std::vector outputPaths; + for (const auto& executable : executables) outputPaths.push_back(Io::NativePath(executable.OutputPath)); + for (const auto& artifact : guestArtifacts) outputPaths.push_back(artifact.Path); + outputPaths.insert(outputPaths.end(), registryPaths.begin(), registryPaths.end()); + if (desktopEntry) { + outputPaths.push_back(desktopEntry->IconPath); + outputPaths.push_back(desktopEntry->EntryPath); + } + if (args.windowsDiagnostics) outputPaths.push_back(std::filesystem::absolute("windows-diagnostics-imports.txt")); + for (std::size_t index = 0; index < outputPaths.size(); ++index) { + for (std::size_t previous = 0; previous < index; ++previous) { + const auto& first = outputPaths[index]; + const auto& second = outputPaths[previous]; + if (SameOutputDestination(first, second)) + throw Domain::RelinkerException("Output paths collide: " + Io::Utf8Path(first) + " and " + Io::Utf8Path(second)); + } + } + if (args.windowsDiagnostics) fileWriter.Write("windows-diagnostics-imports.txt", executables.front().Diagnostics); const auto outputCount = guestArtifacts.size() + executables.size(); progress.Report({Domain::ProgressStage::Write, absPath, 0, outputCount}); diff --git a/core/relinker/relinker/tests/test_output_collisions.py b/core/relinker/relinker/tests/test_output_collisions.py new file mode 100644 index 0000000000..2026412310 --- /dev/null +++ b/core/relinker/relinker/tests/test_output_collisions.py @@ -0,0 +1,153 @@ +import itertools +import json +import os +from pathlib import Path +import stat +import subprocess +import sys +import tempfile + +from test_guest_intel_trampolines import PLAIN_SITE, guest_fixture, main_fixture +from test_linux_desktop_entry_output import png + + +def prepare(case, windows): + case.mkdir() + source = case / "input.elf" + source.write_bytes(main_fixture()) + modules = case / "sce_module" + modules.mkdir() + module = modules / "sample.prx" + module.write_bytes(guest_fixture(PLAIN_SITE)) + output = case / ("output.exe" if windows else "output.elf") + paths = { + "executable": output, + "module": case / "app0/sce_module/sample.prx.guest.prx", + "registry": case / "output.registry.json", + "module-registry": case / "output.sample.prx.guest.prx.registry.json", + } + if windows: + paths["diagnostics"] = case / "windows-diagnostics-imports.txt" + else: + (case / "sce_sys").mkdir() + (case / "sce_sys/icon0.png").write_bytes(png(1, 1)) + paths["icon"] = Path(f"{output}.png") + paths["desktop"] = Path(f"{output}.desktop") + for path in paths.values(): + path.parent.mkdir(parents=True, exist_ok=True) + options = ["--windows", "--windows-diagnostics"] if windows else [] + command = [*options, "--registry", str(source), str(output)] + return command, paths, {source: source.read_bytes(), module: module.read_bytes()} + + +def case_sensitive(directory): + probe = directory / "CaseProbe" + probe.write_bytes(b"probe") + sensitive = not (directory / "caseprobe").exists() + probe.unlink() + return sensitive + + +def main(): + relinker = Path(sys.argv[1]).resolve() + checked = 0 + with tempfile.TemporaryDirectory(prefix="anyps5-output-collisions-") as directory: + work = Path(directory) + for windows in (False, True): + command, paths, inputs = prepare(work / f"{windows}-control", windows) + links = paths["executable"].parent / "separate-links" + links.mkdir() + for name, path in paths.items(): + path.write_bytes(b"previous output") + os.link(path, links / name) + result = subprocess.run([str(relinker), *command], cwd=paths["executable"].parent, + capture_output=True, text=True, timeout=20) + assert result.returncode == 0, (result.stdout, result.stderr) + for name in ("executable", "module"): + assert paths[name].read_bytes().startswith(b"MZ" if windows else b"\x7fELF") + for name in ("registry", "module-registry"): + json.loads(paths[name].read_text()) + for path, original in inputs.items(): + assert path.read_bytes() == original + assert not list(paths["executable"].parent.rglob(".anyps5-output-identity-*")) + for first, second in itertools.combinations(paths, 2): + case = work / f"{windows}-{first}-{second}" + command, outputs, inputs = prepare(case, windows) + previous = b"previous output" + outputs[first].write_bytes(previous) + os.link(outputs[first], outputs[second]) + for name, path in outputs.items(): + if name not in (first, second): + path.write_bytes(previous) + result = subprocess.run([str(relinker), *command], cwd=case, + capture_output=True, text=True, timeout=20) + assert result.returncode == 2 and "Output paths collide" in result.stderr, ( + windows, first, second, result.returncode, result.stdout, result.stderr) + for path in outputs.values(): + assert path.read_bytes() == previous, path + for path, original in inputs.items(): + assert path.read_bytes() == original, path + assert not list(case.rglob(".anyps5-output-identity-*")) + checked += 1 + names = ["windows-diagnostics-imports.txt"] + if os.name == "nt" and not case_sensitive(work): + names.append("WINDOWS-DIAGNOSTICS-IMPORTS.TXT") + for index, name in enumerate(names): + for existing in (False, True): + case = work / f"diagnostics-executable-{index}-{existing}" + case.mkdir() + source = case / "input.elf" + source.write_bytes(main_fixture()) + output = case / name + if existing: + output.write_bytes(b"previous output") + result = subprocess.run([str(relinker), "--windows", "--windows-diagnostics", "--skip-sce-module", + str(source), str(output)], cwd=case, + capture_output=True, text=True, timeout=20) + assert result.returncode == 2 and "Output paths collide" in result.stderr, (result.stdout, result.stderr) + assert output.read_bytes() == b"previous output" if existing else not output.exists() + assert source.read_bytes() == main_fixture() + assert not list(case.rglob(".anyps5-output-identity-*")) + for existing in (False, True): + case = work / f"case-sensitive-{existing}" + case.mkdir() + if not case_sensitive(case): + print("Distinct case-variant outputs skipped: case-sensitive directory unavailable") + break + source = case / "input.elf" + source.write_bytes(main_fixture()) + output = case / "WINDOWS-DIAGNOSTICS-IMPORTS.TXT" + diagnostics = case / "windows-diagnostics-imports.txt" + if existing: + output.write_bytes(b"previous executable") + diagnostics.write_bytes(b"previous diagnostics") + result = subprocess.run([str(relinker), "--windows", "--windows-diagnostics", "--skip-sce-module", + str(source), str(output)], cwd=case, + capture_output=True, text=True, timeout=20) + assert result.returncode == 0, (result.stdout, result.stderr) + assert output.read_bytes().startswith(b"MZ") and diagnostics.exists() + assert not os.path.samefile(output, diagnostics) + assert not list(case.rglob(".anyps5-output-identity-*")) + if os.name == "nt": + case = work / "readonly-identity" + command, paths, inputs = prepare(case, True) + output = paths["executable"] + output.write_bytes(b"previous output") + os.link(output, paths["module"]) + output.chmod(stat.S_IREAD) + try: + result = subprocess.run([str(relinker), *command], cwd=case, + capture_output=True, text=True, timeout=20) + assert result.returncode == 2 and "Cannot compare read-only output identities" in result.stderr, ( + result.stdout, result.stderr) + assert output.read_bytes() == b"previous output" + assert paths["module"].read_bytes() == b"previous output" + assert not output.stat().st_mode & stat.S_IWRITE + assert not list(case.rglob(".anyps5-output-identity-*")) + finally: + output.chmod(stat.S_IWRITE) + print(f"Output collision tests passed: {checked} hard-link pairs, direct paths, and identity controls") + + +if __name__ == "__main__": + main() diff --git a/core/relinker/relinker/tests/test_secondary_output_collisions.py b/core/relinker/relinker/tests/test_secondary_output_collisions.py new file mode 100644 index 0000000000..a0db558788 --- /dev/null +++ b/core/relinker/relinker/tests/test_secondary_output_collisions.py @@ -0,0 +1,101 @@ +import itertools +import json +import os +from pathlib import Path +import subprocess +import sys +import tempfile + +from test_guest_intel_trampolines import main_fixture +from test_output_collisions import prepare + + +def prepare_secondary(case, windows): + command, paths, inputs = prepare(case, windows) + options = [] + for name in ("secondary-one", "secondary-two"): + source = case / f"{name}.bin" + source.write_bytes(main_fixture()) + inputs[source] = source.read_bytes() + options.extend(["--executable", source.name]) + paths[name] = case / f"{source.name}{paths['executable'].suffix}" + paths[f"{name}-registry"] = case / f"{source.name}.registry.json" + if not windows: + icon = case / "sce_sys/icon0.png" + inputs[icon] = icon.read_bytes() + return [*command[:-2], *options, *command[-2:]], paths, inputs + + +def main(): + relinker = Path(sys.argv[1]).resolve() + secondary = {"secondary-one", "secondary-two"} + checked = 0 + with tempfile.TemporaryDirectory(prefix="anyps5-secondary-collisions-") as directory: + work = Path(directory) + for windows in (False, True): + case = work / f"{windows}-control" + command, paths, inputs = prepare_secondary(case, windows) + links = case / "separate-links" + links.mkdir() + for name, path in paths.items(): + path.write_bytes(b"previous output") + os.link(path, links / name) + result = subprocess.run([str(relinker), *command], cwd=case, + capture_output=True, text=True, timeout=20) + assert result.returncode == 0, (result.stdout, result.stderr) + for name in ("executable", "module", *sorted(secondary)): + assert paths[name].read_bytes().startswith(b"MZ" if windows else b"\x7fELF"), paths[name] + for name, path in paths.items(): + if "registry" in name: + json.loads(path.read_text()) + for path, original in inputs.items(): + assert path.read_bytes() == original, path + assert not list(case.rglob(".anyps5-output-identity-*")) + + for first, second in itertools.combinations(paths, 2): + if not secondary.intersection((first, second)): + continue + case = work / f"{windows}-{first}-{second}" + command, outputs, inputs = prepare_secondary(case, windows) + previous = b"previous output" + outputs[first].write_bytes(previous) + os.link(outputs[first], outputs[second]) + for name, path in outputs.items(): + if name not in (first, second): + path.write_bytes(previous) + result = subprocess.run([str(relinker), *command], cwd=case, + capture_output=True, text=True, timeout=20) + assert result.returncode == 2 and "Output paths collide" in result.stderr, ( + windows, first, second, result.returncode, result.stdout, result.stderr) + for path in outputs.values(): + assert path.read_bytes() == previous, path + for path, original in inputs.items(): + assert path.read_bytes() == original, path + assert not list(case.rglob(".anyps5-output-identity-*")) + checked += 1 + + for existing in (False, True): + case = work / f"diagnostics-secondary-{existing}" + command, paths, inputs = prepare(case, True) + source = case / "windows-diagnostics-imports" + source.write_bytes(main_fixture()) + inputs[source] = source.read_bytes() + paths["executable"] = case / "output.txt" + paths["secondary-registry"] = case / "windows-diagnostics-imports.registry.json" + if existing: + for path in paths.values(): + path.write_bytes(b"previous output") + result = subprocess.run([str(relinker), *command[:-2], "--executable", source.name, + command[-2], str(paths["executable"])], cwd=case, + capture_output=True, text=True, timeout=20) + assert result.returncode == 2 and "Output paths collide" in result.stderr, (result.stdout, result.stderr) + for path in paths.values(): + assert path.read_bytes() == b"previous output" if existing else not path.exists(), path + for path, original in inputs.items(): + assert path.read_bytes() == original, path + assert not list(case.rglob(".anyps5-output-identity-*")) + print(f"Secondary output collision tests passed: {checked} hard-link pairs, direct paths, and identity controls") + + +if __name__ == "__main__": + main()