fix: confine installer writes to the selected target

README.md

@@ -76,6 +76,14 @@ bash install.sh --ingredient protean-drafts --target ./protean-installed
 
 The installer verifies each pinned tag and tree before promotion, then reads every written file back.
 
+The explicit target root may be a deliberate symlink; it is resolved once. All
+selected staging, promotion, cleanup, and contribution-state paths must stay
+beneath that physical root. Existing child links that escape it are refused
+with install exit 5 before any selected ingredient writes or staging cleanup.
+Links to destinations within the root remain supported. The installer preflights
+the full selection and rechecks mutation boundaries; these path checks do not
+protect against concurrent hostile link replacement between a check and use.
+
 ## What's in the repo
 
 | Folder | What it is |

build/install.py

@@ -491,14 +491,73 @@ def read_descriptor(entry_dir):
         return json.load(fh)
 
 
-def copy_paths(entry_dir, staging, targets):
+def confined_path(target, path):
+    """Check both spelling and physical destination against the resolved root.
+
+    Rechecked at mutation boundaries; this is not descriptor-based IO and does
+    not defend against a concurrent actor replacing links between check and use.
+    """
+    path = os.path.abspath(path)
+    try:
+        inside = (os.path.commonpath([target, path]) == target
+                  and os.path.commonpath([target, os.path.realpath(path)]) == target)
+    except (ValueError, OSError):
+        inside = False
+    if not inside:
+        raise Failure(5, "target path escapes the resolved target root: %s" % path)
+    return path
+
+
+def confined_tree(target, path):
+    """Check existing descendants too, before recursively deleting a tree."""
+    confined_path(target, path)
+    if os.path.isdir(path):
+        for root, dirs, files in os.walk(path, followlinks=False):
+            for name in dirs + files:
+                confined_path(target, os.path.join(root, name))
+
+
+def selected_paths(entry, entry_dir):
+    """Expand every selected payload before the first target mutation."""
+    paths = set(entry["installTargets"])
+    for rel in entry["installTargets"]:
+        src = os.path.join(entry_dir, rel)
+        if os.path.isdir(src):
+            for root, dirs, files in os.walk(src, followlinks=True):
+                for name in dirs + files:
+                    paths.add(os.path.relpath(os.path.join(root, name), entry_dir))
+        elif not os.path.isfile(src):
+            raise Failure(5, "declared install target missing in the ingredient: %s" % rel)
+    return paths
+
+
+def preflight_target(target, order, entries, sources):
+    confined_tree(target, os.path.join(target, STAGING))
+    for slug in order:
+        staging = os.path.join(target, STAGING, slug)
+        confined_path(target, staging)
+        for rel in selected_paths(entries[slug], sources[slug]):
+            confined_path(target, os.path.join(target, rel))
+            confined_path(target, os.path.join(staging, rel))
+        descriptor = read_descriptor(sources[slug])
+        for gate in descriptor.get("gates") or []:
+            confined_path(target, os.path.join(staging, gate.get("cwd", ".")))
+    confined_path(target, os.path.join(target, STATE_REL))
+    confined_path(target, os.path.join(target, STATE_TMPL_REL))
+
+
+def copy_paths(entry_dir, staging, targets, target):
     written = []
     for rel in targets:
         src = os.path.join(entry_dir, rel)
-        dest = os.path.join(staging, rel)
+        dest = confined_path(target, os.path.join(staging, rel))
         if os.path.isdir(src):
             if os.path.isdir(dest):
-                shutil.rmtree(dest)
+                confined_tree(target, dest)
+                if os.path.islink(dest):
+                    os.unlink(dest)
+                else:
+                    shutil.rmtree(dest)
             shutil.copytree(src, dest)
             for root, _dirs, files in os.walk(dest):
                 for name in files:
@@ -512,7 +571,7 @@ def copy_paths(entry_dir, staging, targets):
     return sorted(written)
 
 
-def run_gates(descriptor, entry_dir, staging):
+def run_gates(descriptor, entry_dir, staging, target):
     gates = descriptor.get("gates") or []
     ran = []
     for gate in gates:
@@ -521,7 +580,7 @@ def run_gates(descriptor, entry_dir, staging):
         cmd = gate.get("cmd")
         if not cmd:
             continue
-        cwd = os.path.join(staging, gate.get("cwd", "."))
+        cwd = confined_path(target, os.path.join(staging, gate.get("cwd", ".")))
         result = run(["bash", "-c", cmd], cwd=cwd)
         ran.append({"name": gate.get("name"), "cmd": cmd, "exit": result.returncode})
         if result.returncode != 0:
@@ -532,8 +591,8 @@ def run_gates(descriptor, entry_dir, staging):
 
 def promote(staging, target, written):
     for rel in written:
-        src = os.path.join(staging, rel)
-        dest = os.path.join(target, rel)
+        src = confined_path(target, os.path.join(staging, rel))
+        dest = confined_path(target, os.path.join(target, rel))
         os.makedirs(os.path.dirname(dest), exist_ok=True)
         if os.path.exists(dest):
             os.remove(dest)
@@ -560,18 +619,27 @@ def read_back(entry_dir, target, written):
 def install_one(entry, cache, opts, plan, entry_dir):
     slug = entry["slug"]
     target = plan["target"]
+    preflight_target(target, [slug], {slug: entry}, {slug: entry_dir})
     os.makedirs(target, exist_ok=True)
     if not os.access(target, os.W_OK):
         raise Failure(5, "target is not writable: %s" % target)
     descriptor = read_descriptor(entry_dir)
     staging_root = os.path.join(target, STAGING, slug)
     if os.path.isdir(staging_root):
-        shutil.rmtree(staging_root)
+        confined_tree(target, staging_root)
+        if os.path.islink(staging_root):
+            os.unlink(staging_root)
+        else:
+            shutil.rmtree(staging_root)
     os.makedirs(staging_root, exist_ok=True)
-    written = copy_paths(entry_dir, staging_root, entry["installTargets"])
-    gates = run_gates(descriptor, entry_dir, staging_root)
+    written = copy_paths(entry_dir, staging_root, entry["installTargets"], target)
+    gates = run_gates(descriptor, entry_dir, staging_root, target)
     promote(staging_root, target, written)
-    shutil.rmtree(os.path.join(target, STAGING), ignore_errors=True)
+    confined_tree(target, os.path.join(target, STAGING))
+    if os.path.islink(os.path.join(target, STAGING)):
+        os.unlink(os.path.join(target, STAGING))
+    else:
+        shutil.rmtree(os.path.join(target, STAGING), ignore_errors=True)
     return written, gates
 
 
@@ -613,11 +681,11 @@ def contribution_state(target, mode, operator_ref):
     own installed template instead of carrying a copy of it. An existing record
     is left untouched: an install never silently widens a permission.
     """
-    dest = os.path.join(target, STATE_REL)
+    dest = confined_path(target, os.path.join(target, STATE_REL))
     if os.path.exists(dest):
         return None, ("%s already exists and was left unchanged; an install never overwrites "
                       "an existing contribution-state record" % STATE_REL)
-    template = os.path.join(target, STATE_TMPL_REL)
+    template = confined_path(target, os.path.join(target, STATE_TMPL_REL))
     if not os.path.isfile(template):
         if mode == "on":
             raise Failure(5, "contribution mode 'on' was requested, but %s is not installed, "
@@ -639,6 +707,7 @@ def contribution_state(target, mode, operator_ref):
         row = contribution_switch_row(now, mode, operator_ref)
         text = "\n".join(row if "empty state" in line.lower() else line
                          for line in text.splitlines()) + "\n"
+    confined_path(target, dest)
     os.makedirs(os.path.dirname(dest), exist_ok=True)
     with open(dest, "w", encoding="utf-8") as fh:
         fh.write(text)
@@ -695,7 +764,7 @@ def main(argv):
         warn(order_note)
 
     plan = {
-        "target": target_root(opts),
+        "target": os.path.realpath(target_root(opts)),
         "selection_label": "--all" if opts["all"] else ", ".join(selection),
         "no_deps": opts["no_deps"],
         "degraded": degraded,
@@ -711,12 +780,18 @@ def main(argv):
     info("order: %s" % " -> ".join(order))
     installed = []
     warnings = []
+    sources = {}
     for slug in order:
         entry = seen[slug]
         entry_dir, how = fetch_entry(entry, cache, opts["offline"])
         info("fetch: %s [%s]" % (slug, how))
         verify_entry(entry, entry_dir)
         info("verify: %s [sha ok, tree ok]" % slug)
+        sources[slug] = entry_dir
+    preflight_target(plan["target"], order, seen, sources)
+    for slug in order:
+        entry = seen[slug]
+        entry_dir = sources[slug]
         written, gates = install_one(entry, cache, opts, plan, entry_dir)
         info("install: %s [staged, %d gate(s) passed, %d file(s) promoted]"
              % (slug, len(gates), len(written)))

build/verify-all.py

@@ -88,6 +88,9 @@ def main():
     gates.append(("contribution mode tests",
                   [sys.executable, "-m", "unittest", "tests.test_contribution_mode"]))
 
+    gates.append(("target confinement tests",
+                  [sys.executable, "-m", "unittest", "tests.test_target_confinement"]))
+
     # Run all gates
     results = []
     for name, cmd in gates:

tests/test_target_confinement.py

@@ -0,0 +1,102 @@
+"""Offline target-boundary regressions using real pinned ingredient fixtures."""
+import os
+import sys
+import unittest
+
+sys.path.insert(0, os.path.dirname(os.path.abspath(__file__)))
+from test_kit_lock import FixtureCase, INSTALL, lock_entry, run, warm_cache, write_lock
+
+INSTALL = os.environ.get("INSTALL_SOURCE", INSTALL)
+
+
+class TestTargetConfinement(FixtureCase):
+    def setUp(self):
+        super().setUp()
+        self.target = os.path.join(self.base, "target")
+        self.outside = os.path.join(self.base, "outside")
+        os.makedirs(self.target)
+        os.makedirs(self.outside)
+        self.canary = os.path.join(self.outside, "canary")
+        with open(self.canary, "w") as fh:
+            fh.write("outside user bytes")
+
+    def lock(self, slugs=("protean-one",), **kwargs):
+        entries = []
+        for slug in slugs:
+            fixture = self.ingredient(slug, **kwargs)
+            warm_cache(self.cache, slug, fixture)
+            entries.append(lock_entry(slug, fixture))
+        return write_lock(os.path.join(self.base, "kit.lock.json"), entries)
+
+    def install(self, lock, target=None):
+        return run(INSTALL, "--all", "--target", target or self.target, "--offline",
+                   "--cache", self.cache, "--lock", lock)
+
+    def refused(self, result):
+        self.assertEqual(result.returncode, 5, result.stdout + result.stderr)
+        with open(self.canary) as fh:
+            self.assertEqual(fh.read(), "outside user bytes")
+        self.assertEqual(os.listdir(self.outside), ["canary"])
+        self.assertFalse(os.path.exists(os.path.join(self.target, "gates")))
+
+    def test_payload_parent_escape(self):
+        os.symlink(self.outside, os.path.join(self.target, "payload"))
+        self.refused(self.install(self.lock()))
+
+    def test_payload_leaf_escape(self):
+        parent = os.path.join(self.target, "payload", "protean-one")
+        os.makedirs(parent)
+        os.symlink(self.canary, os.path.join(parent, "file.txt"))
+        self.refused(self.install(self.lock()))
+
+    def test_staging_root_escape(self):
+        os.symlink(self.outside, os.path.join(self.target, ".protean-staging"))
+        self.refused(self.install(self.lock()))
+
+    def test_unused_staging_cleanup_child_escape(self):
+        staging = os.path.join(self.target, ".protean-staging")
+        os.makedirs(staging)
+        os.symlink(self.outside, os.path.join(staging, "old-run"))
+        self.refused(self.install(self.lock()))
+
+    def test_dangling_contribution_state_escape(self):
+        os.makedirs(os.path.join(self.target, "records"))
+        dest = os.path.join(self.outside, "new-record")
+        os.symlink(dest, os.path.join(self.target, "records", "CONTRIB-STATE.md"))
+        self.refused(self.install(self.lock(extra_files={"records/CONTRIB-STATE.md.tmpl": "template"},
+                                             install_targets=["records", "gates/protean-one"])))
+        self.assertFalse(os.path.exists(dest))
+
+    def test_existing_contribution_state_escape(self):
+        os.makedirs(os.path.join(self.target, "records"))
+        os.symlink(self.canary, os.path.join(self.target, "records", "CONTRIB-STATE.md"))
+        self.refused(self.install(self.lock()))
+
+    def test_later_unsafe_ingredient_does_not_partially_install(self):
+        os.makedirs(os.path.join(self.target, "payload"))
+        os.symlink(self.outside, os.path.join(self.target, "payload", "protean-two"))
+        self.refused(self.install(self.lock(("protean-one", "protean-two"))))
+        self.assertFalse(os.path.exists(os.path.join(self.target, "payload", "protean-one")))
+
+    def test_explicit_root_link_is_allowed(self):
+        link = os.path.join(self.base, "chosen-root")
+        os.symlink(self.target, link)
+        result = self.install(self.lock(), link)
+        self.assertEqual(result.returncode, 0, result.stdout + result.stderr)
+        self.assertTrue(os.path.isfile(os.path.join(self.target, "payload", "protean-one", "file.txt")))
+
+    def test_confined_payload_parent_link_is_allowed(self):
+        internal = os.path.join(self.target, "internal")
+        os.makedirs(internal)
+        os.symlink(internal, os.path.join(self.target, "payload"))
+        result = self.install(self.lock())
+        self.assertEqual(result.returncode, 0, result.stdout + result.stderr)
+        self.assertTrue(os.path.isfile(os.path.join(internal, "protean-one", "file.txt")))
+
+    def test_normal_install(self):
+        result = self.install(self.lock())
+        self.assertEqual(result.returncode, 0, result.stdout + result.stderr)
+
+
+if __name__ == "__main__":
+    unittest.main()