fix(sleep): confine per-skill adoption to the staged skills roots

skillopt_sleep/cycle.py

@@ -958,6 +958,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 (),
         )
         if ev is not None:
             ev.log("stage", "staged", staging_dir=staging_dir,

skillopt_sleep/staging.py

@@ -365,6 +365,28 @@ def _safe_live_path(path: object) -> str:
     return candidate
 
 
+def _live_target_within_roots(live: str, roots: Iterable[str]) -> bool:
+    """Return True when ``live`` resolves inside one of ``roots``.
+
+    ``_safe_live_path`` only proves a target is absolute, traversal-free and
+    ``*.md``; it accepts any such path on the machine. Containment is a separate
+    question and the manifest is not a trust boundary -- a tampered
+    ``live_skill_path`` with self-consistent pins otherwise redirects an adopt
+    onto an arbitrary file. Callers pass the roots recorded when the night was
+    staged and must run this AFTER the existing ``realpath(live) == live``
+    identity check, so a symlinked ancestor cannot make an outside target look
+    contained.
+    """
+    real_live = os.path.realpath(live)
+    for root in roots:
+        if not isinstance(root, str) or not root.strip():
+            continue
+        real_root = os.path.realpath(os.path.abspath(os.path.expanduser(root)))
+        if _path_is_within(real_live, real_root):
+            return True
+    return False
+
+
 def proposal_filename(skill_name: str) -> str:
     """Staged filename for one skill's proposal (unique per skill name)."""
     return f"proposed_SKILL.{skill_name}.md"
@@ -950,6 +972,7 @@ def write_staging(
     report_md: str,
     out_dir: str = "",
     skill_proposals: Iterable[SkillProposal] = (),
+    skill_roots: Iterable[str] = (),
 ) -> str:
     """Write proposals + report into staging/<ts>/ and return that path.
 
@@ -1031,6 +1054,24 @@ def write_staging(
     }
     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 = [
+                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 legacy:
         manifest["legacy"] = legacy
     artifacts: List[tuple[str, str]] = [
@@ -1237,6 +1278,32 @@ def staged_skills(staging_dir: str) -> List[Dict[str, Any]]:
     return rows
 
 
+def staged_skill_roots(staging_dir: str) -> List[str]:
+    """The skills roots 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")
+    roots = manifest.get("skill_roots")
+    if not isinstance(roots, list) or not roots:
+        raise StagingError(
+            "staging manifest is missing 'skill_roots'; it was written by an older "
+            "version that could not confine adoption. Discard and restage this night."
+        )
+    out: List[str] = []
+    for root in roots:
+        if not isinstance(root, str) or not root.strip():
+            raise StagingError("staging manifest 'skill_roots' must be non-empty strings")
+        if not os.path.isabs(root):
+            raise StagingError(f"staging manifest 'skill_roots' entry is not absolute: {root}")
+        out.append(root)
+    return out
+
+
 def _selected_rows(
     rows: Sequence[Dict[str, Any]], skill_names: Optional[Sequence[str]]
 ) -> List[Dict[str, Any]]:
@@ -1385,7 +1452,12 @@ def _adopt_target_ok(
         )
 
 
-def _adopt_live_target_ok(name: str, live: str, expected_realpath: str) -> None:
+def _adopt_live_target_ok(
+    name: str,
+    live: str,
+    expected_realpath: str,
+    roots: Iterable[str] = (),
+) -> None:
     _adopt_target_ok(
         repr(name),
         live,
@@ -1397,6 +1469,13 @@ def _adopt_live_target_ok(name: str, live: str, expected_realpath: str) -> None:
         raise StagingError(
             f"live skill path for {name!r} is not {name}/SKILL.md: {live}"
         )
+    # Shape is not location. Run this only after ``_adopt_target_ok`` has proved
+    # ``realpath(live) == live``, so a symlinked ancestor cannot fake containment.
+    if roots and not _live_target_within_roots(live, roots):
+        raise StagingError(
+            f"live skill path for {name!r} is outside the skills roots recorded "
+            f"when this night was staged: {live}"
+        )
 
 
 def _planned_live_directories(
@@ -2628,6 +2707,7 @@ def adopt_skills(
     initial_rows = _selected_rows(initial_all_rows, skill_names)
     if not initial_rows:
         return []
+    skill_roots = staged_skill_roots(staging_dir)
     initial_live_paths: List[str] = []
     for row in initial_rows:
         live = _safe_live_path(row.get("live_skill_path"))
@@ -2680,7 +2760,7 @@ def adopt_skills(
                     f"staged skill {name!r} is missing safe live baseline pins; "
                     "discard and restage this night"
                 )
-            _adopt_live_target_ok(name, live, expected_realpath)
+            _adopt_live_target_ok(name, live, expected_realpath, skill_roots)
 
             expected_file = proposal_filename(name)
             if row.get("proposed_file") != expected_file:

tests/test_sleep_adopt_skill_subset.py

@@ -79,6 +79,79 @@ def __init__(
         )
 
 
+class TestAdoptionIsConfinedToTheStagedRoots(unittest.TestCase):
+    """A manifest is data, not a trust boundary.
+
+    ``_safe_live_path`` only proves a target is absolute, traversal-free and
+    ``*.md``; it accepts any such path on the machine. Adoption must therefore
+    re-check that each live target still sits under a skills root recorded when
+    the night was staged, or a tampered ``live_skill_path`` with self-consistent
+    pins redirects the write onto an arbitrary file.
+    """
+
+    def _retarget(self, staging, skill_name, new_live):
+        manifest_path = os.path.join(staging, "manifest.json")
+        with open(manifest_path, encoding="utf-8") as handle:
+            manifest = json.load(handle)
+        for row in manifest["skills"]:
+            if row["skill_name"] == skill_name:
+                row["live_skill_path"] = new_live
+                row["live_realpath"] = new_live
+                if row.get("live_sha256"):
+                    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"] = ""
+        with open(manifest_path, "w", encoding="utf-8") as handle:
+            json.dump(manifest, handle)
+
+    def test_manifest_retargeted_onto_an_outside_file_is_refused(self):
+        with tempfile.TemporaryDirectory() as tmp:
+            night = TwoSkillNight(tmp)
+            outside = os.path.join(tmp, "outside", "alpha", "SKILL.md")
+            os.makedirs(os.path.dirname(outside), exist_ok=True)
+            _write(outside, "# victim\n")
+            self._retarget(night.staging, "alpha", outside)
+            with self.assertRaises(StagingError) as ctx:
+                adopt_skills(night.staging, ["alpha"])
+            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_manifest_retargeted_to_create_a_new_outside_file_is_refused(self):
+        with tempfile.TemporaryDirectory() as tmp:
+            night = TwoSkillNight(tmp)
+            outside_dir = os.path.join(tmp, "outside", "alpha")
+            os.makedirs(outside_dir, exist_ok=True)
+            outside = os.path.join(outside_dir, "SKILL.md")
+            self._retarget(night.staging, "alpha", outside)
+            with self.assertRaises(StagingError):
+                adopt_skills(night.staging, ["alpha"])
+            self.assertFalse(os.path.exists(outside))
+
+    def test_a_manifest_without_recorded_roots_is_refused(self):
+        with tempfile.TemporaryDirectory() as tmp:
+            night = TwoSkillNight(tmp)
+            manifest_path = os.path.join(night.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_skills(night.staging, ["alpha"])
+            self.assertIn("skill_roots", str(ctx.exception))
+
+    def test_the_ordinary_in_root_adoption_still_succeeds(self):
+        with tempfile.TemporaryDirectory() as tmp:
+            night = TwoSkillNight(tmp)
+            adopt_skills(night.staging, ["alpha", "beta"])
+            with open(night.alpha_live, encoding="utf-8") as handle:
+                self.assertEqual(handle.read(), "# alpha v2\n")
+
+
 class TestStagedSkills(unittest.TestCase):
     def test_rows_are_readable_from_the_manifest(self):
         with tempfile.TemporaryDirectory() as tmp: