address review round 5: ancestor-resolves-outside check + TOCTOU-safe write

- Per-component walk now also re-resolves each existing ancestor and
  requires it to land under project_root.resolve(). This catches
  non-symlink directory aliases (e.g. Windows junctions / mount points)
  that resolve outside the project even though no component is itself
  a symlink.
- ZIP write now uses os.open(O_WRONLY|O_CREAT|O_EXCL|O_NOFOLLOW, 0o600).
  O_NOFOLLOW (POSIX) refuses a symlink swapped in between the cache
  validation and the write, closing the TOCTOU window. O_EXCL guarantees
  the unique UUID filename is freshly created and cannot be pre-staged.
  O_NOFOLLOW falls back to 0 on Windows.
- Add TestExtensionAddFromAncestorEscape: simulates a junction-like
  alias by patching Path.resolve() so .cache resolves outside; command
  must refuse with 'escapes project root'.
- Add TestExtensionAddFromTOCTOUWrite: races a symlink swap into the
  cache between validation and write; O_NOFOLLOW/O_EXCL must reject.

Author: mnriem

src/specify_cli/__init__.py

@@ -3646,19 +3646,29 @@ def extension_add(
                 # (consistent with shared_infra._ensure_safe_shared_directory):
                 # `mkdir(parents=True)` would silently traverse a symlinked ancestor,
                 # so each component is checked and created individually instead.
+                # Each existing component is also re-resolved and required to land
+                # under project_root, which catches non-symlink directory aliases
+                # (e.g. Windows junctions / mount points) that resolve outside.
                 import uuid as _uuid
                 download_dir = project_root / ".specify" / "extensions" / ".cache" / "downloads"
+                project_root_resolved = project_root.resolve()
                 _current = project_root
                 for _part in download_dir.relative_to(project_root).parts:
                     _current = _current / _part
                     if _current.is_symlink():
                         console.print("[red]Error:[/red] Refusing to use symlinked download cache directory")
                         raise typer.Exit(1)
-                    if not _current.exists():
-                        _current.mkdir()
-                        if _current.is_symlink():
-                            console.print("[red]Error:[/red] Refusing to use symlinked download cache directory")
+                    if _current.exists():
+                        try:
+                            _current.resolve().relative_to(project_root_resolved)
+                        except (OSError, ValueError):
+                            console.print("[red]Error:[/red] Download cache directory escapes project root")
                             raise typer.Exit(1)
+                        continue
+                    _current.mkdir()
+                    if _current.is_symlink():
+                        console.print("[red]Error:[/red] Refusing to use symlinked download cache directory")
+                        raise typer.Exit(1)
 
                 safe_name = Path(extension).name.replace("/", "_").replace("\\", "_") or "download"
                 safe_name = safe_name[:64]  # cap length to avoid filesystem errors
@@ -3676,7 +3686,24 @@ def extension_add(
 
                     with _open_url(from_url, timeout=60) as response:
                         zip_data = response.read()
-                    zip_path.write_bytes(zip_data)
+                    # Open with O_NOFOLLOW | O_EXCL to defeat any TOCTOU symlink
+                    # swap that may have happened between the cache validation
+                    # above and this write. O_EXCL also guarantees the unique
+                    # UUID filename is freshly created, never reused. O_NOFOLLOW
+                    # is a no-op on Windows where the constant is absent.
+                    _flags = os.O_WRONLY | os.O_CREAT | os.O_EXCL | getattr(os, "O_NOFOLLOW", 0)
+                    _fd = os.open(zip_path, _flags, 0o600)
+                    try:
+                        with os.fdopen(_fd, "wb") as _zf:
+                            _zf.write(zip_data)
+                    except BaseException:
+                        # os.fdopen took ownership only on success; on failure
+                        # before the with-block, the fd may still be open.
+                        try:
+                            os.close(_fd)
+                        except OSError:
+                            pass
+                        raise
 
                     # Install from downloaded ZIP
                     manifest = manager.install_from_zip(zip_path, speckit_version, priority=priority)

tests/test_extension_add_path_traversal.py

@@ -5,6 +5,7 @@
 escape the intended cache directory, causing arbitrary file writes and deletes.
 """
 
+from pathlib import Path
 from unittest.mock import MagicMock, patch
 
 import pytest
@@ -145,3 +146,87 @@ def test_symlinked_ancestor_is_refused(self, tmp_path, monkeypatch, ancestor_par
         mock_install.assert_not_called()
         # No download was written through the symlink
         assert list(outside.iterdir()) == []
+
+
+class TestExtensionAddFromAncestorEscape:
+    """A non-symlink ancestor whose ``.resolve()`` lands outside the project must be refused.
+
+    Simulates the Windows junction / mount-point case using a manually-pre-resolved
+    directory swap: we cannot create real junctions on POSIX, but we can patch the
+    ``.resolve()`` of an existing ancestor to land outside the project root. The
+    production guard must reject it before any download happens.
+    """
+
+    def test_ancestor_resolves_outside_project_is_refused(self, tmp_path, monkeypatch):
+        project_dir = tmp_path / "project"
+        project_dir.mkdir()
+        outside = tmp_path / "outside"
+        outside.mkdir()
+        monkeypatch.chdir(project_dir)
+
+        # Pre-create the cache ancestors as real directories so the per-component
+        # walk reaches the resolve()/relative_to() check on each existing one.
+        cache_root = project_dir / ".specify" / "extensions" / ".cache"
+        cache_root.mkdir(parents=True)
+
+        real_resolve = Path.resolve
+
+        def fake_resolve(self, *a, **kw):
+            # Pretend the .cache ancestor resolves outside the project.
+            if self == cache_root:
+                return real_resolve(outside, *a, **kw)
+            return real_resolve(self, *a, **kw)
+
+        with patch.object(Path, "resolve", fake_resolve), \
+             patch("specify_cli.authentication.http.open_url", return_value=_mock_open_url()), \
+             patch("specify_cli.extensions.ExtensionManager.install_from_zip") as mock_install:
+            result = runner.invoke(
+                app,
+                ["extension", "add", "my-ext", "--from", "https://example.com/ext.zip"],
+            )
+
+        assert result.exit_code != 0
+        assert "escapes project root" in (result.output or "").lower()
+        mock_install.assert_not_called()
+        assert list(outside.iterdir()) == []
+
+
+class TestExtensionAddFromTOCTOUWrite:
+    """The download write must not follow a symlink swapped in after validation."""
+
+    def test_swapped_zip_path_symlink_is_refused(self, project_dir):
+        """If zip_path is replaced by a symlink between validation and write, refuse."""
+        cache_dir = project_dir / ".specify" / "extensions" / ".cache" / "downloads"
+        outside = project_dir.parent / "outside"
+        outside.mkdir(exist_ok=True)
+        target = outside / "stolen.bin"
+
+        # Patch os.open to simulate the TOCTOU window: just before the real
+        # os.open runs, replace the target path with a symlink pointing outside.
+        # With O_NOFOLLOW the os.open call must raise instead of writing through.
+        import os as _os
+        real_open = _os.open
+
+        def racing_open(path, flags, mode=0o777):
+            try:
+                p = Path(path)
+                if p.parent == cache_dir and not p.exists():
+                    p.symlink_to(target)
+            except OSError:
+                pass
+            return real_open(path, flags, mode)
+
+        with patch("specify_cli.authentication.http.open_url", return_value=_mock_open_url()), \
+             patch("specify_cli.extensions.ExtensionManager.install_from_zip") as mock_install, \
+             patch("os.open", side_effect=racing_open):
+            result = runner.invoke(
+                app,
+                ["extension", "add", "my-ext", "--from", "https://example.com/ext.zip"],
+            )
+
+        # Either O_NOFOLLOW (POSIX) or O_EXCL (all platforms) must reject the
+        # swapped symlink. The command must fail and nothing must be written
+        # through the symlink to `outside`.
+        assert result.exit_code != 0
+        mock_install.assert_not_called()
+        assert not target.exists(), "write followed swapped symlink"