diff --git a/scripts/nvx_tools/collect_alpine_sources.py b/scripts/nvx_tools/collect_alpine_sources.py index b7ce29fa1..b82ed4374 100644 --- a/scripts/nvx_tools/collect_alpine_sources.py +++ b/scripts/nvx_tools/collect_alpine_sources.py @@ -8,9 +8,7 @@ import os import re import shutil -import stat import subprocess -import sys import tarfile from pathlib import Path, PurePosixPath from typing import cast @@ -19,7 +17,7 @@ AlpineBuildConstants, BuildConstants, ) -from .common import write_sha256_sums +from .common import is_link_or_junction, write_sha256_sums class SourceError(RuntimeError): @@ -374,22 +372,9 @@ def _fetch_upstream_sources(output: Path, alpine_version: str) -> None: ) -def _is_link(path: Path) -> bool: - # Path.is_symlink() misses Windows junctions, which redirect a path the same way. - try: - metadata = path.lstat() - except (FileNotFoundError, NotADirectoryError): - return False - if stat.S_ISLNK(metadata.st_mode): - return True - if sys.platform == "win32": - return metadata.st_reparse_tag == stat.IO_REPARSE_TAG_MOUNT_POINT - return False - - def _resolve_source_output(output: Path) -> Path: # Check the path as given, so a link there is refused instead of followed. - if _is_link(output) or (output.exists() and not output.is_dir()): + if is_link_or_junction(output) or (output.exists() and not output.is_dir()): raise SourceError(f"Alpine source output is not a directory: {output}") if output.exists(): unexpected = sorted( @@ -407,7 +392,9 @@ def _resolve_source_output(output: Path) -> Path: def _reset_source_output(output: Path) -> None: for generated in (output / "recipes", output / "upstream"): - if _is_link(generated) or (generated.exists() and not generated.is_dir()): + if is_link_or_junction(generated) or ( + generated.exists() and not generated.is_dir() + ): raise SourceError( f"Alpine source generated path is not a directory: {generated}" ) @@ -421,7 +408,9 @@ def _reset_source_output(output: Path) -> None: "root-owned files, so remove it manually" ) from error for generated in (output / "manifest.json", output / "SHA256SUMS"): - if _is_link(generated) or (generated.exists() and not generated.is_file()): + if is_link_or_junction(generated) or ( + generated.exists() and not generated.is_file() + ): raise SourceError( f"Alpine source generated path is not a file: {generated}" ) diff --git a/scripts/nvx_tools/collect_ubuntu_sources.py b/scripts/nvx_tools/collect_ubuntu_sources.py index ddd391a76..4ca825640 100644 --- a/scripts/nvx_tools/collect_ubuntu_sources.py +++ b/scripts/nvx_tools/collect_ubuntu_sources.py @@ -22,6 +22,7 @@ ScriptError, download, download_verified, + is_link_or_junction, require_tool, run_checked, sha256_file, @@ -645,7 +646,7 @@ def _encoded_package_directory(name: str, version: str) -> str: def _validate_source_output(output: Path) -> None: - if output.is_symlink() or (output.exists() and not output.is_dir()): + if is_link_or_junction(output) or (output.exists() and not output.is_dir()): raise ScriptError(f"Ubuntu source output is not a directory: {output}") if not output.exists(): return @@ -661,15 +662,22 @@ def _validate_source_output(output: Path) -> None: def _reset_source_output(output: Path) -> None: + # Refuse a link here: rmtree below would follow it into another directory. + if is_link_or_junction(output) or (output.exists() and not output.is_dir()): + raise ScriptError(f"Ubuntu source output is not a directory: {output}") for generated in (output / "packages", output / "metadata"): - if generated.is_symlink() or (generated.exists() and not generated.is_dir()): + if is_link_or_junction(generated) or ( + generated.exists() and not generated.is_dir() + ): raise ScriptError( f"Ubuntu source generated path is not a directory: {generated}" ) if generated.exists(): shutil.rmtree(generated) for generated in (output / "manifest.json", output / "SHA256SUMS"): - if generated.is_symlink() or (generated.exists() and not generated.is_file()): + if is_link_or_junction(generated) or ( + generated.exists() and not generated.is_file() + ): raise ScriptError( f"Ubuntu source generated path is not a file: {generated}" ) diff --git a/scripts/nvx_tools/common.py b/scripts/nvx_tools/common.py index 5d7f573bf..52cef2c1e 100644 --- a/scripts/nvx_tools/common.py +++ b/scripts/nvx_tools/common.py @@ -10,6 +10,7 @@ import shutil import stat import subprocess +import sys import time import urllib.error import urllib.parse @@ -30,6 +31,22 @@ class ScriptError(RuntimeError): """Raised for an actionable command-line workflow failure.""" +def is_link_or_junction(path: Path) -> bool: + """Return whether ``path`` is a symbolic link or, on Windows, a junction. + + ``Path.is_symlink()`` misses junctions, which redirect a path the same way. + """ + try: + metadata = path.lstat() + except (FileNotFoundError, NotADirectoryError): + return False + if stat.S_ISLNK(metadata.st_mode): + return True + if sys.platform == "win32": + return metadata.st_reparse_tag == stat.IO_REPARSE_TAG_MOUNT_POINT + return False + + def strict_json_object(pairs: list[tuple[str, object]]) -> dict[str, object]: result: dict[str, object] = {} for key, value in pairs: diff --git a/scripts/test_nvx_tools.py b/scripts/test_nvx_tools.py index 1cd1f93f9..961e7a0ee 100644 --- a/scripts/test_nvx_tools.py +++ b/scripts/test_nvx_tools.py @@ -9244,6 +9244,30 @@ def test_ubuntu_source_requirements_deduplicate_binary_manifests(self): ), ) + @unittest.skipUnless( + sys.platform == "win32", "junctions are Windows reparse points" + ) + def test_ubuntu_source_output_refuses_a_junction(self): + with tempfile.TemporaryDirectory() as temporary: + root = Path(temporary) + target = root / "target" + (target / "packages").mkdir(parents=True) + (target / "packages" / "keep.txt").write_text("keep", encoding="utf-8") + junction = root / "output" + subprocess.run( + ["cmd", "/c", "mklink", "/J", str(junction), str(target)], + check=True, + capture_output=True, + ) + + with self.assertRaisesRegex(common.ScriptError, "not a directory"): + collect_ubuntu_sources._validate_source_output(junction) + with self.assertRaisesRegex(common.ScriptError, "not a directory"): + collect_ubuntu_sources._reset_source_output(junction) + self.assertEqual( + (target / "packages" / "keep.txt").read_text(encoding="utf-8"), "keep" + ) + def test_ubuntu_dsc_validation_matches_source_index(self): payload = b"source" digest = hashlib.sha256(payload).hexdigest()