security: harden completion gates and artifact writes

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,22 +153,67 @@ 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 {}
 
 
 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,14 +227,19 @@ 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))
         resolves = review.get("resolves", [])
         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()

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

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