Merge pull request #4802 from ArjenS-LA/fix/relinker-output-collisions

fix(relinker): reject collisions between generated outputs

core/relinker/CMakeLists.txt

@@ -549,6 +549,10 @@ if(BUILD_TESTING)
             ${CMAKE_CURRENT_SOURCE_DIR}/relinker/tests/test_input_magic.py $<TARGET_FILE:relinker>)
         add_test(NAME input_preservation COMMAND ${Python3_EXECUTABLE}
             ${CMAKE_CURRENT_SOURCE_DIR}/relinker/tests/test_input_preservation.py $<TARGET_FILE:relinker>)
+        add_test(NAME output_collisions COMMAND ${Python3_EXECUTABLE}
+            ${CMAKE_CURRENT_SOURCE_DIR}/relinker/tests/test_output_collisions.py $<TARGET_FILE:relinker>)
+        add_test(NAME secondary_output_collisions COMMAND ${Python3_EXECUTABLE}
+            ${CMAKE_CURRENT_SOURCE_DIR}/relinker/tests/test_secondary_output_collisions.py $<TARGET_FILE:relinker>)
         add_test(NAME input_file COMMAND ${Python3_EXECUTABLE}
             ${CMAKE_CURRENT_SOURCE_DIR}/relinker/tests/test_input_file.py $<TARGET_FILE:relinker>)
         add_test(NAME null_symbol_relocations COMMAND ${Python3_EXECUTABLE}

core/relinker/main.cpp

@@ -14,6 +14,7 @@
 #include <filesystem>
 #include <iostream>
 #include <memory>
+#include <random>
 #include <string>
 #include <vector>
 
@@ -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<char>(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<char>(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<std::filesystem::path> 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});

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()

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()