fix(workflows): confine workflow and step catalog cache writes

docs/reference/workflows.md

@@ -423,6 +423,14 @@ installed locally.
 
 Workflow catalogs control where `search` and `add` look for workflows. Catalogs are checked in priority order.
 
+Workflow and step catalogs cache fetched JSON and freshness metadata for one
+hour in `.specify/workflows/.cache/` and `.specify/workflows/steps/.cache/`,
+respectively. Cache I/O does not follow symlinked directory components or
+cache files. Writes atomically replace cache files rather than truncate their
+referents, so a pre-existing cache-file symlink can be replaced without changing
+its target. If a cache cannot be safely written, a warning is logged and
+discovery continues using the fetched catalog.
+
 > **A project's `.specify/workflow-catalogs.yml` can point `add` and `search` at a catalog you didn't choose.** Before running a workflow from an unfamiliar project, run `specify workflow catalog list` (and `specify workflow step catalog list` for the step catalogs its steps can pull in) — a project supplying that config is not evidence its workflows or steps were vetted. Maintainers do not audit `run` fields; read a workflow's shell steps yourself before running it (see [Who maintains workflows?](#who-maintains-workflows)).
 
 ### List Catalogs

src/specify_cli/workflows/_catalog_cache.py

@@ -0,0 +1,195 @@
+"""Confined filesystem I/O shared by workflow and step catalog caches."""
+
+from __future__ import annotations
+
+import errno
+import json
+import os
+import stat
+from collections.abc import Callable, Iterator
+from contextlib import ExitStack, contextmanager
+from pathlib import Path
+from typing import Any
+from uuid import uuid4
+
+
+@contextmanager
+def _windows_handle(path: Path, *, directory: bool) -> Iterator[int]:
+    """Pin a non-reparse path by withholding Windows delete sharing."""
+    import ctypes
+    from ctypes import wintypes
+
+    class AttributeTagInfo(ctypes.Structure):
+        _fields_ = [("attributes", wintypes.DWORD), ("tag", wintypes.DWORD)]
+
+    kernel = ctypes.WinDLL("kernel32", use_last_error=True)
+    create = kernel.CreateFileW
+    create.argtypes = [
+        wintypes.LPCWSTR,
+        wintypes.DWORD,
+        wintypes.DWORD,
+        wintypes.LPVOID,
+        wintypes.DWORD,
+        wintypes.DWORD,
+        wintypes.HANDLE,
+    ]
+    create.restype = wintypes.HANDLE
+    info = kernel.GetFileInformationByHandleEx
+    info.argtypes = [wintypes.HANDLE, ctypes.c_int, wintypes.LPVOID, wintypes.DWORD]
+    info.restype = wintypes.BOOL
+    close = kernel.CloseHandle
+    close.argtypes = [wintypes.HANDLE]
+    close.restype = wintypes.BOOL
+
+    # OPEN_REPARSE_POINT inspects the link itself; BACKUP_SEMANTICS permits
+    # directory handles. Holding every ancestor without FILE_SHARE_DELETE
+    # prevents a junction/symlink swap during path-based create/replace.
+    # Attribute-only handles bypass share checking: include FILE_LIST_DIRECTORY
+    # with FILE_READ_ATTRIBUTES so even an empty directory cannot be renamed.
+    handle = create(
+        str(path),
+        (0x80 | 0x1) if directory else 0x80000000,
+        0x3,
+        None,
+        3,
+        0x00200000 | 0x02000000,
+        None,
+    )
+    if handle == wintypes.HANDLE(-1).value:
+        raise ctypes.WinError(ctypes.get_last_error())
+    try:
+        attributes = AttributeTagInfo()
+        if not info(handle, 9, ctypes.byref(attributes), ctypes.sizeof(attributes)):
+            raise ctypes.WinError(ctypes.get_last_error())
+        if attributes.attributes & 0x400:
+            raise OSError(
+                errno.ELOOP, "Refusing a reparse-point catalog cache path", str(path)
+            )
+        if bool(attributes.attributes & 0x10) != directory:
+            raise OSError(errno.EINVAL, "Unexpected catalog cache file type", str(path))
+        yield handle
+    finally:
+        close(handle)
+
+
+@contextmanager
+def _cache_directory(
+    root: Path, directory: Path, *, create: bool
+) -> Iterator[int | Path]:
+    relative = directory.relative_to(root)
+    if ".." in relative.parts:
+        raise OSError(errno.EINVAL, "Catalog cache escapes project root")
+    root = root.resolve()
+    if os.name == "nt":
+        with ExitStack() as stack:
+            current = root
+            stack.enter_context(_windows_handle(current, directory=True))
+            for part in relative.parts:
+                current = current / part
+                try:
+                    stack.enter_context(_windows_handle(current, directory=True))
+                except FileNotFoundError:
+                    if not create:
+                        raise
+                    try:
+                        current.mkdir()
+                    except FileExistsError:
+                        pass
+                    stack.enter_context(_windows_handle(current, directory=True))
+            yield current
+        return
+
+    if not (
+        getattr(os, "O_NOFOLLOW", 0)
+        and getattr(os, "O_DIRECTORY", 0)
+        and {os.open, os.mkdir, os.rename, os.unlink} <= os.supports_dir_fd
+    ):
+        raise OSError(
+            errno.ENOTSUP, "Safe catalog cache I/O is unsupported on this platform"
+        )
+    flags = os.O_RDONLY | os.O_DIRECTORY | os.O_NOFOLLOW | getattr(os, "O_CLOEXEC", 0)
+    parent = os.open(root, flags)
+    try:
+        for part in relative.parts:
+            try:
+                child = os.open(part, flags, dir_fd=parent)
+            except FileNotFoundError:
+                if not create:
+                    raise
+                try:
+                    os.mkdir(part, dir_fd=parent)
+                except FileExistsError:
+                    pass
+                child = os.open(part, flags, dir_fd=parent)
+            os.close(parent)
+            parent = child
+        yield parent
+    finally:
+        os.close(parent)
+
+
+def read_cache_json(
+    root: Path,
+    path: Path,
+    *,
+    object_pairs_hook: Callable[[list[tuple[str, Any]]], dict[str, Any]] | None = None,
+) -> Any:
+    """Read a regular cache leaf without following links or swapped ancestors."""
+    with _cache_directory(root, path.parent, create=False) as directory_fd:
+        if isinstance(directory_fd, Path):
+            path = directory_fd / path.name
+            with (
+                _windows_handle(path, directory=False),
+                open(path, encoding="utf-8") as stream,
+            ):
+                return json.load(stream, object_pairs_hook=object_pairs_hook)
+        flags = (
+            os.O_RDONLY | os.O_NOFOLLOW | os.O_NONBLOCK | getattr(os, "O_CLOEXEC", 0)
+        )
+        fd = os.open(path.name, flags, dir_fd=directory_fd)
+        try:
+            if not stat.S_ISREG(os.fstat(fd).st_mode):
+                raise OSError(
+                    errno.EINVAL, "Catalog cache leaf is not a regular file", str(path)
+                )
+            stream = os.fdopen(fd, encoding="utf-8")
+        except BaseException:
+            os.close(fd)
+            raise
+        with stream:
+            return json.load(stream, object_pairs_hook=object_pairs_hook)
+
+
+def write_cache_json(
+    root: Path, path: Path, payload: dict[str, Any], *, indent: int | None = None
+) -> None:
+    """Atomically replace a cache leaf, never truncating its previous inode."""
+    content = json.dumps(payload, indent=indent)
+    with _cache_directory(root, path.parent, create=True) as directory:
+        directory_fd = None if isinstance(directory, Path) else directory
+        temporary = f".{path.name}.{uuid4().hex}.tmp"
+        staged = directory / temporary if isinstance(directory, Path) else temporary
+        destination = (
+            directory / path.name if isinstance(directory, Path) else path.name
+        )
+        flags = os.O_WRONLY | os.O_CREAT | os.O_EXCL | getattr(os, "O_CLOEXEC", 0)
+        flags |= getattr(os, "O_NOFOLLOW", 0)
+        fd = os.open(staged, flags, 0o600, dir_fd=directory_fd)
+        try:
+            try:
+                stream = os.fdopen(fd, "w", encoding="utf-8")
+            except BaseException:
+                os.close(fd)
+                raise
+            with stream:
+                stream.write(content)
+            # replace operates on directory entries, not the destination's
+            # referent, even if a symlink is installed just before this call.
+            os.replace(
+                staged, destination, src_dir_fd=directory_fd, dst_dir_fd=directory_fd
+            )
+        finally:
+            try:
+                os.unlink(staged, dir_fd=directory_fd)
+            except FileNotFoundError:
+                pass

