diff --git a/design/file-helper.md b/design/file-helper.md index f5497ef256..b9dbcddd2f 100644 --- a/design/file-helper.md +++ b/design/file-helper.md @@ -69,6 +69,16 @@ adapters. Commands using another established root call `project_files` on their actually selected root. Retain a helper only for the current operation; a later invocation resolves a new helper, so manual edits take effect. +The `specify_cli.integration_state` helpers are an explicit managed-metadata +exception to this handoff. Their reads and writes of `.specify/integration.json` +always use deny mode, including contained links when `allow_symlinks` is true. +This preserves the +[integration lifecycle's restriction](integration.md#storage-loading-and-lifecycle) +against symlinked write destinations before snapshotting and mutation. +This guarantee does not cover unmigrated consumers such as +`bundles.project.active_integration`, whose containment-checked reads can still +follow contained state-file links. + Existing Path-returning resolvers preflight policy without changing their return type or domain constructors. They do not implicitly equip unmigrated domain consumers with an operation helper. Failures are translated into existing diff --git a/src/specify_cli/integration_state.py b/src/specify_cli/integration_state.py index 45df5394ef..ae29b833a8 100644 --- a/src/specify_cli/integration_state.py +++ b/src/specify_cli/integration_state.py @@ -7,9 +7,16 @@ from pathlib import Path from typing import Any +from .file_helper import FileHelper, FileHelperError, SymlinkDeniedError + INTEGRATION_JSON = ".specify/integration.json" INTEGRATION_STATE_SCHEMA = 1 +_SYMLINK_STATE_ERROR = ( + f"Refusing symlinked integration state at {INTEGRATION_JSON}. " + "Replace linked entries in this path with a real directory or regular file; " + "allow_symlinks does not apply to integration state." +) @dataclass(frozen=True) @@ -35,6 +42,14 @@ def _read_integration_json_data( schema, encoding, or filesystem failures. """ path = project_root / INTEGRATION_JSON + try: + files = FileHelper(project_root) + except FileNotFoundError: + return None, None + except (OSError, RuntimeError) as exc: + return None, IntegrationReadError( + kind="os", detail=f"Cannot establish project root for {INTEGRATION_JSON}: {exc}" + ) # Avoid Path.exists() / Path.is_file() as a pre-check: both return False # on some OSErrors (e.g. permission errors during stat), which would # silently treat an unreadable-but-present file as missing. Attempt the @@ -42,7 +57,7 @@ def _read_integration_json_data( # other OSErrors (which become loud errors via the IntegrationReadError # path). try: - raw = path.read_text(encoding="utf-8") + raw = files.read_text(INTEGRATION_JSON, encoding="utf-8") except FileNotFoundError: return None, None except IsADirectoryError as exc: @@ -52,7 +67,9 @@ def _read_integration_json_data( ) except UnicodeDecodeError as exc: return None, IntegrationReadError(kind="decode", detail=str(exc)) - except OSError as exc: + except SymlinkDeniedError: + return None, IntegrationReadError(kind="os", detail=_SYMLINK_STATE_ERROR) + except (OSError, FileHelperError) as exc: return None, IntegrationReadError(kind="os", detail=str(exc)) try: data = json.loads(raw) @@ -223,9 +240,20 @@ def write_integration_json( installed_integrations: list[str] | None = None, settings: dict[str, dict[str, Any]] | None = None, ) -> None: - """Write ``.specify/integration.json`` with legacy-compatible state.""" + """Write legacy-compatible state beneath the trusted, existing project root. + + State reads and writes deny links below that root, independently of any + lifecycle transaction or operational symlink opt-in. + """ + files = FileHelper(project_root) + # Reject the full destination before an observer can snapshot its target. + try: + files.read_bytes(INTEGRATION_JSON) + except FileNotFoundError: + pass + except SymlinkDeniedError as exc: + raise SymlinkDeniedError(_SYMLINK_STATE_ERROR) from exc dest = project_root / INTEGRATION_JSON - dest.parent.mkdir(parents=True, exist_ok=True) integration_key = clean_integration_key(integration_key) installed = dedupe_integration_keys(installed_integrations or []) @@ -251,6 +279,10 @@ def write_integration_json( from .integrations._file_changes import after_file_change, before_file_change - before_file_change(dest) - dest.write_text(json.dumps(data, indent=2) + "\n", encoding="utf-8") - after_file_change(dest) + try: + before_file_change(dest) + files.mkdir(".specify", exist_ok=True) + files.overwrite_text(INTEGRATION_JSON, json.dumps(data, indent=2) + "\n", encoding="utf-8") + after_file_change(dest) + except SymlinkDeniedError as exc: + raise SymlinkDeniedError(_SYMLINK_STATE_ERROR) from exc diff --git a/tests/integrations/test_integration_state.py b/tests/integrations/test_integration_state.py index ebedc1056c..ceaa9cc845 100644 --- a/tests/integrations/test_integration_state.py +++ b/tests/integrations/test_integration_state.py @@ -2,13 +2,20 @@ import json +import pytest + +from specify_cli._file_policy import project_files, save_project_options +from specify_cli.file_helper import SymlinkDeniedError from specify_cli.integration_state import ( INTEGRATION_JSON, default_integration_key, integration_setting, normalize_integration_state, + try_read_integration_json, + try_read_integration_json_with_raw, write_integration_json, ) +from specify_cli.integrations._file_changes import file_change_observer def test_normalize_integration_state_strips_default_key_without_duplicates(): @@ -86,6 +93,161 @@ def test_write_integration_json_strips_integration_key(tmp_path): assert state["installed_integrations"] == ["claude"] +@pytest.fixture(params=[ + "external-file", + "dangling-file", + "contained-file", + "cyclic-file", + "external-directory", + "dangling-directory", + "contained-directory", +]) +def linked_integration_state(tmp_path, request): + project = tmp_path / "project" + project.mkdir() + outside = tmp_path / "outside" + outside.mkdir() + (outside / "integration.json").write_text('{"integration": "claude"}\n', encoding="utf-8") + contained = project / "state" + contained.mkdir() + (contained / "integration.json").write_text('{"integration": "claude"}\n', encoding="utf-8") + kind = request.param + if kind.endswith("directory"): + target = { + "external-directory": outside, + "dangling-directory": outside / "missing", + "contained-directory": contained, + }[kind] + link = project / ".specify" + else: + (project / ".specify").mkdir() + link = project / INTEGRATION_JSON + target = { + "external-file": outside / "integration.json", + "dangling-file": outside / "missing.json", + "contained-file": contained / "integration.json", + "cyclic-file": link, + }[kind] + try: + link.symlink_to(target, target_is_directory=kind.endswith("directory")) + except OSError as exc: + pytest.skip(f"symlinks unavailable: {exc}") + return project, link, target, outside, contained + + +@pytest.mark.parametrize("reader", [try_read_integration_json, try_read_integration_json_with_raw]) +def test_read_integration_json_rejects_symlinks(linked_integration_state, reader): + project, link, _, _, _ = linked_integration_state + + *states, error = reader(project) + + assert all(state is None for state in states) + assert error is not None + assert error.kind == "os" + assert "symlink" in error.detail.lower() + assert "Replace linked entries" in error.detail + assert "allow_symlinks does not apply to integration state" in error.detail + assert "explicitly allow symlinks" not in error.detail + assert link.is_symlink() + + +def test_write_integration_json_rejects_symlinks_before_observer(linked_integration_state): + project, link, _, outside, contained = linked_integration_state + original_target = link.readlink() + original = (outside / "integration.json").read_bytes() + contained_original = (contained / "integration.json").read_bytes() + changes = [] + token = file_change_observer.set(lambda *args: changes.append(args)) + try: + with pytest.raises(SymlinkDeniedError, match="symlink") as exc_info: + write_integration_json(project, version="1.2.3", integration_key="codex") + finally: + file_change_observer.reset(token) + + assert changes == [] + assert "Replace linked entries" in str(exc_info.value) + assert "allow_symlinks does not apply to integration state" in str(exc_info.value) + assert "explicitly allow symlinks" not in str(exc_info.value) + assert link.is_symlink() + assert link.readlink() == original_target + assert (outside / "integration.json").read_bytes() == original + assert (contained / "integration.json").read_bytes() == contained_original + assert not (outside / "missing.json").exists() + assert not (outside / "missing").exists() + + +@pytest.mark.parametrize("linked_integration_state", ["contained-file"], indirect=True) +def test_contained_state_link_is_denied_despite_symlink_opt_in(linked_integration_state): + project, link, target, _, _ = linked_integration_state + original = target.read_bytes() + original_target = link.readlink() + save_project_options(project, {"allow_symlinks": True}) + + files = project_files(project) + assert files.allow_symlinks is True + assert files.read_bytes(INTEGRATION_JSON) == original + + for reader in (try_read_integration_json, try_read_integration_json_with_raw): + *states, error = reader(project) + assert all(state is None for state in states) + assert error is not None + assert error.kind == "os" + assert "allow_symlinks does not apply to integration state" in error.detail + + with pytest.raises(SymlinkDeniedError, match="allow_symlinks does not apply"): + write_integration_json(project, version="1.2.3", integration_key="codex") + + assert link.is_symlink() + assert link.readlink() == original_target + assert target.read_bytes() == original + + +def test_write_integration_json_preserves_observer_and_regular_file(tmp_path): + destination = tmp_path / INTEGRATION_JSON + destination.parent.mkdir() + destination.write_text('{"integration": "claude"}\n', encoding="utf-8") + original = destination.read_bytes() + changes = [] + token = file_change_observer.set( + lambda path, before, removal: changes.append((path, before, removal, path.read_bytes())) + ) + try: + write_integration_json(tmp_path, version="1.2.3", integration_key="codex") + finally: + file_change_observer.reset(token) + + assert changes == [ + (destination, True, False, original), + (destination, False, False, destination.read_bytes()), + ] + state, raw, error = try_read_integration_json_with_raw(tmp_path) + assert error is None + assert state == raw + assert state["integration"] == "codex" + + +@pytest.mark.parametrize("reader", [try_read_integration_json, try_read_integration_json_with_raw]) +def test_read_integration_json_missing_is_not_an_error(tmp_path, reader): + assert all(value is None for value in reader(tmp_path)) + + +def test_integration_state_accepts_trusted_root_alias(tmp_path): + project = tmp_path / "project" + project.mkdir() + alias = tmp_path / "alias" + try: + alias.symlink_to(project, target_is_directory=True) + except OSError as exc: + pytest.skip(f"symlinks unavailable: {exc}") + + write_integration_json(alias, version="1.2.3", integration_key="claude") + state, error = try_read_integration_json(alias) + + assert error is None + assert state["integration"] == "claude" + assert (project / INTEGRATION_JSON).is_file() + + def test_with_integration_setting_recomputes_separator_from_retained_options(): """Updating only script_type must not drop an options-dependent separator. diff --git a/tests/specify_cli/integrations/test_command_install.py b/tests/specify_cli/integrations/test_command_install.py index abbbc2a3ff..568ea8bbd8 100644 --- a/tests/specify_cli/integrations/test_command_install.py +++ b/tests/specify_cli/integrations/test_command_install.py @@ -26,6 +26,45 @@ ) class TestIntegrationInstall: + @pytest.mark.parametrize("existing_target", [True, False]) + @pytest.mark.parametrize("allow_symlinks", [False, True]) + @pytest.mark.parametrize("contained_target", [False, True]) + def test_install_rejects_symlinked_state_before_setup( + self, tmp_path, existing_target, allow_symlinks, contained_target, + ): + project = _init_project(tmp_path, "claude") + options_path = project / ".specify" / "init-options.json" + options = json.loads(options_path.read_text(encoding="utf-8")) + options["allow_symlinks"] = allow_symlinks + options_path.write_text(json.dumps(options), encoding="utf-8") + destination = project / ".specify" / "integration.json" + target = project / "state.json" if contained_target else tmp_path / "external-state.json" + original = b'{"integration": "claude"}\n' + if existing_target: + target.write_bytes(original) + destination.unlink() + try: + destination.symlink_to(target) + except OSError as exc: + pytest.skip(f"symlinks unavailable: {exc}") + + result = _run_in_project(project, [ + "integration", "install", "codex", "--script", "sh", + ]) + + assert result.exit_code != 0, result.output + output = " ".join(strip_ansi(result.output).split()) + assert "Replace linked entries" in output + assert "allow_symlinks does not apply to integration state" in output + assert "explicitly allow symlinks" not in output + assert destination.is_symlink() + if existing_target: + assert target.read_bytes() == original + else: + assert not target.exists() + assert not (project / ".agents").exists() + assert not (project / ".specify" / "integrations" / "codex.manifest.json").exists() + def test_historical_catalog_versions_are_not_install_options(self): result = runner.invoke( app, ["integration", "install", "copilot", "--version", "1.0.0"] diff --git a/tests/specify_cli/integrations/test_command_status.py b/tests/specify_cli/integrations/test_command_status.py index 2a0f1be471..9f8212b36d 100644 --- a/tests/specify_cli/integrations/test_command_status.py +++ b/tests/specify_cli/integrations/test_command_status.py @@ -398,7 +398,7 @@ def fail_first_project_root_resolve(self, *args, **kwargs): assert result.stdout == "" assert "Cannot establish project root" in json.loads(result.stderr)["error"] - def test_status_loads_manifests_when_project_root_resolution_keeps_failing( + def test_status_reports_state_error_when_project_root_resolution_keeps_failing( self, copilot_project, monkeypatch, @@ -414,13 +414,15 @@ def fail_project_root_resolve(self, *args, **kwargs): from specify_cli.integration_status import build_integration_status_report - # The tolerant domain diagnostic remains available; CLI policy preflight - # now requires an established root before reaching this diagnostic. + # State I/O must establish its boundary even outside CLI policy preflight. payload = build_integration_status_report(copilot_project) - assert payload["status"] == "warning" - assert payload["manifests"]["copilot"]["readable"] is True - assert payload["manifests"]["speckit"]["readable"] is True + assert payload["status"] == "error" + assert payload["manifests"] == {} assert any(item["code"] == "project-root-unresolved" for item in payload["findings"]) + state_error = next( + item for item in payload["findings"] if item["code"] == "integration-state-unreadable" + ) + assert "Cannot establish project root" in state_error["message"] def test_status_uses_lexical_manifest_paths_when_project_root_resolution_falls_back(self, tmp_path): from specify_cli.integration_status import _manifest_file_status