fix(sources): refuse a Windows junction as the Ubuntu source output

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}"
             )

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}"
             )

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:

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