Kevin Veen-BirkenbachandClaude Opus 5.5
1b31b10df6
pkgmgr review writes one self-contained HTML page for a repository: commits ahead of the base, the staged, unstaged and untracked diff per file, and the checks of the change set. Repository text reaches the packaged Jinja template only as autoescaped data, git runs in scan mode (now also without replace objects), and the page is written atomically through mkstemp and os.replace. 🎯 Fix: 92 % 💥 Break: 3 % 💸 Cost: new ⏱️ Time: new 🛡️ Security: 100 % 🏆 Score: 89 🚦 Gate: cost new, time new: cleared by the operator's commit word for C (2026-11-07) Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
230 lines
8.7 KiB
Python
230 lines
8.7 KiB
Python
from __future__ import annotations
|
|
|
|
import os
|
|
import stat
|
|
import subprocess
|
|
import tempfile
|
|
import unittest
|
|
from pathlib import Path
|
|
from unittest.mock import patch
|
|
|
|
from pkgmgr.actions.review import ReviewError, write_review
|
|
from tests.integration.test_review import GIT_ENV, git, write
|
|
|
|
SECRET = b"SECRET-KEY-MATERIAL\n"
|
|
SIGNATURE = (
|
|
"gpgsig -----BEGIN PGP SIGNATURE-----\n \n iQEzBAABCAAdFiEE\n"
|
|
" -----END PGP SIGNATURE-----"
|
|
)
|
|
MODULE = '[submodule "{0}"]\n\tpath = {0}\n\turl = ./{0}\n{1}'
|
|
|
|
|
|
class ReviewRepoCase(unittest.TestCase):
|
|
def setUp(self) -> None:
|
|
self._tmp = tempfile.TemporaryDirectory()
|
|
self.root = os.path.realpath(self._tmp.name)
|
|
self.repo = os.path.join(self.root, "repo")
|
|
self.marker = os.path.join(self.root, "marker")
|
|
self.secret = os.path.join(self.root, "secret", "id_ed25519")
|
|
write(self.secret, SECRET)
|
|
git(self.root, "init", "-b", "main", self.repo)
|
|
write(self._path("app.py"), b"print('base')\n")
|
|
write(self._path("lib.py"), b"print('lib')\n")
|
|
git(self.repo, "add", "-A")
|
|
git(self.repo, "commit", "-m", "init")
|
|
git(self.repo, "checkout", "-b", "feature")
|
|
write(self._path("app.py"), b"print('feature')\n")
|
|
git(self.repo, "commit", "-am", "feature")
|
|
write(self._path("app.py"), b"print('staged')\n")
|
|
git(self.repo, "add", "app.py")
|
|
write(self._path("app.py"), b"print('unstaged')\n")
|
|
write(self._path("new.txt"), b"untracked\n")
|
|
|
|
def tearDown(self) -> None:
|
|
self._tmp.cleanup()
|
|
|
|
def _path(self, *parts: str) -> str:
|
|
return os.path.join(self.repo, *parts)
|
|
|
|
def _render(self, base: str | None, notes: list[str]) -> str:
|
|
path = write_review(
|
|
cwd=self.repo,
|
|
out=os.path.join(self.root, "out", "page.html"),
|
|
max_file_lines=50,
|
|
base=base,
|
|
notes=notes,
|
|
)
|
|
return Path(path).read_text(encoding="utf-8")
|
|
|
|
def _trap(self, path: str, tail: str) -> str:
|
|
write(path, f"#!/bin/sh\ntouch '{self.marker}'\n{tail}\n".encode())
|
|
os.chmod(path, 0o755)
|
|
return path
|
|
|
|
def _assert_fires(self, *args: str) -> None:
|
|
subprocess.run(
|
|
["git", *args],
|
|
cwd=self.repo,
|
|
env=GIT_ENV,
|
|
stdin=subprocess.DEVNULL,
|
|
capture_output=True,
|
|
check=False,
|
|
)
|
|
self.assertTrue(os.path.exists(self.marker), f"plain git {args[0]} fires")
|
|
os.remove(self.marker)
|
|
|
|
def _submodule(self, parent: str, name: str, extra: str = "") -> str:
|
|
path = os.path.join(parent, name)
|
|
git(parent, "init", "-b", "main", path)
|
|
write(os.path.join(path, "f.txt"), b"one\n")
|
|
git(path, "add", "-A")
|
|
git(path, "commit", "-m", "one")
|
|
head = git(path, "rev-parse", "HEAD")
|
|
write(os.path.join(parent, ".gitmodules"), MODULE.format(name, extra).encode())
|
|
git(parent, "update-index", "--add", "--cacheinfo", f"160000,{head},{name}")
|
|
git(parent, "add", ".gitmodules")
|
|
git(parent, "commit", "-m", name)
|
|
return path
|
|
|
|
def _object(self, kind: str, content: str) -> str:
|
|
return subprocess.run(
|
|
["git", "hash-object", "-t", kind, "-w", "--stdin"],
|
|
cwd=self.repo,
|
|
env=GIT_ENV,
|
|
input=content,
|
|
capture_output=True,
|
|
text=True,
|
|
check=True,
|
|
).stdout.strip()
|
|
|
|
|
|
class TestReviewHardening(ReviewRepoCase):
|
|
def test_fsmonitor_never_runs(self) -> None:
|
|
monitor = self._trap(os.path.join(self.root, "fsmonitor"), "exit 1")
|
|
git(self.repo, "config", "core.fsmonitor", monitor)
|
|
self._assert_fires("diff")
|
|
|
|
self._render("main", [])
|
|
|
|
self.assertFalse(os.path.exists(self.marker))
|
|
|
|
def test_repository_filter_never_runs(self) -> None:
|
|
write(self._path(".git", "info", "attributes"), b"* filter=trap\n")
|
|
clean = self._trap(os.path.join(self.root, "clean"), "exec cat")
|
|
git(self.repo, "config", "filter.trap.clean", clean)
|
|
self._assert_fires("diff")
|
|
|
|
html = self._render("main", [])
|
|
|
|
self.assertFalse(os.path.exists(self.marker))
|
|
self.assertIn('id="unstaged" data-files="0"', html)
|
|
self.assertIn("filter.trap.clean", html)
|
|
self.assertNotIn("print('unstaged')", html)
|
|
self.assertIn("<dd>(not computed, see the warnings)</dd>", html)
|
|
|
|
def test_submodule_filter_never_runs(self) -> None:
|
|
sub = self._submodule(self.repo, "sub")
|
|
inner = self._submodule(sub, "inner")
|
|
clean = self._trap(os.path.join(self.root, "clean"), "exec cat")
|
|
for path in (sub, inner):
|
|
write(os.path.join(path, ".git", "info", "attributes"), b"* filter=x\n")
|
|
write(os.path.join(path, "f.txt"), b"two\n")
|
|
git(path, "config", "filter.x.clean", clean)
|
|
self._assert_fires("diff")
|
|
|
|
html = self._render(None, [])
|
|
|
|
self.assertFalse(os.path.exists(self.marker))
|
|
self.assertIn('id="unstaged" data-files="0"', html)
|
|
self.assertIn("the config of submodule sub defines filter.x.clean", html)
|
|
self.assertIn("the config of submodule sub/inner defines filter.x.clean", html)
|
|
self.assertIn("<dd>(not computed, see the warnings)</dd>", html)
|
|
|
|
def test_signature_program_never_runs(self) -> None:
|
|
header, message = git(self.repo, "cat-file", "commit", "HEAD").split("\n\n", 1)
|
|
signed = self._object("commit", f"{header}\n{SIGNATURE}\n\n{message}\n")
|
|
git(self.repo, "update-ref", "HEAD", signed)
|
|
program = self._trap(os.path.join(self.root, "gpg"), "exit 1")
|
|
git(self.repo, "config", "gpg.program", program)
|
|
self._assert_fires("log", "-1", "--format=%G?")
|
|
|
|
html = self._render("main", [])
|
|
|
|
self.assertFalse(os.path.exists(self.marker))
|
|
self.assertIn('id="commits" data-commits="1"', html)
|
|
self.assertNotIn('<span class="sig', html)
|
|
|
|
def test_hooks_never_run(self) -> None:
|
|
hooks = os.path.join(self.root, "hooks")
|
|
self._trap(os.path.join(hooks, "post-index-change"), "exit 0")
|
|
git(self.repo, "config", "core.hooksPath", hooks)
|
|
self._assert_fires("update-index", "--force-write-index")
|
|
|
|
self._render("main", [])
|
|
|
|
self.assertFalse(os.path.exists(self.marker))
|
|
|
|
def test_note_symlinks_are_refused(self) -> None:
|
|
link = os.path.join(self.root, "score.txt")
|
|
os.symlink(self.secret, link)
|
|
parent = os.path.join(self.root, "notes")
|
|
os.symlink(os.path.dirname(self.secret), parent)
|
|
|
|
for note in (link, os.path.join(parent, "id_ed25519")):
|
|
with (
|
|
self.subTest(note=note),
|
|
self.assertRaisesRegex(ReviewError, "cannot read note"),
|
|
):
|
|
self._render(None, [note])
|
|
|
|
def test_untracked_parent_swap_is_not_followed(self) -> None:
|
|
write(self._path("sub", "id_ed25519"), b"harmless\n")
|
|
real_run = subprocess.run
|
|
swapped: list[list[str]] = []
|
|
|
|
def swap(cmd, *args, **kwargs):
|
|
result = real_run(cmd, *args, **kwargs)
|
|
if "--others" in cmd and not swapped:
|
|
os.rename(self._path("sub"), os.path.join(self.root, "sub-real"))
|
|
os.symlink(os.path.dirname(self.secret), self._path("sub"))
|
|
swapped.append(cmd)
|
|
return result
|
|
|
|
with patch("subprocess.run", swap):
|
|
html = self._render(None, [])
|
|
|
|
self.assertTrue(swapped)
|
|
self.assertNotIn(SECRET.decode().strip(), html)
|
|
self.assertIn("Not a regular file, contents not shown.", html)
|
|
|
|
def test_out_symlink_is_replaced_not_followed(self) -> None:
|
|
keys = b"ssh-ed25519 AAAA operator\n"
|
|
victim = os.path.join(self.root, "authorized_keys")
|
|
write(victim, keys)
|
|
missing = os.path.join(self.root, "missing_keys")
|
|
out = os.path.join(self.root, "out")
|
|
os.makedirs(out)
|
|
os.symlink(victim, os.path.join(out, "page.html"))
|
|
os.symlink(missing, os.path.join(out, "dangling.html"))
|
|
|
|
for name in ("page.html", "dangling.html"):
|
|
with self.subTest(out=name):
|
|
path = write_review(
|
|
cwd=self.repo,
|
|
out=os.path.join(out, name),
|
|
max_file_lines=50,
|
|
base=None,
|
|
notes=[],
|
|
)
|
|
info = os.lstat(path)
|
|
self.assertTrue(stat.S_ISREG(info.st_mode))
|
|
self.assertEqual(stat.S_IMODE(info.st_mode), 0o600)
|
|
|
|
self.assertEqual(Path(victim).read_bytes(), keys)
|
|
self.assertFalse(os.path.lexists(missing))
|
|
self.assertEqual(sorted(os.listdir(out)), ["dangling.html", "page.html"])
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|