diff --git a/src/specify_cli/integrations/manifest.py b/src/specify_cli/integrations/manifest.py index 6a9eee668..7c2b75095 100644 --- a/src/specify_cli/integrations/manifest.py +++ b/src/specify_cli/integrations/manifest.py @@ -115,6 +115,7 @@ class IntegrationManifest: self.project_root = project_root.resolve() self.version = version self._files: dict[str, str] = {} # rel_path → sha256 hex + self._recovered_files: set[str] = set() self._installed_at: str = "" # -- Manifest file location ------------------------------------------- @@ -146,9 +147,16 @@ class IntegrationManifest: self._files[normalized] = hashlib.sha256(content).hexdigest() return abs_path - def record_existing(self, rel_path: str | Path) -> None: + def record_existing(self, rel_path: str | Path, *, recovered: bool = False) -> None: """Record the hash of an already-existing regular file at *rel_path*. + When ``recovered=True``, the path is also marked in the manifest's + ``recovered_files`` list to signal that the file's on-disk hash was + *observed* during install (because the file already existed and was not + overwritten), not *produced* by the install. Future ``refresh_managed`` + runs should consult ``is_recovered`` before treating the recorded hash + as a managed baseline. + Raises: ValueError: if *rel_path* resolves outside the project root, is a symlink, or is not a regular file. A directory or other @@ -163,16 +171,27 @@ class IntegrationManifest: # canonical error messages used elsewhere. if rel.is_absolute() or ".." in rel.parts: _validate_rel_path(rel, self.project_root) - # Defensive: _validate_rel_path always raises on these inputs, - # but make the contract explicit if it is ever loosened. - raise ValueError(f"Manifest path escapes project root: {rel}") - # Check ``is_symlink()`` on the un-resolved path because - # ``_validate_rel_path`` resolves the path (which would follow - # the symlink and silently record the target instead). - if (self.project_root / rel).is_symlink(): + # _validate_rel_path raised for any actually-escaping path. If we reach + # here the path normalizes inside root (e.g. ``dir/../file.txt``). + # Reject anyway: manifest keys must be canonical so ``check_modified`` + # and ``uninstall`` cannot key the same file under two paths. raise ValueError( - f"Refusing to record symlinked manifest path: {rel}" + f"Manifest paths must be canonical; '..' segments are not " + f"allowed (got {rel})" ) + # Walk each path component before resolution so a symlinked ancestor + # (e.g. ``linked_dir/file.txt`` where ``linked_dir`` is a symlink) + # cannot be silently followed by ``_validate_rel_path().resolve()`` + # down to a target outside the project root. ``_ensure_safe_manifest_directory`` + # uses the same pattern. + _walk = self.project_root + for part in rel.parts: + _walk = _walk / part + if _walk.is_symlink(): + raise ValueError( + f"Refusing to record symlinked manifest path: {rel} " + f"(symlinked at {_walk.relative_to(self.project_root).as_posix()})" + ) abs_path = _validate_rel_path(rel, self.project_root) if not abs_path.is_file(): raise ValueError( @@ -180,6 +199,8 @@ class IntegrationManifest: ) normalized = abs_path.relative_to(self.project_root).as_posix() self._files[normalized] = _sha256(abs_path) + if recovered: + self._recovered_files.add(normalized) # -- Querying --------------------------------------------------------- @@ -188,6 +209,24 @@ class IntegrationManifest: """Return a copy of the ``{rel_path: sha256}`` mapping.""" return dict(self._files) + @property + def recovered_files(self) -> set[str]: + """Return a copy of the set of paths recorded with ``recovered=True``. + + These entries had their hashes observed (not produced) during install + because the file already existed on disk and the install skipped it. + Their on-disk bytes may be user customizations — callers that would + overwrite based on hash equality (e.g. ``refresh_managed``) MUST check + ``is_recovered`` first. + """ + return set(self._recovered_files) + + def is_recovered(self, rel_path: str | Path) -> bool: + """Return True if *rel_path* was recorded via ``record_existing(recovered=True)``.""" + rel = Path(rel_path) + normalized = rel.as_posix() + return normalized in self._recovered_files + def check_modified(self) -> list[str]: """Return relative paths of tracked files whose content changed on disk.""" modified: list[str] = [] @@ -294,6 +333,11 @@ class IntegrationManifest: "version": self.version, "installed_at": self._installed_at, "files": self._files, + **( + {"recovered_files": sorted(self._recovered_files)} + if self._recovered_files + else {} + ), } path = self.manifest_path content = json.dumps(data, indent=2) + "\n" @@ -345,6 +389,16 @@ class IntegrationManifest: inst._installed_at = data.get("installed_at", "") inst._files = files + recovered = data.get("recovered_files", []) + if not isinstance(recovered, list) or not all( + isinstance(p, str) for p in recovered + ): + raise ValueError( + f"Integration manifest 'recovered_files' at {path} must be a " + "list of string paths" + ) + inst._recovered_files = set(recovered) + stored_key = data.get("integration", "") if stored_key and stored_key != key: raise ValueError( diff --git a/src/specify_cli/shared_infra.py b/src/specify_cli/shared_infra.py index 673fba3f6..8071df69a 100644 --- a/src/specify_cli/shared_infra.py +++ b/src/specify_cli/shared_infra.py @@ -368,7 +368,7 @@ def install_shared_infra( # ``manifest.files`` performs on every access. if dst_path.is_file() and rel not in prior_hashes: try: - manifest.record_existing(rel) + manifest.record_existing(rel, recovered=True) except (OSError, ValueError) as exc: # Tolerate races / permission issues / non-file # collisions so one weird path does not abort @@ -409,7 +409,7 @@ def install_shared_infra( # performs on every access. if dst.is_file() and rel not in prior_hashes: try: - manifest.record_existing(rel) + manifest.record_existing(rel, recovered=True) except (OSError, ValueError) as exc: # Tolerate races / permission issues / non-file # collisions so one weird path does not abort diff --git a/tests/integrations/test_manifest.py b/tests/integrations/test_manifest.py index a3fb14e3d..e4d7c5ce7 100644 --- a/tests/integrations/test_manifest.py +++ b/tests/integrations/test_manifest.py @@ -296,3 +296,83 @@ class TestManifestLoadValidation: path.write_text("{not valid json", encoding="utf-8") with pytest.raises(ValueError, match="invalid JSON"): IntegrationManifest.load("bad", tmp_path) + + +class TestManifestRecoveredFiles: + """Coverage for the ``recovered_files`` channel added in #2483. + + When ``shared_infra`` skips an existing file (because the user already has + it on disk) it now records the file with ``recovered=True``. The path + appears in ``manifest.recovered_files`` and ``is_recovered(path)`` returns + True. ``refresh_managed`` (out of scope for this PR) consults this list + before treating the recorded hash as a managed baseline, defending against + silent overwrite of user customizations after manifest loss. + """ + + def test_record_existing_default_is_not_recovered(self, tmp_path): + (tmp_path / "f.txt").write_text("x", encoding="utf-8") + m = IntegrationManifest("test", tmp_path) + m.record_existing("f.txt") + assert m.is_recovered("f.txt") is False + assert m.recovered_files == set() + + def test_record_existing_with_recovered_flag(self, tmp_path): + (tmp_path / "f.txt").write_text("x", encoding="utf-8") + m = IntegrationManifest("test", tmp_path) + m.record_existing("f.txt", recovered=True) + assert m.is_recovered("f.txt") is True + assert m.recovered_files == {"f.txt"} + # File still hashed normally so check_modified/uninstall keep working + assert m.files["f.txt"] == _sha256(tmp_path / "f.txt") + + def test_recovered_files_round_trips_through_save_load(self, tmp_path): + (tmp_path / "a.txt").write_text("aaa", encoding="utf-8") + (tmp_path / "b.txt").write_text("bbb", encoding="utf-8") + m = IntegrationManifest("test", tmp_path, version="9.9") + m.record_existing("a.txt", recovered=True) + m.record_existing("b.txt") # not recovered + m.save() + loaded = IntegrationManifest.load("test", tmp_path) + assert loaded.is_recovered("a.txt") is True + assert loaded.is_recovered("b.txt") is False + assert loaded.recovered_files == {"a.txt"} + + def test_save_omits_empty_recovered_files(self, tmp_path): + m = IntegrationManifest("test", tmp_path) + m.record_file("f.txt", "x") + path = m.save() + data = json.loads(path.read_text(encoding="utf-8")) + assert "recovered_files" not in data + + def test_load_rejects_non_list_recovered_files(self, tmp_path): + path = tmp_path / ".specify" / "integrations" / "bad.manifest.json" + path.parent.mkdir(parents=True) + path.write_text( + json.dumps({"files": {}, "recovered_files": "not-a-list"}), + encoding="utf-8", + ) + with pytest.raises(ValueError, match="recovered_files"): + IntegrationManifest.load("bad", tmp_path) + + +class TestRecordExistingNewGuards: + """Coverage for the two new guards added by Copilot's 2026-05-18 review.""" + + def test_rejects_symlinked_ancestor(self, tmp_path): + real_dir = tmp_path / "real_dir" + real_dir.mkdir() + (real_dir / "file.txt").write_text("payload", encoding="utf-8") + (tmp_path / "linked_dir").symlink_to(real_dir, target_is_directory=True) + m = IntegrationManifest("test", tmp_path) + with pytest.raises(ValueError, match="symlinked"): + m.record_existing("linked_dir/file.txt") + + def test_rejects_inside_root_dotdot_with_explicit_message(self, tmp_path): + # ``dir/../file.txt`` normalizes inside root, so the old "escapes + # project root" message was misleading. The new message names the + # actual reason: canonicalization. + (tmp_path / "dir").mkdir() + (tmp_path / "file.txt").write_text("x", encoding="utf-8") + m = IntegrationManifest("test", tmp_path) + with pytest.raises(ValueError, match=r"canonical|'\.\.' segments"): + m.record_existing("dir/../file.txt")