Harden Piranha utility mutation campaigns

README.md

@@ -252,17 +252,30 @@ observation hooks proposed for the safe levels.
 python -m code_covenant.cli piranha path/to/project \
     --sandbox-root /tmp/piranha-sandbox \
     --i-understand-piranha-is-destructive \
-    --engine shell \
-    --engine-command "python optimizer.py --before {before} --after {after}" \
+    --engine utility \
+    --test-command "python -m pytest -q -p no:cacheprovider" \
     --rounds 5 \
     --duration 300 \
     --max-attempts-per-round 10
 ```
 
-Copies the source into an isolated workspace, then runs aggressive
-auto-merge optimisation rounds **inside the sandbox only**. The original
-source is byte-identical before and after. Writes `piranha_lab_report.md`
-with champion state and best / worst lineage.
+The default utility engine proposes conservative AST-derived rewrites (Boolean
+and guard-return collapse, `any()` reductions, and filtered comprehensions). Promotion is
+fail-closed: exact CPC bytes and public signatures must survive, capabilities
+cannot expand, immutable tests must pass in an OS-confined process, and every
+declared trusted metric must be present and strictly Pareto-improve over the
+current champion. The original source remains byte-identical.
+
+Piranha refuses existing unowned sandbox roots, source symlinks, unsafe CPC
+module paths, missing behavior tests, and unsupported confinement hosts. The
+built-in host backend is macOS `sandbox-exec`; use a dedicated VM/container on
+other hosts. Run the adversarial check-of-the-check with:
+
+```bash
+python -m code_covenant.tools.piranha_utility_verify
+```
+
+See `docs/PIRANHA_UTILITY_CAMPAIGNS.md` for model-engine and campaign guidance.
 
 ## Try the portfolio runner
 

code_covenant/gate/folder.py

@@ -22,6 +22,8 @@
 # ║   - Must create all parent directories                       ║
 # ║   - Must not overwrite an existing proposal folder           ║
 # ║   - Must write JSON with stable key ordering                 ║
+# ║   - Module and proposal id must be safe single path names    ║
+# ║   - Proposal paths must stay under a real proposals_dir      ║
 # ║                                                              ║
 # ║ invariants:                                                  ║
 # ║   - The returned path exists and contains the seven files    ║
@@ -31,6 +33,7 @@
 # ║                                                              ║
 # ║ forbidden_changes:                                           ║
 # ║   - Do not permit overwriting an existing proposal           ║
+# ║   - Do not follow proposal or artifact symlinks              ║
 # ║                                                              ║
 # ║ optimization_targets:                                        ║
 # ║   - [clarity] obvious, stable on-disk layout                 ║
@@ -47,6 +50,7 @@
 from __future__ import annotations
 
 import json
+import re
 from pathlib import Path
 from typing import Any
 
@@ -59,17 +63,38 @@
     "decision.json",
     "rationale.md",
 )
+_MODULE_RE = re.compile(r"[A-Za-z_][A-Za-z0-9_]*(?:\.[A-Za-z_][A-Za-z0-9_]*)*")
+_PROPOSAL_ID_RE = re.compile(r"[A-Za-z0-9][A-Za-z0-9._-]{0,127}")
 
 
-def build_proposal_folder(
-    proposals_dir: Path, module: str, proposal_id: str
-) -> Path:
+def build_proposal_folder(proposals_dir: Path, module: str, proposal_id: str) -> Path:
     """Create the per-proposal directory path and return it. Raises on conflict."""
-    safe_module = module or "unknown_module"
-    folder = proposals_dir / safe_module / proposal_id
-    if folder.exists():
+    _validate_module(module)
+    _validate_proposal_id(proposal_id)
+
+    raw_base = proposals_dir.expanduser().absolute()
+    if raw_base.is_symlink():
+        raise ValueError(f"proposals_dir must not be a symlink: {raw_base}")
+    raw_base.mkdir(parents=True, exist_ok=True)
+    if raw_base.is_symlink() or not raw_base.is_dir():
+        raise NotADirectoryError(f"proposals_dir is not a real directory: {raw_base}")
+    base = raw_base.resolve()
+
+    module_dir = base / module
+    _validate_contained(module_dir, base)
+    if module_dir.is_symlink():
+        raise ValueError(f"proposal module directory must not be a symlink: {module_dir}")
+    if module_dir.exists() and not module_dir.is_dir():
+        raise NotADirectoryError(f"proposal module path is not a directory: {module_dir}")
+    module_dir.mkdir(exist_ok=True)
+    _validate_real_contained_directory(module_dir, base)
+
+    folder = module_dir / proposal_id
+    _validate_contained(folder, base)
+    if folder.exists() or folder.is_symlink():
         raise FileExistsError(f"proposal folder already exists: {folder}")
-    folder.mkdir(parents=True, exist_ok=False)
+    folder.mkdir(exist_ok=False)
+    _validate_real_contained_directory(folder, base)
     return folder
 
 
@@ -85,23 +110,88 @@ def write_proposal_artifacts(
     rationale: str,
 ) -> None:
     """Write every standard proposal artifact into folder."""
-    (folder / "before.py").write_text(before_source, encoding="utf-8")
-    (folder / "after.py").write_text(after_source, encoding="utf-8")
-    (folder / "diff.patch").write_text(diff_text, encoding="utf-8")
-    (folder / "metrics.json").write_text(
-        json.dumps(metrics, indent=2, sort_keys=True) + "\n", encoding="utf-8"
+    raw_folder = folder.expanduser().absolute()
+    if raw_folder.is_symlink() or raw_folder.resolve() != raw_folder:
+        raise ValueError(f"proposal folder must not traverse symlinks: {raw_folder}")
+    if not raw_folder.is_dir():
+        raise NotADirectoryError(f"proposal folder is not a directory: {raw_folder}")
+    _write_new_text(raw_folder / "before.py", before_source)
+    _write_new_text(raw_folder / "after.py", after_source)
+    _write_new_text(raw_folder / "diff.patch", diff_text)
+    _write_new_text(
+        raw_folder / "metrics.json",
+        json.dumps(metrics, indent=2, sort_keys=True) + "\n",
     )
-    (folder / "constraint_grade.json").write_text(
+    _write_new_text(
+        raw_folder / "constraint_grade.json",
         json.dumps(constraint_grade, indent=2, sort_keys=True) + "\n",
-        encoding="utf-8",
     )
-    (folder / "decision.json").write_text(
-        json.dumps(decision, indent=2, sort_keys=True) + "\n", encoding="utf-8"
+    _write_new_text(
+        raw_folder / "decision.json",
+        json.dumps(decision, indent=2, sort_keys=True) + "\n",
     )
-    (folder / "rationale.md").write_text(rationale or "(no rationale supplied)\n",
-                                         encoding="utf-8")
+    _write_new_text(raw_folder / "rationale.md", rationale or "(no rationale supplied)\n")
 
 
 def required_artifact_files() -> tuple[str, ...]:
     """Return the tuple of filenames every proposal folder is expected to hold."""
     return _ARTIFACT_FILES
+
+
+def existing_proposal_folder(proposals_dir: Path, module: str, proposal_id: str) -> Path:
+    """Resolve an existing folder without permitting traversal or symlinks."""
+    _validate_module(module)
+    _validate_proposal_id(proposal_id)
+    raw_base = proposals_dir.expanduser().absolute()
+    if raw_base.is_symlink():
+        raise ValueError(f"proposals_dir is not a real directory: {raw_base}")
+    if not raw_base.exists():
+        raise FileNotFoundError(f"proposals_dir does not exist: {raw_base}")
+    if not raw_base.is_dir():
+        raise NotADirectoryError(f"proposals_dir is not a directory: {raw_base}")
+    base = raw_base.resolve()
+    folder = base / module / proposal_id
+    _validate_contained(folder, base)
+    _validate_real_contained_directory(folder, base)
+    return folder
+
+
+def _validate_module(module: str) -> None:
+    if not isinstance(module, str) or len(module) > 255 or not _MODULE_RE.fullmatch(module):
+        raise ValueError(f"invalid CPC module for proposal folder: {module!r}")
+
+
+def _validate_proposal_id(proposal_id: str) -> None:
+    if (
+        not isinstance(proposal_id, str)
+        or not _PROPOSAL_ID_RE.fullmatch(proposal_id)
+        or proposal_id in {".", ".."}
+    ):
+        raise ValueError(f"invalid proposal_id for proposal folder: {proposal_id!r}")
+
+
+def _validate_contained(path: Path, root: Path) -> None:
+    try:
+        path.resolve().relative_to(root.resolve())
+    except ValueError as exc:
+        raise ValueError(f"proposal path escapes proposals_dir: {path}") from exc
+
+
+def _validate_real_contained_directory(path: Path, root: Path) -> None:
+    if path.is_symlink():
+        raise ValueError(f"proposal path is not a real directory: {path}")
+    if not path.exists():
+        raise FileNotFoundError(f"proposal path does not exist: {path}")
+    if not path.is_dir():
+        raise NotADirectoryError(f"proposal path is not a directory: {path}")
+    resolved = path.resolve()
+    _validate_contained(resolved, root)
+    if resolved != path:
+        raise ValueError(f"proposal path traverses a symlink: {path}")
+
+
+def _write_new_text(path: Path, content: str) -> None:
+    # Exclusive creation prevents a pre-planted artifact symlink from being
+    # followed and also preserves the no-overwrite proposal invariant.
+    with path.open("x", encoding="utf-8") as stream:
+        stream.write(content)

code_covenant/gate/gates.py

@@ -51,6 +51,10 @@
 import time
 from dataclasses import dataclass, field
 from pathlib import Path
+from typing import TYPE_CHECKING
+
+if TYPE_CHECKING:
+    from code_covenant.piranha.execution import ConfinementPolicy
 
 
 @dataclass
@@ -90,9 +94,10 @@ def test_gate(
     command: str,
     cwd: Path | None = None,
     timeout_s: float = 120.0,
+    confinement: "ConfinementPolicy | None" = None,
 ) -> GateResult:
     """Run a test command under a timeout and return a GateResult."""
-    return _run_subprocess_gate("tests", command, cwd, timeout_s)
+    return _run_subprocess_gate("tests", command, cwd, timeout_s, confinement)
 
 
 test_gate.__test__ = False  # type: ignore[attr-defined]  # not a pytest test case despite the name
@@ -102,9 +107,10 @@ def benchmark_gate(
     command: str,
     cwd: Path | None = None,
     timeout_s: float = 120.0,
+    confinement: "ConfinementPolicy | None" = None,
 ) -> GateResult:
     """Run a benchmark command and return a GateResult (no comparison)."""
-    return _run_subprocess_gate("benchmark", command, cwd, timeout_s)
+    return _run_subprocess_gate("benchmark", command, cwd, timeout_s, confinement)
 
 
 def _decode_io(value: bytes | str | None) -> str:
@@ -120,6 +126,7 @@ def _run_subprocess_gate(
     command: str,
     cwd: Path | None,
     timeout_s: float,
+    confinement: "ConfinementPolicy | None" = None,
 ) -> GateResult:
     """Shared subprocess runner for test and benchmark gates."""
     start = time.perf_counter()
@@ -133,14 +140,26 @@ def _run_subprocess_gate(
             error=f"could not parse command: {exc}",
         )
     try:
-        completed = subprocess.run(  # noqa: S603 (caller-provided command, by design)
-            argv,
-            cwd=str(cwd) if cwd else None,
-            capture_output=True,
-            text=True,
-            timeout=timeout_s,
-            check=False,
-        )
+        if confinement is not None:
+            if cwd is None:
+                raise ValueError("confined gates require an explicit cwd")
+            from code_covenant.piranha.execution import run_confined
+
+            completed = run_confined(
+                argv,
+                cwd=cwd,
+                timeout_s=timeout_s,
+                policy=confinement,
+            )
+        else:
+            completed = subprocess.run(  # noqa: S603 (caller-provided command, by design)
+                argv,
+                cwd=str(cwd) if cwd else None,
+                capture_output=True,
+                text=True,
+                timeout=timeout_s,
+                check=False,
+            )
     except FileNotFoundError as exc:
         return GateResult(
             name=name,
@@ -157,6 +176,13 @@ def _run_subprocess_gate(
             stdout=_decode_io(exc.stdout),
             stderr=_decode_io(exc.stderr),
         )
+    except (OSError, RuntimeError, ValueError) as exc:
+        return GateResult(
+            name=name,
+            ok=False,
+            duration_s=time.perf_counter() - start,
+            error=f"confinement failure: {exc}",
+        )
     return GateResult(
         name=name,
         ok=completed.returncode == 0,

code_covenant/gate/grading.py

@@ -133,9 +133,11 @@ def decide(
     if not compile_result.ok:
         return Decision("reject", f"compile gate failed: {compile_result.error}")
     if test_result is not None and not test_result.ok:
-        return Decision("reject", "test gate failed")
-    if benchmark_result is not None and benchmark_result.error:
-        return Decision("reject", f"benchmark gate error: {benchmark_result.error}")
+        detail = test_result.error or test_result.stderr[-200:] or f"exit {test_result.exit_code}"
+        return Decision("reject", f"test gate failed: {detail}")
+    if benchmark_result is not None and not benchmark_result.ok:
+        detail = benchmark_result.error or f"exit {benchmark_result.exit_code}"
+        return Decision("reject", f"benchmark gate failed: {detail}")
     if after_validator_issues:
         return Decision(
             "reject",