fix: preflight shared infra and future state schemas

Author: PascalThuet

src/specify_cli/__init__.py

@@ -61,6 +61,7 @@
 )
 from .integration_state import (
     INTEGRATION_JSON,
+    INTEGRATION_STATE_SCHEMA,
     dedupe_integration_keys as _dedupe_integration_keys,
     default_integration_key as _default_integration_key,
     installed_integration_keys as _installed_integration_keys,
@@ -1918,6 +1919,14 @@ def _read_integration_json(project_root: Path) -> dict[str, Any]:
         console.print(f"[red]Error:[/red] {path} must contain a JSON object, got {type(data).__name__}.")
         console.print(f"Please fix or delete {INTEGRATION_JSON} and retry.")
         raise typer.Exit(1)
+    schema = data.get("integration_state_schema")
+    if isinstance(schema, int) and not isinstance(schema, bool) and schema > INTEGRATION_STATE_SCHEMA:
+        console.print(
+            f"[red]Error:[/red] {path} uses integration state schema {schema}, "
+            f"but this CLI only supports schema {INTEGRATION_STATE_SCHEMA}."
+        )
+        console.print("Please upgrade Spec Kit before modifying integrations.")
+        raise typer.Exit(1)
     return _normalize_integration_state(data)
 
 

src/specify_cli/integration_state.py

@@ -66,6 +66,12 @@ def normalize_integration_settings(settings: Any) -> dict[str, dict[str, Any]]:
     return normalized
 
 
+def _normalized_integration_state_schema(value: Any) -> int:
+    if isinstance(value, int) and not isinstance(value, bool) and value > INTEGRATION_STATE_SCHEMA:
+        return value
+    return INTEGRATION_STATE_SCHEMA
+
+
 def normalize_integration_state(data: dict[str, Any]) -> dict[str, Any]:
     """Normalize legacy and multi-install integration metadata."""
     legacy_key = clean_integration_key(data.get("integration"))
@@ -81,7 +87,9 @@ def normalize_integration_state(data: dict[str, Any]) -> dict[str, Any]:
     settings = normalize_integration_settings(data.get("integration_settings"))
 
     normalized = dict(data)
-    normalized["integration_state_schema"] = INTEGRATION_STATE_SCHEMA
+    normalized["integration_state_schema"] = _normalized_integration_state_schema(
+        data.get("integration_state_schema")
+    )
     if default_key:
         normalized["integration"] = default_key
         normalized["default_integration"] = default_key

src/specify_cli/shared_infra.py

@@ -66,15 +66,54 @@ def _shared_destination_label(project_path: Path, dest: Path) -> str:
         return str(dest)
 
 
-def _ensure_safe_shared_destination(project_path: Path, dest: Path) -> None:
-    """Refuse shared infra writes that would escape or follow symlinks."""
-    root = project_path.resolve()
+def _shared_relative_path(project_path: Path, dest: Path) -> Path:
     try:
-        dest.parent.resolve().relative_to(root)
-    except (OSError, ValueError):
+        rel = dest.relative_to(project_path)
+    except ValueError:
+        label = _shared_destination_label(project_path, dest)
+        raise ValueError(f"Shared infrastructure path escapes project root: {label}") from None
+
+    if rel.is_absolute() or ".." in rel.parts:
         label = _shared_destination_label(project_path, dest)
-        raise ValueError(f"Shared infrastructure destination escapes project root: {label}") from None
+        raise ValueError(f"Shared infrastructure path escapes project root: {label}")
+    return rel
+
+
+def _ensure_safe_shared_directory(project_path: Path, directory: Path, *, create: bool = True) -> None:
+    """Create a shared infra directory without following symlinked parents."""
+    root = project_path.resolve()
+    rel = _shared_relative_path(project_path, directory)
+    current = project_path
+
+    for part in rel.parts:
+        current = current / part
+        label = _shared_destination_label(project_path, current)
+        if current.is_symlink():
+            raise ValueError(f"Refusing to use symlinked shared infrastructure directory: {label}")
+        if current.exists():
+            if not current.is_dir():
+                raise ValueError(f"Shared infrastructure directory path is not a directory: {label}")
+            try:
+                current.resolve().relative_to(root)
+            except (OSError, ValueError):
+                raise ValueError(f"Shared infrastructure directory escapes project root: {label}") from None
+            continue
+        if not create:
+            raise ValueError(f"Shared infrastructure directory does not exist: {label}")
+        current.mkdir()
+        if current.is_symlink():
+            raise ValueError(f"Refusing to use symlinked shared infrastructure directory: {label}")
+        try:
+            current.resolve().relative_to(root)
+        except (OSError, ValueError):
+            raise ValueError(f"Shared infrastructure directory escapes project root: {label}") from None
 
+
+def _ensure_safe_shared_destination(project_path: Path, dest: Path) -> None:
+    """Refuse shared infra writes that would escape or follow symlinks."""
+    root = project_path.resolve()
+    _shared_relative_path(project_path, dest)
+    _ensure_safe_shared_directory(project_path, dest.parent, create=False)
     label = _shared_destination_label(project_path, dest)
     if dest.is_symlink():
         raise ValueError(f"Refusing to overwrite symlinked shared infrastructure path: {label}")
@@ -90,10 +129,6 @@ def _write_shared_text(project_path: Path, dest: Path, content: str) -> None:
     _write_shared_bytes(project_path, dest, content.encode("utf-8"))
 
 
-def _copy_shared_file(project_path: Path, src: Path, dest: Path) -> None:
-    _write_shared_bytes(project_path, dest, src.read_bytes(), mode=src.stat().st_mode & 0o777)
-
-
 def _write_shared_bytes(
     project_path: Path,
     dest: Path,
@@ -134,9 +169,10 @@ def refresh_shared_templates(
     tracked_files = manifest.files
     modified = set(manifest.check_modified())
     skipped_files: list[str] = []
+    planned_updates: list[tuple[Path, str, str]] = []
 
     dest_templates = project_path / ".specify" / "templates"
-    dest_templates.mkdir(parents=True, exist_ok=True)
+    _ensure_safe_shared_directory(project_path, dest_templates)
     for src in templates_src.iterdir():
         if not src.is_file() or src.name == "vscode-settings.json" or src.name.startswith("."):
             continue
@@ -151,6 +187,9 @@ def refresh_shared_templates(
 
         content = src.read_text(encoding="utf-8")
         content = IntegrationBase.resolve_command_refs(content, invoke_separator)
+        planned_updates.append((dst, rel, content))
+
+    for dst, rel, content in planned_updates:
         _write_shared_text(project_path, dst, content)
         manifest.record_existing(rel)
 
@@ -178,16 +217,18 @@ def install_shared_infra(
     """Install shared scripts and templates into *project_path*."""
     manifest = load_speckit_manifest(project_path, version=version, console=console)
     skipped_files: list[str] = []
+    planned_copies: list[tuple[Path, str, bytes, int]] = []
+    planned_templates: list[tuple[Path, str, str]] = []
 
     scripts_src = shared_scripts_source(core_pack=core_pack, repo_root=repo_root)
     if scripts_src.is_dir():
         dest_scripts = project_path / ".specify" / "scripts"
-        dest_scripts.mkdir(parents=True, exist_ok=True)
+        _ensure_safe_shared_directory(project_path, dest_scripts)
         variant_dir = "bash" if script_type == "sh" else "powershell"
         variant_src = scripts_src / variant_dir
         if variant_src.is_dir():
             dest_variant = dest_scripts / variant_dir
-            dest_variant.mkdir(parents=True, exist_ok=True)
+            _ensure_safe_shared_directory(project_path, dest_variant)
             for src_path in variant_src.rglob("*"):
                 if not src_path.is_file():
                     continue
@@ -199,15 +240,14 @@ def install_shared_infra(
                     skipped_files.append(str(dst_path.relative_to(project_path)))
                     continue
 
-                dst_path.parent.mkdir(parents=True, exist_ok=True)
-                _copy_shared_file(project_path, src_path, dst_path)
+                _ensure_safe_shared_directory(project_path, dst_path.parent)
                 rel = dst_path.relative_to(project_path).as_posix()
-                manifest.record_existing(rel)
+                planned_copies.append((dst_path, rel, src_path.read_bytes(), src_path.stat().st_mode & 0o777))
 
     templates_src = shared_templates_source(core_pack=core_pack, repo_root=repo_root)
     if templates_src.is_dir():
         dest_templates = project_path / ".specify" / "templates"
-        dest_templates.mkdir(parents=True, exist_ok=True)
+        _ensure_safe_shared_directory(project_path, dest_templates)
         for src in templates_src.iterdir():
             if not src.is_file() or src.name == "vscode-settings.json" or src.name.startswith("."):
                 continue
@@ -220,9 +260,16 @@ def install_shared_infra(
 
             content = src.read_text(encoding="utf-8")
             content = IntegrationBase.resolve_command_refs(content, invoke_separator)
-            _write_shared_text(project_path, dst, content)
             rel = dst.relative_to(project_path).as_posix()
-            manifest.record_existing(rel)
+            planned_templates.append((dst, rel, content))
+
+    for dst_path, rel, content, mode in planned_copies:
+        _write_shared_bytes(project_path, dst_path, content, mode=mode)
+        manifest.record_existing(rel)
+
+    for dst, rel, content in planned_templates:
+        _write_shared_text(project_path, dst, content)
+        manifest.record_existing(rel)
 
     if skipped_files:
         console.print(

tests/integrations/test_cli.py

@@ -9,6 +9,11 @@
 from tests.conftest import strip_ansi
 
 
+class _NoopConsole:
+    def print(self, *args, **kwargs):
+        pass
+
+
 def _normalize_cli_output(output: str) -> str:
     output = strip_ansi(output)
     output = " ".join(output.split())
@@ -349,6 +354,95 @@ def test_shared_template_refresh_refuses_symlinked_destination(self, tmp_path):
 
         assert outside.read_text(encoding="utf-8") == "# outside\n"
 
+    @pytest.mark.skipif(not hasattr(os, "symlink"), reason="symlinks are unavailable")
+    def test_shared_infra_refuses_symlinked_specify_directory_before_mkdir(self, tmp_path):
+        """Shared infra directory creation must not follow a symlinked .specify."""
+        from specify_cli import _install_shared_infra
+
+        project = tmp_path / "symlink-dir-test"
+        project.mkdir()
+        outside = tmp_path / "outside-specify"
+        outside.mkdir()
+        os.symlink(outside, project / ".specify")
+
+        with pytest.raises(ValueError, match="symlinked shared infrastructure directory"):
+            _install_shared_infra(project, "sh", force=True)
+
+        assert not (outside / "scripts").exists()
+        assert not (outside / "templates").exists()
+
+    @pytest.mark.skipif(not hasattr(os, "symlink"), reason="symlinks are unavailable")
+    def test_shared_template_refresh_preflights_before_writing(self, tmp_path):
+        """Template refresh validates all destinations before writing any file."""
+        from specify_cli.shared_infra import refresh_shared_templates
+
+        project = tmp_path / "preflight-refresh-test"
+        project.mkdir()
+        templates_dir = project / ".specify" / "templates"
+        templates_dir.mkdir(parents=True)
+
+        core_pack = tmp_path / "core-pack"
+        templates_src = core_pack / "templates"
+        templates_src.mkdir(parents=True)
+        (templates_src / "a-template.md").write_text("# new a\n", encoding="utf-8")
+        (templates_src / "z-template.md").write_text("# new z\n", encoding="utf-8")
+
+        existing = templates_dir / "a-template.md"
+        existing.write_text("# old a\n", encoding="utf-8")
+        outside = tmp_path / "outside-z.md"
+        outside.write_text("# outside\n", encoding="utf-8")
+        os.symlink(outside, templates_dir / "z-template.md")
+
+        with pytest.raises(ValueError, match="Refusing to overwrite symlinked"):
+            refresh_shared_templates(
+                project,
+                version="test",
+                core_pack=core_pack,
+                repo_root=tmp_path / "unused",
+                console=_NoopConsole(),
+                invoke_separator=".",
+                force=True,
+            )
+
+        assert existing.read_text(encoding="utf-8") == "# old a\n"
+        assert outside.read_text(encoding="utf-8") == "# outside\n"
+
+    @pytest.mark.skipif(not hasattr(os, "symlink"), reason="symlinks are unavailable")
+    def test_shared_infra_install_preflights_before_writing(self, tmp_path):
+        """Full shared infra installs validate destinations before writing any file."""
+        from specify_cli.shared_infra import install_shared_infra
+
+        project = tmp_path / "preflight-install-test"
+        project.mkdir()
+        scripts_dir = project / ".specify" / "scripts" / "bash"
+        scripts_dir.mkdir(parents=True)
+
+        core_pack = tmp_path / "core-pack"
+        scripts_src = core_pack / "scripts" / "bash"
+        scripts_src.mkdir(parents=True)
+        (scripts_src / "a.sh").write_text("# new a\n", encoding="utf-8")
+        (scripts_src / "z.sh").write_text("# new z\n", encoding="utf-8")
+
+        existing = scripts_dir / "a.sh"
+        existing.write_text("# old a\n", encoding="utf-8")
+        outside = tmp_path / "outside-z.sh"
+        outside.write_text("# outside\n", encoding="utf-8")
+        os.symlink(outside, scripts_dir / "z.sh")
+
+        with pytest.raises(ValueError, match="Refusing to overwrite symlinked"):
+            install_shared_infra(
+                project,
+                "sh",
+                version="test",
+                core_pack=core_pack,
+                repo_root=tmp_path / "unused",
+                console=_NoopConsole(),
+                force=True,
+            )
+
+        assert existing.read_text(encoding="utf-8") == "# old a\n"
+        assert outside.read_text(encoding="utf-8") == "# outside\n"
+
     def test_shared_infra_no_warning_when_forced(self, tmp_path, capsys):
         """No skip warning when force=True (all files overwritten)."""
         from specify_cli import _install_shared_infra

tests/integrations/test_integration_state.py

@@ -38,6 +38,20 @@ def test_normalize_integration_state_strips_legacy_key_fallback():
     assert state["installed_integrations"] == ["codex"]
 
 
+def test_normalize_integration_state_preserves_newer_schema():
+    state = normalize_integration_state(
+        {
+            "integration_state_schema": 99,
+            "integration": "claude",
+            "installed_integrations": ["claude"],
+            "future_field": {"keep": True},
+        }
+    )
+
+    assert state["integration_state_schema"] == 99
+    assert state["future_field"] == {"keep": True}
+
+
 def test_default_integration_key_strips_raw_state_values():
     assert default_integration_key({"default_integration": " claude "}) == "claude"
     assert default_integration_key({"integration": " codex "}) == "codex"

tests/integrations/test_integration_subcommand.py

@@ -96,6 +96,24 @@ def test_list_shows_multi_install_safe_status(self, tmp_path):
         assert _integration_list_row_cells(result.output, "claude")[-1] == "yes"
         assert _integration_list_row_cells(result.output, "copilot")[-1] == "no"
 
+    def test_list_rejects_newer_integration_state_schema(self, tmp_path):
+        project = _init_project(tmp_path, "claude")
+        int_json = project / ".specify" / "integration.json"
+        data = json.loads(int_json.read_text(encoding="utf-8"))
+        data["integration_state_schema"] = 99
+        int_json.write_text(json.dumps(data), encoding="utf-8")
+
+        old_cwd = os.getcwd()
+        try:
+            os.chdir(project)
+            result = runner.invoke(app, ["integration", "list"])
+        finally:
+            os.chdir(old_cwd)
+
+        assert result.exit_code != 0
+        assert "schema 99" in result.output
+        assert "only supports schema 1" in result.output
+
 
 # ── install ──────────────────────────────────────────────────────────