diff --git a/docs/configuration.md b/docs/configuration.md index 3e1ca61..6d96abc 100644 --- a/docs/configuration.md +++ b/docs/configuration.md @@ -230,6 +230,17 @@ The configurable stages are `issue_review`, `implement`, `verify`, `review`, `se `issue_review.mode: auto` (the default) makes a new `run_cycle.py`, `cc-orchestrator`, or `cc-orca-orchestrator` cycle review its work item with `cc-issue-review` before implementing it, except a task declared trivial and not security-sensitive. `off` keeps the original flow. `--issue-review`, or `issue_review=auto|off` in an orchestrator invocation, overrides it for one run. Only a confirmed `READY` reaches implementation. See [Review lifecycle → Issue review](review-cycle.md#issue-review-before-implementation). +Issue review's GitHub network grant does not grant stored credentials. Codex +denies `~/.config/gh` in this non-publishing role. Claude ordinary issue review +keeps its normal tooling; if registry domains or a Docker grant enable its Bash +sandbox, that sandbox also denies the gh store. For authenticated issue reads +on private repositories from either sandbox, set +`code_cycle.user_choices.credential_reads: ["~/.config/gh"]` in the trusted +project configuration. Without that exception, stored-credential authentication +is unavailable to sandboxed gh commands; a resulting access failure does not +mean the work item needs refinement. The exception does not authorize publication. +See [credential-read exceptions](dependency-preparation.md#credential-read-exceptions). + ### CI policy `ci.policy` decides how a cycle treats the code host's checks. `strict`, the default, is the original serial flow: the implementation and each resolution wait for their pushed head's checks, and a review approves only once they passed. `parallel` starts the review while the checks run, sends a review that requests changes to resolution at once, and approves a clean review only after the runtime's one CI gate saw every required check of that same SHA pass. Parallel mode never disables CI; it removes the time spent waiting for it before any useful work can start. @@ -394,6 +405,10 @@ On Windows, the `~/.config/code-cycle-toolkit` directory is `%APPDATA%\code-cycl See [dependency preparation](dependency-preparation.md) for automatic frozen Node installs, contained lifecycle-build recovery, persistent setup evidence, and the `code_cycle.user_choices` configuration shared by dispatched stages. +Its `credential_reads` list accepts explicit credential-store exceptions, such +as `["~/.ssh"]`; the default is no exceptions. See +[credential-read exceptions](dependency-preparation.md#credential-read-exceptions) +for the accepted stores, evidence and sandbox scope. --- diff --git a/docs/dependency-preparation.md b/docs/dependency-preparation.md index 08e2a55..87ffe55 100644 --- a/docs/dependency-preparation.md +++ b/docs/dependency-preparation.md @@ -19,8 +19,10 @@ silently installing with another manager. Workspace package manifests determine the package-local dependency directories, including `apps/web/node_modules`. Independent nested lock roots are prepared separately. PnP outputs are supported. -Preparation runs through `codex sandbox` for Codex, and the installed `srt` -(Anthropic sandbox runtime) for Claude. Neither path falls back to a host install. +On Linux/WSL2 preparation runs through the installed `srt` (Anthropic sandbox +runtime) for both executors. There is no Linux Codex fallback. Until the `srt` +path is validated on macOS and Windows, Codex retains `codex sandbox` there, +with the same credential denials. Neither path falls back to a host install. The install runs in a scratch copy of the lock root's tracked files. The sandbox can write only that scratch directory and the download cache; the lockfile and manifests stay read-only there too. The checkout and its Git metadata are never @@ -38,8 +40,8 @@ failure, not a successful install or a model failure. The host supplies the native executor and package manager. Before native stage dispatch, Linux/WSL2 recovery reuses working `srt` and `rg` binaries. Missing or unusable tools are installed in immutable bundles below -`~/.local/lib/code-cycle-tools/`; `srt` recovery runs only for Claude stages. -The stage, host isolation probe and Claude preparation process receive their +`~/.local/lib/code-cycle-tools/`; both executors require `srt` for Linux preparation. +The stage, host isolation probe and preparation process receive their managed executable directories prepended to their own `PATH`, preserving the original host directories in their existing order. Tools already selected from the host `PATH` do not promote their directories. This does not change the runtime's @@ -172,11 +174,44 @@ Claude receives both the allowed-tool patterns and the command hook policy. Codex retains normal tools/MCP access and receives the matching explicit network profile. Default code-host domains are granted to publishing stages. Per the owner's #279 decision, Codex issue review also receives `github.com` and -`api.github.com`, with the same existing gh credentials, while its workspace and -publication contract remain read-only. This is a credential trust decision: -the domain grant itself does not make an HTTP credential read-only. Other +`api.github.com`. Under #294 its sandboxed commands need the explicit +`credential_reads: ["~/.config/gh"]` exception to use stored gh credentials; +its workspace and publication contract remain read-only. The domain grant itself +does not make an HTTP credential read-only. Other non-publishing roles receive only registry access, without a publication profile. +### Credential-read exceptions + +Every Codex stage sandbox, existing Claude Bash sandbox and preparation sandbox +denies reads of `~/.ssh`, `~/.aws`, +`~/.codex/auth.json`, `~/.claude/.credentials.json` and `~/.claude.json`. +`~/.config/gh` is readable by publishing stages only; preparation always denies +it. These rules restrict child commands, so the agent can still authenticate +outside its command sandbox. Rules name existing resolved targets, including +symlink targets; absent paths are skipped to avoid sandbox mount failures and +persistent placeholder files. A sandbox startup failure never retries without +the denials. Claude's sandbox scope is unchanged: active GitHub policies with +confined review verification, registry domains, publication commands or a Docker +grant. Inactive policies and ordinary Claude issue review keep their earlier +tooling. Static-only recovery disables execution and does not prove denial. + +A project whose tests need a particular store, or whose remote is SSH-only, +must opt out explicitly in the trusted source checkout: + +```yaml +code_cycle: + user_choices: + credential_reads: ["~/.ssh"] +``` + +Only the six store names above are accepted, without wildcards or arbitrary +paths. Each exception removes only that store's denial in stages and preparation; +it does not grant network access or make raw SSH work through the HTTPS proxy. +The stage artifacts, preparation report and stage prompt record the exceptions. +Never put credentials in this configuration. Review clones cannot supply their +own exceptions. See [host permissions](host-permissions.md) for the remaining +trust boundary around publishing credentials. + When runtime preparation detects Node dependencies, the worker inherits the same npm/Yarn registry domains as preparation, plus explicitly selected registry domains. This also applies after incomplete setup: package-manager launchers and diff --git a/docs/host-permissions.md b/docs/host-permissions.md index 374fb98..356fe01 100644 --- a/docs/host-permissions.md +++ b/docs/host-permissions.md @@ -251,13 +251,30 @@ an install with lifecycle scripts disabled, a separate contained lifecycle build only when one is required, persistent evidence and bounded retries; it never inherits another revision's `node_modules`. Non-publishing Codex stages receive no default code-host network route, because readable credentials would permit -writes as well as reads. The exception is Codex issue review, which by the -owner's #279 decision receives `github.com` and `api.github.com` with the -existing `gh` credentials (see +writes as well as reads. Codex issue review, by the owner's #279 decision, +receives `github.com` and `api.github.com`; #294 denies its stored gh +credentials unless the project explicitly opts out (see [dependency preparation](dependency-preparation.md#explicit-user-choices)); its command table still denies publication. Explicitly selected registry domains do not add code-host domains. +Every Codex stage and preparation sandbox denies credential-store reads: +`~/.ssh`, `~/.aws`, `~/.codex/auth.json`, `~/.claude/.credentials.json`, and +`~/.claude.json`. Only publishing stages retain `~/.config/gh` access by +default. Codex uses `none` filesystem rules; Claude and preparation use +`denyRead`. Claude adds these denials where its Bash sandbox already applies: +an active GitHub policy with confined review verification, registry domains, +publication commands or a Docker grant. Inactive policies retain their normal +tooling; adding credential denials does not enable a sandbox for them. The same +condition selects sandbox configuration checks and the Linux functional probe. +Unavailable isolation selects static-only recovery with verification disabled; +this does not establish credential confinement. Existing paths resolve to their +real targets before rules are +emitted; missing paths are omitted. An explicit per-project +`code_cycle.user_choices.credential_reads` list selects exceptions and is +reported in dispatch and preparation evidence. See +[credential-read exceptions](dependency-preparation.md#credential-read-exceptions). + ## When a command is denied The stage report lists each denied command (`blocked command: …`), read from diff --git a/docs/verification.md b/docs/verification.md index 192b012..ac76594 100644 --- a/docs/verification.md +++ b/docs/verification.md @@ -24,6 +24,35 @@ Herdr is optional when running the suite: with `HERDR_ENV` unset, worktree selection uses Git directly; dedicated Herdr tests opt into the flag and mock the command. +### Credential-read confinement (E7) + +The unit suite checks every native stage profile's credential rules, explicit +exceptions, canonical symlink targets, and each missing-denial failure in +`probe_review_confinement`. The installed Codex confinement probe and `srt` +prerequisite probe also test synthetic credential stores without reading host +secrets. Claude settings with inherited `allowRead` are refused because that +setting could cancel a denial. + +On a host that supports native sandbox execution, run the opt-in E7 matrix: + +```bash +CODE_CYCLE_LIVE_CREDENTIAL_PROBE=1 python3 -m unittest discover -s tests -p test_credential_reads.py +``` + +This requires installed `codex` and `srt`. It exercises read/write, publishing, +registry and Docker-grant profile combinations on both executors, prints only +`READABLE`/`denied` states for synthetic stores, and fails on any unexpected +read. Publishing profiles must allow synthetic gh credentials and deny the five +other stores. No host credential contents are accessed. Preserve the output +and platform/version information as boundary evidence. macOS/Windows still need +platform validation before removing their Codex preparation fallback. + +Do not opt in inside a native stage whose outer sandbox cannot support nested +Codex or `srt`. The test skips there; record that boundary as `not_verified`, +and keep unit results separate from real sandbox evidence. Static-only recovery +when the sandbox is unavailable does not establish credential confinement; +verification remains disabled under the existing recovery policy. + ### Codex Git metadata sandbox Writing Codex dispatches from linked worktrees need some of the worktree's Git diff --git a/scripts/dependency_setup.py b/scripts/dependency_setup.py index 1e037b8..3806e1f 100644 --- a/scripts/dependency_setup.py +++ b/scripts/dependency_setup.py @@ -13,11 +13,14 @@ import signal import shutil import subprocess +import sys import tempfile import time from pathlib import Path +from typing import Any import native_prerequisites +import stage_permissions as sp REGISTRIES = ("registry.npmjs.org", "registry.yarnpkg.com") LOCKS = {"pnpm-lock.yaml": "pnpm", "package-lock.json": "npm", @@ -99,8 +102,12 @@ def _manifest(path: Path) -> dict: def sandbox_command(executor: str, binary: str, command: list[str], *, writable: list[Path], protected: list[Path], domains: tuple[str, ...], - settings: Path, path: str | None = None) -> list[str]: - if executor == "codex": + settings: Path, path: str | None = None, + credential_reads: tuple[str, ...] = ()) -> list[str]: + denied = sp.credential_read_denials(credential_reads=credential_reads) + # Linux/WSL2 is validated for srt on both executors. Keep the aligned + # Codex fallback only on platforms whose srt path is not yet validated. + if executor == "codex" and not sys.platform.startswith("linux"): if native_prerequisites.nested_codex(): raise OSError("known nested Codex limitation (#268): app-server socket directory must be " "a user-owned directory with mode 0700; check not_verified without probing") @@ -113,6 +120,7 @@ def sandbox_command(executor: str, binary: str, command: list[str], *, rules.update({str(path): "read" for path in protected if any(path.is_relative_to(output) or output.is_relative_to(path) for output in writable)}) + rules.update({path: "none" for path in denied}) filesystem = "{" + ",".join(json.dumps(key) + "=" + json.dumps(value) for key, value in rules.items()) + "}" network = "{" + ",".join(json.dumps(host) + '=\"allow\"' for host in domains) + "}" @@ -126,13 +134,13 @@ def sandbox_command(executor: str, binary: str, command: list[str], *, # bootstrap or unsandboxed fallback when the host has not installed it. srt = native_prerequisites.executable("srt", path=path) if not srt: - raise OSError("Claude dependency preparation needs the installed srt sandbox runtime") + raise OSError("dependency preparation needs the installed srt sandbox runtime") settings.write_text(json.dumps({ "network": {"allowedDomains": list(domains), "deniedDomains": [], "allowLocalBinding": False}, "filesystem": {"allowWrite": [str(path) for path in writable], "denyWrite": [str(path) for path in protected], - "denyRead": ["~/.ssh", "~/.aws", "~/.config/gh"]}, + "denyRead": list(denied)}, }), encoding="utf-8") # srt shell-quotes each positional argument itself, so a pre-joined string # would run as one word. "--" keeps srt from parsing npm's own flags, such @@ -313,7 +321,8 @@ def _run(argv, *, timeout, **options): def prepare(cwd: str, executor: str, binary: str, *, domains: tuple[str, ...] = (), runner=_run, evidence_root: Path | None = None, sleep=time.sleep, - prerequisite_path: str | None = None) -> dict: + prerequisite_path: str | None = None, + credential_reads: tuple[str, ...] = ()) -> dict: """Return honest setup evidence even on failure; no verdict or CI claim. Logs outlive the disposable clone. Failure is given to the stage so it can @@ -327,7 +336,10 @@ def prepare(cwd: str, executor: str, binary: str, *, base = evidence_root or Path.home() / ".local/state/code-cycle-toolkit/preparation" base.mkdir(parents=True, exist_ok=True) evidence = Path(tempfile.mkdtemp(prefix="setup-", dir=base)) - report = {"status": "prepared", "items": [], "evidence": str(evidence / "report.json")} + report: dict[str, Any] = {"status": "prepared", "items": [], "evidence": str(evidence / "report.json"), + "credential_read_policy": {"exceptions": list(credential_reads), + "denied_paths": list(sp.credential_read_denials( + credential_reads=credential_reads))}} head = subprocess.run(["git", "rev-parse", "HEAD"], cwd=root, capture_output=True, text=True, timeout=30, check=True).stdout.strip() tracked = subprocess.run(["git", "ls-files", "-z"], cwd=root, @@ -357,7 +369,7 @@ def prepare(cwd: str, executor: str, binary: str, *, if key in {"PATH", "SYSTEMROOT", "LANG", "LC_ALL", "HTTP_PROXY", "HTTPS_PROXY", "NO_PROXY", "http_proxy", "https_proxy", "no_proxy"}} - if executor == "claude": + if executor == "claude" or sys.platform.startswith("linux"): env["PATH"] = (native_prerequisites.tool_path(env.get("PATH", "")) if prerequisite_path is None else prerequisite_path) env.update(HOME=str(scratch), TMPDIR=str(scratch), CI="true", @@ -376,7 +388,8 @@ def run(command, label, timeout, *, cwd=staged, environment=env, argv = sandbox_command(executor, binary, command, writable=allowed, protected=protected, domains=tuple(dict.fromkeys( (*REGISTRIES, *domains))), settings=scratch / "sandbox.json", - path=environment.get("PATH") if executor == "claude" else None) + path=environment.get("PATH"), + credential_reads=credential_reads) record["sandbox_command"] = argv result = runner(argv, cwd=cwd, env=environment, capture_output=True, text=True, timeout=timeout, stdin=subprocess.DEVNULL) diff --git a/scripts/executors.py b/scripts/executors.py index 377b3b9..f906841 100644 --- a/scripts/executors.py +++ b/scripts/executors.py @@ -683,7 +683,7 @@ def _docker_host_probe(adapter, *, cwd: str, writes: bool, publishes: bool, git_directories.append(str((Path(cwd) / path_value).resolve())) writable.extend(CodexAdapter.git_write_roots(tuple(dict.fromkeys(git_directories)))) sandbox = sp.review_sandbox_settings( - tuple(writable), network_domains, docker_host=True)["sandbox"] + tuple(writable), network_domains, docker_host=True, publishes=publishes)["sandbox"] network = dict(sandbox["network"]) network.setdefault("deniedDomains", []) filesystem = dict(sandbox["filesystem"]) @@ -1233,6 +1233,24 @@ def readable(self, argv: list[str], read_dirs: tuple[str, ...]) -> list[str]: """ return argv + def _bash_sandbox_preflight(self, command_policy: StagePolicy, + evidence: dict[str, Any], *, path: str, + live: bool) -> StagePolicy: + """Probe each effective Claude sandbox once, including late activation.""" + if (self.name == "claude" and command_policy.uses_bash_sandbox + and sys.platform.startswith("linux") and live + and "functional_check" not in evidence): + ready, detail = native.isolation_probe("bwrap", runner=_run_sandbox_probe, path=path) + evidence["functional_check"] = {"probe": "bwrap", "passed": ready, "evidence": detail} + if not ready or any(tool in evidence["missing"] for tool in ("bwrap", "socat")): + # Keep static editing/reading, never run code or publish through + # a sandbox the host could not reproduce. + evidence.update(mode="static_only", status="unverified") + return replace(command_policy, static_only=True) + if evidence["status"] != "unverified": + evidence["status"] = "verified" + return command_policy + def dispatch(self, target: Target, task: str, *, cwd: str | None = None, timeout: int = DEFAULT_DISPATCH_TIMEOUT_SECONDS, runner=_run, writes: bool = False, publishes: bool = False, @@ -1264,38 +1282,39 @@ def dispatch(self, target: Target, task: str, *, cwd: str | None = None, if command_policy is not None and dispatch_attempt_id is not None: # The hook records its own denials under this attempt (#306). command_policy = replace(command_policy, attempt_id=dispatch_attempt_id) + if command_policy is not None: + command_policy = replace(command_policy, credential_publishing=publishes) policy = command_policy if command_policy is not None and command_policy.active else None requested_verification = command_policy.verification_commands if command_policy is not None else () policy_note: dict[str, Any] = ({} if command_policy is None else {"command_policy": ( "active" if policy is not None else f"inactive: {command_policy.inactive_reason}")}) + credential_reads = command_policy.credential_reads if command_policy is not None else () + policy_note["credential_read_policy"] = { + "exceptions": list(credential_reads), + "denied_paths": list(sp.credential_read_denials( + publishes=publishes, credential_reads=credential_reads)), + "gh_readable": publishes or sp.GH_CREDENTIAL_STORE in credential_reads, + } + if credential_reads: + task += ("\nProject credential-read opt-out (explicit trusted configuration): " + + json.dumps(list(credential_reads)) + + ". Report these exceptions in stage evidence.") # Record native-sandbox prerequisites before dependency setup or agent # dispatch. This evidence is attached to the stage attempt; a missing # capability must not silently turn an unrun check into a pass. prerequisite_path = os.environ.get("PATH", "") + sandbox_evidence: dict[str, Any] = {} if command_policy is not None: sandbox_evidence = native.observe(self.name, recover=runner is _run) prerequisite_path = native.tool_path(prerequisite_path, probes=sandbox_evidence.get("probes", [])) missing = sandbox_evidence["missing"] - needs_bash_sandbox = (self.name == "claude" and policy is not None - and (policy.requires_confined_verification or policy.registry_domains - or policy.publication_commands)) - if needs_bash_sandbox and sys.platform.startswith("linux") and runner is _run: - ready, detail = native.isolation_probe("bwrap", runner=_run_sandbox_probe, path=prerequisite_path) - sandbox_evidence["functional_check"] = {"probe": "bwrap", "passed": ready, - "evidence": detail} - sandbox_evidence["status"] = "verified" if ready else "unverified" - if not ready or any(tool in missing for tool in ("bwrap", "socat")): - # The existing hook + disposable clone remains the static - # boundary. Remove code execution grants, never run tests - # outside a sandbox. Writing stages may still edit, but - # publication requires a working sandbox and stays denied. - command_policy = replace(command_policy, static_only=True) - policy = command_policy - sandbox_evidence["mode"] = "static_only" - sandbox_evidence["status"] = "unverified" - if (self.name == "claude" and sys.platform.startswith("linux") - and runner is _run and "srt" not in missing): + command_policy = self._bash_sandbox_preflight( + command_policy, sandbox_evidence, path=prerequisite_path, live=runner is _run) + policy = command_policy if command_policy.active else None + if (sys.platform.startswith("linux") + and runner is _run and "srt" not in missing + and not sandbox_evidence.get("codex_socket")): ready, detail = native.isolation_probe("srt", runner=_run_sandbox_probe, path=prerequisite_path) sandbox_evidence["preparation_isolation"] = {"passed": ready, "evidence": detail} if not ready: @@ -1389,7 +1408,7 @@ def dispatch(self, target: Target, task: str, *, cwd: str | None = None, else: preparation = dependency_setup.prepare( worker_cwd, self.name, self.binary, domains=command_policy.registry_domains, - prerequisite_path=prerequisite_path) + prerequisite_path=prerequisite_path, credential_reads=credential_reads) except (OSError, ValueError, subprocess.SubprocessError) as error: preparation = {"status": "incomplete", "items": [], "error": str(error)} if preparation["status"] != "not_required": @@ -1401,12 +1420,32 @@ def dispatch(self, target: Target, task: str, *, cwd: str | None = None, (*dependency_setup.REGISTRIES, *command_policy.registry_domains)))) policy = command_policy if policy is not None else None policy_note["dependency_preparation"] = preparation + if command_policy is not None: + # Preparation can activate the sandbox by adding registry domains. + # Recheck the final policy before emission; an earlier probe is + # retained rather than launched twice. + command_policy = self._bash_sandbox_preflight( + command_policy, sandbox_evidence, path=prerequisite_path, live=runner is _run) + policy = command_policy if command_policy.active else None + if command_policy.static_only: + task += ("\nNative Bash isolation is unavailable; keep static guards and report " + "affected verification as not_verified. Do not launch a nested sandbox.") + refused = self.policy_preflight(policy, cwd) + if refused is not None: + return DispatchResult( + DispatchOutcome.FAILED, self.name, target, + detail=failure_excerpt(f"BLOCKED: {refused}", 600), + error_code="precondition", start_state="not_started", lifecycle_state="failed", + artifacts={**policy_note, "command_policy": f"blocked: {failure_excerpt(refused, 300)}", + "policy_preflight_refusal": failure_excerpt(refused, 600)}, + ) argv_options = {} if self.name == "codex" and command_policy is not None: - # The owner explicitly authorizes issue-review GitHub reads with - # the same credentials as publishing stages (#279). Its workspace - # and publication contract remain read-only. Other non-publishing - # roles do not acquire code-host access through registry choices. + argv_options["credential_reads"] = credential_reads + # The owner authorizes issue-review GitHub network reads (#279). + # Stored gh credentials now require an explicit credential_reads + # exception (#294); workspace and publication remain read-only. + # Other non-publishing roles gain no code-host route via registries. issue_domains = (("github.com", "api.github.com") if command_policy.role == "issue_review" and command_policy.code_host == "github" else ()) @@ -1964,6 +2003,11 @@ def probe_git_metadata_write( script = "\n".join(script_lines) + "\n" profile = "code_cycle_publish_write" filesystem = self._publish_filesystem_rules(writable_dirs, cwd=cwd) + credential_rules = ",".join(f'{json.dumps(path)}="none"' for path in + sp.credential_read_denials(publishes=True)) + if credential_rules: + filesystem = (filesystem[:-1] + ("," if filesystem[1:-1].strip() else "") + + credential_rules + "}") argv = [ self.binary, "sandbox", "--permission-profile", profile, "--cd", cwd, @@ -2075,7 +2119,8 @@ def argv(self, target: Target, task: str, cwd: str | None = None, sandbox_grants: tuple[str, ...] = (), ignore_rules: bool = False, review_clone: str | None = None, - network_domains: tuple[str, ...] | None = None) -> list[str]: + network_domains: tuple[str, ...] | None = None, + credential_reads: tuple[str, ...] = ()) -> list[str]: # `codex exec` is read-only unless told otherwise, which is why four # canary runs had an implementer that could not implement: it reported # BLOCKED on "the read-only workspace" and nothing here had ever asked @@ -2121,7 +2166,10 @@ def argv(self, target: Target, task: str, cwd: str | None = None, argv += ["-c", f'default_permissions="{profile}"', "-c", f'permissions.{profile}.extends=":read-only"'] else: - argv += ["-s", "workspace-write" if writes else "read-only"] + profile = "code_cycle_write" if writes else "code_cycle_read" + parent = ":workspace" if writes else ":read-only" + argv += ["-c", f'default_permissions="{profile}"', + "-c", f'permissions.{profile}.extends="{parent}"'] if review_clone and not writes: # A reviewer verifies in its disposable clone (#142): the clone and, # added with the scratch directory later, its TMPDIR are the only @@ -2140,6 +2188,18 @@ def argv(self, target: Target, task: str, cwd: str | None = None, f"permissions.{profile}.filesystem=" f"{self._publish_filesystem_rules(writable_dirs, cwd=cwd)}", ] + # A single filesystem table preserves the Git/clone grants as well as + # denials; a later override must never erase either side of the boundary. + rule = f"permissions.{profile}.filesystem=" + existing = next((i for i, option in enumerate(argv) if option.startswith(rule)), None) + denied = ",".join(f'{json.dumps(path)}="none"' for path in + sp.credential_read_denials(publishes=publishes, + credential_reads=credential_reads)) + if existing is None: + argv += ["-c", f"{rule}{{{denied}}}"] + elif denied: + table = argv[existing][len(rule):] + argv[existing] = f"{rule}{table[:-1]}{',' if table[1:-1].strip() else ''}{denied}}}" if ignore_rules: # Execpolicy rules load only from configuration layers, with no # per-run option, so a stage does not inherit the user's @@ -2172,6 +2232,19 @@ def probe_review_confinement(self, *, runner=None) -> tuple[bool, str]: if os.name != "nt": (clone / "escape").symlink_to(sibling, target_is_directory=True) checks.append(("a symlink out of the clone", clone / "escape" / "link", False)) + # Synthetic E7 stores: never read real credentials, and prove + # denial even on a host where the corresponding store is absent. + home = base / "home" + credential_files = [] + for store in (*sp.CREDENTIAL_STORES, sp.GH_CREDENTIAL_STORE): + target = home / store[2:] + target.parent.mkdir(parents=True, exist_ok=True) + if store in ("~/.ssh", "~/.aws", sp.GH_CREDENTIAL_STORE): + target.mkdir() + target = target / "probe" + target.write_text("synthetic credential probe", encoding="utf-8") + credential_files.append(target) + denied = sp.credential_read_denials(home=home) script = "\n".join([ "import sys", "failed = []", @@ -2179,11 +2252,16 @@ def probe_review_confinement(self, *, runner=None) -> tuple[bool, str]: f"except OSError:\n ok = False\n" f"if ok != {expected}: failed.append({json.dumps(label)})" for label, path, expected in checks), + *(f"try:\n open({json.dumps(str(path))}, 'rb').close(); readable = True\n" + f"except OSError:\n readable = False\n" + f"if readable: failed.append('credential read denial {index}')" + for index, path in enumerate(credential_files)), "sys.exit('; '.join(failed) if failed else 0)", ]) profile = "code_cycle_review_probe" filesystem = (f"{{{json.dumps(str(clone))}=\"write\"," - f"{json.dumps(str(scratch))}=\"write\"}}") + f"{json.dumps(str(scratch))}=\"write\"," + + ",".join(f'{json.dumps(path)}="none"' for path in denied) + "}") argv = [self.binary, "sandbox", "--permission-profile", profile, "-c", f'permissions.{profile}.extends=":read-only"', "-c", f"permissions.{profile}.filesystem={filesystem}", @@ -2720,7 +2798,7 @@ def policy_preflight(self, policy: StagePolicy | None, cwd: str | None) -> str | f"{', '.join(mods)}. Claude Code offers no flag that turns mods " "off for one dispatch while keeping the user's other hooks, so " "disable that plugin for cycle runs") - if policy.requires_confined_verification: + if policy.uses_bash_sandbox: reason = sp.sandbox_override() if reason: return reason @@ -2785,9 +2863,7 @@ def policy_files(self, argv: list[str], policy: StagePolicy | None, *, "BASH_DEFAULT_TIMEOUT_MS": str(sp.CLAUDE_VERIFICATION_TIMEOUT_MS), "BASH_MAX_TIMEOUT_MS": str(sp.CLAUDE_VERIFICATION_TIMEOUT_MS), }) - if runtime_policy.active and not runtime_policy.static_only and (runtime_policy.requires_confined_verification - or runtime_policy.registry_domains or runtime_policy.publication_commands - or runtime_policy.docker): + if runtime_policy.uses_bash_sandbox: writable = [cwd, scratch] if runtime_policy.role in sp.WRITING_ROLES: git_directories = [] @@ -2798,7 +2874,9 @@ def policy_files(self, argv: list[str], policy: StagePolicy | None, *, writable.extend(CodexAdapter.git_write_roots(tuple(dict.fromkeys(git_directories)))) settings.update(sp.review_sandbox_settings( tuple(writable), runtime_policy.registry_domains, - docker_host=runtime_policy.docker)) + docker_host=runtime_policy.docker, + publishes=runtime_policy.credential_publishing, + credential_reads=runtime_policy.credential_reads)) if os.name != "nt": for path in (script, journal_module, policy_file): os.chmod(path, 0o400) diff --git a/scripts/native_prerequisites.py b/scripts/native_prerequisites.py index 0045e7a..995b189 100644 --- a/scripts/native_prerequisites.py +++ b/scripts/native_prerequisites.py @@ -267,6 +267,20 @@ def isolation_probe(kind: str, *, runner, path: str | None = None) -> tuple[bool denied = base / "denied" secret = base / "secret" secret.write_text("probe") + credential_files = [] + credential_denials = [] + if kind == "srt": + import stage_permissions as sp + home = base / "home" + for store in (*sp.CREDENTIAL_STORES, sp.GH_CREDENTIAL_STORE): + target = home / store[2:] + target.parent.mkdir(parents=True, exist_ok=True) + if store in ("~/.ssh", "~/.aws", sp.GH_CREDENTIAL_STORE): + target.mkdir() + target = target / "probe" + target.write_text("synthetic credential probe") + credential_files.append(target) + credential_denials = list(sp.credential_read_denials(home=home)) with socket.socket() as listener: listener.bind(("127.0.0.1", 0)) listener.listen() @@ -277,6 +291,9 @@ def isolation_probe(kind: str, *, runner, path: str | None = None) -> tuple[bool "except OSError: pass\nelse: sys.exit('write escaped')\n" f"try: pathlib.Path({str(secret)!r}).read_text()\n" "except OSError: pass\nelse: sys.exit('read escaped')\n" + + "".join(f"try: open({str(path)!r}, 'rb').close()\n" + "except OSError: pass\nelse: sys.exit('credential read escaped')\n" + for path in credential_files) + "s=socket.socket(); s.settimeout(1)\n" f"try: s.connect({listener.getsockname()!r})\n" "except OSError: pass\nelse: sys.exit('network escaped')\n" @@ -300,7 +317,8 @@ def isolation_probe(kind: str, *, runner, path: str | None = None) -> tuple[bool settings = base / "settings.json" settings.write_text(json.dumps({ "filesystem": {"allowWrite": [str(allowed)], - "denyWrite": [str(denied)], "denyRead": [str(secret)]}, + "denyWrite": [str(denied)], + "denyRead": [str(secret), *credential_denials]}, "network": {"allowedDomains": [], "deniedDomains": []}, })) argv = [binary, "--settings", str(settings), "--", sys.executable, "-c", script] @@ -319,7 +337,7 @@ def isolation_probe(kind: str, *, runner, path: str | None = None) -> tuple[bool def observe(executor: str, *, recover: bool = True) -> dict[str, Any]: probes = [] for tool in TOOLS: - if (tool == "rg" or tool == "srt" and executor == "claude") and recover and sys.platform.startswith("linux"): + if tool in {"rg", "srt"} and recover and sys.platform.startswith("linux"): record = ensure(tool) else: path = shutil.which(tool) @@ -328,7 +346,7 @@ def observe(executor: str, *, recover: bool = True) -> dict[str, Any]: required = {"rg"} if sys.platform.startswith("linux"): required |= {"bwrap", "socat"} - if executor == "claude": + if executor == "claude" or sys.platform.startswith("linux"): required.add("srt") missing = [probe["name"] for probe in probes if probe["name"] in required and not probe.get("path")] diff --git a/scripts/stage_permissions.py b/scripts/stage_permissions.py index a9eac58..3beff53 100644 --- a/scripts/stage_permissions.py +++ b/scripts/stage_permissions.py @@ -45,6 +45,29 @@ REVIEW_ROLES = frozenset({"review", "rereview"}) ISSUE_REVIEW_ROLE = "issue_review" +CREDENTIAL_STORES = ("~/.ssh", "~/.aws", "~/.codex/auth.json", + "~/.claude/.credentials.json", "~/.claude.json") +GH_CREDENTIAL_STORE = "~/.config/gh" + + +def credential_read_denials(*, publishes: bool = False, + credential_reads: tuple[str, ...] = (), + home: Path | None = None) -> tuple[str, ...]: + """Existing canonical targets only: bwrap cannot mask a symlink (#294). + + Inspect metadata, never credential contents. An absent target needs no + mount, and must not be created by the sandbox as a persistent placeholder. + """ + unknown = set(credential_reads) - {*CREDENTIAL_STORES, GH_CREDENTIAL_STORE} + if unknown: + raise PolicyError("credential_reads must name known credential stores") + root = home if home is not None else Path.home() + stores = (*CREDENTIAL_STORES, *((GH_CREDENTIAL_STORE,) if not publishes else ())) + return tuple(dict.fromkeys( + str(target.resolve(strict=True)) for store in stores + if store not in credential_reads + for target in (root / store[2:],) if target.exists())) + #: Read-only shell tools every stage may use, and nothing else from the shell. READ_SHELL = frozenset({"ls", "cat", "head", "tail", "wc", "grep", "find", "jq"}) #: `find` primaries that run a command, delete, or write a file. @@ -143,6 +166,8 @@ class StagePolicy: commands: tuple[str, ...] = () integrations: tuple[str, ...] = () registry_domains: tuple[str, ...] = () + credential_reads: tuple[str, ...] = () + credential_publishing: bool = False # Host-observed unavailable isolation: retain read/publication/edit guards, # but never allow project code execution without the required sandbox. static_only: bool = False @@ -224,6 +249,13 @@ def requires_confined_verification(self) -> bool: """A reviewer that may run tests must do so in a confined clone.""" return self.role in REVIEW_ROLES and bool(self.verification_commands) + @property + def uses_bash_sandbox(self) -> bool: + """Keep Claude sandbox emission and its preflight on the same scope.""" + return self.active and not self.static_only and bool( + self.requires_confined_verification or self.registry_domains + or self.publication_commands or self.docker) + def to_json(self) -> str: return json.dumps(asdict(self), sort_keys=True) @@ -237,7 +269,7 @@ def from_json(cls, text: str) -> "StagePolicy": if unknown: raise PolicyError(f"unknown stage policy keys: {', '.join(unknown)}") data["verify_prefixes"] = tuple(data.get("verify_prefixes") or ()) - for key in ("tools", "commands", "integrations", "registry_domains"): + for key in ("tools", "commands", "integrations", "registry_domains", "credential_reads"): data[key] = tuple(data.get(key) or ()) return cls(**data) @@ -1533,7 +1565,7 @@ def prompt_note(policy: StagePolicy) -> str: f"available with `{'rtk ' if policy.rtk else ''}printenv TMPDIR`." ) choices = {key: getattr(policy, key) for key in - ("tools", "commands", "integrations", "registry_domains") + ("tools", "commands", "integrations", "registry_domains", "credential_reads") if getattr(policy, key)} if choices: tmpdir_note += " Effective project choices: " + json.dumps(choices) + "." @@ -1681,7 +1713,8 @@ def _handles_tool_check(root: Path) -> bool: #: User-level sandbox settings that would widen a confined review, because #: Claude merges arrays across settings sources. SANDBOX_WIDENING_KEYS = ( - ("filesystem", "allowWrite"), ("filesystem", "disabled"), ("excludedCommands",), + ("filesystem", "allowWrite"), ("filesystem", "allowRead"), + ("filesystem", "disabled"), ("excludedCommands",), ("allowUnsandboxedCommands",), ("network", "allowUnixSockets"), ("network", "allowAllUnixSockets"), ("enableWeakerNestedSandbox",), ) @@ -1830,7 +1863,8 @@ def preapproved_notice(tools, limit: int = NOTICE_LIMIT) -> str | None: def review_sandbox_settings(writable: tuple[str, ...], registry_domains: tuple[str, ...] = (), *, - docker_host: bool = False) -> dict: + docker_host: bool = False, publishes: bool = False, + credential_reads: tuple[str, ...] = ()) -> dict: """Claude's Bash sandbox for a reviewer that verifies in its clone. A sandboxed command has no network route except the allowed domains, and a @@ -1840,7 +1874,9 @@ def review_sandbox_settings(writable: tuple[str, ...], "enabled": True, "failIfUnavailable": True, "allowUnsandboxedCommands": False, - "filesystem": {"allowWrite": list(writable)}, + "filesystem": {"allowWrite": list(writable), + "denyRead": list(credential_read_denials( + publishes=publishes, credential_reads=credential_reads))}, "network": { "allowedDomains": list(dict.fromkeys( (*REVIEW_SANDBOX_DOMAINS, *registry_domains))), diff --git a/scripts/user_choices.py b/scripts/user_choices.py index d044c25..b9dcf5a 100644 --- a/scripts/user_choices.py +++ b/scripts/user_choices.py @@ -7,16 +7,16 @@ import re -from stage_permissions import PolicyError, parse_command +from stage_permissions import CREDENTIAL_STORES, GH_CREDENTIAL_STORE, PolicyError, parse_command -KEYS = frozenset({"tools", "commands", "integrations", "registry_domains"}) +KEYS = frozenset({"tools", "commands", "integrations", "registry_domains", "credential_reads"}) def load(config: dict) -> dict[str, tuple[str, ...]]: value = (config.get("code_cycle") or {}).get("user_choices", {}) if not isinstance(value, dict) or set(value) - KEYS: raise PolicyError("code_cycle.user_choices must be a mapping with only " - "tools, commands, integrations and registry_domains") + "tools, commands, integrations, registry_domains and credential_reads") result = {} for key in KEYS: entries = value.get(key, []) @@ -26,6 +26,8 @@ def load(config: dict) -> dict[str, tuple[str, ...]]: normalized = [] for entry in entries: entry = entry.strip() + if key == "credential_reads" and entry not in (*CREDENTIAL_STORES, GH_CREDENTIAL_STORE): + raise PolicyError("credential_reads must name known credential stores") if key == "tools" and not re.fullmatch(r"[a-zA-Z0-9_./+-]+", entry): raise PolicyError("a selected tool must name one executable") if key == "commands": diff --git a/scripts/windows-test-shards.json b/scripts/windows-test-shards.json index 536dcbf..d26980b 100644 --- a/scripts/windows-test-shards.json +++ b/scripts/windows-test-shards.json @@ -47,6 +47,7 @@ "test_ci_failure_access", "test_cycle_journal", "test_dependency_setup", + "test_credential_reads", "test_driver_characterization", "test_execution_profiles", "test_host_permissions", @@ -76,6 +77,7 @@ ], "windows-2": [ "test_behavioral", + "test_credential_reads", "test_dependency_setup", "test_driver_characterization", "test_environment_recovery", diff --git a/tests/test_codex_git_sandbox.py b/tests/test_codex_git_sandbox.py index 8bc4f99..5bc34a9 100644 --- a/tests/test_codex_git_sandbox.py +++ b/tests/test_codex_git_sandbox.py @@ -121,6 +121,13 @@ def setUp(self) -> None: temporary = tempfile.TemporaryDirectory() self.addCleanup(temporary.cleanup) self.root = Path(temporary.name).resolve() + # Exact Git-rule assertions do not inherit the developer's stores; + # credential profile coverage uses a populated synthetic home separately. + home = self.root / "home" + home.mkdir() + home_patch = patch.object(Path, "home", return_value=home) + home_patch.start() + self.addCleanup(home_patch.stop) environment = patch.dict(os.environ, { "GIT_CONFIG_GLOBAL": os.devnull, "GIT_CONFIG_NOSYSTEM": "1", "GIT_AUTHOR_NAME": "Cycle Test", "GIT_AUTHOR_EMAIL": "cycle@example.invalid", diff --git a/tests/test_credential_reads.py b/tests/test_credential_reads.py new file mode 100644 index 0000000..9793e93 --- /dev/null +++ b/tests/test_credential_reads.py @@ -0,0 +1,253 @@ +"""Credential policy uses metadata and synthetic stores, never host secrets.""" +from __future__ import annotations + +import builtins +import itertools +import json +import os +import re +import subprocess +import sys +import tempfile +import unittest +from pathlib import Path +from unittest.mock import patch + +sys.path.insert(0, str(Path(__file__).resolve().parent.parent / "scripts")) +import dependency_setup as ds +import executors as ex +import stage_permissions as sp +import user_choices +from router import Target + + +def rules(argv): + table = next(arg.split("=", 1)[1] for arg in argv + if re.match(r"permissions\.[a-z_]+\.filesystem=", arg)) + return {json.loads(path): json.loads(access) for path, access in + re.findall(r'("(?:\\.|[^"\\])*")=("\w+")', table)} + + +class CredentialReadsTests(unittest.TestCase): + def setUp(self): + temporary = tempfile.TemporaryDirectory() + self.addCleanup(temporary.cleanup) + self.root = Path(temporary.name).resolve() + self.home = self.root / "home" + self.home.mkdir() + for name in (*sp.CREDENTIAL_STORES, sp.GH_CREDENTIAL_STORE): + target = self.home / name[2:] + target.parent.mkdir(parents=True, exist_ok=True) + if name in ("~/.ssh", "~/.aws", sp.GH_CREDENTIAL_STORE): + target.mkdir() + else: + target.write_text("fixture") + self.patch = patch.object(Path, "home", return_value=self.home) + self.patch.start() + self.addCleanup(self.patch.stop) + + def test_every_codex_profile_denies_stores_and_preserves_clone_and_scratch(self): + target = Target("codex", "openai", "fixture", "medium") + expected = {str(self.home / name[2:]) for name in sp.CREDENTIAL_STORES} + clone = self.root / "clone" + clone.mkdir() + for writes, publishes, registry, docker in itertools.product((False, True), repeat=4): + with self.subTest(writes=writes, publishes=publishes, registry=registry, docker=docker): + argv = ex.CodexAdapter().argv( + target, "task", writes=writes, publishes=publishes, + review_clone=str(clone) if not writes else None, + network_domains=("registry.npmjs.org",) if registry else (), + sandbox_grants=("docker_host",) if docker else ()) + with ex._native_review_scratch(ex.CodexAdapter(), argv, {}, + writes=writes, publishes=publishes, + cwd=str(clone)) as (argv, env): + actual = rules(argv) + self.assertTrue(all(actual[path] == "none" for path in expected)) + self.assertEqual("write", actual[env["TMPDIR"]]) + if not writes: + self.assertEqual("write", actual[str(clone)]) + self.assertEqual(publishes, str(self.home / ".config/gh") not in actual) + + def test_claude_sandbox_applies_denials_for_every_role_when_required(self): + for role in ("implement", "resolve", "review", "rereview", "issue_review", "triage"): + with self.subTest(role=role): + adapter = ex.ClaudeAdapter() + argv = ["claude", "--verbose", "-p", "task"] + policy = sp.StagePolicy(role=role, registry_domains=("registry.npmjs.org",), + credential_publishing=role in + sp.WRITING_ROLES | sp.REVIEW_ROLES) + with adapter.policy_files(argv, policy, scratch=str(self.root), + cwd=str(self.home)) as result: + config = json.loads(result[result.index("--settings") + 1]) + denied = config["sandbox"]["filesystem"]["denyRead"] + self.assertEqual(set(sp.credential_read_denials( + publishes=policy.credential_publishing)), set(denied)) + self.assertTrue(config["sandbox"]["failIfUnavailable"]) + + @unittest.skipIf(os.name == "nt", "symlink fixture") + def test_resolves_symlinks_skips_absent_and_never_reads_contents(self): + original = self.home / ".aws" + original.rmdir() + real = self.root / "real-aws" + real.mkdir() + original.symlink_to(real, target_is_directory=True) + (self.home / ".claude.json").unlink() + with patch.object(Path, "read_bytes", side_effect=AssertionError("credential read")): + denied = sp.credential_read_denials() + self.assertIn(str(real), denied) + self.assertNotIn(str(original), denied) + self.assertNotIn(str(self.home / ".claude.json"), denied) + + def test_opt_out_loaded_from_real_config_validated_and_round_tripped(self): + import run_cycle + config = self.root / ".code-cycle.yml" + config.write_text('code_cycle:\n user_choices:\n credential_reads: ["~/.ssh"]\n') + choices = user_choices.load(run_cycle.load_config(config)) + policy = sp.StagePolicy.from_json(sp.StagePolicy(role="implement", **choices).to_json()) + self.assertEqual(("~/.ssh",), policy.credential_reads) + denied = sp.credential_read_denials(credential_reads=policy.credential_reads) + self.assertNotIn(str(self.home / ".ssh"), denied) + self.assertIn(str(self.home / ".aws"), denied) + self.assertIn('"credential_reads": ["~/.ssh"]', sp.prompt_note(policy)) + config.write_text('code_cycle:\n user_choices:\n credential_reads: ["*"]\n') + with self.assertRaises(run_cycle.CycleDriverError): + run_cycle.load_config(config) + + def test_linux_preparation_uses_identical_srt_policy_for_both_executors(self): + settings = self.root / "settings.json" + with patch.object(ds.sys, "platform", "linux"), patch.object( + ds.native_prerequisites, "executable", return_value="/tools/srt"): + configs = [] + for executor in ("codex", "claude"): + argv = ds.sandbox_command(executor, executor, ["npm", "ci"], + writable=[self.root], protected=[settings], + domains=ds.REGISTRIES, settings=settings) + self.assertEqual("/tools/srt", argv[0]) + configs.append(json.loads(settings.read_text())) + self.assertEqual(configs[0], configs[1]) + self.assertEqual(set(sp.credential_read_denials()), + set(configs[0]["filesystem"]["denyRead"])) + with patch.object(ds.sys, "platform", "linux"), patch.object( + ds.native_prerequisites, "executable", return_value=None): + with self.assertRaisesRegex(OSError, "installed srt"): + ds.sandbox_command("codex", "codex", ["npm", "ci"], writable=[], + protected=[], domains=(), settings=settings) + + def test_unvalidated_platform_codex_fallback_has_same_denials(self): + for platform in ("darwin", "win32"): + with patch.object(ds.sys, "platform", platform), patch.object( + ds.native_prerequisites, "nested_codex", return_value=False): + argv = ds.sandbox_command("codex", "codex", ["npm", "ci"], + writable=[], protected=[], domains=(), + settings=self.root / "settings.json") + self.assertEqual({path: "none" for path in sp.credential_read_denials()}, rules(argv)) + + def test_confinement_probe_rejects_each_missing_credential_denial(self): + self.assert_probe_rejects_each_missing_denial() + + @unittest.skipIf(os.name == "nt", "symlink fixture") + def test_confinement_probe_emulation_follows_a_linked_temporary_root(self): + # A runner's temporary root may be reached through a symlink, as Windows + # runners reach it through an 8.3 short name; denials are canonical. + real = self.root / "real-tmp" + real.mkdir() + link = self.root / "linked-tmp" + link.symlink_to(real, target_is_directory=True) + with patch.object(tempfile, "tempdir", str(link)): + self.assert_probe_rejects_each_missing_denial() + + def assert_probe_rejects_each_missing_denial(self): + # Emulate the declared filesystem inside the generated probe. This is a + # policy/probe regression test, not evidence of real OS confinement. + for missing in (None, *range(6)): + def runner(argv, **options): + table = rules(argv) + denied = [path for path, access in table.items() if access == "none"] + if missing is not None: + del table[denied[missing]] + def sandbox_open(path, mode="r", *args, **kwargs): + resolved = str(Path(path).resolve()) + matching = [(root, access) for root, access in table.items() + if Path(resolved).is_relative_to(Path(root).resolve())] + access = max(matching, key=lambda item: len(item[0]))[1] if matching else "read" + if access == "none" or ("w" in mode and access != "write"): + raise PermissionError("synthetic policy denied") + return builtins.open(path, mode, *args, **kwargs) + try: + exec(argv[-1], {"__builtins__": {**vars(builtins), "open": sandbox_open}}) + except SystemExit as error: + return subprocess.CompletedProcess(argv, 0 if error.code == 0 else 1, + "", str(error.code)) + self.fail("probe did not exit") + ok, detail = ex.CodexAdapter().probe_review_confinement(runner=runner) + self.assertEqual(missing is None, ok, detail) + + def test_inherited_allow_read_cannot_cancel_stage_denials(self): + with patch.object(sp, "_read_settings", return_value={ + "sandbox": {"filesystem": {"allowRead": [str(self.home)]}}}): + self.assertIn("allowRead", sp.sandbox_override()) + + def test_opt_out_is_reported_in_dispatch_evidence(self): + target = Target("codex", "openai", "fixture", "medium") + with patch.object(ex.native, "observe", return_value={ + "probes": [], "missing": [], "status": "observed"}): + result = ex.CodexAdapter().dispatch( + target, "inspect", cwd=str(self.home), + command_policy=sp.StagePolicy(role="triage", credential_reads=("~/.aws",)), + runner=lambda argv, **options: subprocess.CompletedProcess( + argv, 0, '{"type":"turn.completed"}', "")) + self.assertEqual(["~/.aws"], result.artifacts["credential_read_policy"]["exceptions"]) + self.assertNotIn(str(self.home / ".aws"), + result.artifacts["credential_read_policy"]["denied_paths"]) + + @unittest.skipUnless(os.environ.get("CODE_CYCLE_LIVE_CREDENTIAL_PROBE") == "1" + and not ds.native_prerequisites.nested_codex(), + "host-only E7 probe; nested native sandboxes are not supported") + def test_live_e7_stage_profiles_on_both_executors(self): + """Opt-in host boundary check; prints states only, with synthetic stores.""" + import shutil + target = Target("codex", "openai", "fixture", "medium") + files = [] + for store in (*sp.CREDENTIAL_STORES, sp.GH_CREDENTIAL_STORE): + path = self.home / store[2:] + if path.is_dir(): + path = path / "probe" + path.write_text("synthetic credential probe") + files.append((store, path)) + for executor, writes, publishes, registry, docker in itertools.product( + ("codex", "claude"), (False, True), (False, True), (False, True), (False, True)): + with self.subTest(executor=executor, writes=writes, publishes=publishes, + registry=registry, docker=docker): + script = "import sys\nfailed=[]\n" + "\n".join( + f"try:\n open({str(path)!r}, 'rb').close(); readable=True\n" + "except OSError:\n readable=False\n" + f"print({store!r}, 'READABLE' if readable else 'denied')\n" + f"if readable != {publishes and store == sp.GH_CREDENTIAL_STORE}: failed.append({store!r})" + for store, path in files) + "\nsys.exit('; '.join(failed) if failed else 0)" + domains = ("registry.npmjs.org",) if registry else () + if executor == "codex": + binary = shutil.which("codex") + self.assertIsNotNone(binary, "host E7 requires codex") + argv = ex.CodexAdapter().argv(target, "probe", writes=writes, + publishes=publishes, network_domains=domains, + sandbox_grants=("docker_host",) if docker else (), + review_clone=str(self.root) if not writes else None) + profile = next(arg.split("=", 1)[1].strip('"') for arg in argv + if arg.startswith("default_permissions=")) + command = [binary, "sandbox", "--permission-profile", profile] + for index, arg in enumerate(argv): + if arg == "-c" and argv[index + 1].startswith(("permissions.", "features.")): + command.extend((arg, argv[index + 1])) + command.extend(("--", sys.executable, "-c", script)) + else: + binary = shutil.which("srt") + self.assertIsNotNone(binary, "host E7 requires srt") + sandbox = sp.review_sandbox_settings((str(self.root),), domains, + publishes=publishes, docker_host=docker)["sandbox"] + settings = self.root / "live-srt.json" + settings.write_text(json.dumps({key: sandbox[key] for key in ("filesystem", "network")})) + command = [binary, "--settings", str(settings), "--", sys.executable, "-c", script] + completed = subprocess.run(command, cwd=self.root, capture_output=True, + text=True, timeout=30) + print(executor, writes, publishes, registry, docker, completed.stdout) + self.assertEqual(0, completed.returncode, completed.stderr) diff --git a/tests/test_dependency_setup.py b/tests/test_dependency_setup.py index fabc019..ab21bbc 100644 --- a/tests/test_dependency_setup.py +++ b/tests/test_dependency_setup.py @@ -20,9 +20,20 @@ import user_choices from router import Target +#: The real resolver, for the test that supplies its own managed srt bundle. +RESOLVE_EXECUTABLE = ds.native_prerequisites.executable + class SetupTests(unittest.TestCase): def setUp(self): + nested = patch.object(ds.native_prerequisites, "nested_codex", return_value=False) + nested.start() + self.addCleanup(nested.stop) + # Linux preparation runs srt for both executors (#294); a host or CI + # runner without it must not change these results. + srt = patch.object(ds.native_prerequisites, "executable", return_value="/tools/srt") + srt.start() + self.addCleanup(srt.stop) self.temp = tempfile.TemporaryDirectory() self.addCleanup(self.temp.cleanup) # prepare() canonicalizes the checkout; Windows temp paths may be 8.3 short names. @@ -38,6 +49,7 @@ def setUp(self): self.git("-c", "user.name=Test", "-c", "user.email=test@example.invalid", "commit", "-qm", "fixture") self.calls = [] + self.sandbox_configs = {} self.failures = [] def git(self, *args): @@ -51,8 +63,10 @@ def package(self, directory, scripts=None): def runner(self, argv, **options): self.calls.append((argv, options)) - self.assertEqual("codex", argv[0]) - self.assertIn("sandbox", argv) + if "--settings" in argv: + self.sandbox_configs[id(argv)] = json.loads( + Path(argv[argv.index("--settings") + 1]).read_text()) + self.assertTrue(argv[0] == "codex" or argv[0].endswith("srt")) command = argv[argv.index("--") + 1:] if command[-1] == "--version": return subprocess.CompletedProcess(argv, 0, "9.1.0\n", "") @@ -63,6 +77,14 @@ def runner(self, argv, **options): return subprocess.CompletedProcess(argv, 0, "installed\n", "") def filesystem(self, argv): + if "--settings" in argv: + config = self.sandbox_configs.get(id(argv)) + if config is None: + config = json.loads(Path(argv[argv.index("--settings") + 1]).read_text()) + filesystem = config["filesystem"] + return "{" + ",".join(json.dumps(path) + '=' + json.dumps(access) + for key, access in (("allowWrite", "write"), ("denyWrite", "read")) + for path in filesystem[key]) + "}" return next(arg for arg in argv if arg.startswith( "permissions.code_cycle_prepare.filesystem=")) @@ -109,7 +131,8 @@ def test_claude_preparation_passes_managed_rg_to_the_process(self): srt.chmod(0o755) previous_path = os.environ.get("PATH") with patch.object(Path, "home", return_value=home), patch.dict( - os.environ, {"TEST_HOST_CREDENTIAL": "fixture"}): + os.environ, {"TEST_HOST_CREDENTIAL": "fixture"}), patch.object( + ds.native_prerequisites, "executable", RESOLVE_EXECUTABLE): selected_path = ds.native_prerequisites.tool_path(os.environ.get("PATH", ""), probes=[ {"name": tool, "path": str(tools / tool / tool), "outcome": "reused", "executable": tool} for tool in ("srt", "rg")]) @@ -158,7 +181,8 @@ def test_retries_transient_failure_records_head_logs_and_package_outputs(self): self.assertTrue(self.writable(argv)) self.assertFalse([path for path in self.writable(argv) if path.is_relative_to(self.root)]) self.assertFalse(Path(options["cwd"]).is_relative_to(self.root)) - self.assertIn("registry.npmjs.org", " ".join(argv)) + if "--settings" not in argv: + self.assertIn("registry.npmjs.org", " ".join(argv)) self.assertEqual("true", options["env"]["npm_config_ignore_scripts"]) self.assertNotIn("GITHUB_TOKEN", options["env"]) @@ -269,6 +293,7 @@ def runner(argv, **options): self.assertIn("without required dependency outputs", result["items"][0]["error"]) self.assertFalse((self.root / "node_modules").exists()) + @patch.object(ds.sys, "platform", "darwin") def test_large_tracked_source_tree_does_not_inflate_codex_filesystem_argument(self): # This source listing alone exceeded Linux's 128 KiB argv-entry limit. source = self.root / "src" @@ -291,6 +316,7 @@ def test_large_tracked_source_tree_does_not_inflate_codex_filesystem_argument(se self.assertIn('permissions.code_cycle_prepare.extends=":read-only"', argv) self.assertEqual("tracked plugin", protected.read_text()) + @patch.object(ds.sys, "platform", "darwin") def test_codex_protects_tracked_ancestors_of_writable_outputs(self): ancestor = self.root / ".yarn" argv = ds.sandbox_command( @@ -324,7 +350,8 @@ def test_nonzero_install_crosses_a_real_process_boundary(self): "if sys.argv[-1] == '--version': print('9.1.0')\n" "else:\n print('fixture install failed', file=sys.stderr)\n sys.exit(12)\n") binary.chmod(0o700) - result = ds.prepare(str(self.root), "codex", str(binary), evidence_root=self.evidence) + with patch.object(ds.native_prerequisites, "executable", return_value=str(binary)): + result = ds.prepare(str(self.root), "codex", str(binary), evidence_root=self.evidence) self.assertEqual("incomplete", result["status"]) attempt = result["items"][0]["attempts"][-1] self.assertEqual(12, attempt["exit_code"]) @@ -342,6 +369,9 @@ def lifecycle_fixture(self, build, ignored=()): def runner(argv, **options): command = argv[argv.index("--") + 1:] calls.append((command, argv, options)) + if "--settings" in argv: + self.sandbox_configs[id(argv)] = json.loads( + Path(argv[argv.index("--settings") + 1]).read_text()) cwd = Path(options["cwd"]) if command[-1] == "--version": return subprocess.CompletedProcess(argv, 0, "9.1.0\n", "") @@ -515,14 +545,14 @@ def test_symlink_outputs_never_grant_writes_outside_checkout(self): self.assertFalse(self.calls) def test_both_sandbox_wrappers_disable_general_network_and_protect_lock(self): - with patch.object(ds.shutil, "which", return_value="/usr/bin/srt"): + with patch.object(ds.native_prerequisites, "executable", return_value="/usr/bin/srt"): for executor in ("claude", "codex"): settings = Path(self.temp.name) / "settings.json" argv = ds.sandbox_command(executor, executor, ["npm", "ci", "--ignore-scripts"], writable=[self.root / "node_modules"], protected=[self.root / "pnpm-lock.yaml"], domains=ds.REGISTRIES, settings=settings) - if executor == "claude": + if executor == "claude" or sys.platform.startswith("linux"): config = json.loads(settings.read_text()) self.assertEqual(list(ds.REGISTRIES), config["network"]["allowedDomains"]) self.assertIn(str(self.root / "pnpm-lock.yaml"), config["filesystem"]["denyWrite"]) @@ -583,7 +613,8 @@ def runner(argv, **options): self.assertEqual("prepared", report["status"]) self.assertEqual(source_head, report["items"][0]["head_sha"]) return result - with patch.object(ds.shutil, "which", return_value=str(binary)): + with patch.object(ds.shutil, "which", return_value=str(binary)), patch.object( + ds.native_prerequisites, "executable", return_value=str(binary)): result = ex._isolated_review_dispatch( adapter, target, role, cwd=str(self.root), dispatch_call=review, command_policy=True) diff --git a/tests/test_executors.py b/tests/test_executors.py index ffc1cb0..67f0033 100644 --- a/tests/test_executors.py +++ b/tests/test_executors.py @@ -850,7 +850,7 @@ def test_codex_read_only_sandbox_already_reads_it_and_keeps_the_task_last(self) read_dirs=("/tmp/code-cycle-diff-x",)) self.assertNotIn("--add-dir", argv) - self.assertIn('permissions.code_cycle_review_read.extends=":read-only"', argv) + self.assertIn('permissions.code_cycle_read.extends=":read-only"', argv) self.assertEqual("review", argv[-1]) @@ -1215,8 +1215,7 @@ def claude(self, **kw): def test_a_writing_stage_may_write_where_it_was_sent(self) -> None: argv = self.codex(writes=True) - self.assertIn("-s", argv) - self.assertEqual("workspace-write", argv[argv.index("-s") + 1]) + self.assertIn('permissions.code_cycle_write.extends=":workspace"', argv) def test_codex_publish_profile_keeps_review_read_only_and_limits_network(self) -> None: argv = ex.CodexAdapter().argv(TARGET, "review", writes=False, publishes=True) @@ -1252,8 +1251,7 @@ def test_no_codex_profile_opens_unix_sockets_without_a_grant(self) -> None: argv = self.codex(**options) self.assertFalse(any("unix_sockets" in arg for arg in argv)) self.assertFalse(any("docker" in arg for arg in argv)) - # A writer without a grant keeps the legacy workspace sandbox. - self.assertIn("workspace-write", self.codex(writes=True)) + self.assertIn('permissions.code_cycle_write.extends=":workspace"', self.codex(writes=True)) def test_docker_grant_allows_unix_sockets_and_keeps_publication_domains(self) -> None: """A filesystem rule on the socket crashes Codex's Linux helper (#239).""" @@ -1358,10 +1356,10 @@ def test_a_reading_stage_says_so_rather_than_relying_on_a_default(self) -> None: """A default that changes is a permission nobody chose.""" argv = self.codex(writes=False) - self.assertEqual("read-only", argv[argv.index("-s") + 1]) + self.assertIn('permissions.code_cycle_read.extends=":read-only"', argv) def test_asking_for_nothing_asks_for_the_smaller_permission(self) -> None: - self.assertEqual("read-only", self.codex()[self.codex().index("-s") + 1]) + self.assertIn('permissions.code_cycle_read.extends=":read-only"', self.codex()) def test_no_argument_list_ever_asks_for_full_access(self) -> None: """The CLIs offer it; nothing here builds a command that requests it. @@ -1780,7 +1778,7 @@ def runner(argv, timeout=None, cwd=None, env=None): self.assertEqual(ex.DispatchOutcome.SUCCEEDED, result.outcome) self.assertEqual("enforced", result.artifacts["read_only_mode"]) - self.assertIn('permissions.code_cycle_review_read.extends=":read-only"', seen["argv"]) + self.assertIn('permissions.code_cycle_read.extends=":read-only"', seen["argv"]) def test_disposable_dispatch_requires_an_isolated_workspace(self) -> None: target = router.parse_target("codex:openai/gpt-6-sol high") @@ -2598,8 +2596,9 @@ def runner(argv, timeout=None, cwd=None, env=None): if adapter.name == "codex": self.assertIn(f'shell_environment_policy.set.TMPDIR={json.dumps(str(scratch))}', argv) if writes and not publishes and not grants: - self.assertIn(str(scratch), argv) - self.assertIn("workspace-write", argv) + self.assertTrue(any(f'{json.dumps(str(scratch))}="write"' in arg + for arg in argv)) + self.assertIn('permissions.code_cycle_write.extends=":workspace"', argv) else: self.assertTrue(any(f'{json.dumps(str(scratch))}="write"' in arg for arg in argv)) @@ -2835,7 +2834,7 @@ def test_a_codex_stage_that_changes_nothing_stays_enforced(self) -> None: self.assertEqual(result.artifacts["workspace_fingerprint_before"], result.artifacts["workspace_fingerprint_after"]) argv = self.argv_file.read_text(encoding="utf-8").split("\0") - self.assertIn('permissions.code_cycle_review_read.extends=":read-only"', argv) + self.assertIn('permissions.code_cycle_read.extends=":read-only"', argv) self.assertFalse({"--dangerously-bypass-approvals-and-sandbox", "--yolo", "--full-auto"} & set(argv)) @@ -2918,11 +2917,11 @@ def test_codex_grants_only_its_scratch_directory_and_preserves_publication(self) observed = [] def capture(argv, timeout=None, cwd=None, env=None): - profile = "code_cycle_publish_read" if publishes else "code_cycle_review_read" + profile = "code_cycle_publish_read" if publishes else "code_cycle_read" fs = next(option for option in argv if option.startswith(f"permissions.{profile}.filesystem=")) - self.assertEqual(f'permissions.{profile}.filesystem={{' - f'{json.dumps(env["TMPDIR"])}="write"}}', fs) + self.assertIn(f'{json.dumps(env["TMPDIR"])}="write"', fs) + self.assertEqual(1, fs.count('="write"')) self.assertIn(f'permissions.{profile}.extends=":read-only"', argv) self.assertIn(f'shell_environment_policy.set.TMPDIR={json.dumps(env["TMPDIR"])}', argv) self.assertNotIn("--add-dir", argv) @@ -2946,8 +2945,10 @@ def test_a_granted_publishing_review_keeps_its_scratch_grant(self) -> None: observed = [] def capture(argv, timeout=None, cwd=None, env=None): - self.assertEqual([f'{rule}{{{json.dumps(env["TMPDIR"])}="write"}}'], - [option for option in argv if option.startswith(rule)]) + tables = [option for option in argv if option.startswith(rule)] + self.assertEqual(1, len(tables)) + self.assertIn(f'{json.dumps(env["TMPDIR"])}="write"', tables[0]) + self.assertEqual(1, tables[0].count('="write"')) self.assertIn("permissions.code_cycle_publish_read.network." "dangerously_allow_all_unix_sockets=true", argv) observed.append(env["TMPDIR"]) diff --git a/tests/test_native_prerequisites.py b/tests/test_native_prerequisites.py index 5dcae59..f0f5c97 100644 --- a/tests/test_native_prerequisites.py +++ b/tests/test_native_prerequisites.py @@ -239,12 +239,12 @@ def test_host_rg_does_not_switch_verification_python_at_process_boundary(self): self.assertEqual(0, result.returncode, result.stderr) self.assertEqual("venv", result.stdout.strip()) - def test_codex_does_not_attempt_srt_recovery(self): + def test_both_executors_recover_srt_for_linux_preparation(self): for executor in ("codex", "claude"): with self.subTest(executor=executor), patch.object(native.sys, "platform", "linux"), patch.object( native, "ensure", return_value={"path": None, "outcome": "not_verified"}) as ensure: native.observe(executor) - self.assertEqual(["rg"] if executor == "codex" else ["srt", "rg"], + self.assertEqual(["srt", "rg"], [call.args[0] for call in ensure.call_args_list]) def test_known_failure_rejects_unrelated_restrictions(self): @@ -335,7 +335,8 @@ def test_selected_system_rg_is_not_shadowed_by_stale_managed_rg(self): def test_nested_preparation_is_classified_without_a_socket_or_process_probe(self): with patch.dict(os.environ, {"CODE_CYCLE_STAGE_EXECUTOR": "codex"}), patch.object( - native.subprocess, "run", side_effect=AssertionError("launched nested process")): + native.subprocess, "run", side_effect=AssertionError("launched nested process")), patch.object( + native.sys, "platform", "darwin"): with self.assertRaisesRegex(OSError, "not_verified without probing"): dependency_setup.sandbox_command( "codex", "codex", ["npm", "ci"], writable=[], protected=[], diff --git a/tests/test_permission_view.py b/tests/test_permission_view.py index 66a9a0a..1353f0f 100644 --- a/tests/test_permission_view.py +++ b/tests/test_permission_view.py @@ -320,7 +320,8 @@ def test_the_cli_supplies_no_choices_when_the_configuration_was_not_read(self): # With a file it supplies what the file says, even when that is nothing. self.assertEqual(("rg",), self.passed_choices( config="code_cycle:\n user_choices:\n tools: [rg]\n")["tools"]) - self.assertEqual({"tools": (), "commands": (), "integrations": (), "registry_domains": ()}, + self.assertEqual({"tools": (), "commands": (), "integrations": (), "registry_domains": (), + "credential_reads": ()}, self.passed_choices(config="code_cycle: {}\n")) self.assertIsNone(self.passed_choices("--no-config")) diff --git a/tests/test_stage_dispatch.py b/tests/test_stage_dispatch.py index 117f2ce..0dd3759 100644 --- a/tests/test_stage_dispatch.py +++ b/tests/test_stage_dispatch.py @@ -89,6 +89,62 @@ def policy(self, role="implement", **fields) -> sp.StagePolicy: class DependencyPreparationDispatchTests(DispatchTestCase): + def test_preparation_registry_grant_gets_final_sandbox_configuration_check(self): + for status in ("prepared", "incomplete"): + with self.subTest(status=status), mock.patch.object( + ex.dependency_setup, "prepare", return_value={"status": status, "items": []}), mock.patch.object( + sp, "sandbox_override", return_value="filesystem.allowRead could cancel denials") as override: + runner = mock.Mock(side_effect=AssertionError("dispatched with inherited allowRead")) + result = ex.ClaudeAdapter().dispatch( + CLAUDE, "review", cwd=str(self.repo), + command_policy=self.policy("review"), runner=runner) + self.assertEqual(ex.DispatchOutcome.FAILED, result.outcome, result.detail) + self.assertIn("allowRead", result.detail) + override.assert_called_once() + runner.assert_not_called() + + def test_preparation_registry_grant_gets_functional_probe_and_static_fallback(self): + for status in ("prepared", "incomplete"): + with self.subTest(status=status): + evidence = {"missing": [], "probes": [], "status": "observed"} + seen = [] + def runner(argv, **options): + seen.append(settings_of(argv)) + self.assertIn("not_verified", argv[argv.index("-p") + 1]) + return completed() + with (mock.patch.object(ex, "_run", side_effect=runner) as launch, + mock.patch.object(ex.native, "observe", return_value=evidence), + mock.patch.object(ex.native, "isolation_probe", side_effect=lambda kind, **kw: + (kind != "bwrap", "fixture isolation")) as probe, + mock.patch.object(ex.dependency_setup, "prepare", return_value={"status": status, "items": []}), + mock.patch.object(ex.sys, "platform", "linux")): + result = ex.ClaudeAdapter().dispatch( + CLAUDE, "review", cwd=str(self.repo), + command_policy=self.policy("review"), runner=launch) + self.assertEqual(ex.DispatchOutcome.SUCCEEDED, result.outcome, result.detail) + self.assertEqual(["srt", "bwrap"], [call.args[0] for call in probe.call_args_list]) + self.assertEqual("static_only", result.artifacts["native_sandbox_preflight"]["mode"]) + self.assertNotIn("sandbox", seen[0]) + + def test_docker_only_sandbox_uses_functional_preflight_and_static_fallback(self): + evidence = {"missing": [], "probes": [], "status": "observed"} + seen = [] + def runner(argv, **options): + seen.append(settings_of(argv)) + return completed() + with (mock.patch.object(ex, "_run", side_effect=runner) as launch, + mock.patch.object(ex.native, "observe", return_value=evidence), + mock.patch.object(ex.native, "isolation_probe", return_value=(False, "namespace unavailable")) as probe, + mock.patch.object(ex.sys, "platform", "linux")): + result = ex.ClaudeAdapter().dispatch( + CLAUDE, "inspect", cwd=str(self.repo), + command_policy=self.policy("issue_review", docker=True), runner=launch) + self.assertEqual(ex.DispatchOutcome.SUCCEEDED, result.outcome, result.detail) + self.assertEqual(["bwrap", "srt"], [call.args[0] for call in probe.call_args_list]) + self.assertEqual("static_only", result.artifacts["native_sandbox_preflight"]["mode"]) + self.assertNotIn("sandbox", seen[0]) + self.assertEqual("not_verified", result.artifacts["docker_host_probe"]["status"]) + def test_claude_selected_prerequisites_reach_all_consumers(self): evidence = {"missing": [], "status": "observed", "probes": [ {"name": "srt", "path": "/tools/srt", "outcome": "reused"}, @@ -172,6 +228,7 @@ def runner(argv, **options): if expected: prepare.assert_called_once_with(str(self.repo), executor, executor, domains=("packages.example.org",), + credential_reads=(), prerequisite_path=os.environ.get("PATH", "")) self.assertEqual(report, result.artifacts["dependency_preparation"]) else: @@ -293,6 +350,40 @@ def runner(argv, **options): class ClaudePolicyTests(DispatchTestCase): + def test_inactive_policies_do_not_gain_sandbox_configuration_preflight(self): + (self.claude_home / "settings.json").write_text( + json.dumps({"sandbox": {"filesystem": {"allowRead": ["/custom"]}}}), + encoding="utf-8") + for fields in ({"code_host": "bitbucket"}, {"issue_provider": "plane"}, + {"issue_provider": "jira"}): + with self.subTest(fields=fields): + seen = [] + def runner(argv, **options): + seen.append(settings_of(argv)) + return completed() + result = ex.ClaudeAdapter().dispatch( + CLAUDE, "implement", cwd=str(self.repo), writes=True, publishes=True, + command_policy=self.policy(**fields), runner=runner) + self.assertEqual(ex.DispatchOutcome.SUCCEEDED, result.outcome, result.detail) + self.assertNotIn("sandbox", seen[0]) + + def test_issue_review_sandbox_is_only_added_and_checked_when_required(self): + with mock.patch.object(sp, "sandbox_override", return_value="widened sandbox") as override: + for fields, expected in (({}, False), + ({"registry_domains": ("registry.npmjs.org",)}, True), + ({"docker": True}, True), + ({"static_only": True, "docker": True}, False)): + with self.subTest(fields=fields): + policy = self.policy("issue_review", **fields) + override.reset_mock() + reason = ex.ClaudeAdapter().policy_preflight(policy, str(self.repo)) + self.assertEqual("widened sandbox" if expected else None, reason) + self.assertEqual(expected, override.called) + argv = ["claude", "--verbose", "-p", "task"] + with ex.ClaudeAdapter().policy_files( + argv, policy, scratch=str(self.base), cwd=str(self.repo)) as result: + self.assertEqual(expected, "sandbox" in settings_of(result)) + def test_project_choices_reach_both_barriers_and_private_registry_profile(self) -> None: def runner(argv, timeout=None, cwd=None, env=None): settings = settings_of(argv) @@ -695,7 +786,10 @@ def test_a_review_profile_writes_only_its_clone(self) -> None: ignore_rules=True) self.assertIn('default_permissions="code_cycle_review_read"', argv) self.assertIn('permissions.code_cycle_review_read.extends=":read-only"', argv) - self.assertIn('permissions.code_cycle_review_read.filesystem={"/clone"="write"}', argv) + filesystem = next(option for option in argv if option.startswith( + "permissions.code_cycle_review_read.filesystem=")) + self.assertIn('"/clone"="write"', filesystem) + self.assertEqual(1, filesystem.count('="write"')) self.assertNotIn("workspace-write", argv) self.assertIn("--ignore-rules", argv)