src/specify_cli/workflows/catalog/_domain.py

@@ -11,6 +11,7 @@
 
 import hashlib
 import json
+import logging
 import os
 import stat
 import tempfile
@@ -25,6 +26,9 @@
     MAX_JSON_CATALOG_BYTES as MAX_JSON_CATALOG_BYTES,
     read_response_limited,
 )
+from .._catalog_cache import read_cache_json, write_cache_json
+
+logger = logging.getLogger(__name__)
 
 
 def _max_json_catalog_bytes() -> int:
@@ -499,16 +503,13 @@ def _get_cache_paths(self, url: str) -> tuple[Path, Path]:
     def _is_url_cache_valid(self, url: str) -> bool:
         """Check if cached data for a URL is still fresh."""
         _, meta_file = self._get_cache_paths(url)
-        if not meta_file.exists():
-            return False
         try:
-            with open(meta_file, encoding="utf-8") as f:
-                meta = json.load(f)
+            meta = read_cache_json(self.project_root, meta_file)
             if not isinstance(meta, dict):
                 return False
             fetched_at = float(meta.get("fetched_at", 0))
             return (time.time() - fetched_at) < self.CACHE_DURATION
-        except (json.JSONDecodeError, OSError, TypeError, ValueError):
+        except (UnicodeError, json.JSONDecodeError, OSError, TypeError, ValueError):
             return False
 
     def _fetch_single_catalog(
@@ -519,8 +520,7 @@ def _fetch_single_catalog(
 
         if not force_refresh and self._is_url_cache_valid(entry.url):
             try:
-                with open(cache_file, encoding="utf-8") as f:
-                    cached = json.load(f)
+                cached = read_cache_json(self.project_root, cache_file)
                 if isinstance(cached, dict):
                     return cached
             except (UnicodeDecodeError, json.JSONDecodeError, OSError):
@@ -584,15 +584,13 @@ def _validate_redirect(_old_url: str, new_url: str) -> None:
                 )
         except Exception as exc:
             # Fall back to cache if available
-            if cache_file.exists():
-                try:
-                    with open(cache_file, encoding="utf-8") as f:
-                        cached = json.load(f)
-                    if isinstance(cached, dict):
-                        return cached
-                except (json.JSONDecodeError, ValueError, OSError):
-                    # Stale-cache read failed; let the original fetch error propagate.
-                    pass
+            try:
+                cached = read_cache_json(self.project_root, cache_file)
+                if isinstance(cached, dict):
+                    return cached
+            except (UnicodeError, json.JSONDecodeError, ValueError, OSError):
+                # Stale-cache read failed; let the original fetch error propagate.
+                pass
             raise WorkflowCatalogError(
                 f"Failed to fetch catalog from {entry.url}: {exc}"
             ) from exc
@@ -604,13 +602,13 @@ def _validate_redirect(_old_url: str, new_url: str) -> None:
 
         # Write cache
         try:
-            self.cache_dir.mkdir(parents=True, exist_ok=True)
-            with open(cache_file, "w", encoding="utf-8") as f:
-                json.dump(data, f, indent=2)
-            with open(meta_file, "w", encoding="utf-8") as f:
-                json.dump({"url": entry.url, "fetched_at": time.time()}, f)
-        except OSError:
-            pass  # Proceed without caching if disk write fails
+            write_cache_json(self.project_root, cache_file, data, indent=2)
+            write_cache_json(
+                self.project_root, meta_file,
+                {"url": entry.url, "fetched_at": time.time()},
+            )
+        except OSError as exc:
+            logger.warning("Could not safely cache workflow catalog %s: %s", entry.url, exc)
 
         return data