From f3376be9eb8122918eb19082a10666c79d473560 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E7=A8=8B=E8=AD=A6?= Date: Wed, 2 Sep 2026 22:54:40 +0800 Subject: [PATCH] security: harden completion gates and artifact writes --- harness/co_math/gating.py | 100 +++++++++++++++++++++++--- harness/co_math/messages.py | 21 +++++- harness/co_math/reports.py | 30 +++++++- tests/test_security_gates.py | 135 +++++++++++++++++++++++++++++++++++ 4 files changed, 270 insertions(+), 16 deletions(-) create mode 100644 tests/test_security_gates.py diff --git a/harness/co_math/gating.py b/harness/co_math/gating.py index 6133ce1..8200e62 100644 --- a/harness/co_math/gating.py +++ b/harness/co_math/gating.py @@ -1,6 +1,8 @@ from __future__ import annotations import json +import hashlib +import re from pathlib import Path from typing import Any @@ -82,9 +84,13 @@ def check_gate( def _selected_workstreams(root: Path, workstream_id: str | None) -> list[Path]: workstreams_dir = root / "workstreams" + if workstreams_dir.is_symlink(): + return [] if workstream_id: + if Path(workstream_id).name != workstream_id or workstream_id in {".", ".."}: + return [] path = workstreams_dir / workstream_id - return [path] if path.exists() else [] + return [path] if path.is_dir() and not path.is_symlink() else [] if not workstreams_dir.exists(): return [] return sorted( @@ -101,15 +107,23 @@ def _workstream_issues(workstream: Path) -> list[str]: report = workstream / "report.md" reviews = _load_reviews(workstream) - if not report.exists(): + if not status: + issues.append(f"{label}: status.yaml is missing or invalid.") + elif str(status.get("status", "")).lower() != "complete": + issues.append(f"{label}: status.yaml is not marked complete.") + + if not _regular_file(report): issues.append(f"{label}: report.md is missing.") + report_hash = _sha256(report) if _regular_file(report) else None + approved_reviewers = [ - review.get("reviewer", path.stem) + str(review.get("reviewer", "")).strip() for path, review in reviews - if review.get("approved") is True + if review.get("approved") is True and isinstance(review.get("reviewer"), str) ] - coordinator = status.get("coordinator", "workstream_coordinator") + coordinator_value = status.get("coordinator") + coordinator = coordinator_value.strip() if isinstance(coordinator_value, str) else "" independent_approvals = [ reviewer for reviewer in approved_reviewers if reviewer != coordinator ] @@ -118,12 +132,14 @@ def _workstream_issues(workstream: Path) -> list[str]: resolved_reviews = _resolved_review_names(reviews) for path, review in reviews: + if review.get("report_sha256") != report_hash: + issues.append(f"{label}: review in {path.name} is not bound to the current report.") if review.get("severity") == "blocking" and not _review_is_resolved( path, review, resolved_reviews ): issues.append(f"{label}: blocking review in {path.name}.") - if report.exists(): + if _regular_file(report): text = report.read_text(encoding="utf-8") if not _has_section(text, "Provenance"): issues.append(f"{label}: report.md is missing a Provenance section.") @@ -137,7 +153,7 @@ def _workstream_issues(workstream: Path) -> list[str]: def _load_status(workstream: Path) -> dict[str, Any]: path = workstream / "status.yaml" - if path.exists(): + if _regular_file(path): data = read_yaml(path) return data if isinstance(data, dict) else {} return {} @@ -145,14 +161,59 @@ def _load_status(workstream: Path) -> dict[str, Any]: def _load_reviews(workstream: Path) -> list[tuple[Path, dict[str, Any]]]: reviews_dir = workstream / "reviews" - if not reviews_dir.exists(): + if reviews_dir.is_symlink() or not reviews_dir.is_dir(): return [] reviews = [] for path in sorted(reviews_dir.glob("*.json")): + if not _regular_file(path): + reviews.append((path, {"_invalid": "Review must be a regular file."})) + continue try: data = json.loads(path.read_text(encoding="utf-8")) - except json.JSONDecodeError: - data = {"severity": "blocking", "comment": "Review JSON is invalid."} + if not isinstance(data, dict): + raise ValueError("review must be a JSON object") + required = { + "approved", + "severity", + "issue_type", + "reviewer", + "comment", + "suggested_fix", + "report_sha256", + } + optional = {"resolves", "resolved"} + if not required.issubset(data) or set(data) - required - optional: + raise ValueError("review has missing or unknown fields") + if not isinstance(data["approved"], bool) or data["severity"] not in {"blocking", "major", "minor", "none"}: + raise ValueError("review has invalid approval or severity") + if not all( + isinstance(data[field], str) and data[field].strip() + for field in ( + "issue_type", + "reviewer", + "comment", + "suggested_fix", + "report_sha256", + ) + ): + raise ValueError("review text fields must be non-empty strings") + report_sha256 = data["report_sha256"] + if not re.fullmatch(r"[0-9a-f]{64}", report_sha256): + raise ValueError("report_sha256 must be a SHA-256 hex digest") + if "resolved" in data and not isinstance(data["resolved"], bool): + raise ValueError("resolved must be a boolean") + if "resolves" in data: + resolves = data["resolves"] + if not isinstance(resolves, list) or not all( + isinstance(target, str) + and target.endswith(".json") + and Path(target).name == target + and target not in {".", ".."} + for target in resolves + ): + raise ValueError("resolves must contain review filenames in this directory") + except (json.JSONDecodeError, ValueError): + data = {"_invalid": "Review JSON is invalid or has an invalid schema.", "severity": "blocking"} reviews.append((path, data)) return reviews @@ -166,6 +227,11 @@ def _is_workstream_dir(path: Path) -> bool: def _resolved_review_names(reviews: list[tuple[Path, dict[str, Any]]]) -> set[str]: resolved: set[str] = set() + review_names = { + key + for path, _review in reviews + for key in (path.name, path.stem) + } for path, review in reviews: if review.get("resolved") is True: resolved.update((path.name, path.stem)) @@ -173,7 +239,7 @@ def _resolved_review_names(reviews: list[tuple[Path, dict[str, Any]]]) -> set[st if not isinstance(resolves, list): continue for target in resolves: - if not isinstance(target, str) or not target.strip(): + if not isinstance(target, str) or target not in review_names: continue target_path = Path(target.strip()) resolved.update((target.strip(), target_path.name, target_path.stem)) @@ -193,3 +259,15 @@ def _review_is_resolved( def _has_section(text: str, section: str) -> bool: expected = f"## {section}".lower() return any(line.strip().lower() == expected for line in text.splitlines()) + + +def _regular_file(path: Path) -> bool: + return not path.is_symlink() and path.is_file() + + +def _sha256(path: Path) -> str: + digest = hashlib.sha256() + with path.open("rb") as handle: + for chunk in iter(lambda: handle.read(1024 * 1024), b""): + digest.update(chunk) + return digest.hexdigest() diff --git a/harness/co_math/messages.py b/harness/co_math/messages.py index 38a6db9..b7e2a54 100644 --- a/harness/co_math/messages.py +++ b/harness/co_math/messages.py @@ -29,6 +29,8 @@ def append_message( "uncertainty": uncertainty or [], } path = Path(workspace) / "project" / "messages.jsonl" + if path.is_symlink() or (path.exists() and not path.is_file()): + raise ValueError(f"messages.jsonl is missing or unsafe: {path}") path.parent.mkdir(parents=True, exist_ok=True) with path.open("a", encoding="utf-8") as handle: handle.write(json.dumps(record, ensure_ascii=False) + "\n") @@ -37,10 +39,25 @@ def append_message( def read_messages(workspace: str | Path) -> list[MessageRecord]: path = Path(workspace) / "project" / "messages.jsonl" + if path.is_symlink() or (path.exists() and not path.is_file()): + raise ValueError(f"messages.jsonl is missing or unsafe: {path}") if not path.exists(): return [] records: list[MessageRecord] = [] - for line in path.read_text(encoding="utf-8").splitlines(): + for line_number, line in enumerate(path.read_text(encoding="utf-8").splitlines(), 1): if line.strip(): - records.append(json.loads(line)) + try: + record = json.loads(line) + except json.JSONDecodeError as exc: + raise ValueError(f"messages.jsonl line {line_number} is invalid JSON") from exc + if not isinstance(record, dict): + raise ValueError(f"messages.jsonl line {line_number} must be an object") + required = {"timestamp", "sender", "recipient", "type", "content", "provenance", "uncertainty"} + if set(record) != required or record["type"] not in VALID_MESSAGE_TYPES: + raise ValueError(f"messages.jsonl line {line_number} has an invalid schema") + if not all(isinstance(record[field], str) for field in ("timestamp", "sender", "recipient", "type", "content")): + raise ValueError(f"messages.jsonl line {line_number} has invalid text fields") + if not all(isinstance(record[field], list) and all(isinstance(item, str) for item in record[field]) for field in ("provenance", "uncertainty")): + raise ValueError(f"messages.jsonl line {line_number} has invalid evidence fields") + records.append(record) return records diff --git a/harness/co_math/reports.py b/harness/co_math/reports.py index 8df96b0..4a57a1f 100644 --- a/harness/co_math/reports.py +++ b/harness/co_math/reports.py @@ -1,5 +1,7 @@ from __future__ import annotations +import os +import tempfile from pathlib import Path from .gating import check_workstream_completion @@ -9,6 +11,8 @@ def render_final(workspace: str | Path) -> Path: root = Path(workspace) final_dir = root / "final" + if final_dir.is_symlink() or (final_dir.exists() and not final_dir.is_dir()): + raise ValueError(f"Final directory is missing or unsafe: {final_dir}") final_dir.mkdir(parents=True, exist_ok=True) sections: list[str] = [] @@ -28,13 +32,33 @@ def render_final(workspace: str | Path) -> Path: raise ValueError("No reviewed workstream reports are ready to render.") output = final_dir / "working_paper.md" - output.write_text( + content = ( "# Working Paper\n\n" f"- rendered_at: {utc_timestamp()}\n" "- status: draft_from_reviewed_workstreams\n" "- note: This is a working paper, not a chat summary.\n\n" + "\n\n---\n\n".join(sections) - + "\n", - encoding="utf-8", + + "\n" ) + _atomic_write_regular_file(output, content) return output + + +def _atomic_write_regular_file(path: Path, content: str) -> None: + """Write beside the destination, then replace it without following a symlink.""" + if path.is_symlink() or (path.exists() and not path.is_file()): + raise ValueError(f"Final report is not a regular file: {path}") + descriptor, temporary_name = tempfile.mkstemp( + dir=path.parent, prefix=f".{path.name}.", text=True + ) + temporary_path = Path(temporary_name) + try: + with os.fdopen(descriptor, "w", encoding="utf-8") as handle: + handle.write(content) + handle.flush() + os.fsync(handle.fileno()) + # os.replace replaces a symlink itself; it does not follow the link. + os.replace(temporary_path, path) + except Exception: + temporary_path.unlink(missing_ok=True) + raise diff --git a/tests/test_security_gates.py b/tests/test_security_gates.py new file mode 100644 index 0000000..9a66fbd --- /dev/null +++ b/tests/test_security_gates.py @@ -0,0 +1,135 @@ +from __future__ import annotations + +import hashlib +import json +import tempfile +import unittest +from pathlib import Path + +from harness.co_math.gating import check_workstream_completion +from harness.co_math.messages import append_message +from harness.co_math.reports import render_final +from harness.co_math.workspace import write_yaml + + +REPORT = """# Report + +## Provenance +Evidence. + +## Uncertainty +None. + +## Failed Explorations +None. +""" + + +class SecurityGateTests(unittest.TestCase): + def make_workstream(self, root: Path, *, status: str = "complete") -> Path: + workstream = root / "workstreams" / "WS-G1-001-test" + (workstream / "reviews").mkdir(parents=True) + write_yaml( + workstream / "status.yaml", + {"status": status, "coordinator": "author", "title": "Test"}, + ) + report = workstream / "report.md" + report.write_text(REPORT, encoding="utf-8") + digest = hashlib.sha256(report.read_bytes()).hexdigest() + review = { + "approved": True, + "severity": "none", + "issue_type": "logic", + "reviewer": "reviewer", + "comment": "Approved.", + "suggested_fix": "No change.", + "report_sha256": digest, + } + (workstream / "reviews" / "approval.json").write_text( + json.dumps(review), encoding="utf-8" + ) + return workstream + + def test_completion_requires_complete_status(self) -> None: + with tempfile.TemporaryDirectory() as directory: + root = Path(directory) + self.make_workstream(root, status="active") + gate = check_workstream_completion(root) + self.assertFalse(gate.passed) + self.assertTrue(any("not marked complete" in issue for issue in gate.issues)) + + def test_reviewer_must_differ_from_coordinator(self) -> None: + with tempfile.TemporaryDirectory() as directory: + root = Path(directory) + workstream = self.make_workstream(root) + review_path = workstream / "reviews" / "approval.json" + review = json.loads(review_path.read_text(encoding="utf-8")) + review["reviewer"] = "author" + review_path.write_text(json.dumps(review), encoding="utf-8") + gate = check_workstream_completion(root) + self.assertFalse(gate.passed) + self.assertTrue(any("independent reviewer" in issue for issue in gate.issues)) + + def test_later_review_can_resolve_named_blocking_review(self) -> None: + with tempfile.TemporaryDirectory() as directory: + root = Path(directory) + workstream = self.make_workstream(root) + digest = hashlib.sha256((workstream / "report.md").read_bytes()).hexdigest() + blocking = { + "approved": False, + "severity": "blocking", + "issue_type": "logic", + "reviewer": "reviewer-1", + "comment": "Fix this.", + "suggested_fix": "Add evidence.", + "report_sha256": digest, + } + resolution = { + "approved": True, + "severity": "none", + "issue_type": "logic", + "reviewer": "reviewer-2", + "comment": "Resolved.", + "suggested_fix": "No change.", + "report_sha256": digest, + "resolves": ["blocking.json"], + } + reviews = workstream / "reviews" + (reviews / "blocking.json").write_text(json.dumps(blocking), encoding="utf-8") + (reviews / "resolution.json").write_text(json.dumps(resolution), encoding="utf-8") + self.assertTrue(check_workstream_completion(root).passed) + + def test_messages_append_rejects_symlink(self) -> None: + with tempfile.TemporaryDirectory() as directory: + root = Path(directory) + project = root / "project" + project.mkdir() + victim = root / "victim" + victim.write_text("unchanged", encoding="utf-8") + (project / "messages.jsonl").symlink_to(victim) + with self.assertRaises(ValueError): + append_message( + root, + sender="a", + recipient="b", + message_type="status", + content="test", + ) + self.assertEqual(victim.read_text(encoding="utf-8"), "unchanged") + + def test_final_render_rejects_symlink_output(self) -> None: + with tempfile.TemporaryDirectory() as directory: + root = Path(directory) + self.make_workstream(root) + final = root / "final" + final.mkdir() + victim = root / "victim" + victim.write_text("unchanged", encoding="utf-8") + (final / "working_paper.md").symlink_to(victim) + with self.assertRaises(ValueError): + render_final(root) + self.assertEqual(victim.read_text(encoding="utf-8"), "unchanged") + + +if __name__ == "__main__": + unittest.main() \ No newline at end of file