diff --git a/skillopt_sleep/cycle.py b/skillopt_sleep/cycle.py index f1d7bf02..abf269bc 100644 --- a/skillopt_sleep/cycle.py +++ b/skillopt_sleep/cycle.py @@ -1014,7 +1014,7 @@ def run_sleep_cycle( report_md=report_md, out_dir=staging_dir_pre, skill_proposals=skill_proposals, - skill_roots=skill_search_roots(cfg) if skill_proposals else (), + skill_roots=skill_search_roots(cfg), ) if ev is not None: ev.log("stage", "staged", staging_dir=staging_dir, diff --git a/skillopt_sleep/staging.py b/skillopt_sleep/staging.py index 615c5ac8..1e7087ff 100644 --- a/skillopt_sleep/staging.py +++ b/skillopt_sleep/staging.py @@ -1076,6 +1076,7 @@ def write_staging( manifest = { "schema": _MANIFEST_SCHEMA, "schema_version": _MANIFEST_VERSION, + "project_root": os.path.abspath(project), "live_skill_path": live_skill_path, "live_memory_path": live_memory_path, # PyPI v0.2.0 adopted these top-level flags without integrity pins. @@ -1086,26 +1087,30 @@ def write_staging( "has_managed_memory": proposed_memory is not None, "accepted": report.accepted, } - if skill_rows: - manifest["skills"] = skill_rows - # The roots the fan-out actually resolved from. Recorded so adoption can - # re-check containment instead of trusting each row's live path. - recorded_roots = [ - os.path.abspath(os.path.expanduser(str(root))) - for root in skill_roots - if isinstance(root, str) and str(root).strip() - ] - if not recorded_roots: - # Low-level callers may not know the search roots. Every live target - # is //SKILL.md, so the root each resolved path sits in - # is derivable here -- at stage time, from paths we just resolved - # ourselves, never from the manifest we are about to trust later. - recorded_roots = [ + recorded_roots = [ + os.path.abspath(os.path.expanduser(str(root))) + for root in skill_roots + if isinstance(root, str) and str(root).strip() + ] + if not recorded_roots: + if skill_rows: + recorded_roots.extend( os.path.dirname(os.path.dirname(os.path.abspath(str(row["live_skill_path"])))) for row in skill_rows if str(row.get("live_skill_path") or "").strip() - ] - manifest["skill_roots"] = list(dict.fromkeys(recorded_roots)) + ) + if live_skill_path and str(live_skill_path).strip(): + live_abs = os.path.abspath(live_skill_path) + dir1 = os.path.dirname(live_abs) + dir2 = os.path.dirname(dir1) + proj_abs = os.path.abspath(project) + if dir1 == proj_abs or dir2 == proj_abs: + recorded_roots.append(dir1) + else: + recorded_roots.append(dir2) + manifest["skill_roots"] = list(dict.fromkeys(recorded_roots)) + if skill_rows: + manifest["skills"] = skill_rows if legacy: manifest["legacy"] = legacy artifacts: List[tuple[str, str]] = [ @@ -1338,6 +1343,27 @@ def staged_skill_roots(staging_dir: str) -> List[str]: return out +def staged_project_root(staging_dir: str) -> str: + """The project root recorded when this night was staged.""" + manifest_path = os.path.join(staging_dir, "manifest.json") + try: + with open(manifest_path, encoding="utf-8") as f: + manifest = json.load(f) + except (OSError, UnicodeError, json.JSONDecodeError, ValueError) as exc: + raise StagingError(f"cannot read staging manifest: {exc}") from exc + if not isinstance(manifest, dict): + raise StagingError("staging manifest must be a JSON object") + root = manifest.get("project_root") + if not isinstance(root, str) or not root.strip(): + raise StagingError( + "staging manifest is missing 'project_root'; it was written by an older " + "version that could not confine adoption. Discard and restage this night." + ) + if not os.path.isabs(root): + raise StagingError(f"staging manifest 'project_root' entry is not absolute: {root}") + return root + + def _selected_rows( rows: Sequence[Dict[str, Any]], skill_names: Optional[Sequence[str]] ) -> List[Dict[str, Any]]: @@ -3026,6 +3052,12 @@ def adopt(staging_dir: str) -> List[str]: initial_rows = _legacy_rows(initial_manifest) if not initial_rows: return [] + skill_roots: List[str] = [] + if "skill" in initial_rows: + skill_roots = staged_skill_roots(staging_dir) + project_root = "" + if "memory" in initial_rows: + project_root = staged_project_root(staging_dir) initial_paths: List[str] = [] for row in initial_rows.values(): live = _safe_live_path(row.get("live_path")) @@ -3088,6 +3120,16 @@ def adopt(staging_dir: str) -> List[str]: expected_realpath, expected_basename=expected_basename, ) + allowed_roots = [project_root] if label == "memory" else skill_roots + if not _live_target_within_roots(live, allowed_roots): + if label == "memory": + raise StagingError( + f"live target for legacy memory is outside the project root: {live}" + ) + raise StagingError( + f"live target for legacy skill is outside the skills roots recorded " + f"when this night was staged: {live}" + ) staged = os.path.join(staging_dir, expected_file) if _is_link_or_junction(staged) or not os.path.isfile(staged): raise StagingError(f"legacy {label} proposal is missing or a symlink") diff --git a/tests/test_sleep_adopt_skill_subset.py b/tests/test_sleep_adopt_skill_subset.py index f6c32af5..a81e9b76 100644 --- a/tests/test_sleep_adopt_skill_subset.py +++ b/tests/test_sleep_adopt_skill_subset.py @@ -23,6 +23,8 @@ has_pending_staged_managed, latest_staging, pending_staged_skills, + staged_project_root, + staged_skill_roots, staged_skills, write_staging, ) @@ -163,6 +165,147 @@ def test_the_ordinary_in_root_adoption_still_succeeds(self): self.assertEqual(handle.read(), "# alpha v2\n") +class TestLegacyAdoptionIsConfinedToTheStagedRoots(unittest.TestCase): + """Refuse legacy SKILL.md/CLAUDE.md targets that escape staged roots. + + Legacy adoption (``adopt``) must confine live skill targets to the recorded + skills roots and live memory targets (``CLAUDE.md``) to the recorded project + root. A tampered or malicious manifest must never redirect writes outside + these roots or create directories outside them. + """ + + def _legacy_night(self, tmp): + live_root = os.path.join(_canonical(tmp), "live") + skill = os.path.join(live_root, "skill", "SKILL.md") + memory = os.path.join(live_root, "CLAUDE.md") + _write(skill, "# skill v1\n") + _write(memory, "# memory v1\n") + staging = write_staging( + tmp, + report=SleepReport(night=1, project=tmp, accepted=True), + proposed_skill="# skill v2\n", + proposed_memory="# memory v2\n", + live_skill_path=skill, + live_memory_path=memory, + report_md="# report\n", + ) + return staging, skill, memory + + def _retarget(self, staging, target_label, new_live): + new_live = _canonical(new_live) + manifest_path = os.path.join(staging, "manifest.json") + with open(manifest_path, encoding="utf-8") as handle: + manifest = json.load(handle) + row = manifest["legacy"][target_label] + row["live_path"] = new_live + row["live_realpath"] = new_live + if os.path.exists(new_live): + with open(new_live, "rb") as h: + row["live_sha256"] = hashlib.sha256(h.read()).hexdigest() + else: + row["live_sha256"] = "" + if target_label == "skill": + manifest["live_skill_path"] = new_live + else: + manifest["live_memory_path"] = new_live + with open(manifest_path, "w", encoding="utf-8") as handle: + json.dump(manifest, handle) + + def test_legacy_skill_retargeted_onto_an_outside_file_is_refused(self): + with tempfile.TemporaryDirectory() as tmp: + staging, skill, memory = self._legacy_night(tmp) + outside = os.path.join(tmp, "outside", "victim", "SKILL.md") + os.makedirs(os.path.dirname(outside), exist_ok=True) + _write(outside, "# victim\n") + self._retarget(staging, "skill", outside) + with self.assertRaises(StagingError) as ctx: + adopt(staging) + self.assertIn("outside the skills roots", str(ctx.exception)) + # Fails closed: the victim file is untouched. + with open(outside, encoding="utf-8") as handle: + self.assertEqual(handle.read(), "# victim\n") + + def test_legacy_skill_retargeted_to_create_a_new_outside_file_is_refused(self): + with tempfile.TemporaryDirectory() as tmp: + staging, skill, memory = self._legacy_night(tmp) + outside_dir = os.path.join(tmp, "outside", "victim") + outside = os.path.join(outside_dir, "SKILL.md") + self._retarget(staging, "skill", outside) + with self.assertRaises(StagingError): + adopt(staging) + self.assertFalse(os.path.exists(outside)) + + def test_legacy_memory_retargeted_onto_an_outside_file_is_refused(self): + with tempfile.TemporaryDirectory() as tmp: + staging, skill, memory = self._legacy_night(tmp) + outside_root = tempfile.mkdtemp() + try: + outside = os.path.join(outside_root, "CLAUDE.md") + _write(outside, "# victim memory\n") + self._retarget(staging, "memory", outside) + with self.assertRaises(StagingError) as ctx: + adopt(staging) + self.assertIn("outside the project root", str(ctx.exception)) + with open(outside, encoding="utf-8") as handle: + self.assertEqual(handle.read(), "# victim memory\n") + finally: + import shutil + shutil.rmtree(outside_root, ignore_errors=True) + + def test_legacy_memory_retargeted_to_create_a_new_outside_file_is_refused(self): + with tempfile.TemporaryDirectory() as tmp: + staging, skill, memory = self._legacy_night(tmp) + outside_root = tempfile.mkdtemp() + try: + outside = os.path.join(outside_root, "nested", "CLAUDE.md") + self._retarget(staging, "memory", outside) + with self.assertRaises(StagingError): + adopt(staging) + self.assertFalse(os.path.exists(outside)) + finally: + import shutil + shutil.rmtree(outside_root, ignore_errors=True) + + def test_legacy_manifest_without_recorded_skill_roots_is_refused(self): + with tempfile.TemporaryDirectory() as tmp: + staging, skill, memory = self._legacy_night(tmp) + manifest_path = os.path.join(staging, "manifest.json") + with open(manifest_path, encoding="utf-8") as handle: + manifest = json.load(handle) + del manifest["skill_roots"] + with open(manifest_path, "w", encoding="utf-8") as handle: + json.dump(manifest, handle) + with self.assertRaises(StagingError) as ctx: + adopt(staging) + self.assertIn("skill_roots", str(ctx.exception)) + + def test_legacy_manifest_without_recorded_project_root_is_refused(self): + with tempfile.TemporaryDirectory() as tmp: + staging, skill, memory = self._legacy_night(tmp) + manifest_path = os.path.join(staging, "manifest.json") + with open(manifest_path, encoding="utf-8") as handle: + manifest = json.load(handle) + del manifest["project_root"] + with open(manifest_path, "w", encoding="utf-8") as handle: + json.dump(manifest, handle) + with self.assertRaises(StagingError) as ctx: + adopt(staging) + self.assertIn("project_root", str(ctx.exception)) + + def test_staged_project_root_reads_and_validates(self): + with tempfile.TemporaryDirectory() as tmp: + staging, skill, memory = self._legacy_night(tmp) + self.assertEqual(staged_project_root(staging), os.path.abspath(tmp)) + + def test_ordinary_legacy_adoption_still_succeeds(self): + with tempfile.TemporaryDirectory() as tmp: + staging, skill, memory = self._legacy_night(tmp) + updated = adopt(staging) + self.assertEqual(updated, [skill, memory]) + self.assertEqual(_read(skill), "# skill v2\n") + self.assertEqual(_read(memory), "# memory v2\n") + + class TestStagedSkills(unittest.TestCase): def test_rows_are_readable_from_the_manifest(self): with tempfile.TemporaryDirectory() as tmp: