Skip to content
Open
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
10 changes: 10 additions & 0 deletions design/file-helper.md
Original file line number Diff line number Diff line change
Expand Up @@ -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
Expand Down
46 changes: 39 additions & 7 deletions src/specify_cli/integration_state.py
Original file line number Diff line number Diff line change
Expand Up @@ -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)
Expand All @@ -35,14 +42,22 @@ 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
# read directly and distinguish FileNotFoundError (genuinely absent) from
# 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:
Expand All @@ -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)
Expand Down Expand Up @@ -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 [])
Expand All @@ -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
162 changes: 162 additions & 0 deletions tests/integrations/test_integration_state.py
Original file line number Diff line number Diff line change
Expand Up @@ -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():
Expand Down Expand Up @@ -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.

Expand Down
39 changes: 39 additions & 0 deletions tests/specify_cli/integrations/test_command_install.py
Original file line number Diff line number Diff line change
Expand Up @@ -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"]
Expand Down
14 changes: 8 additions & 6 deletions tests/specify_cli/integrations/test_command_status.py
Original file line number Diff line number Diff line change
Expand Up @@ -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,
Expand All @@ -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
Expand Down
Loading