fix(sleep): confine legacy SKILL.md and CLAUDE.md adoption to staged roots

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,

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 <root>/<name>/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")

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: