fix: authenticate cached ingredient files before installation

build/install.py

@@ -466,6 +466,48 @@ def fetch_entry(entry, cache, offline):
         raise
 
 
+def verify_cache_checkout(entry, entry_dir):
+    """Compare the files we execute/copy with the pinned blobs, not the index.
+
+    A matching HEAD and object tree do not authenticate a mutable checkout.
+    Read the worktree directly so skip-worktree/assume-unchanged and ignored
+    files cannot hide an altered or extra payload from the installer.
+    """
+    tracked = set()
+    tree = git(["ls-tree", "-r", "-z", "--full-tree", entry["sha"]], cwd=entry_dir)
+    for item in tree.split("\0"):
+        if not item:
+            continue
+        metadata, rel = item.split("\t", 1)
+        mode, kind, blob = metadata.split()
+        tracked.add(rel)
+        path = os.path.join(entry_dir, rel)
+        if kind != "blob" or not os.path.lexists(path):
+            raise Failure(4, "%s: cache file missing or unsupported: %s" % (entry["slug"], rel))
+        # The installer copies regular bytes, not links. A pinned link pins
+        # only its text; dereferencing it would copy bytes outside this tree.
+        if mode == "120000":
+            raise Failure(4, "%s: cache contains an unsupported symlink: %s" % (entry["slug"], rel))
+        if os.path.islink(path) or not os.path.isfile(path):
+            raise Failure(4, "%s: cache file type changed: %s" % (entry["slug"], rel))
+        with open(path, "rb") as handle:
+            data = handle.read()
+        executable = bool(os.stat(path).st_mode & 0o111)
+        if executable != (mode == "100755"):
+            raise Failure(4, "%s: cache file mode changed: %s" % (entry["slug"], rel))
+        actual = hashlib.sha1(b"blob " + str(len(data)).encode("ascii") + b"\0" + data).hexdigest()
+        if actual != blob:
+            raise Failure(4, "%s: cache file differs from pinned blob: %s" % (entry["slug"], rel))
+    for root, directories, files in os.walk(entry_dir, followlinks=False):
+        if root == entry_dir:
+            directories[:] = [name for name in directories if name != ".git"]
+        links = [name for name in directories if os.path.islink(os.path.join(root, name))]
+        for name in files + links:
+            rel = os.path.relpath(os.path.join(root, name), entry_dir)
+            if rel not in tracked:
+                raise Failure(4, "%s: cache contains an unpinned file: %s" % (entry["slug"], rel))
+
+
 def verify_entry(entry, entry_dir):
     slug = entry["slug"]
     head = run(["git", "rev-parse", "HEAD"], cwd=entry_dir)
@@ -476,6 +518,7 @@ def verify_entry(entry, entry_dir):
     if actual != expected:
         raise Failure(4, "%s: treeSha256 mismatch\nexpected: %s\nactual:   %s"
                       % (slug, expected, actual))
+    verify_cache_checkout(entry, entry_dir)
     return True
 
 

tests/test_kit_lock.py

@@ -231,6 +231,58 @@ def test_offline_warm_cache_installs_and_reads_back(self):
         self.assertTrue(os.path.isfile(os.path.join(target, "payload", "protean-one", "file.txt")))
         self.assertFalse(os.path.isdir(os.path.join(target, ".protean-staging")))
 
+    def test_tampered_cache_is_refused_before_any_target_write(self):
+        for mutation in ("tracked", "untracked", "ignored", "hidden-tracked"):
+            with self.subTest(mutation=mutation), tempfile.TemporaryDirectory() as base:
+                os.makedirs(os.path.join(base, "repos"))
+                cache = os.path.join(base, "cache")
+                fixture = build_ingredient(os.path.join(base, "repos"), "protean-one")
+                cached = warm_cache(cache, "protean-one", fixture)
+                lock = write_lock(os.path.join(base, "kit.lock.json"),
+                                  [lock_entry("protean-one", fixture)])
+                payload = os.path.join(cached, "payload", "protean-one", "file.txt")
+                if mutation == "hidden-tracked":
+                    git(["update-index", "--skip-worktree", "payload/protean-one/file.txt"], cwd=cached)
+                if mutation in ("tracked", "hidden-tracked"):
+                    with open(payload, "w", encoding="utf-8") as handle:
+                        handle.write("TAMPERED CACHE PAYLOAD\n")
+                else:
+                    name = "injected.txt" if mutation == "untracked" else "injected.pyc"
+                    if mutation == "ignored":
+                        with open(os.path.join(cached, ".git", "info", "exclude"), "a") as handle:
+                            handle.write("*.pyc\n")
+                    with open(os.path.join(os.path.dirname(payload), name), "w") as handle:
+                        handle.write("UNPINNED CACHE PAYLOAD\n")
+                target = os.path.join(base, "installed")
+                result = run(INSTALL, "--all", "--target", target, "--offline", "--cache", cache,
+                             "--lock", lock)
+                self.assertEqual(result.returncode, 4, result.stdout + result.stderr)
+                self.assertIn("cache", result.stderr)
+                self.assertFalse(os.path.exists(target), "integrity failure must precede writes")
+
+    def test_pinned_symlink_cannot_copy_unpinned_external_bytes(self):
+        fixture = self.ingredient("protean-one")
+        outside = os.path.join(self.base, "outside-marker.txt")
+        with open(outside, "w") as handle:
+            handle.write("UNPINNED EXTERNAL MARKER\n")
+        os.symlink(outside, os.path.join(fixture["path"], "payload", "protean-one", "linked.txt"))
+        git(["add", "payload/protean-one/linked.txt"], cwd=fixture["path"])
+        git(IDENTITY + ["commit", "--quiet", "-m", "synthetic symlink"], cwd=fixture["path"])
+        fixture["sha"] = git(["rev-parse", "HEAD"], cwd=fixture["path"])
+        git(IDENTITY + ["tag", "-a", "v1.0.1", "-m", "synthetic symlink fixture"], cwd=fixture["path"])
+        warm_cache(self.cache, "protean-one", fixture)
+        entry = lock_entry("protean-one", fixture)
+        entry.update(ref="refs/tags/v1.0.1", version="1.0.1")
+        lock = write_lock(os.path.join(self.base, "kit.lock.json"), [entry])
+        target = os.path.join(self.base, "installed")
+        result = run(INSTALL, "--all", "--target", target, "--offline", "--cache", self.cache,
+                     "--lock", lock)
+        self.assertEqual(result.returncode, 4, result.stdout + result.stderr)
+        self.assertIn("symlink", result.stderr)
+        self.assertFalse(os.path.exists(target))
+        with open(outside) as handle:
+            self.assertEqual(handle.read(), "UNPINNED EXTERNAL MARKER\n")
+
     def test_single_ingredient_dry_run_writes_nothing(self):
         fixture, lock = self.fixture_lock()
         target = os.path.join(self.base, "dry-target")