diff --git a/.claude/notes/lint-rules.md b/.claude/notes/lint-rules.md index 02491bb4d..b7ab4f06c 100644 --- a/.claude/notes/lint-rules.md +++ b/.claude/notes/lint-rules.md @@ -667,7 +667,7 @@ shipped. An id is a permanent documentation anchor: a suppression comment carryi an older branch, review or commit message must never start meaning something new. CE023 is retired the same way, after the rule was deleted with the package it guarded. -Claim 068 next, and note 065 IS TAKEN without being in `ALL_RULES`: doc-surface and +Claim 069 next, and note 065 IS TAKEN without being in `ALL_RULES`: doc-surface and whole-tree rules are `@pytest.mark.lint` classes in `tests/test_custom_lint.py` rather than `BaseRule`s, so `runner.py`'s uniqueness assert cannot see them. Enumerating them in a comment is how that note fell behind CE044, so grep instead: diff --git a/.claude/notes/permissions.md b/.claude/notes/permissions.md index fc1d64c3e..93cfd2b8c 100644 --- a/.claude/notes/permissions.md +++ b/.claude/notes/permissions.md @@ -29,10 +29,6 @@ read-write throwaway copy; when it was bind-mounted `:ro` the chmod returned ERO no tmpfs mask and has not been one since: see [isolation.md](isolation.md) § Why the framework mounts are writable copies. -What the window does NOT hide is the task DEFINITION. `task.yaml` is also staged at -`/work/input` for the in-container orchestrator and that mount is untouched, so hiding the -criteria from the agent remains a separate, unsolved problem. - Criteria address reference files with the `$REFERENCE_DIR` token (same resolver as `$TASK_DIR`) and the `REFERENCE_DIR` env var for `run_command`; `reference_comparison` names one file via `reference_file`. @@ -64,10 +60,6 @@ criteria and judges run. The task directory is shielded by the same window: unde it is mounted as a throwaway COPY at a fixed container path, which is what makes it chmod-able without touching the user's checked-out `tasks/` tree. -It shields grading MATERIAL that happens to live in the task directory (a `reference/` -subdirectory, fixtures), not the task DEFINITION: `task.yaml` is separately staged at -`/work/input`, which the agent can still read. - ### Why a stack and not a refcount Windows nest with *different* modes, so what an exit has to restore is the mode of the diff --git a/docs/DOCKER_ISOLATION.md b/docs/DOCKER_ISOLATION.md index f898e3b56..83a5cb1e8 100644 --- a/docs/DOCKER_ISOLATION.md +++ b/docs/DOCKER_ISOLATION.md @@ -339,6 +339,45 @@ If the task declares a `reference:` block, a throwaway **copy** of its directory See [Reference Solutions](TASK_DEFINITION_GUIDE.md#reference-solutions). +### Two more passive-read anti-cheat blocks + +These are `driver: docker` only. `driver: tempdir` shares the host uid and has no +filesystem isolation, so neither applies there (nor can — there is nothing to +mask). Both are defense-in-depth passive-read blocks, consistent with the +reference window's posture above; neither contains an adversarial agent. + +- **The staged grading inputs are deleted after load.** The host stages the + post-override `TaskDefinition` (with `success_criteria`) at `/work/input/task.yaml` + **and** a `context.json` whose `source_yaml` is the raw task text — criteria + verbatim, both at the top level and inside every `config_lineage` entry — for the + in-container orchestrator to load once at startup. The agent runs in the same + container, so leaving *either* readable would hand it the grading answer key + (deleting only `task.yaml` leaves the identical criteria one file over in + `context.json`). The in-container entry point deletes **both** immediately after + they are consumed — `context.json` is parsed into memory in the command body and + `task.yaml` by `load_task`, both before the delete (gated on + `CODER_EVAL_IN_CONTAINER`). They are read exactly once — grading reads criteria + from the in-memory task, never from disk. The `/work/input` mount is therefore + read-write (a `:ro` mount rejects `rm` with EROFS). `prior.json` is kept: it is + read later on the regrade path, and a regrade runs no agent so it is not a leak. + +- **Auto-mounted plugin trees are default-deny masked.** An `agent.plugins[].path` + (or a `TemplateDirSource.path` that is itself a plugin root) is auto-mounted at + its host path `:ro` so the plugin loads. Eval material colocated under that tree + as siblings of the skills dir — sibling task YAMLs, reference solutions, test + fixtures — would otherwise be readable. So the runner keeps the whole root + mounted but layers an empty `--tmpfs` over every child dir OUTSIDE the keep-set + (`.claude-plugin` + the manifest-declared skill dirs). Everything that is not + the plugin surface is masked by default, so an unknown or new eval layout can + never leak; `tests/`, `node_modules/`, and reference solutions are masked for + free. A root agent cannot `umount` a tmpfs (`CAP_SYS_ADMIN` is not in Docker's + default set), so this mask is *stronger* than the mode-000 reference window. Two + residuals the mask cannot cover — an eval def or reference COLOCATED inside a + skill dir (masking it would hide the skill), and a `task_id:` YAML **file** loose + at the plugin root (a tmpfs masks a directory, not a single file) — are caught by + lint rule CE068 (keep eval material out of skill dirs and off the plugin root; + put it under a sibling `tests/`). + Inside the container, the entrypoint invokes `coder-eval _run-task-internal` (hidden subcommand), which loads the staged YAML + context, runs the standard in-process Orchestrator (driver auto-coerced back to `tempdir`), and writes `task.json` to the output mount. Host reads it and feeds the existing aggregation pipeline. A `result_kind` discriminator on `CriterionResult` ensures `ClassificationCriterionResult` subclasses survive the JSON round-trip — without it, host-side aggregation would silently lose `observed_label`/`expected_label`. diff --git a/pyproject.toml b/pyproject.toml index 6a435bac7..4548c18cf 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -305,6 +305,7 @@ external = [ "CE064", "CE065", "CE066", + "CE068", ] # custom architectural lint rules (tests/lint/) [tool.ruff.lint.pylint] diff --git a/src/coder_eval/agents/_skills.py b/src/coder_eval/agents/_skills.py index 7ca52d6f2..9967c8cb4 100644 --- a/src/coder_eval/agents/_skills.py +++ b/src/coder_eval/agents/_skills.py @@ -24,7 +24,7 @@ _SKILL_FILE = "SKILL.md" -def _manifest_skill_dirs(root: Path) -> list[Path]: +def manifest_skill_dirs(root: Path) -> list[Path]: """Skill directories a Claude-plugin root declares, in manifest order. Reads the ``skills`` field of ``/.claude-plugin/plugin.json`` (a string @@ -81,7 +81,7 @@ def _plugin_skill_dirs( hint, ) continue - candidates = [directory for directory in _manifest_skill_dirs(root) if directory.is_dir()] + candidates = [directory for directory in manifest_skill_dirs(root) if directory.is_dir()] # A path that is ALREADY a bare skills directory has no `skills/` subdir, # so use it as-is. Deliberately NOT a fallback for a root that HAS one. # Rationale: .claude/notes/agents.md § Skills, per harness diff --git a/src/coder_eval/cli/run_task_internal_command.py b/src/coder_eval/cli/run_task_internal_command.py index 7f0fe1b3d..56c60a515 100644 --- a/src/coder_eval/cli/run_task_internal_command.py +++ b/src/coder_eval/cli/run_task_internal_command.py @@ -17,6 +17,7 @@ import asyncio import contextlib import logging +import os from pathlib import Path from typing import Any @@ -45,6 +46,17 @@ logger = logging.getLogger(__name__) +def _scrub_staged_inputs(task_yaml: Path, context_json: Path) -> None: + """Delete the staged ``task.yaml`` and ``context.json`` once loaded, inside a container only. + + Both carry the task's ``success_criteria``, and the agent runs in this same container. + ``prior.json`` is left for the regrade path, which reads it later and runs no agent. + """ + if os.environ.get(IN_CONTAINER_ENV) == "1": + task_yaml.unlink(missing_ok=True) + context_json.unlink(missing_ok=True) + + def heartbeat_is_alive(current: str, last_counter: str, current_mtime: float, last_mtime: float) -> bool: """True when the heartbeat shows a fresh signal of life. @@ -176,6 +188,7 @@ def run_task_internal_command( # `task_file` is then pointed under the task_dir mount, so the `TASK_DIR` the # Orchestrator exposes to `run_command` criteria resolves there, not /work/input. task, _ = load_task(task_yaml) + _scrub_staged_inputs(task_yaml, context_json) # The path below is never re-read; it only seeds Orchestrator's TASK_DIR. runtime_task_file = task_dir / "task.yaml" if task_dir.is_dir() else task_yaml diff --git a/src/coder_eval/isolation/docker_runner.py b/src/coder_eval/isolation/docker_runner.py index 581347688..e3217dad0 100644 --- a/src/coder_eval/isolation/docker_runner.py +++ b/src/coder_eval/isolation/docker_runner.py @@ -140,6 +140,13 @@ def _rewrite_loopback_for_container(url: str) -> str | None: # Rationale: .claude/notes/isolation.md § The stdout line limit STDOUT_LINE_LIMIT_BYTES = 64 * 1024 * 1024 # 64 MiB +_MASK_WARNING = "Masking non-skill path %s under plugin root %s (anti-cheat: only skills stay readable)." +_MASK_STANDDOWN_WARNING = ( + "Anti-cheat mask stood down for plugin root %s: its whole tree is the declared skill surface " + '(e.g. manifest `skills: "."`), so nothing is masked. Any eval material colocated here is READABLE ' + "to the agent — move it outside the plugin root." +) + async def _heartbeat_loop(heartbeat_path: Path) -> None: """Write a monotonic counter to ``heartbeat_path`` every interval until cancelled. @@ -656,8 +663,9 @@ async def run(self) -> EvaluationResult: await asyncio.to_thread(self._prepare_task_dir_mount, staging) # AFTER staging, BEFORE the container starts: the DAC caps are dropped, so # every framework-owned mount must be reachable through its `other` bits. + # Writable so the entry point can delete the staged task.yaml/context.json. # Rationale: .claude/notes/isolation.md § grant_container_access - await asyncio.to_thread(grant_container_access, input_dir, writable=False) + await asyncio.to_thread(grant_container_access, input_dir, writable=True) await asyncio.to_thread(grant_container_access, output_dir, writable=True) if self.grade_workspace is not None: # The one mount whose files the harness did NOT create, so the owner @@ -1218,6 +1226,84 @@ def _reference_mount_args(self) -> list[str]: # every turn, which a `:ro` mount rejects with EROFS. return ["-v", f"{self._reference_mount_src}:{CONTAINER_REFERENCE_DIR}"] + def _resolve_mount_path(self, raw_path: str) -> Path: + """Resolve an auto-mount source to an absolute host path. + + A relative path resolves against the task-file dir, as CE068 resolves it, not the CWD. + """ + expanded = Path(os.path.expandvars(os.path.expanduser(raw_path))) + if not expanded.is_absolute() and self.rt.task_file is not None: + expanded = self.rt.task_file.parent / expanded + return expanded.resolve() + + def _append_auto_mounts(self, argv: list[str]) -> None: + """Bind-mount the plugin, template and system-prompt paths a task references, ``:ro`` at their host path. + + A plugin root also gets every non-skill child dir masked with an empty tmpfs. The + reference is deliberately NOT here: it has its own mount (``_reference_mount_args``). + """ + mounted: set[Path] = set() + # Warned, not refused: `plugin.path` / `reference.directory` / + # `template_sources` are user-controlled strings, and legitimate uses exist. + # Rationale: .claude/notes/isolation.md § Extra mounts and reserved destinations + sensitive_sources = self._sensitive_source_paths() + + # Lazy: eval_material imports agents._skills, whose package imports this module. + from coder_eval.isolation.eval_material import mask_dirs + + # Masked dir -> its plugin root. Emitted only once every bind is known, so a + # nested plugin root's bind can win over its parent's mask of the same path. + mask_targets: dict[Path, Path] = {} + + def _auto_mount(raw_path: str | None, *, dir_only: bool = True) -> None: + if not raw_path: + return + resolved = self._resolve_mount_path(raw_path) + # File paths get mounted as the parent dir so a single -v covers + # the file; container-side reads still resolve at the same path. + target = resolved if (dir_only or resolved.is_dir()) else resolved.parent + if target in mounted or not target.is_dir(): + return + for sensitive in sensitive_sources: + if target == sensitive or sensitive in target.parents: + logger.warning( + "Auto-mounting sensitive host path %s into container; fix task YAML if unintended.", + target, + ) + break + mounted.add(target) + argv.extend(["-v", f"{target}:{target}:ro"]) + masks = mask_dirs(target) + if not masks and (target / ".claude-plugin" / "plugin.json").is_file(): + logger.warning(_MASK_STANDDOWN_WARNING, target) + for masked_dir in masks: + mask_targets.setdefault(masked_dir, target) + + plugins = (self.rt.task.agent.plugins if self.rt.task.agent else None) or [] + for plugin in plugins: + _auto_mount(plugin.get("path") if isinstance(plugin, dict) else None) + + from coder_eval.models import TemplateDirSource + + sandbox_cfg = self.rt.task.sandbox + for source in (sandbox_cfg.template_sources or []) if sandbox_cfg else []: + if isinstance(source, TemplateDirSource): + _auto_mount(source.path) + + # Defensive: normally inlined into system_prompt by load_task / experiment + # resolution, but a variant could inject an absolute path that survives. + agent_cfg = self.rt.task.agent + if agent_cfg and agent_cfg.system_prompt_file: + _auto_mount(agent_cfg.system_prompt_file, dir_only=False) + + # A deeper --tmpfs wins over the enclosing :ro bind regardless of argv order; + # a mask that is also a bind would be a duplicate mount point, so the bind wins. + for masked_dir, root in sorted(mask_targets.items()): + if masked_dir in mounted: + continue + argv.extend(["--tmpfs", str(masked_dir)]) + logger.warning(_MASK_WARNING, masked_dir, root) + def _build_argv( self, input_dir: Path, output_dir: Path, *, container_name: str, image: str | None = None ) -> list[str]: @@ -1300,7 +1386,9 @@ def _build_argv( # Rationale: .claude/notes/isolation.md § Environment forwarding argv += ["--env", "TELEMETRY_ENABLED=false"] - argv += ["-v", f"{input_dir.resolve()}:{CONTAINER_INPUT_DIR}:ro"] + # Read-WRITE: the entry point deletes the staged task.yaml and context.json + # after load, and `rm` fails with EROFS on a `:ro` bind mount. + argv += ["-v", f"{input_dir.resolve()}:{CONTAINER_INPUT_DIR}"] # The host run_dir at the container's standard output location, so the # in-container Orchestrator writes straight to the host filesystem. argv += ["-v", f"{output_dir}:{CONTAINER_OUTPUT_DIR}"] @@ -1330,51 +1418,7 @@ def _build_argv( host_claude_dir = Path.home() / ".claude" argv += ["-v", f"{self._claude_mount_src}:{host_claude_dir}"] - # Host paths the task references (plugin dirs, resolved template dirs), at - # the SAME path inside the container. The reference is deliberately NOT - # here -- it has its own mount and is masked out of the task_dir mount. - # ``mounted`` dedupes overlapping entries. - mounted: set[Path] = set() - # Warned, not refused: `plugin.path` / `reference.directory` / - # `template_sources` are user-controlled strings, and legitimate uses exist. - # Rationale: .claude/notes/isolation.md § Extra mounts and reserved destinations - sensitive_sources = self._sensitive_source_paths() - - def _auto_mount(raw_path: str | None, *, dir_only: bool = True) -> None: - if not raw_path: - return - resolved = Path(os.path.expandvars(os.path.expanduser(raw_path))).resolve() - # File paths get mounted as the parent dir so a single -v covers - # the file; container-side reads still resolve at the same path. - target = resolved if (dir_only or resolved.is_dir()) else resolved.parent - if target in mounted or not target.is_dir(): - return - for sensitive in sensitive_sources: - if target == sensitive or sensitive in target.parents: - logger.warning( - "Auto-mounting sensitive host path %s into container; fix task YAML if unintended.", - target, - ) - break - mounted.add(target) - argv.extend(["-v", f"{target}:{target}:ro"]) - - plugins = (self.rt.task.agent.plugins if self.rt.task.agent else None) or [] - for plugin in plugins: - _auto_mount(plugin.get("path") if isinstance(plugin, dict) else None) - - from coder_eval.models import TemplateDirSource - - sandbox_cfg = self.rt.task.sandbox - for source in (sandbox_cfg.template_sources or []) if sandbox_cfg else []: - if isinstance(source, TemplateDirSource): - _auto_mount(source.path) - - # Defensive: normally inlined into system_prompt by load_task / experiment - # resolution, but a variant could inject an absolute path that survives. - agent_cfg = self.rt.task.agent - if agent_cfg and agent_cfg.system_prompt_file: - _auto_mount(agent_cfg.system_prompt_file, dir_only=False) + self._append_auto_mounts(argv) # HAZARD: task.reference.directory is deliberately NOT auto-mounted at its # host path. That would bind the REAL tree in beside the shielded copy, so diff --git a/src/coder_eval/isolation/eval_material.py b/src/coder_eval/isolation/eval_material.py new file mode 100644 index 000000000..46517cfed --- /dev/null +++ b/src/coder_eval/isolation/eval_material.py @@ -0,0 +1,58 @@ +"""Allowlist (default-deny) mask for auto-mounted Claude-plugin trees. + +``mask_dirs`` names the child dirs of a mounted plugin root to tmpfs-mask so that only the +plugin surface (``.claude-plugin`` + the manifest-declared skill dirs) stays readable. +See docs/DOCKER_ISOLATION.md for the two residuals the mask cannot cover. +""" + +from __future__ import annotations + +from pathlib import Path + +from coder_eval.agents._skills import manifest_skill_dirs + + +_PLUGIN_MANIFEST_RELPATH = (".claude-plugin", "plugin.json") + + +def mask_dirs(root: Path) -> list[Path]: + """Directories under a mounted plugin ``root`` to tmpfs-mask, or ``[]`` for a non-plugin root. + + A nested skills path keeps its ancestors unmasked but masks their other children; + symlinked children are skipped. + """ + root = root.resolve() + if not (root / Path(*_PLUGIN_MANIFEST_RELPATH)).is_file(): + return [] + + keep = {(root / ".claude-plugin").resolve(), *manifest_skill_dirs(root)} + + # A kept path or an ancestor of one: never masked, descended into instead. + protected: set[Path] = set() + for kept in keep: + protected.add(kept) + for ancestor in kept.parents: + if ancestor == root: + break + if root in ancestor.parents: + protected.add(ancestor) + + # Never descend into a kept path. Under manifest `skills: "."` the root itself + # is kept, so nothing is masked (the caller warns). + descend = {p for p in ({root} | protected) if p not in keep} + + masked: set[Path] = set() + for directory in descend: + if not directory.is_dir() or directory.is_symlink(): + continue + for child in directory.iterdir(): + if child.is_symlink(): + continue + if not child.is_dir(): + continue + resolved = child.resolve() + if resolved in protected: + continue + masked.add(resolved) + + return sorted(masked) diff --git a/src/coder_eval/orchestrator.py b/src/coder_eval/orchestrator.py index 8bc75cbfa..4567ae84c 100644 --- a/src/coder_eval/orchestrator.py +++ b/src/coder_eval/orchestrator.py @@ -1957,8 +1957,8 @@ async def _communicate_attempt() -> TurnRecord: # ANTI-CHEAT WINDOW. The reference and the task dir sit at mode 000 for every # communicate attempt, retries included (this wrapper is outside # execute_with_retry), and are restored on every exit path. The SANDBOX owns - # whether a chmod window means anything for its driver. It does NOT hide the - # task DEFINITION: task.yaml is also staged at /work/input, an unsolved gap. + # whether a chmod window means anything for its driver. The task DEFINITION + # staged at /work/input is not windowed; the entry point deletes it after load. # Rationale: .claude/notes/permissions.md § Reference solutions and the anti-cheat window assert self.sandbox is not None async with self.sandbox.set_permissions([self._reference_dir, self.sandbox.task_dir]): diff --git a/tests/lint/runner.py b/tests/lint/runner.py index 51aeebaae..1dcd17964 100644 --- a/tests/lint/runner.py +++ b/tests/lint/runner.py @@ -60,7 +60,7 @@ # CE062 and CE023 are RETIRED ids and must stay unused — the ids above jump 061 to # 063. An id is a permanent documentation anchor: a suppression comment carrying one # in an older branch or commit message must never start meaning something new. -# Claim 068 next; 065 is taken without being in ALL_RULES. To see the whole space: +# Claim 069 next; 065 is taken without being in ALL_RULES. To see the whole space: # `grep -E '^class Test(CE[0-9]{3})' tests/test_custom_lint.py`. # Rationale: .claude/notes/lint-rules.md § The CE id space type RuleClass = type[BaseRule] diff --git a/tests/test_custom_lint.py b/tests/test_custom_lint.py index 46cf50f9c..db114448c 100644 --- a/tests/test_custom_lint.py +++ b/tests/test_custom_lint.py @@ -5223,3 +5223,178 @@ def test_is_suppressible(self, tmp_path): encoding="utf-8", ) assert not check_file(target, [TurnBracketOnTheClock]) + + +class TestCE068EvalMaterialReadableUnderPluginRoot: + """CE068 — no eval material left READABLE under an auto-mounted plugin root. + + Flags a `task_id:` YAML or a resolved `reference.directory` on the surface + `eval_material.mask_dirs` leaves unmasked. Move it under a sibling `tests/`. + """ + + ROOT = Path(__file__).parent.parent + + @staticmethod + def _plugin_roots_for_task(task, task_file: Path) -> list[Path]: + """Plugin roots reachable from a task's plugin / template paths, resolved as the runtime auto-mount does.""" + import os + + from coder_eval.models import TemplateDirSource + + raws: list[str] = [] + agent = task.agent + for plugin in (agent.plugins if agent else None) or []: + raw = plugin.get("path") if isinstance(plugin, dict) else None + if raw: + raws.append(str(raw)) + sandbox = task.sandbox + for source in (sandbox.template_sources if sandbox else None) or []: + if isinstance(source, TemplateDirSource): + raws.append(str(source.path)) + + roots: list[Path] = [] + for raw in raws: + expanded = Path(os.path.expandvars(os.path.expanduser(raw))) + root = (expanded if expanded.is_absolute() else task_file.parent / expanded).resolve() + if (root / ".claude-plugin" / "plugin.json").is_file(): + roots.append(root) + return roots + + @classmethod + def _offenders(cls, task, task_file: Path) -> list[str]: + import os + + from coder_eval.isolation.eval_material import mask_dirs + from coder_eval.orchestration.evaluation import resolve_host_reference_dir + + offenders: list[str] = [] + ref_dir = resolve_host_reference_dir(task, task_file) + + for root in cls._plugin_roots_for_task(task, task_file): + # The exact set the runtime tmpfs-masks. Anything NOT under one of + # these stays readable to the agent -- so if it is eval material, it + # leaks. + masked = {Path(m).resolve() for m in mask_dirs(root)} + + def _is_masked(p: Path, _masked: set[Path] = masked) -> bool: + rp = p.resolve() + return any(m == rp or m in rp.parents for m in _masked) + + # Walk the readable surface only: prune masked subtrees so we neither + # waste time nor false-flag a task def the runtime already hides. + for dirpath, dirnames, filenames in os.walk(root): + here = Path(dirpath) + dirnames[:] = [d for d in dirnames if not _is_masked(here / d)] + for fn in filenames: + if not (fn.endswith(".yaml") or fn.endswith(".yml")) or fn == "metadata.yaml": + continue + yaml_file = here / fn + try: + text = yaml_file.read_text(encoding="utf-8") + except OSError: + continue + # Cheap key test: a task definition declares a top-level task_id. + if re.search(r"(?m)^task_id\s*:", text): + offenders.append(f"readable task def under plugin root -> {yaml_file}") + + # A resolved reference dir under this root that the runtime leaves readable. + if ref_dir is not None and (ref_dir == root or root in ref_dir.parents) and not _is_masked(ref_dir): + offenders.append(f"readable reference.directory -> {ref_dir}") + + return offenders + + @pytest.mark.parametrize( + "path", + sorted(p for p in (Path(__file__).parent.parent / "tasks").rglob("*.yaml") if p.name != "metadata.yaml"), + ids=lambda p: p.relative_to(Path(__file__).parent.parent).as_posix(), + ) + def test_repo_tasks_keep_eval_material_masked(self, path: Path): + from coder_eval.orchestration.task_loader import load_task + + task, _ = load_task(path) + offenders = self._offenders(task, path) + assert not offenders, ( + f"{path}: {offenders} stay READABLE under an auto-mounted plugin root. The plugin-tree mask " + "masks non-skill child dirs, but cannot mask eval material inside a skill dir (would hide the " + "skill) or a loose YAML file at the root (a tmpfs masks a dir, not a file) — so the agent under " + "`driver: docker` reads its own grading answer key. Move the eval def / reference under a sibling " + "`tests/` (or any non-kept dir), where the allowlist masks it." + ) + + def _synthetic_plugin_task(self, tmp_path: Path, *, place_task_inside_skill: bool): + from coder_eval.models import TaskDefinition + + plugin_root = tmp_path / "plugin" + (plugin_root / ".claude-plugin").mkdir(parents=True) + (plugin_root / ".claude-plugin" / "plugin.json").write_text(json.dumps({"name": "demo"}), encoding="utf-8") + skill = plugin_root / "skills" / "demo" + skill.mkdir(parents=True) + (skill / "SKILL.md").write_text("# skill", encoding="utf-8") + + if place_task_inside_skill: + (skill / "leak.yaml").write_text("task_id: leaked\n", encoding="utf-8") + else: + sibling = plugin_root / "tests" / "tasks" + sibling.mkdir(parents=True) + (sibling / "ok.yaml").write_text("task_id: fine\n", encoding="utf-8") + + task_file = tmp_path / "task.yaml" + task_file.write_text("# task", encoding="utf-8") + task = TaskDefinition( + task_id="host", + description="d", + initial_prompt="p", + agent={"type": "claude-code", "plugins": [{"type": "local", "path": str(plugin_root)}]}, + success_criteria=[], + ) + return task, task_file + + def test_detects_a_task_def_inside_a_skill_dir(self, tmp_path): + """Positive sensor — a task def inside a kept skill dir stays readable.""" + task, task_file = self._synthetic_plugin_task(tmp_path, place_task_inside_skill=True) + offenders = self._offenders(task, task_file) + assert offenders and any("readable task def" in o for o in offenders), offenders + + def test_detects_a_loose_task_def_file_at_plugin_root(self, tmp_path): + """Positive sensor (L1) — a `task_id:` YAML FILE loose at the plugin root + is unmasked (a tmpfs masks a dir, not a single file) and must be flagged.""" + task, task_file = self._synthetic_plugin_task(tmp_path, place_task_inside_skill=False) + # add a loose criteria file directly at the plugin root (not in any masked dir) + plugin_root = tmp_path / "plugin" + (plugin_root / "answers.yaml").write_text("task_id: loose_leak\nsuccess_criteria: []\n", encoding="utf-8") + offenders = self._offenders(task, task_file) + assert offenders and any("answers.yaml" in o for o in offenders), offenders + + def test_allows_eval_material_outside_skill_dirs(self, tmp_path): + """Negative sensor — a task def under a sibling `tests/` is masked at runtime.""" + task, task_file = self._synthetic_plugin_task(tmp_path, place_task_inside_skill=False) + assert self._offenders(task, task_file) == [] + + def test_detects_a_reference_dir_inside_a_skill_dir(self, tmp_path): + from coder_eval.models import ReferenceSource, TaskDefinition + + plugin_root = tmp_path / "plugin" + (plugin_root / ".claude-plugin").mkdir(parents=True) + (plugin_root / ".claude-plugin" / "plugin.json").write_text(json.dumps({"name": "demo"}), encoding="utf-8") + ref = plugin_root / "skills" / "demo" / "solution" + ref.mkdir(parents=True) + + task_file = tmp_path / "task.yaml" + task_file.write_text("# task", encoding="utf-8") + task = TaskDefinition( + task_id="host", + description="d", + initial_prompt="p", + agent={"type": "claude-code", "plugins": [{"type": "local", "path": str(plugin_root)}]}, + reference=ReferenceSource(directory="plugin/skills/demo/solution"), + success_criteria=[], + ) + offenders = self._offenders(task, task_file) + assert offenders and any("reference" in o for o in offenders), offenders + + def test_shares_manifest_skill_dirs_with_the_runtime_allowlist(self): + """SSOT: CE068 and eval_material.mask_dirs must agree on skill dirs.""" + from coder_eval.agents._skills import manifest_skill_dirs + from coder_eval.isolation import eval_material + + assert eval_material.manifest_skill_dirs is manifest_skill_dirs diff --git a/tests/test_docker_runner_mounts.py b/tests/test_docker_runner_mounts.py index 609681ea3..89a95694a 100644 --- a/tests/test_docker_runner_mounts.py +++ b/tests/test_docker_runner_mounts.py @@ -35,7 +35,13 @@ _validate_extra_mount, grant_container_access, ) -from coder_eval.models import FileExistsCriterion, ReferenceSource, SandboxConfig, TaskDefinition +from coder_eval.models import ( + CONTAINER_INPUT_DIR, + FileExistsCriterion, + ReferenceSource, + SandboxConfig, + TaskDefinition, +) # DockerRunner targets Linux containers from POSIX hosts. On Windows the test @@ -715,6 +721,193 @@ def test_container_paths_reexported_from_docker_runner(self): assert CONTAINER_OUTPUT_DIR == "/work/output" +class TestAutoMountAllowlistMask: + """Auto-mounted plugin trees are default-deny masked (Fix B). + + A plugin root is mounted whole (:ro, so it loads), but every child dir that + is not the plugin surface (.claude-plugin + declared skill dirs) is masked + with an empty tmpfs so colocated eval material (sibling task YAMLs, reference + solutions, tests/, node_modules/) can never be read by the agent. + """ + + import json as _json + + @staticmethod + def _mounts(argv: list[str]) -> list[str]: + return [argv[i + 1] for i, a in enumerate(argv) if a == "-v"] + + @staticmethod + def _tmpfs(argv: list[str]) -> list[str]: + return [argv[i + 1] for i, a in enumerate(argv) if a == "--tmpfs"] + + def _plugin_root(self, root: Path, *, skills=None) -> Path: + (root / ".claude-plugin").mkdir(parents=True) + payload = {"name": "demo"} + if skills is not None: + payload["skills"] = skills + (root / ".claude-plugin" / "plugin.json").write_text(self._json.dumps(payload), encoding="utf-8") + return root + + def _runner(self, tmp_path: Path, *, plugins=None, template_sources=None) -> DockerRunner: + from coder_eval.models import DockerDriverConfig + + agent_kwargs: dict = {"type": "claude-code"} + if plugins is not None: + agent_kwargs["plugins"] = plugins + sandbox_kwargs: dict = {"driver": "docker", "docker": DockerDriverConfig()} + if template_sources is not None: + sandbox_kwargs["template_sources"] = template_sources + + task = TaskDefinition( + task_id="test", + description="test task", + initial_prompt="test", + agent=agent_kwargs, # dict -> validated into the AgentConfig union + sandbox=SandboxConfig(**sandbox_kwargs), + success_criteria=[FileExistsCriterion(description="c", path="t.txt")], + ) + rt = MagicMock() + rt.task = task + rt.run_dir = tmp_path / "run" + rt.task_file = None + return DockerRunner(rt) + + def _argv(self, runner: DockerRunner, tmp_path: Path) -> list[str]: + input_dir = tmp_path / "input" + output_dir = tmp_path / "output" + input_dir.mkdir(exist_ok=True) + output_dir.mkdir(exist_ok=True) + return runner._build_argv(input_dir, output_dir, container_name="c", image="img") + + def test_plugin_root_masks_non_skill_children(self, tmp_path: Path): + root = self._plugin_root(tmp_path / "plugin") + (root / "skills" / "demo").mkdir(parents=True) + (root / "tests").mkdir() + (root / "reference").mkdir() + (root / "node_modules").mkdir() + + runner = self._runner(tmp_path, plugins=[{"type": "local", "path": str(root)}]) + argv = self._argv(runner, tmp_path) + + tmpfs = self._tmpfs(argv) + # The whole root is :ro-mounted so the plugin loads. + assert f"{root.resolve()}:{root.resolve()}:ro" in self._mounts(argv) + # Non-skill children masked; skill surface + manifest not. + assert str((root / "tests").resolve()) in tmpfs + assert str((root / "reference").resolve()) in tmpfs + assert str((root / "node_modules").resolve()) in tmpfs + assert str((root / "skills").resolve()) not in tmpfs + assert str((root / ".claude-plugin").resolve()) not in tmpfs + + def test_tmpfs_targets_are_under_mounted_host_root(self, tmp_path: Path): + # Codex symlinks / Antigravity search paths dereference the ORIGINAL + # mounted host path, so the mask must sit on that path, not a copy. + root = self._plugin_root(tmp_path / "plugin") + (root / "skills" / "demo").mkdir(parents=True) + (root / "tests").mkdir() + + runner = self._runner(tmp_path, plugins=[{"type": "local", "path": str(root)}]) + argv = self._argv(runner, tmp_path) + + for masked in self._tmpfs(argv): + assert Path(masked).is_relative_to(root.resolve()) + + def test_template_source_plugin_root_is_masked(self, tmp_path: Path): + from coder_eval.models import TemplateDirSource + + root = self._plugin_root(tmp_path / "tpl") + (root / "skills" / "demo").mkdir(parents=True) + (root / "tests").mkdir() + + runner = self._runner(tmp_path, template_sources=[TemplateDirSource(path=str(root))]) + argv = self._argv(runner, tmp_path) + + assert str((root / "tests").resolve()) in self._tmpfs(argv) + + def test_non_plugin_template_dir_emits_no_mask(self, tmp_path: Path): + from coder_eval.models import TemplateDirSource + + plain = tmp_path / "plain_tpl" + (plain / "some_files").mkdir(parents=True) + + runner = self._runner(tmp_path, template_sources=[TemplateDirSource(path=str(plain))]) + argv = self._argv(runner, tmp_path) + + assert self._tmpfs(argv) == [] + + def test_no_plugins_no_mask(self, tmp_path: Path): + runner = self._runner(tmp_path) + argv = self._argv(runner, tmp_path) + + assert self._tmpfs(argv) == [] + + def test_nested_plugin_root_bind_wins_over_mask(self, tmp_path: Path): + # M2: plugin B nested under plugin A. A's mask would `--tmpfs /nested_b` + # while B's mount does `-v /nested_b:...:ro` -- an identical Docker mount + # destination the daemon rejects. The bind must win (B loads + masks its + # own children), so A's mask of B is dropped. + a = self._plugin_root(tmp_path / "plugin_a") + (a / "skills" / "demo").mkdir(parents=True) + (a / "tests").mkdir() + b = self._plugin_root(a / "nested_b") + (b / "skills" / "demo").mkdir(parents=True) + (b / "tests").mkdir() + + runner = self._runner( + tmp_path, + plugins=[{"type": "local", "path": str(a)}, {"type": "local", "path": str(b)}], + ) + argv = self._argv(runner, tmp_path) + mounts, tmpfs = self._mounts(argv), self._tmpfs(argv) + + # B is bind-mounted (so it loads) and NOT tmpfs-masked (no duplicate dest). + assert f"{b.resolve()}:{b.resolve()}:ro" in mounts + assert str(b.resolve()) not in tmpfs + # A's own non-skill child is still masked; B masks its own. + assert str((a / "tests").resolve()) in tmpfs + assert str((b / "tests").resolve()) in tmpfs + # No --tmpfs target collides with a bind destination (the M2 crash). + bind_dests = {m.split(":")[1] for m in mounts if m.count(":") >= 2} + assert not (set(tmpfs) & bind_dests) + + def test_skills_at_root_logs_mask_standdown(self, tmp_path: Path, caplog): + # M3: a plugin root whose whole tree is the skill surface (manifest + # `skills: "."`) voids the mask -- warn so it isn't silent. + root = self._plugin_root(tmp_path / "plugin", skills=".") + (root / "tests").mkdir() + + runner = self._runner(tmp_path, plugins=[{"type": "local", "path": str(root)}]) + with caplog.at_level("WARNING"): + argv = self._argv(runner, tmp_path) + + assert self._tmpfs(argv) == [] # nothing masked + assert any("stood down" in r.getMessage() for r in caplog.records), [r.getMessage() for r in caplog.records] + + # --- M1: _resolve_mount_path (the L2 relative-vs-CWD resolution) --- + + def test_resolve_mount_path_relative_uses_task_file_dir(self, tmp_path: Path): + runner = self._runner(tmp_path) + runner.rt.task_file = tmp_path / "suite" / "task.yaml" + (tmp_path / "suite" / "plugin").mkdir(parents=True) + assert runner._resolve_mount_path("plugin") == (tmp_path / "suite" / "plugin").resolve() + + def test_resolve_mount_path_absolute_is_unchanged(self, tmp_path: Path): + runner = self._runner(tmp_path) + runner.rt.task_file = tmp_path / "suite" / "task.yaml" + abs_path = tmp_path / "elsewhere" + assert runner._resolve_mount_path(str(abs_path)) == abs_path.resolve() + + def test_resolve_mount_path_none_task_file_falls_back_to_cwd(self, tmp_path: Path): + runner = self._runner(tmp_path) + runner.rt.task_file = None + assert runner._resolve_mount_path("rel") == (Path.cwd() / "rel").resolve() + + def test_resolve_mount_path_expands_env_vars(self, tmp_path: Path, monkeypatch): + monkeypatch.setenv("MY_PLUGIN_DIR", str(tmp_path / "pdir")) + runner = self._runner(tmp_path) + assert runner._resolve_mount_path("$MY_PLUGIN_DIR") == (tmp_path / "pdir").resolve() + + class TestReferenceMountAntiCheat: """The reference must reach the harness but never the agent under evaluation.""" @@ -1050,6 +1243,55 @@ async def fake_exec(*argv, **kwargs): "and DAC_OVERRIDE is dropped" ) + async def test_run_grants_input_dir_writable(self, tmp_path: Path, monkeypatch): + """ANTI-CHEAT: the in-container entry point DELETES the staged task.yaml. + + `unlink` needs `other`-write on the input DIRECTORY through the dropped + DAC caps, so run() must grant the input dir writable (like the run dir), + not read-only. A read-only grant leaves the delete failing EACCES and the + agent able to `cat` its own grading criteria. + """ + monkeypatch.setenv("CODER_EVAL_NO_CLAUDE_MOUNT", "1") + run_dir = tmp_path / "run" + run_dir.mkdir(mode=0o755) + task = TaskDefinition( + task_id="grant-input", + description="test task", + initial_prompt="test", + sandbox=SandboxConfig(), + success_criteria=[FileExistsCriterion(description="c", path="t.txt")], + ) + rt = MagicMock() + rt.task = task + rt.run_dir = run_dir + rt.replicate_index = 0 + rt.variant_id = "default" + rt.config_lineage = {} + rt.source_yaml = "# task" + rt.task_file = tmp_path / "task.yaml" + rt.task_file.write_text("# task", encoding="utf-8") + runner = DockerRunner(rt) + + seen: dict[str, list] = {"grants": []} + + def fake_grant(root, *, writable): + seen["grants"].append((Path(root).name, writable)) + return [] + + async def fake_exec(*argv, **kwargs): + raise FileNotFoundError("docker not present in this test") + + monkeypatch.setattr("coder_eval.isolation.docker_runner.grant_container_access", fake_grant) + monkeypatch.setattr("asyncio.create_subprocess_exec", fake_exec) + with pytest.raises(Exception): # noqa: B017 - the launch failure itself is not under test + await runner.run() + + # The input dir (a mkdtemp under staging; basename is "input") must be + # granted writable=True so the in-container delete of task.yaml succeeds. + input_grants = [w for name, w in seen["grants"] if name == "input"] + assert input_grants, "run() must grant the input dir container access" + assert all(input_grants), "the input dir must be granted writable (the container deletes task.yaml from it)" + class TestTaskDirCopyMount: """$TASK_DIR is a shielded copy at a fixed container path, not the host tree. @@ -1174,3 +1416,17 @@ def test_no_task_file_emits_no_mount(self, tmp_path: Path): runner._prepare_task_dir_mount(staging) assert runner._task_dir_mount_src is None + + def test_input_mount_is_read_write(self, tmp_path: Path): + """ANTI-CHEAT (Fix A): the container deletes the staged task.yaml. + + A `:ro` input mount rejects `rm` with EROFS, so the mount must be + read-write. The output mount is already writable for symmetry. + """ + argv, _ = self._prepared_argv(tmp_path) + + input_specs = [m for m in self._mounts(argv) if m.endswith(CONTAINER_INPUT_DIR)] + assert len(input_specs) == 1 + assert not input_specs[0].endswith(":ro"), "the input mount must be writable so task.yaml can be deleted" + output_specs = [m for m in self._mounts(argv) if m.endswith(CONTAINER_OUTPUT_DIR)] + assert output_specs and not output_specs[0].endswith(":ro") diff --git a/tests/test_eval_material.py b/tests/test_eval_material.py new file mode 100644 index 000000000..52f82c11f --- /dev/null +++ b/tests/test_eval_material.py @@ -0,0 +1,147 @@ +"""Unit tests for the allowlist (default-deny) mask over auto-mounted plugin trees. + +`eval_material.mask_dirs(root)` returns the child dirs to tmpfs-mask so that only +`.claude-plugin` + the manifest-declared skill dirs stay readable under a mounted +Claude-plugin root. Everything else (sibling task YAMLs, reference solutions, +`tests/`, `node_modules/`) is masked by default so it can never leak. +""" + +from __future__ import annotations + +import json +import os +from pathlib import Path + +import pytest + +from coder_eval.isolation.eval_material import mask_dirs + + +def _plugin_root(root: Path, *, skills: str | list[str] | None = None) -> Path: + """Create a `.claude-plugin/plugin.json`, optionally declaring a skills path.""" + manifest_dir = root / ".claude-plugin" + manifest_dir.mkdir(parents=True) + payload: dict = {"name": "demo"} + if skills is not None: + payload["skills"] = skills + (manifest_dir / "plugin.json").write_text(json.dumps(payload), encoding="utf-8") + return root + + +class TestMaskDirs: + def test_non_plugin_root_returns_empty(self, tmp_path: Path): + # No .claude-plugin/plugin.json -> not a plugin root -> nothing masked. + (tmp_path / "some_dir").mkdir() + assert mask_dirs(tmp_path) == [] + + def test_default_skills_layout_masks_non_skill_children(self, tmp_path: Path): + root = _plugin_root(tmp_path) + (root / "skills" / "demo").mkdir(parents=True) + (root / "skills" / "demo" / "SKILL.md").write_text("x", encoding="utf-8") + (root / "skills" / "demo" / "helper.py").write_text("x", encoding="utf-8") + (root / "tests").mkdir() + (root / "reference").mkdir() + (root / "node_modules").mkdir() + (root / "fixtures").mkdir() + + masked = set(mask_dirs(root)) + + assert masked == { + (root / "tests").resolve(), + (root / "reference").resolve(), + (root / "node_modules").resolve(), + (root / "fixtures").resolve(), + } + # Skill surface and manifest stay readable. + assert (root / "skills").resolve() not in masked + assert (root / ".claude-plugin").resolve() not in masked + + def test_manifest_skills_at_root_masks_nothing(self, tmp_path: Path): + # Degenerate layout: the manifest declares the plugin ROOT as its skills + # dir (`skills: "."`). The whole root is then the skill surface, so masking + # any child would hide the skill -> mask nothing (fail-safe: denial, not a + # leak). Guards against collapsing the keep-set and breaking plugin load. + root = _plugin_root(tmp_path, skills=".") + (root / "some_dir").mkdir() + (root / "tests").mkdir() + assert mask_dirs(root) == [] + + def test_skill_internal_assets_stay_readable(self, tmp_path: Path): + root = _plugin_root(tmp_path) + (root / "skills" / "demo" / "assets").mkdir(parents=True) + (root / "tests").mkdir() + + masked = set(mask_dirs(root)) + + # A skill's own subdirs are inside a kept dir -> never masked. + assert (root / "skills" / "demo" / "assets").resolve() not in masked + assert (root / "tests").resolve() in masked + + def test_nested_skills_path_does_not_over_expose_src(self, tmp_path: Path): + root = _plugin_root(tmp_path, skills="src/skills") + (root / "src" / "skills" / "demo").mkdir(parents=True) + (root / "src" / "secret_lib").mkdir(parents=True) + (root / "tests").mkdir() + + masked = set(mask_dirs(root)) + + # src/skills is kept; src/secret_lib (a sibling under src) is masked; src + # itself is NOT masked (it is an ancestor of the kept skills dir). + assert (root / "src" / "skills").resolve() not in masked + assert (root / "src").resolve() not in masked + assert (root / "src" / "secret_lib").resolve() in masked + assert (root / "tests").resolve() in masked + + def test_multiple_declared_skill_dirs_all_kept(self, tmp_path: Path): + root = _plugin_root(tmp_path, skills=["skills", "extra_skills"]) + (root / "skills" / "a").mkdir(parents=True) + (root / "extra_skills" / "b").mkdir(parents=True) + (root / "tests").mkdir() + + masked = set(mask_dirs(root)) + + assert (root / "skills").resolve() not in masked + assert (root / "extra_skills").resolve() not in masked + assert (root / "tests").resolve() in masked + + def test_symlinked_child_is_skipped(self, tmp_path: Path): + root = _plugin_root(tmp_path) + (root / "skills" / "demo").mkdir(parents=True) + outside = tmp_path.parent / "outside_target" + outside.mkdir(exist_ok=True) + os.symlink(outside, root / "evil_link") + + masked = set(mask_dirs(root)) + + # A symlinked child is not a valid tmpfs mountpoint and resolves against + # the container fs -> not masked. + assert (root / "evil_link") not in masked + assert outside.resolve() not in masked + + def test_reference_dir_under_root_is_masked(self, tmp_path: Path): + root = _plugin_root(tmp_path) + (root / "skills" / "demo").mkdir(parents=True) + (root / "solution").mkdir() # a reference dir sibling of skills + + assert (root / "solution").resolve() in set(mask_dirs(root)) + + def test_files_at_root_are_not_masked(self, tmp_path: Path): + # Only directories can be tmpfs mountpoints; a stray file is left alone. + root = _plugin_root(tmp_path) + (root / "skills" / "demo").mkdir(parents=True) + (root / "README.md").write_text("x", encoding="utf-8") + + assert (root / "README.md").resolve() not in set(mask_dirs(root)) + + +class TestSharedResolver: + def test_uses_shared_manifest_skill_dirs(self): + # SSOT: eval_material and CE068 must agree on "what is a skill dir". + from coder_eval.agents._skills import manifest_skill_dirs + from coder_eval.isolation import eval_material + + assert eval_material.manifest_skill_dirs is manifest_skill_dirs + + +if __name__ == "__main__": # pragma: no cover + raise SystemExit(pytest.main([__file__, "-v"])) diff --git a/tests/test_run_task_internal.py b/tests/test_run_task_internal.py new file mode 100644 index 000000000..271fcc482 --- /dev/null +++ b/tests/test_run_task_internal.py @@ -0,0 +1,75 @@ +"""Tests for the in-container entry point (`coder-eval _run-task-internal`). + +CE048: the command itself is a Typer command whose parameter defaults are +`OptionInfo` sentinels, so it must never be invoked in-process. These tests +target the extracted, pure `_scrub_staged_inputs` helper instead. +""" + +from __future__ import annotations + +from pathlib import Path + +from coder_eval.cli.run_task_internal_command import _scrub_staged_inputs +from coder_eval.models import IN_CONTAINER_ENV + + +def _stage(tmp_path: Path) -> tuple[Path, Path]: + """Write a staged task.yaml + context.json, both carrying the criteria.""" + task_yaml = tmp_path / "task.yaml" + task_yaml.write_text("task_id: x\nsuccess_criteria: []\n", encoding="utf-8") + context_json = tmp_path / "context.json" + # source_yaml is the raw task text (criteria verbatim) -- the second copy the + # scrub must remove, at the top level and inside config_lineage. + context_json.write_text( + '{"variant_id": "v", "source_yaml": "task_id: x\\nsuccess_criteria: [SECRET]\\n"}', + encoding="utf-8", + ) + return task_yaml, context_json + + +class TestScrubStagedInputs: + """Anti-cheat: BOTH staged grading-answer-key copies (task.yaml's criteria AND + context.json's source_yaml) are deleted after load so the in-container agent + cannot read its own criteria back from either file.""" + + def test_deletes_both_when_in_container(self, tmp_path: Path, monkeypatch): + monkeypatch.setenv(IN_CONTAINER_ENV, "1") + task_yaml, context_json = _stage(tmp_path) + + _scrub_staged_inputs(task_yaml, context_json) + + assert not task_yaml.exists() + assert not context_json.exists(), "context.json (source_yaml has criteria) must be deleted too" + + def test_survives_when_not_in_container(self, tmp_path: Path, monkeypatch): + # A host/tempdir invocation shares our uid and has no filesystem + # isolation; the gate keeps the delete docker-only by construction. + monkeypatch.delenv(IN_CONTAINER_ENV, raising=False) + task_yaml, context_json = _stage(tmp_path) + + _scrub_staged_inputs(task_yaml, context_json) + + assert task_yaml.exists() + assert context_json.exists() + + def test_env_var_set_but_not_one_leaves_files(self, tmp_path: Path, monkeypatch): + # Only the exact "1" sentinel arms the delete, mirroring the reference + # window gate (Sandbox.enforces_permission_windows). + monkeypatch.setenv(IN_CONTAINER_ENV, "0") + task_yaml, context_json = _stage(tmp_path) + + _scrub_staged_inputs(task_yaml, context_json) + + assert task_yaml.exists() + assert context_json.exists() + + def test_missing_files_do_not_raise(self, tmp_path: Path, monkeypatch): + monkeypatch.setenv(IN_CONTAINER_ENV, "1") + task_yaml = tmp_path / "does-not-exist.yaml" + context_json = tmp_path / "also-missing.json" + + # missing_ok=True: a re-entrant call must not crash on an absent file. + _scrub_staged_inputs(task_yaml, context_json) + + assert not task_yaml.exists() + assert not context_json.exists()