fix: harden portable PR walkthrough runner (#2942)
This commit is contained in:
Carlos Florêncio
@@ -1,100 +0,0 @@
|
||||
#!/usr/bin/env python3
|
||||
"""Read one bounded UTF-8 regular file from an immutable Git commit."""
|
||||
|
||||
import argparse
|
||||
import os
|
||||
from pathlib import PurePosixPath
|
||||
import re
|
||||
import subprocess
|
||||
import sys
|
||||
|
||||
|
||||
MAX_BYTES = 512 * 1024
|
||||
SHA_RE = re.compile(r"^[0-9a-f]{40}$")
|
||||
REGULAR_MODES = {"100644", "100755"}
|
||||
|
||||
|
||||
class ReadError(Exception):
|
||||
pass
|
||||
|
||||
|
||||
def git(repo: str, *args: str, text: bool = False) -> subprocess.CompletedProcess:
|
||||
env = {"PATH": os.environ.get("PATH", "/usr/bin:/bin"), "LC_ALL": "C"}
|
||||
return subprocess.run(
|
||||
["git", *args],
|
||||
cwd=repo,
|
||||
env=env,
|
||||
capture_output=True,
|
||||
text=text,
|
||||
check=False,
|
||||
)
|
||||
|
||||
|
||||
def validate_path(value: str) -> str:
|
||||
path = PurePosixPath(value)
|
||||
parts = value.split("/")
|
||||
if (
|
||||
not value
|
||||
or value.startswith(("/", ":"))
|
||||
or "\\" in value
|
||||
or "\x00" in value
|
||||
or any(part in {"", ".", ".."} for part in parts)
|
||||
or path.is_absolute()
|
||||
):
|
||||
raise ReadError("invalid repository-relative path")
|
||||
return value
|
||||
|
||||
|
||||
def read_blob(repo: str, head_sha: str, path: str) -> str:
|
||||
if not SHA_RE.fullmatch(head_sha):
|
||||
raise ReadError("head SHA must be 40 lowercase hexadecimal characters")
|
||||
path = validate_path(path)
|
||||
|
||||
commit = git(repo, "cat-file", "-e", f"{head_sha}^{{commit}}")
|
||||
if commit.returncode != 0:
|
||||
raise ReadError("head SHA is not an available commit")
|
||||
|
||||
listing = git(repo, "--literal-pathspecs", "ls-tree", "-z", head_sha, "--", path)
|
||||
records = [record for record in listing.stdout.split(b"\0") if record]
|
||||
if listing.returncode != 0 or len(records) != 1:
|
||||
raise ReadError("path is not a regular file at the requested commit")
|
||||
metadata, separator, matched_path = records[0].partition(b"\t")
|
||||
fields = metadata.split()
|
||||
if not separator or len(fields) != 3 or matched_path != path.encode("utf-8"):
|
||||
raise ReadError("path is not a regular file at the requested commit")
|
||||
mode, object_type, object_id = (field.decode("ascii") for field in fields)
|
||||
if mode not in REGULAR_MODES or object_type != "blob":
|
||||
raise ReadError("path is not a regular file at the requested commit")
|
||||
|
||||
size_result = git(repo, "cat-file", "-s", object_id, text=True)
|
||||
if size_result.returncode != 0:
|
||||
raise ReadError("could not inspect file object")
|
||||
size = int(size_result.stdout.strip())
|
||||
if size > MAX_BYTES:
|
||||
raise ReadError(f"file exceeds {MAX_BYTES} bytes")
|
||||
|
||||
content = git(repo, "cat-file", "blob", object_id)
|
||||
if content.returncode != 0:
|
||||
raise ReadError("could not read file object")
|
||||
try:
|
||||
return content.stdout.decode("utf-8")
|
||||
except UnicodeDecodeError as exc:
|
||||
raise ReadError("file must be UTF-8 text") from exc
|
||||
|
||||
|
||||
def main() -> int:
|
||||
parser = argparse.ArgumentParser()
|
||||
parser.add_argument("--repo", required=True)
|
||||
parser.add_argument("--head-sha", required=True)
|
||||
parser.add_argument("--path", required=True)
|
||||
args = parser.parse_args()
|
||||
try:
|
||||
sys.stdout.write(read_blob(args.repo, args.head_sha, args.path))
|
||||
except (OSError, ReadError, ValueError) as exc:
|
||||
print(f"PR file read failed: {exc}", file=sys.stderr)
|
||||
return 1
|
||||
return 0
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
raise SystemExit(main())
|
||||
@@ -1,112 +0,0 @@
|
||||
#!/usr/bin/env python3
|
||||
"""Tests for immutable, regular-file reads from an untrusted PR commit."""
|
||||
|
||||
from pathlib import Path
|
||||
import os
|
||||
import subprocess
|
||||
import sys
|
||||
import tempfile
|
||||
import unittest
|
||||
|
||||
|
||||
ROOT = Path(__file__).resolve().parents[1]
|
||||
SCRIPT = ROOT / "scripts" / "pr-walkthrough-read-file"
|
||||
|
||||
|
||||
class PRWalkthroughReadFileTest(unittest.TestCase):
|
||||
def setUp(self) -> None:
|
||||
self.temp = tempfile.TemporaryDirectory()
|
||||
self.addCleanup(self.temp.cleanup)
|
||||
self.repo = Path(self.temp.name) / "repo"
|
||||
self.repo.mkdir()
|
||||
self.git("init")
|
||||
self.git("config", "user.name", "Walkthrough Test")
|
||||
self.git("config", "user.email", "walkthrough@example.invalid")
|
||||
(self.repo / "src").mkdir()
|
||||
(self.repo / "src" / "regular.txt").write_text("trusted commit bytes\n", encoding="utf-8")
|
||||
(self.repo / "executable.sh").write_text("#!/bin/sh\n", encoding="utf-8")
|
||||
os.chmod(self.repo / "executable.sh", 0o755)
|
||||
(self.repo / "binary.dat").write_bytes(b"\xff\xfe")
|
||||
(self.repo / "large.txt").write_text("x" * (512 * 1024 + 1), encoding="utf-8")
|
||||
os.symlink("/etc/passwd", self.repo / "outside-link")
|
||||
self.git("add", ".")
|
||||
self.git("commit", "-m", "test fixture")
|
||||
self.head = self.git("rev-parse", "HEAD").stdout.strip()
|
||||
|
||||
def git(self, *args: str) -> subprocess.CompletedProcess[str]:
|
||||
return subprocess.run(
|
||||
["git", *args],
|
||||
cwd=self.repo,
|
||||
text=True,
|
||||
capture_output=True,
|
||||
check=True,
|
||||
)
|
||||
|
||||
def read(self, path: str, *, head: str | None = None) -> subprocess.CompletedProcess[str]:
|
||||
return subprocess.run(
|
||||
[
|
||||
sys.executable,
|
||||
str(SCRIPT),
|
||||
"--repo",
|
||||
str(self.repo),
|
||||
"--head-sha",
|
||||
head or self.head,
|
||||
"--path",
|
||||
path,
|
||||
],
|
||||
text=True,
|
||||
capture_output=True,
|
||||
check=False,
|
||||
)
|
||||
|
||||
def test_reads_regular_file_from_exact_commit_not_worktree(self) -> None:
|
||||
(self.repo / "src" / "regular.txt").write_text("changed worktree bytes\n", encoding="utf-8")
|
||||
|
||||
result = self.read("src/regular.txt")
|
||||
|
||||
self.assertEqual(result.returncode, 0, result.stderr)
|
||||
self.assertEqual(result.stdout, "trusted commit bytes\n")
|
||||
|
||||
def test_reads_executable_regular_file(self) -> None:
|
||||
result = self.read("executable.sh")
|
||||
|
||||
self.assertEqual(result.returncode, 0, result.stderr)
|
||||
self.assertEqual(result.stdout, "#!/bin/sh\n")
|
||||
|
||||
def test_rejects_traversal_absolute_and_pathspec_paths(self) -> None:
|
||||
for path in ("../outside", "/etc/passwd", "src//regular.txt", ":(glob)**"):
|
||||
with self.subTest(path=path):
|
||||
result = self.read(path)
|
||||
self.assertNotEqual(result.returncode, 0)
|
||||
self.assertIn("invalid repository-relative path", result.stderr)
|
||||
|
||||
def test_rejects_symlink(self) -> None:
|
||||
result = self.read("outside-link")
|
||||
|
||||
self.assertNotEqual(result.returncode, 0)
|
||||
self.assertIn("regular file", result.stderr)
|
||||
|
||||
def test_rejects_binary_file(self) -> None:
|
||||
result = self.read("binary.dat")
|
||||
|
||||
self.assertNotEqual(result.returncode, 0)
|
||||
self.assertIn("UTF-8 text", result.stderr)
|
||||
|
||||
def test_rejects_oversized_file(self) -> None:
|
||||
result = self.read("large.txt")
|
||||
|
||||
self.assertNotEqual(result.returncode, 0)
|
||||
self.assertIn("exceeds 524288 bytes", result.stderr)
|
||||
|
||||
def test_rejects_missing_file_and_invalid_sha(self) -> None:
|
||||
missing = self.read("missing.txt")
|
||||
invalid_sha = self.read("src/regular.txt", head="main")
|
||||
|
||||
self.assertNotEqual(missing.returncode, 0)
|
||||
self.assertIn("not a regular file", missing.stderr)
|
||||
self.assertNotEqual(invalid_sha.returncode, 0)
|
||||
self.assertIn("40 lowercase hexadecimal", invalid_sha.stderr)
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
unittest.main()
|
||||
@@ -1,109 +0,0 @@
|
||||
#!/usr/bin/env python3
|
||||
"""Render agent-supplied walkthrough JSON with trusted pull-request identity."""
|
||||
|
||||
import json
|
||||
import os
|
||||
import re
|
||||
import subprocess
|
||||
import sys
|
||||
import tempfile
|
||||
from pathlib import Path
|
||||
|
||||
|
||||
TRUSTED_DIR = Path(".opencode-walkthrough")
|
||||
OUTPUT_DIR = Path("docs/pr-walkthrough")
|
||||
REPO_RE = re.compile(r"^[A-Za-z0-9_.-]+/[A-Za-z0-9_.-]+$")
|
||||
|
||||
|
||||
def required_env(name: str) -> str:
|
||||
value = os.environ.get(name, "").strip()
|
||||
if not value:
|
||||
raise ValueError(f"{name} is required")
|
||||
return value
|
||||
|
||||
|
||||
def trusted_identity() -> dict[str, object]:
|
||||
number_text = required_env("PR_NUMBER")
|
||||
if not number_text.isdigit() or int(number_text) < 1:
|
||||
raise ValueError("PR_NUMBER must be a positive integer")
|
||||
number = int(number_text)
|
||||
repo = required_env("PR_REPO")
|
||||
if not REPO_RE.fullmatch(repo):
|
||||
raise ValueError("PR_REPO must be an owner/repository slug")
|
||||
expected_url = f"https://github.com/{repo}/pull/{number}"
|
||||
url = required_env("PR_URL").rstrip("/")
|
||||
if url != expected_url:
|
||||
raise ValueError(f"PR_URL must be {expected_url}")
|
||||
return {
|
||||
"number": number,
|
||||
"title": required_env("PR_TITLE"),
|
||||
"url": url,
|
||||
"files_url": f"{url}/files",
|
||||
"repo": repo,
|
||||
"base": required_env("PR_BASE"),
|
||||
"head": required_env("PR_HEAD"),
|
||||
}
|
||||
|
||||
|
||||
def bind_trusted_metadata(data: object) -> dict[str, object]:
|
||||
if not isinstance(data, dict):
|
||||
raise ValueError("walkthrough JSON must be an object")
|
||||
pr = data.get("pr")
|
||||
if not isinstance(pr, dict):
|
||||
pr = {}
|
||||
data["pr"] = pr
|
||||
pr.update(trusted_identity())
|
||||
changes = data.get("changes")
|
||||
if isinstance(changes, list):
|
||||
for change in changes:
|
||||
if isinstance(change, dict):
|
||||
change.pop("file_url", None)
|
||||
return data
|
||||
|
||||
|
||||
def render(data: dict[str, object]) -> tuple[Path, Path]:
|
||||
build_script = TRUSTED_DIR / "build.py"
|
||||
shell = TRUSTED_DIR / "shell.html"
|
||||
if not build_script.is_file() or not shell.is_file():
|
||||
raise ValueError("trusted walkthrough renderer assets are missing")
|
||||
|
||||
number = int(data["pr"]["number"])
|
||||
OUTPUT_DIR.mkdir(parents=True, exist_ok=True)
|
||||
json_path = OUTPUT_DIR / f"pr-{number}.json"
|
||||
html_path = OUTPUT_DIR / f"pr-{number}.html"
|
||||
|
||||
with tempfile.TemporaryDirectory(prefix=".render-", dir=OUTPUT_DIR) as temp_name:
|
||||
temp_dir = Path(temp_name)
|
||||
temp_json = temp_dir / json_path.name
|
||||
temp_html = temp_dir / html_path.name
|
||||
temp_json.write_text(json.dumps(data, indent=2) + "\n", encoding="utf-8")
|
||||
result = subprocess.run(
|
||||
[sys.executable, str(build_script), str(temp_json), str(temp_html)],
|
||||
text=True,
|
||||
capture_output=True,
|
||||
check=False,
|
||||
)
|
||||
if result.returncode != 0:
|
||||
message = result.stderr.strip() or result.stdout.strip() or "renderer failed"
|
||||
raise ValueError(message)
|
||||
if not temp_html.is_file() or temp_html.stat().st_size == 0:
|
||||
raise ValueError("renderer did not produce non-empty HTML")
|
||||
temp_html.replace(html_path)
|
||||
temp_json.replace(json_path)
|
||||
return json_path, html_path
|
||||
|
||||
|
||||
def main() -> int:
|
||||
try:
|
||||
data = json.loads(sys.stdin.read())
|
||||
bound = bind_trusted_metadata(data)
|
||||
json_path, html_path = render(bound)
|
||||
except (json.JSONDecodeError, OSError, ValueError) as exc:
|
||||
print(f"walkthrough render failed: {exc}", file=sys.stderr)
|
||||
return 1
|
||||
print(f"rendered {json_path} and {html_path}")
|
||||
return 0
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
raise SystemExit(main())
|
||||
@@ -1,98 +0,0 @@
|
||||
#!/usr/bin/env python3
|
||||
import json
|
||||
import os
|
||||
import shutil
|
||||
import subprocess
|
||||
import sys
|
||||
import tempfile
|
||||
import unittest
|
||||
from pathlib import Path
|
||||
|
||||
|
||||
ROOT = Path(__file__).resolve().parents[1]
|
||||
SCRIPT = ROOT / "scripts" / "pr-walkthrough-render"
|
||||
REFERENCES = ROOT / ".agents" / "skills" / "pr-walkthrough" / "references"
|
||||
|
||||
|
||||
class PRWalkthroughRenderTest(unittest.TestCase):
|
||||
def setUp(self) -> None:
|
||||
self.tmp = tempfile.TemporaryDirectory()
|
||||
self.addCleanup(self.tmp.cleanup)
|
||||
self.worktree = Path(self.tmp.name)
|
||||
trusted = self.worktree / ".opencode-walkthrough"
|
||||
trusted.mkdir()
|
||||
shutil.copy2(REFERENCES / "build.py", trusted / "build.py")
|
||||
shutil.copy2(REFERENCES / "shell.html", trusted / "shell.html")
|
||||
self.data = json.loads((REFERENCES / "example.json").read_text(encoding="utf-8"))
|
||||
self.env = {
|
||||
**os.environ,
|
||||
"PR_NUMBER": "42",
|
||||
"PR_TITLE": "Trusted pull request title",
|
||||
"PR_URL": "https://github.com/kdlbs/kandev/pull/42",
|
||||
"PR_REPO": "kdlbs/kandev",
|
||||
"PR_BASE": "main",
|
||||
"PR_HEAD": "feature/walkthrough",
|
||||
}
|
||||
|
||||
def run_render(self, data: object) -> subprocess.CompletedProcess[str]:
|
||||
return subprocess.run(
|
||||
[sys.executable, str(SCRIPT)],
|
||||
cwd=self.worktree,
|
||||
env=self.env,
|
||||
input=json.dumps(data),
|
||||
text=True,
|
||||
capture_output=True,
|
||||
check=False,
|
||||
)
|
||||
|
||||
def test_binds_trusted_pr_identity_and_removes_model_link_overrides(self) -> None:
|
||||
self.data["pr"].update(
|
||||
{
|
||||
"number": 999,
|
||||
"title": "Untrusted title",
|
||||
"url": "javascript:alert(1)",
|
||||
"files_url": "https://attacker.invalid/files",
|
||||
"repo": "attacker/repo; curl attacker.invalid",
|
||||
"base": "wrong-base",
|
||||
"head": "wrong-head",
|
||||
}
|
||||
)
|
||||
self.data["changes"][0]["file_url"] = "javascript:alert(2)"
|
||||
|
||||
result = self.run_render(self.data)
|
||||
|
||||
self.assertEqual(result.returncode, 0, result.stderr)
|
||||
json_path = self.worktree / "docs" / "pr-walkthrough" / "pr-42.json"
|
||||
html_path = self.worktree / "docs" / "pr-walkthrough" / "pr-42.html"
|
||||
rendered_data = json.loads(json_path.read_text(encoding="utf-8"))
|
||||
self.assertEqual(
|
||||
rendered_data["pr"],
|
||||
{
|
||||
**self.data["pr"],
|
||||
"number": 42,
|
||||
"title": "Trusted pull request title",
|
||||
"url": "https://github.com/kdlbs/kandev/pull/42",
|
||||
"repo": "kdlbs/kandev",
|
||||
"base": "main",
|
||||
"head": "feature/walkthrough",
|
||||
}
|
||||
| {"files_url": "https://github.com/kdlbs/kandev/pull/42/files"},
|
||||
)
|
||||
self.assertNotIn("file_url", rendered_data["changes"][0])
|
||||
html = html_path.read_text(encoding="utf-8")
|
||||
self.assertIn("gh pr review 42 --repo kdlbs/kandev --approve", html)
|
||||
self.assertNotIn("attacker.invalid", html)
|
||||
self.assertNotIn("javascript:", html)
|
||||
|
||||
def test_renderer_failure_leaves_no_partial_outputs(self) -> None:
|
||||
result = self.run_render({"pr": {}})
|
||||
|
||||
self.assertNotEqual(result.returncode, 0)
|
||||
output = self.worktree / "docs" / "pr-walkthrough"
|
||||
self.assertFalse((output / "pr-42.json").exists())
|
||||
self.assertFalse((output / "pr-42.html").exists())
|
||||
self.assertIn("why.problem is required", result.stderr)
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
unittest.main()
|
||||
Reference in New Issue
Block a user