diff --git a/.github/workflows/adapter_tests.yml b/.github/workflows/adapter_tests.yml index abff7f20..48845f1d 100644 --- a/.github/workflows/adapter_tests.yml +++ b/.github/workflows/adapter_tests.yml @@ -70,7 +70,7 @@ jobs: - name: Install the adapters, the NAT extra and the test group from the lockfile run: uv sync --locked --python 3.12 --extra nat --group test - name: Run every suite - run: uv run python run_conformance.py claude cursor nat + run: uv run python run_conformance.py claude codex cursor nat tests-without-nat: # The documented setup for Claude Code and Cursor users, who never install NAT. @@ -89,8 +89,8 @@ jobs: version: "0.9.9" - name: Install the adapters and the test group, without the NAT extra run: uv sync --locked --python 3.12 --group test - - name: Run the shared and Claude Code and Cursor suites - run: uv run python run_conformance.py claude cursor + - name: Run the shared, Claude Code, Codex and Cursor suites + run: uv run python run_conformance.py claude codex cursor - name: Check that imports flow integrations, client, core run: uv run lint-imports diff --git a/CONTRIBUTING.md b/CONTRIBUTING.md index aa5a3693..f8a51b1e 100644 --- a/CONTRIBUTING.md +++ b/CONTRIBUTING.md @@ -115,7 +115,7 @@ For changes under `adapters/` (the reference adapters, example Guardian, and the ```bash cd adapters # its own uv project and lockfile uv sync --locked --extra nat --group test -uv run python run_conformance.py claude cursor nat # shared checks + those adapter suites +uv run python run_conformance.py claude codex cursor nat # shared checks + those adapter suites ``` Without the NAT runtime, sync without `--extra nat` and leave `nat` off the command. The command reports passes, skips, and failures separately, and fails on any failure or unexpected skip. The same workflow also builds the package and checks the installed wheel outside the repository (`adapters/scripts/check_wheel.py`). See `adapters/README.md` for what the suite proves and its scope. diff --git a/adapters/README.md b/adapters/README.md index fa117d4c..200cd546 100644 --- a/adapters/README.md +++ b/adapters/README.md @@ -21,7 +21,7 @@ Running the Guardian — terminal window, `launchd`, `systemd`, container — is Known limitations kept in view: Cursor's real-framework wiring is a manual GUI procedure, and Wrapped MCP is validated for namespace shape only, not full wrapping. ACS-Core requires `protocols/MCP/*` for any session that involves MCP and lets only MCP `tools/call` collapse into the generic tool hooks. These adapters perform that collapse and wrap no other MCP surface, so a session that uses MCP falls short of ACS-Core there. Issue #33 tracks the reference-stack side. The adapters also emit no Trace events and do not claim ACS-Trace: `ACS_AUDIT` lines are local audit records, not OpenTelemetry or OCSF output. -One authoritative command runs the shared Guardian checks, shared-library tests, and the adapter suites you name. At least one platform is required; there is no implicit run-all default. Name one, two, or all three platforms in any order: +One authoritative command runs the shared Guardian checks, shared-library tests, and the adapter suites you name. At least one platform is required; there is no implicit run-all default. Name one or more platforms in any order: ```bash cd adapters @@ -35,7 +35,7 @@ uv run python run_conformance.py claude cursor # All platforms. NAT needs the nat extra, or its skips count as # unexpected and fail. -uv run python run_conformance.py claude cursor nat +uv run python run_conformance.py claude codex cursor nat ``` `claude-code` is also accepted as an alias for `claude`. Repeating a platform does not run it twice. The Guardian, shared-library (`_common`), `core`, `client` and `package` checks run once for every command, regardless of the platform selection. @@ -63,7 +63,7 @@ Breaking any adapter's method mapping, envelope/payload field, or signing turns The adapters are **translators**. Each one speaks its framework's hook protocol on one side and ACS JSON-RPC on the other. The framework's agent code is untouched. The Guardian's policy code is untouched. The adapter is the bilingual layer between them. -### The general pattern (same for all three adapters) +### The general pattern (same for every adapter) For each event the framework fires: @@ -182,20 +182,20 @@ Same as above, but with `command: "rm -rf /home/u"`: - **Step 5:** Adapter emits `{"hookSpecificOutput": {"hookEventName": "PreToolUse", "permissionDecision": "deny", "permissionDecisionReason": "destructive Bash pattern in: rm -rf /home/u", "additionalContext": "ACS blocked this tool call: destructive Bash pattern in: rm -rf /home/u"}, "systemMessage": "ACS blocked this tool call: destructive Bash pattern in: rm -rf /home/u"}` - **Step 6:** Claude Code reads `permissionDecision: "deny"`, does not execute the Bash tool, and surfaces the reason: *"The command was blocked: a policy denied the Bash tool call, so it never ran."* -### What changes across the three adapters +### What changes across the adapters The general pattern is identical. The framework-specific translation differs: -| | Claude Code | Cursor | NAT | -|---|---|---|---| -| **Where the adapter lives** | Separate shell process spawned per hook | Separate shell process spawned per hook | In-process Python class, same memory space as the agent | -| **How the framework sends the event** | JSON on stdin; event type is a field inside the JSON (`hook_event_name`) | JSON on stdin; event type passed as a CLI argument (one command per event in `hooks.json`) | Python method call: `pre_invoke(context)` with `context.function_context.name` | -| **Native event field names** | `tool_name`, `tool_input`, `tool_response` | `tool_name`, `tool_input`, `tool_output`, `command` (for shell) | `context.function_context.name`, the input in `context.modified_args[0]` or `context.modified_kwargs` | -| **Native allow/deny output** | Nothing on allow; `{"hookSpecificOutput": {"permissionDecision": "deny"}}` on stdout to block | `{}` on allow for a tool call, `{"permission": "deny"}` on stdout to block, `{"continue": false}` for a prompt, exit code 0 | Raise `ACSGuardianDenied`, or set `context.action = InvocationAction.SKIP` where NAT has it | -| **Native modify mechanism** | `hookSpecificOutput.updatedInput` with `permissionDecision: "ask"` | `updated_input` | Rewrite the input in `context.modified_args[0]` when it is passed positionally, else in `context.modified_kwargs`; replace `context.output` for output | -| **Process model** | OS spawns a Python process for every hook event | OS spawns a Python process for every hook event | Zero IPC; everything in the same Python interpreter | +| | Claude Code | Codex | Cursor | NAT | +|---|---|---|---|---| +| **Where the adapter lives** | Separate shell process spawned per hook | Separate shell process spawned per hook | Separate shell process spawned per hook | In-process Python class, same memory space as the agent | +| **How the framework sends the event** | JSON on stdin; event type is a field inside the JSON (`hook_event_name`) | JSON on stdin; event type is a field inside the JSON (`hook_event_name`); only `PreToolUse` is mapped | JSON on stdin; event type passed as a CLI argument (one command per event in `hooks.json`) | Python method call: `pre_invoke(context)` with `context.function_context.name` | +| **Native event field names** | `tool_name`, `tool_input`, `tool_response` | `tool_name`, `tool_input`, `tool_use_id`, `turn_id` | `tool_name`, `tool_input`, `tool_output`, `command` (for shell) | `context.function_context.name`, the input in `context.modified_args[0]` or `context.modified_kwargs` | +| **Native allow/deny output** | Nothing on allow; `{"hookSpecificOutput": {"permissionDecision": "deny"}}` on stdout to block | Nothing on allow; `{"hookSpecificOutput": {"permissionDecision": "deny"}}` on stdout to block | `{}` on allow for a tool call, `{"permission": "deny"}` on stdout to block, `{"continue": false}` for a prompt, exit code 0 | Raise `ACSGuardianDenied`, or set `context.action = InvocationAction.SKIP` where NAT has it | +| **Native modify mechanism** | `hookSpecificOutput.updatedInput` with `permissionDecision: "ask"` | `hookSpecificOutput.updatedInput` with `permissionDecision: "allow"`, since Codex has no `ask` | `updated_input` | Rewrite the input in `context.modified_args[0]` when it is passed positionally, else in `context.modified_kwargs`; replace `context.output` for output | +| **Process model** | OS spawns a Python process for every hook event | OS spawns a Python process for every hook event | OS spawns a Python process for every hook event | Zero IPC; everything in the same Python interpreter | -The Guardian-side wire format is **the same** for all three. The adapter is bilingual: it knows the framework's protocol on one side and ACS on the other. +The Guardian-side wire format is **the same** for every adapter. The adapter is bilingual: it knows the framework's protocol on one side and ACS on the other. ### Decision honoring is a framework property @@ -209,7 +209,7 @@ ACS standardizes the wire format and the decision contract. Adapters live where 2. Knows ACS (always the same). 3. Translates between them. -The framework's agent code is untouched. The Guardian's policy code is untouched. The adapter is the bilingual translator that makes them speak. **One Guardian, one ACS contract, three adapters that translate three different protocols into that contract.** Add a new framework, write a new adapter, the Guardian doesn't change. +The framework's agent code is untouched. The Guardian's policy code is untouched. The adapter is the bilingual translator that makes them speak. **One Guardian, one ACS contract, and one adapter per framework protocol, each translating into that contract.** Add a new framework, write a new adapter, the Guardian doesn't change. --- @@ -225,9 +225,9 @@ adapters/ │ ├── core/ # no I/O: envelopes, signing, decisions, error codes, bundled schemas │ ├── client/ # I/O: settings and fail posture, the secret, the handshake, the round trip, session state and turns, audit │ ├── integrations// -│ │ ├── hook.py # the installed hook command (Claude Code, Cursor) +│ │ ├── hook.py # the installed hook command (Claude Code, Codex, Cursor) │ │ ├── adapter.py # the hook, or the NAT middleware -│ │ └── wire.py # writes the framework's own config +│ │ └── wire.py # writes the framework's own config (not Codex yet) │ ├── guardian/example_guardian.py # the example Guardian: wire handling, replay, skew and the session chain │ ├── examples/destructive_bash.py # the example Guardian's toy policy │ └── testing/ # CaptureGuardian and the shared test harness @@ -238,7 +238,7 @@ adapters/ Imports flow `integrations → client → core`, and none of the three imports `guardian`, `examples` or `testing`; `uv run lint-imports` checks both rules in CI. -Installing the package (`uv tool install ./adapters`, or `pip install ./adapters`) puts these commands on your PATH: `acs-claude-code-hook`, `acs-cursor-hook `, `acs-claude-code-wire`, `acs-cursor-wire`, `acs-nat-wire` and `acs-example-guardian`. For NAT, install the `nat` extra (`pip install './adapters[nat]'`); NAT then finds `_type: acs_guardian` through the package's `nat.components` entry point. +Installing the package (`uv tool install ./adapters`, or `pip install ./adapters`) puts these commands on your PATH: `acs-claude-code-hook`, `acs-codex-hook`, `acs-cursor-hook `, `acs-claude-code-wire`, `acs-cursor-wire`, `acs-nat-wire` and `acs-example-guardian`. For NAT, install the `nat` extra (`pip install './adapters[nat]'`); NAT then finds `_type: acs_guardian` through the package's `nat.components` entry point. --- diff --git a/adapters/docs/codex/README.md b/adapters/docs/codex/README.md new file mode 100644 index 00000000..569962bb --- /dev/null +++ b/adapters/docs/codex/README.md @@ -0,0 +1,122 @@ +# ACS adapter: Codex + +This first slice for [#89](https://github.com/GenAI-Security-Project/agent-control-standard/issues/89) lets an ACS Guardian allow, deny or rewrite a Codex shell command, file patch or subagent spawn before it runs. Codex runs `acs-codex-hook` as its `PreToolUse` hook, and the command sends the call to the Guardian as a signed `steps/toolCallRequest`. + +```text +Codex PreToolUse -> signed steps/toolCallRequest -> Guardian +Codex runs, blocks or rewrites the call <- signed decision +``` + +The adapter advertises only `steps/toolCallRequest`, the Guardian URL's transport, and no conformance profile. Session and turn lifecycle, tool results, compaction, subagent lifecycle events and Wrapped MCP are outside this slice. See [mapping.md](mapping.md). + +## Install and run a local demo + +The live procedure below passed on **Codex CLI 0.142.5, macOS**, on 2026-11-05. That is a tested version, not a minimum. + +Commands below assume `$ACS_REPO` points at your clone of `GenAI-Security-Project/agent-control-standard`: + +```bash +export ACS_REPO=/path/to/your/clone +uv tool install "$ACS_REPO/adapters" # or: pipx install "$ACS_REPO/adapters" +``` + +Installing the package puts `acs-codex-hook` and `acs-example-guardian` on your PATH. + +### 1. Generate the shared HMAC secret + +The adapter and the Guardian read the same file. §10 requires at least 32 bytes, and the adapter refuses a file that is a symlink or readable by anyone but its owner. + +```bash +mkdir -p ~/.acs +openssl rand -hex 32 > ~/.acs/hmac.key +chmod 600 ~/.acs/hmac.key +``` + +### 2. Run the Guardian + +`ACS_HMAC_KEY_ID` is the label both sides give the secret. A request signed under a label the Guardian does not hold gets an unsigned `-32004`, and the adapter denies the call. + +```bash +ACS_HMAC_SECRET_FILE=~/.acs/hmac.key ACS_HMAC_KEY_ID=2026-11 \ + acs-example-guardian --port 8787 +``` + +The example Guardian's policy is a teaching policy, not a complete shell policy. + +### 3. Add the hook + +Merge the `PreToolUse` entry from [hooks.json.example](hooks.json.example) into your project's `.codex/hooks.json`, keeping any hooks already there. Replace each path with an absolute one, and quote any path that contains spaces. `command -v acs-codex-hook` prints the installed command's path. + +The example sets `ACS_DEFAULT_DENY=1` and matches three tools. `Bash` covers Codex's `exec_command` path, `apply_patch` covers its file edits, and `spawn_agent` covers subagent spawns. Leave out `apply_patch` and a file write the Guardian denies as a shell command can go through as a patch. It runs synchronously, so do not add `async: true`. + +Keep the `sh -c '… || { …; exit 2; }'` wrapper. Codex runs the call after any hook failure except exit 2, so without the wrapper a missing or broken `acs-codex-hook` lets every call through. The wrapper turns a hook that cannot run into a block, under either posture. To turn ACS off, set `ACS_DISABLED=1` or remove the hook; a missing adapter does not do it. `sh` sets the `ACS_*` variables itself, so the hook needs no `env` process. + +### 4. Trust the hook + +Review the exact hook with Codex `/hooks` and trust the project configuration. Codex skips an untrusted hook. Then try a harmless command such as `printf hello`. + +## Configuration + +| Setting | Behavior | +|---|---| +| `ACS_GUARDIAN_URL` | The Guardian, over HTTP(S); defaults to `http://127.0.0.1:8787/acs` | +| `ACS_HMAC_SECRET_FILE` | The shared signing secret, at least 32 bytes, mode `0600`. `ACS_HMAC_SECRET` also works, but prefer the file | +| `ACS_HMAC_KEY_ID` | The label that names the secret (§10). Without it, or without a secret, every call is denied before anything is sent | +| `ACS_DEFAULT_DENY` | `0` lets a call that gets no usable decision proceed, audited. Unset, or any other value, denies it | +| `ACS_STARTUP_POSTURE` | What happens when no Guardian answers the handshake: `proceed` lets the call run, audited, with nothing sent to the Guardian. Unset, or any other value, denies it | +| `ACS_AUDIT_FILE` | A durable audit file, written as well as stderr | +| `ACS_HANDSHAKE_CACHE` | Where session decisions are stored | +| `ACS_AGENT_ID` | The deployment's own agent identity | +| `ACS_DISABLED` | `1` sends nothing and lets every call proceed, audited as `acs_disabled_bypass` | + +A Guardian's negotiated `on_decision_failure: deny` also denies, and neither side can weaken the other's deny. A missing or unreadable secret, invalid hook input, an invalid signature, a mis-bound response and a refusal code deny under either posture. Each substitution and failure writes an `ACS_AUDIT` line. + +When a valid ServerHello leaves `steps/toolCallRequest` out of `methods_evaluated`, the Guardian has not agreed to evaluate tool calls. The adapter records `method_not_evaluated`, sends no tool-call request, and leaves the decision to Codex's own permissions, even with `ACS_DEFAULT_DENY=1`. + +The handshake gets 5 seconds. A negotiated decision timeout up to 10 seconds is supported, and a longer one denies before the step is sent. Keep the hook's own timeout above both (the example uses 30). Codex runs the call when a hook times out, and the wrapper cannot change that. A hook that Codex kills, disables or does not trust cannot deny anything, so a fail-closed adapter does not make Codex a fail-closed enforcement boundary. + +## Run the tests + +```bash +cd "$ACS_REPO/adapters" +uv sync --locked --group test +uv run python run_conformance.py codex +``` + +This runs the shared checks and the Codex suite with no model account or Codex install. The suite runs the adapter as a subprocess against `CaptureGuardian`, which checks the bytes it sends against the repository schemas and recomputes their signatures. It covers each decision, input preservation, request binding, both postures, timeouts, malformed messages, the hook command's own fallback and the declared capabilities. + +## Verify enforcement in the real Codex CLI + +```bash +cd "$ACS_REPO/adapters" +uv run python tests/integrations/codex/live_verify.py --evidence-dir /tmp/acs-codex-evidence +``` + +This check needs the Codex CLI, and it fails rather than skips when the CLI is missing or any assertion fails. It runs the real CLI, the installed `acs-codex-hook` and a signed test Guardian, under the matcher and wrapper from [hooks.json.example](hooks.json.example). A scripted Responses endpoint on localhost emits fixed tool calls, so there is no model inference and no external model request. Codex offers `apply_patch` only to a model its catalog says can use it, so the patch cases name `gpt-5.5`, and the localhost endpoint still answers them. + +The fixture uses a dummy provider credential, ignores user config and rules, disables plugins, and supplies the reviewed test hook on each run. Its `--dangerously-bypass-hook-trust` flag applies only to these runs; normal setup uses `/hooks` review. System-managed hooks can still affect the run, so use a test machine or session. + +| Case | Required result | +|---|---| +| Shell ALLOW | Only `original.marker` exists, with its expected content | +| Shell DENY | Neither marker exists, and Codex returns the Guardian's reason to the model | +| Shell MODIFY | Only `replacement.marker` exists, with its expected content | +| Patch ALLOW, DENY, MODIFY | The same results, for a patch that adds the marker | +| Spawn ALLOW | The subagent starts and runs its command, which the Guardian also sees as a `Bash` call naming the subagent | +| Spawn DENY | The subagent never starts, and Codex returns the Guardian's reason to the model | +| Spawn MODIFY | The subagent starts with the Guardian's rewritten task | +| Hook cannot run | The command does not run, because the wrapper exits 2 | + +Every case that reaches the Guardian must send only signed, schema-valid requests, and every case needs a completed CLI run. The check saves a summary, the wire captures, the model requests and the CLI logs. It proves the `exec_command`/`Bash`, `apply_patch` and `spawn_agent` paths, not every tool path or a hosted-model session. + +## Coverage limits + +The example configuration covers shell commands, file patches and subagent spawns. Object-valued MCP and other local-function arguments are covered by the subprocess tests but not the live check. Include them in a deployment's coverage claim only after changing the matcher and verifying them on the platform. A non-object tool input is denied, because the adapter cannot represent it as ACS named arguments without inventing meaning. + +The other multi-agent tools are not matched. `send_input` can give a running subagent new work and `resume_agent` can restart a closed one. Neither message is gated, though each tool call the subagent then makes still is. Codex also names them inconsistently in hook input (`multi_agent_v1wait_agent`, but `spawn_agent`), so check the names on your version before matching them. + +A subagent's own tool calls go through the same hook, under the parent's session ID. The adapter records Codex's `agent_id` and `agent_type` as `metadata.codex_agent_id` and `metadata.codex_agent_type`, so the Guardian can tell a subagent's call from the parent's. `SubagentStart` cannot stop a spawn on Codex 0.142.5, which is why the gate sits on `spawn_agent`. + +Hosted tools such as web search bypass these hooks. `write_stdin` does not run `PreToolUse` again when it feeds a running process, and some specialized tool paths can opt out. Hook files, the adapter, the shared secret and the stored session decisions also need protection from the observed agent at the deployment level. This adapter does not isolate processes or mediate every action. + +The hook contract was checked against the [Codex hook documentation](https://learn.chatgpt.com/docs/hooks) on 2026-11-03. See its tool coverage and trust sections when you configure a runtime. diff --git a/adapters/docs/codex/hooks.json.example b/adapters/docs/codex/hooks.json.example new file mode 100644 index 00000000..a533d968 --- /dev/null +++ b/adapters/docs/codex/hooks.json.example @@ -0,0 +1,17 @@ +{ + "hooks": { + "PreToolUse": [ + { + "matcher": "^(Bash|apply_patch|spawn_agent)$", + "hooks": [ + { + "type": "command", + "command": "sh -c 'ACS_GUARDIAN_URL=http://127.0.0.1:8787/acs ACS_HMAC_SECRET_FILE=/absolute/path/to/.acs/hmac.key ACS_HMAC_KEY_ID=2026-11 ACS_AUDIT_FILE=/absolute/path/to/.acs/audit.log ACS_DEFAULT_DENY=1 /absolute/path/to/acs-codex-hook || { echo ACS hook could not run, so the call is blocked >&2; exit 2; }'", + "timeout": 30, + "statusMessage": "Checking with the ACS Guardian" + } + ] + } + ] + } +} diff --git a/adapters/docs/codex/mapping.md b/adapters/docs/codex/mapping.md new file mode 100644 index 00000000..7f75deaf --- /dev/null +++ b/adapters/docs/codex/mapping.md @@ -0,0 +1,44 @@ +# Codex to ACS mapping: first slice + +## Hook and identity mapping + +| Input | ACS representation | +|---|---| +| `PreToolUse` | `steps/toolCallRequest`, a synchronous decision before the call runs | +| `tool_name` | `payload.tool.name`, with Codex's own name kept (`Bash` for shell and `exec_command`, `apply_patch` for a file patch, `spawn_agent` for a subagent spawn) | +| `tool_input` object | Each argument becomes `{ "value": original_value }`, with no invented provenance | +| `session_id` | Kept when it is a UUID; otherwise a UUID5 in the `codex:session` namespace. The original goes in `metadata.codex_session_id` | +| `turn_id` | Kept as `metadata.turn_id`. No `turnStart` or `turnEnd` is sent for it | +| `tool_use_id` | With `turn_id`, a UUID5 scoped to the session becomes `request_id`; the original goes in `metadata.codex_tool_use_id` | +| `cwd`, `model`, `permission_mode` | Metadata, when they are strings | +| `agent_id`, `agent_type` | `metadata.codex_agent_id` and `metadata.codex_agent_type`, on a subagent's steps | +| Any other hook event | Nothing sent; audited as `unmapped_hook_event` | + +The session ID, turn ID, tool-call ID, tool name and an object input are required. An event missing any of them is denied with no traffic, under either posture, rather than sent with invented identity. The adapter never reads the transcript, whose format is not a stable hook interface. + +`tool_use_id` is the model's own call ID, so a later turn can reuse it, and a subagent runs under the parent's session in a turn of its own. The request ID includes `turn_id`, so a reused call ID in another turn is not taken for a replay. + +The handshake runs on the first tool call of a session, and its result is stored under an HMAC for the session's later calls. It advertises exactly one step method, the Guardian URL's transport, no Wrapped MCP and no conformance profile. Signing alone is not ACS-Core conformance. No session or turn lifecycle, tool results, heartbeat or `steps/subagentStart` are emitted. A subagent spawn reaches the Guardian as a `steps/toolCallRequest` for `spawn_agent`, since Codex runs a spawn as a tool call and ignores `continue: false` from `SubagentStart`. + +If a valid ServerHello leaves `steps/toolCallRequest` out of `methods_evaluated`, the Guardian is not evaluating tool calls. The adapter audits `method_not_evaluated`, sends no tool-call request and prints nothing, which leaves Codex's own approval policy in charge. That is not a decision failure, so neither the local nor the negotiated deny posture turns it into a denial. A ServerHello that fails its checks is still no answer, and the startup posture settles the call. + +## Decision mapping + +| Guardian decision | Codex `PreToolUse` output | +|---|---| +| ALLOW | Nothing, so Codex's own approval policy and sandbox still decide | +| DENY | `permissionDecision: "deny"` with the Guardian's reasoning | +| MODIFY with only `parameter_overrides` | The overrides merged onto the original arguments, as `permissionDecision: "allow"` with the complete `updatedInput` | +| Any other MODIFY | Audited DENY, never a partial edit | +| ASK or DEFER | Audited DENY, since Codex cannot hold a tool call for approval | +| A decision outside the five | DENY under either posture | + +For `Bash` and `apply_patch`, only a `command` override is accepted, and the merged `command` must stay a string. Other tools get the merged argument object, and Codex owns each tool's argument schema. The adapter supports no redaction and no wholesale content rewrite. + +Codex applies `updatedInput` only together with `permissionDecision: "allow"`, so a MODIFY is the one decision that prints an allow. That allow does not skip Codex's own approval: on Codex CLI 0.142.5 under `-a untrusted`, Codex still asked before a rewritten command ran, and the prompt showed the rewritten command. + +Do not return `permissionDecision: "ask"` to Codex. Codex parses it but does not support it yet: it marks the hook run as failed and continues the tool call. `PermissionRequest` is a different event, and it cannot carry an ASK for calls that never need approval. + +A response must bind both the JSON-RPC `id` and the ACS `request_id` to the outstanding request, carry a valid signature under the session key, and pass `response-envelope.json`. A signature failure, a mis-bound response and a refusal code deny under either posture. A restrictive decision in a result that fails the schema denies too. Transport failures, timeouts, signed errors that are not refusals and an unreadable allow follow the failure posture, and each is audited. + +This slice makes no claim about hosted tools, input written to a running process, tool paths that skip hooks, or edits to its own configuration. See [README.md](README.md) for tested coverage and setup. diff --git a/adapters/pyproject.toml b/adapters/pyproject.toml index 58b164d9..d9cf8e06 100644 --- a/adapters/pyproject.toml +++ b/adapters/pyproject.toml @@ -5,7 +5,7 @@ build-backend = "hatchling.build" [project] name = "acs-adapters" version = "0.2.0" -description = "Reference adapters that connect Claude Code, Cursor and NVIDIA NeMo Agent Toolkit to an ACS Guardian" +description = "Reference adapters that connect Claude Code, Codex, Cursor and NVIDIA NeMo Agent Toolkit to an ACS Guardian" readme = "README.md" license = "Apache-2.0" requires-python = ">=3.11" @@ -19,6 +19,7 @@ nat = ["nvidia-nat-core>=1.7,<1.8", "ruamel.yaml>=0.19,<0.20"] [project.scripts] acs-claude-code-hook = "acs.integrations.claude_code.hook:main" acs-claude-code-wire = "acs.integrations.claude_code.wire:main" +acs-codex-hook = "acs.integrations.codex.hook:main" acs-cursor-hook = "acs.integrations.cursor.hook:main" acs-cursor-wire = "acs.integrations.cursor.wire:main" acs-nat-wire = "acs.integrations.nat.wire:main" @@ -28,6 +29,7 @@ acs-example-guardian = "acs.guardian.example_guardian:main" # distributions register here the same way. [project.entry-points."acs.integrations"] claude_code = "acs.integrations.claude_code" +codex = "acs.integrations.codex" cursor = "acs.integrations.cursor" nat = "acs.integrations.nat" diff --git a/adapters/run_conformance.py b/adapters/run_conformance.py index 1e1bb07f..1bb90aa2 100755 --- a/adapters/run_conformance.py +++ b/adapters/run_conformance.py @@ -21,6 +21,7 @@ PLATFORM_SUITES = { "claude": ("claude-code", HERE, ("discover", "tests/integrations/claude_code")), + "codex": ("codex", HERE, ("discover", "tests/integrations/codex")), "cursor": ("cursor", HERE, ("discover", "tests/integrations/cursor")), "nat": ("nat", HERE, ("discover", "tests/integrations/nat")), } @@ -28,6 +29,7 @@ PLATFORM_ALIASES = { "claude": "claude", "claude-code": "claude", + "codex": "codex", "cursor": "cursor", "nat": "nat", } @@ -74,7 +76,7 @@ def _parse_args(argv: list[str] | None = None) -> argparse.Namespace: metavar="PLATFORM", nargs="+", choices=tuple(PLATFORM_ALIASES), - help="one or more of: claude, cursor, nat", + help="one or more of: claude, codex, cursor, nat", ) parser.add_argument( "--strict", diff --git a/adapters/scripts/check_wheel.py b/adapters/scripts/check_wheel.py index 9270c2ec..906e3781 100644 --- a/adapters/scripts/check_wheel.py +++ b/adapters/scripts/check_wheel.py @@ -18,7 +18,7 @@ from pathlib import Path REPO_SCHEMAS = Path(__file__).resolve().parents[2] / "specification" / "v0.1.0" -SCRIPTS = ("acs-claude-code-hook", "acs-claude-code-wire", "acs-cursor-hook", +SCRIPTS = ("acs-claude-code-hook", "acs-claude-code-wire", "acs-codex-hook", "acs-cursor-hook", "acs-cursor-wire", "acs-nat-wire", "acs-example-guardian") failures: list[str] = [] @@ -60,12 +60,14 @@ def base_checks(env_dir: Path, wheel_version: str, cwd: Path) -> None: p = run([py, "-c", "import json; from importlib.metadata import entry_points as e; " "print(json.dumps(sorted(x.name for x in e(group='acs.integrations'))))"], cwd) - check("acs.integrations lists claude_code, cursor and nat", - p.returncode == 0 and json.loads(p.stdout) == ["claude_code", "cursor", "nat"], p.stdout + p.stderr) + check("acs.integrations lists claude_code, codex, cursor and nat", + p.returncode == 0 and json.loads(p.stdout) == ["claude_code", "codex", "cursor", "nat"], + p.stdout + p.stderr) for script in SCRIPTS: check(f"{script} is installed", (bindir / script).is_file()) - for script, label in (("acs-claude-code-hook", "claude-code"), ("acs-cursor-hook", "cursor")): + for script, label in (("acs-claude-code-hook", "claude-code"), ("acs-codex-hook", "codex"), + ("acs-cursor-hook", "cursor")): p = run([str(bindir / script), "--version"], cwd) check(f"{script} --version", p.returncode == 0 and f"({label}) {wheel_version}" in p.stdout, p.stdout + p.stderr) for script in ("acs-claude-code-wire", "acs-cursor-wire"): @@ -87,6 +89,12 @@ def base_checks(env_dir: Path, wheel_version: str, cwd: Path) -> None: check("acs-claude-code-hook without rfc8785 audits and denies", p.returncode == 0 and '"permissionDecision": "deny"' in p.stdout and "adapter_unavailable" in p.stderr, p.stdout + p.stderr[-300:]) + codex_event = json.dumps({"hook_event_name": "PreToolUse", "session_id": "s", "turn_id": "t", + "tool_use_id": "c", "tool_name": "Bash", "tool_input": {"command": "ls"}}) + p = run([str(bindir / "acs-codex-hook")], cwd, input=codex_event, env=degraded) + check("acs-codex-hook without rfc8785 audits and denies", + p.returncode == 0 and '"permissionDecision": "deny"' in p.stdout + and "adapter_unavailable" in p.stderr, p.stdout + p.stderr[-300:]) p = run([str(bindir / "acs-cursor-hook"), "preToolUse"], cwd, input=event, env=degraded) check("acs-cursor-hook without rfc8785 audits and denies", p.returncode == 0 and '"permission": "deny"' in p.stdout diff --git a/adapters/src/acs/integrations/codex/__init__.py b/adapters/src/acs/integrations/codex/__init__.py new file mode 100644 index 00000000..799859b6 --- /dev/null +++ b/adapters/src/acs/integrations/codex/__init__.py @@ -0,0 +1 @@ +"""Codex hook integration.""" diff --git a/adapters/src/acs/integrations/codex/adapter.py b/adapters/src/acs/integrations/codex/adapter.py new file mode 100644 index 00000000..dac7a81c --- /dev/null +++ b/adapters/src/acs/integrations/codex/adapter.py @@ -0,0 +1,280 @@ +#!/usr/bin/env python3 +"""Codex hook adapter: a PreToolUse event on stdin becomes a signed steps/toolCallRequest, and the Guardian's answer becomes the hook's output.""" +from __future__ import annotations + +import hashlib +import json +import os +import sys +import uuid +from typing import Any + +from acs import __version__ + +from acs.client import ( + Config, + Resolution, + Undecided, + audit_event, + audit_modify_unsupported, + open_session, + round_trip, +) +from acs.core import ( + Decision, + coerce_uuid, + method_evaluated, + modify_composition_violation, + normalize_decision, + request_envelope, +) + +ADAPTER_VERSION = __version__ + +METHOD = "steps/toolCallRequest" +MAX_EVENT_BYTES = 1_000_000 +# hooks.json.example gives the hook 30 s, which has to cover the handshake and the decision. +MAX_DECISION_S = 10.0 +# Codex's shell and patch tools take their input as one command string, and updatedInput can rewrite only that. +COMMAND_TOOLS = frozenset({"Bash", "apply_patch"}) +# The adapter denies an event without these rather than invent the identity the Guardian decides on. +REQUIRED_FIELDS = ("session_id", "turn_id", "tool_use_id", "tool_name") + + +class InvalidEvent(ValueError): + """A hook event the adapter cannot represent without inventing identity or arguments.""" + + +def _strict_json(raw: str) -> Any: + """Parse with duplicate keys, NaN and infinities refused, since each lets the Guardian and Codex read different input.""" + def constant(value: str) -> Any: + raise InvalidEvent("nonfinite_json") + + def unique(pairs: list[tuple[str, Any]]) -> dict[str, Any]: + result: dict[str, Any] = {} + for key, value in pairs: + if key in result: + raise InvalidEvent("duplicate_json_key") + result[key] = value + return result + + try: + return json.loads(raw, parse_constant=constant, object_pairs_hook=unique) + except InvalidEvent: + raise + except ValueError as e: + raise InvalidEvent("invalid_json") from e + + +def _check_input(tool_name: str, tool_input: Any) -> None: + if not isinstance(tool_input, dict): + raise InvalidEvent("tool_input_not_object") + if tool_name in COMMAND_TOOLS and not isinstance(tool_input.get("command"), str): + raise InvalidEvent("command_not_string") + + +def _check_event(event: dict[str, Any]) -> None: + for name in REQUIRED_FIELDS: + value = event.get(name) + if not isinstance(value, str) or not value.strip(): + raise InvalidEvent(f"missing_{name}") + _check_input(event["tool_name"], event.get("tool_input")) + + +def _response(decision: str, reason: str = "", updated_input: dict | None = None) -> dict[str, Any]: + hso: dict[str, Any] = {"hookEventName": "PreToolUse", "permissionDecision": decision} + if reason: + hso["permissionDecisionReason"] = reason + if updated_input is not None: + hso["updatedInput"] = updated_input + return {"hookSpecificOutput": hso} + + +def _emit(payload: dict[str, Any]) -> None: + if payload: + print(json.dumps(payload, allow_nan=False)) + + +def _session_id(event: dict[str, Any]) -> str: + # Codex session ids need not be UUIDs, and an ACS session_id must be one. + return coerce_uuid(event["session_id"], namespace_prefix="codex:session") + + +def _agent_id(event: dict[str, Any]) -> str: + explicit = os.environ.get("ACS_AGENT_ID") + if explicit: + return explicit + cwd = event.get("cwd") + if not isinstance(cwd, str) or not cwd: + return "codex:unknown" + return f"codex:{hashlib.sha256(cwd.encode('utf-8', 'surrogatepass')).hexdigest()[:8]}" + + +def _metadata(event: dict[str, Any], session_id: str) -> dict[str, Any]: + metadata: dict[str, Any] = { + "agent_id": _agent_id(event), + "session_id": session_id, + # Codex's own turn id; the adapter sends no turnStart or turnEnd for it. + "turn_id": event["turn_id"], + "platform": "codex", + "adapter_version": ADAPTER_VERSION, + "codex_session_id": event["session_id"], + "codex_tool_use_id": event["tool_use_id"], + } + for name in ("cwd", "permission_mode", "model"): + if isinstance(event.get(name), str) and event[name]: + metadata[name] = event[name] + # A subagent's steps carry the parent's session_id; these name the subagent that took them. + for name in ("agent_id", "agent_type"): + if isinstance(event.get(name), str) and event[name]: + metadata[f"codex_{name}"] = event[name] + return metadata + + +def build_request(event: dict[str, Any]) -> dict[str, Any]: + session_id = _session_id(event) + # tool_use_id is the model's own call id, which can repeat across sessions and turns, and a + # subagent runs under the parent's session in a turn of its own. + name = f"codex:tool_use:{event['turn_id']}:{event['tool_use_id']}" + request_id = str(uuid.uuid5(uuid.UUID(session_id), name)) + payload = { + "tool": {"name": event["tool_name"]}, + "arguments": {k: {"value": v} for k, v in event["tool_input"].items()}, + } + return request_envelope(METHOD, request_id=request_id, metadata=_metadata(event, session_id), payload=payload) + + +def translate(decision: Decision, event: dict[str, Any]) -> dict[str, Any]: + tool_name = event["tool_name"] + if not decision.known: + # A decision outside the five dispositions denies under either posture (issue #32). + audit_event("unusable_disposition", disposition=decision.disposition or "(missing)", hook="PreToolUse") + return _response("deny", f"unusable Guardian disposition '{decision.disposition}'; failing closed") + if decision.disposition == "allow": + # Nothing: Codex then runs the call under its own approval policy and sandbox. + return {} + if decision.disposition == "deny": + return _response("deny", decision.reasoning) + if decision.disposition in ("ask", "defer"): + # Codex marks a PreToolUse ask as a failed hook and runs the call, so neither can be passed on. + audit_event("unenforceable_decision", decision=decision.disposition, hook="PreToolUse", tool=tool_name, + reason=decision.reasoning or "(none)", substituted="deny") + return _response("deny", f"{decision.disposition.upper()} substituted to DENY, since Codex cannot hold " + f"a tool call for approval: {decision.reasoning or '(no reasoning)'}") + return _translate_modify(decision, event) + + +def _translate_modify(decision: Decision, event: dict[str, Any]) -> dict[str, Any]: + tool_name = event["tool_name"] + modifications = decision.modifications + # §6.3: a modifications object that breaks the composition rules is a DENY, never half-applied. + violation = modify_composition_violation(modifications) + if violation: + audit_event("modify_composition_invalid", hook="PreToolUse", detail=violation) + return _response("deny", f"MODIFY refused ({violation}); failing closed per §6.3") + overrides = modifications.get("parameter_overrides") + # updatedInput replaces the arguments and nothing else, so any other operation would be dropped (§6.5). + if set(modifications) != {"parameter_overrides"} or not isinstance(overrides, dict): + audit_modify_unsupported("Codex applies only parameter_overrides", hook="PreToolUse", tool=tool_name) + return _response("deny", f"MODIFY substituted to DENY, since Codex applies only parameter_overrides: " + f"{decision.reasoning or '(no reasoning)'}") + if tool_name in COMMAND_TOOLS and set(overrides) - {"command"}: + # Codex rewrites only the command of a shell or patch call, so another override would be ignored. + audit_modify_unsupported(f"{tool_name} takes only a command override", hook="PreToolUse", tool=tool_name) + return _response("deny", f"MODIFY substituted to DENY, since Codex rewrites only the command of " + f"{tool_name}: {decision.reasoning or '(no reasoning)'}") + updated = {**event["tool_input"], **overrides} + try: + _check_input(tool_name, updated) + except InvalidEvent as e: + audit_modify_unsupported(f"the rewritten input fails its check ({e})", hook="PreToolUse", tool=tool_name) + return _response("deny", f"MODIFY substituted to DENY, since the rewritten input is unusable ({e})") + # Codex applies updatedInput only with allow (hooks reference, PreToolUse). + return _response("allow", decision.reasoning, updated_input=updated) + + +def main() -> int: + config = Config.from_env(os.environ) + if sys.argv[1:2] == ["--version"]: + print(f"acs-adapter (codex) {ADAPTER_VERSION}") + return 0 + if config.disabled: + # Every step proceeds ungoverned, so the bypass goes to the durable audit sink (§6.4). + audit_event("acs_disabled_bypass", + detail="ACS_DISABLED=1 — hook bypassed with no Guardian " + "traffic; every step proceeds ungoverned") + return 0 + + raw = sys.stdin.read() + try: + if len(raw.encode("utf-8", "surrogatepass")) > MAX_EVENT_BYTES: + raise InvalidEvent("event_too_large") + event = _strict_json(raw) + if not isinstance(event, dict): + raise InvalidEvent("event_not_object") + hook_name = event.get("hook_event_name") + if hook_name != "PreToolUse": + if not isinstance(hook_name, str) or not hook_name: + raise InvalidEvent("missing_hook_event_name") + # Only PreToolUse is mapped; another event is audited, so an ungoverned one is not a quiet one. + audit_event("unmapped_hook_event", hook=hook_name, session_id=event.get("session_id")) + return 0 + _check_event(event) + except InvalidEvent as e: + return _refuse(str(e)) + + session_id = _session_id(event) + try: + return _govern(event, session_id, config) + except Exception as e: # noqa: BLE001 + # An exit 1 would let Codex run the call, so an adapter error denies it under either posture. + sys.stderr.write(f"acs-adapter: adapter error: {e}\n") + return _settle(config.posture.closed().resolve(Undecided("adapter_exception", {"error": str(e)})), + session_id) + + +def _govern(event: dict[str, Any], session_id: str, config: Config) -> int: + # §4.1: a refused or unanswered session sends the Guardian nothing; the step is settled here. + session = open_session(config=config, session_id=session_id, agent_id=_agent_id(event), platform="codex", + methods_implemented=[METHOD], profiles_supported=[]) + gated = config.gate(session, METHOD) + if gated is not None: + return _settle(gated, session_id) + server_hello = session.server_hello + posture = config.posture.negotiated(server_hello) + if not method_evaluated(server_hello, METHOD): + # ALLOW-by-default (handshake.json): Codex's own approval policy decides, under either posture. + audit_event("method_not_evaluated", method=METHOD, hook="PreToolUse", session_id=session_id) + return 0 + timeout_s = posture.timeout_for(METHOD) + if timeout_s > MAX_DECISION_S: + return _refuse("unsupported_decision_timeout", session_id=session_id, timeout_ms=int(timeout_s * 1000)) + + result = round_trip(build_request(event), guardian_url=config.guardian_url, posture=posture, + session_id=session_id, gate=True) + if isinstance(result, Resolution): + return _settle(result, session_id) + _emit(translate(normalize_decision(result.response.get("result")), event)) + return 0 + + +def _refuse(cause: str, **fields: Any) -> int: + """Deny a call the adapter cannot put to the Guardian faithfully, under either posture: nothing decided it.""" + audit_event("codex_call_refused", cause=cause, hook="PreToolUse", **fields) + _emit(_response("deny", f"ACS adapter: {cause}")) + return 0 + + +def _settle(resolution: Resolution, session_id: str | None) -> int: + """Carry out a resolution: deny unless it proceeds, and audit which rule decided (§6.4).""" + if resolution.message: + sys.stderr.write(f"acs-adapter: {resolution.message}\n") + if not resolution.proceed: + _emit(_response("deny", f"ACS adapter: {resolution.label} ({resolution.cause})")) + audit_event(resolution.audit_type, cause=resolution.cause, hook="PreToolUse", + session_id=session_id, **resolution.fields) + return 0 + + +if __name__ == "__main__": + sys.exit(main()) diff --git a/adapters/src/acs/integrations/codex/hook.py b/adapters/src/acs/integrations/codex/hook.py new file mode 100644 index 00000000..a639e7a8 --- /dev/null +++ b/adapters/src/acs/integrations/codex/hook.py @@ -0,0 +1,36 @@ +"""The acs-codex-hook command: run the adapter, or record why it could not load and apply the posture.""" +from __future__ import annotations + +import json +import sys + +from acs.integrations._bootstrap import run_hook + + +def _event_name(raw: str) -> str: + """The event's hook_event_name; an event that does not parse counts as PreToolUse, the one event this command is registered on.""" + try: + event = json.loads(raw or "{}") + except ValueError: + return "PreToolUse" + name = event.get("hook_event_name") if isinstance(event, dict) else None + return name if isinstance(name, str) and name else "PreToolUse" + + +def _deny(event_name: str, reason: str) -> int: + if event_name == "PreToolUse": + print(json.dumps({"hookSpecificOutput": { + "hookEventName": "PreToolUse", + "permissionDecision": "deny", + "permissionDecisionReason": reason, + }})) + return 0 + + +def main() -> int: + return run_hook("acs.integrations.codex.adapter", event_key="hook", + event_name=_event_name, deny=_deny) + + +if __name__ == "__main__": + sys.exit(main()) diff --git a/adapters/tests/integrations/codex/__init__.py b/adapters/tests/integrations/codex/__init__.py new file mode 100644 index 00000000..e69de29b diff --git a/adapters/tests/integrations/codex/live_verify.py b/adapters/tests/integrations/codex/live_verify.py new file mode 100644 index 00000000..94de886b --- /dev/null +++ b/adapters/tests/integrations/codex/live_verify.py @@ -0,0 +1,402 @@ +#!/usr/bin/env python3 +"""Run the real Codex CLI with the installed acs-codex-hook and a signed test Guardian, against a scripted model on localhost. + +No model account or external model call is needed. This tests host enforcement, +not model judgment. Run explicitly; it is not silently skipped in the unit gate. +The hook runs under the matcher and the exit-2 fallback from hooks.json.example, +so the check covers the configuration a deployment copies. +""" +from __future__ import annotations + +import argparse +import http.server +import json +import os +import re +import shlex +import shutil +import subprocess +import sys +import tempfile +import threading +import unittest +from pathlib import Path + +import acs +from acs.testing.capture_guardian import CaptureGuardian + +HOOK = Path(sys.executable).parent / "acs-codex-hook" +EXAMPLE = Path(__file__).resolve().parents[3] / "docs" / "codex" / "hooks.json.example" +SUBAGENT_TASK = "ACS-SUBAGENT-TASK: create the marker" +REPLACEMENT_TASK = "ACS-SUBAGENT-TASK: create the replacement marker" +# A model in Codex's own catalog, named only so Codex offers it apply_patch; the local fixture still answers. +PATCH_MODEL = "gpt-5.5" +UUID = re.compile(r"[0-9a-f]{8}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{12}") + + +def example_hook() -> tuple[str, str]: + """The example's PreToolUse matcher, and the fallback that follows the hook command inside its sh -c.""" + entry = json.loads(EXAMPLE.read_text(encoding="utf-8"))["hooks"]["PreToolUse"][0] + words = shlex.split(entry["hooks"][0]["command"]) + if words[:2] != ["sh", "-c"] or len(words) != 3 or " || " not in words[2]: + raise SystemExit(f"{EXAMPLE} must run the hook as sh -c ' || '") + return entry["matcher"], words[2].split(" || ", 1)[1] + + +def call(name: str, arguments: dict, call_id: str, namespace: str | None = None) -> dict: + item = {"type": "function_call", "id": "fc_" + call_id, "call_id": call_id, + "name": name, "arguments": json.dumps(arguments)} + if namespace: + item["namespace"] = namespace + return item + + +def message(text: str) -> dict: + return {"type": "message", "id": "msg_fixture", "role": "assistant", "status": "completed", + "content": [{"type": "output_text", "text": text, "annotations": []}]} + + +def outputs(req: dict) -> list[dict]: + return [item for item in req.get("input", []) + if item.get("type") in ("function_call_output", "custom_tool_call_output")] + + +def patch_adding(name: str) -> str: + return f"*** Begin Patch\n*** Add File: {name}\n+{name.split('.')[0]}\n*** End Patch\n" + + +def is_subagent(req: dict) -> bool: + return any(item.get("type") == "message" and item.get("role") == "user" + and "ACS-SUBAGENT-TASK" in json.dumps(item.get("content")) for item in req.get("input", [])) + + +class ScriptedModel: + """Answer each Responses request over SSE with the item decide(request) returns.""" + def __init__(self, decide): + self.requests = [] + self.errors = [] + model = self + + class Handler(http.server.BaseHTTPRequestHandler): + def do_GET(self): + raw = b'{"models":[]}' + self.send_response(200) + self.send_header("Content-Type", "application/json") + self.send_header("Content-Length", str(len(raw))) + self.end_headers() + self.wfile.write(raw) + + def do_POST(self): + req = json.loads(self.rfile.read(int(self.headers["Content-Length"]))) + # The local provider must not receive an account credential. + if self.headers.get("Authorization") != "Bearer acs-local-fixture-no-credential": + model.errors.append("unexpected authorization header") + model.requests.append(req) + item = decide(req) + resp = {"id": f"resp_{len(model.requests)}", "status": "completed", "output": [item], + "usage": {"input_tokens": 1, "output_tokens": 1, "total_tokens": 2}} + events = [{"type": "response.created", "response": {"id": resp["id"]}}, + {"type": "response.output_item.done", "output_index": 0, "item": item}, + {"type": "response.completed", "response": resp}] + raw = "".join("data: " + json.dumps(ev) + "\n\n" for ev in events).encode() + self.send_response(200) + self.send_header("Content-Type", "text/event-stream") + self.send_header("Content-Length", str(len(raw))) + self.end_headers() + self.wfile.write(raw) + + def log_message(self, *args): + pass + + self.server = http.server.ThreadingHTTPServer(("127.0.0.1", 0), Handler) + self.thread = threading.Thread(target=self.server.serve_forever, daemon=True) + + @property + def tool_outputs(self) -> list[dict]: + seen = {} + for req in self.requests: + seen.update((item["call_id"], item) for item in outputs(req)) + return list(seen.values()) + + def advertised(self) -> set[str]: + return {tool.get("name") for tool in self.requests[0].get("tools", [])} if self.requests else set() + + def __enter__(self): + self.thread.start() + return self + + def __exit__(self, *args): + self.server.shutdown() + self.server.server_close() + self.thread.join() + + +def run_codex(codex: str, directory: Path, hook: str, decide, *, multi_agent: bool, + name: str, evidence_dir: Path, model_name: str = "acs-fixture", + ) -> tuple[subprocess.CompletedProcess, ScriptedModel]: + matcher, fallback = example_hook() + hook_command = shlex.join(["sh", "-c", hook + " || " + fallback]) + hook_config = ("hooks={PreToolUse=[{matcher=" + json.dumps(matcher) + ",hooks=[{type=\"command\",command=" + + json.dumps(hook_command) + ",timeout=30}]}]}") + with ScriptedModel(decide) as model: + provider = ('model_providers.acs_fixture={name="ACS local fixture",base_url=' + + json.dumps(f"http://127.0.0.1:{model.server.server_port}/v1") + + ',wire_api="responses",requires_openai_auth=false,env_key="ACS_FIXTURE_TOKEN",supports_websockets=false,' + 'request_max_retries=0,stream_max_retries=0}') + args = [codex, "exec", "--ignore-user-config", "--ignore-rules", "--ephemeral", + "--skip-git-repo-check", "--dangerously-bypass-hook-trust", "-C", str(directory), + "--json", "--sandbox", "workspace-write", "-m", model_name, + "-c", 'model_provider="acs_fixture"', "-c", provider, + "-c", 'approval_policy="never"', "-c", 'web_search="disabled"', + "-c", hook_config, "--enable", "hooks", "--enable", "unified_exec", + "--enable" if multi_agent else "--disable", "multi_agent", + "--disable", "plugins", "--disable", "apps", "--disable", "shell_snapshot", + "--disable", "code_mode", "--disable", "code_mode_only", + "Execute the fixture calls supplied by the local test server."] + env = {key: value for key, value in os.environ.items() if not key.startswith("ACS_")} + env["ACS_FIXTURE_TOKEN"] = "acs-local-fixture-no-credential" + proc = subprocess.run(args, input="", env=env, text=True, capture_output=True, timeout=120) + # Preserve diagnostic output even when an assertion fails. + (evidence_dir / f"{name}.stdout.jsonl").write_text(proc.stdout) + (evidence_dir / f"{name}.stderr.log").write_text(proc.stderr) + (evidence_dir / f"{name}.model.json").write_text(json.dumps(model.requests, indent=2) + "\n") + return proc, model + + +def hook_for(guardian: CaptureGuardian, directory: Path) -> str: + """The hook as the example writes it: sh assignments, then the command, with no env process.""" + settings = {"ACS_GUARDIAN_URL": guardian.url(), "ACS_HMAC_SECRET": guardian.hmac_secret, + "ACS_HMAC_KEY_ID": guardian.key_id, "ACS_HANDSHAKE_CACHE": str(directory / "cache"), + "ACS_SESSION_STATE_DIR": str(directory / "state"), "ACS_DEFAULT_DENY": "1"} + return " ".join([*(f"{key}={shlex.quote(value)}" for key, value in settings.items()), shlex.quote(str(HOOK))]) + + +def decide_with(guardian: CaptureGuardian, decision: str, modifications: dict | None = None) -> None: + def verdict(req): + result = guardian._default_allow(req) + if req["params"]["payload"]["tool"]["name"] != "spawn_agent" or decision == "allow": + return result + # response-envelope.json requires reasoning on a deny and a modify. + result.update(decision=decision, reasoning="ACS live fixture decision") + if modifications: + result["modifications"] = modifications + return result + guardian.handlers["steps/toolCallRequest"] = verdict + + +def save_wire(guardian: CaptureGuardian, name: str, evidence_dir: Path) -> None: + (evidence_dir / f"{name}.wire.json").write_text(json.dumps( + {"requests": [r.parsed for r in guardian.captures], "responses": guardian.sent}, indent=2) + "\n") + + +def verify_bash(codex: str, decision: str, evidence_dir: Path) -> dict: + check = unittest.TestCase() + name = f"bash-{decision}" + with tempfile.TemporaryDirectory(prefix="acs-codex-live-") as tmp, CaptureGuardian() as guardian: + directory = Path(tmp) + original = directory / "original.marker" + replacement = directory / "replacement.marker" + command = "printf original > " + shlex.quote(str(original)) + rewritten = "printf replacement > " + shlex.quote(str(replacement)) + + def verdict(req): + result = guardian._default_allow(req) + # response-envelope.json requires reasoning on a deny and a modify. + result.update(decision=decision, reasoning="ACS live fixture decision") + if decision == "modify": + result["modifications"] = {"parameter_overrides": {"command": rewritten}} + return result + guardian.handlers["steps/toolCallRequest"] = verdict + + def decide(req): + if outputs(req): + return message("Fixture completed.") + return call("exec_command", {"cmd": command, "workdir": tmp, "login": False, + "max_output_tokens": 200}, "call_fixture") + proc, model = run_codex(codex, directory, hook_for(guardian, directory), decide, + multi_agent=False, name=name, evidence_dir=evidence_dir) + check.assertEqual(proc.returncode, 0, proc.stderr) + check.assertEqual(model.errors, []) + check.assertIn("exec_command", model.advertised(), "Codex did not advertise exec_command") + check.assertEqual(len(model.requests), 2, "fixture must invoke one tool, then finish") + check.assertEqual(len(model.tool_outputs), 1, "host must return the tool result to the model") + check.assertEqual(guardian.methods(), ["handshake/hello", "steps/toolCallRequest"]) + guardian.assert_all_valid(check) + payload = guardian.payload_of("steps/toolCallRequest") + check.assertEqual(payload["tool"]["name"], "Bash") + check.assertEqual(payload["arguments"]["command"]["value"], command) + check.assertEqual(original.exists(), decision == "allow") + check.assertEqual(replacement.exists(), decision == "modify") + if original.exists(): + check.assertEqual(original.read_text(), "original") + if replacement.exists(): + check.assertEqual(replacement.read_text(), "replacement") + if decision == "deny": + check.assertIn("ACS live fixture decision", json.dumps(model.tool_outputs)) + save_wire(guardian, name, evidence_dir) + return {"case": name, "original_created": original.exists(), + "replacement_created": replacement.exists(), "signed_envelopes_valid": True, + "guardian_methods": guardian.methods(), "model_requests": len(model.requests), + "codex_exit_code": proc.returncode} + + +def verify_patch(codex: str, decision: str, evidence_dir: Path) -> dict: + """Codex offers apply_patch only to a model its catalog says can use it, so this case names a catalog model.""" + check = unittest.TestCase() + name = f"patch-{decision}" + with tempfile.TemporaryDirectory(prefix="acs-codex-live-") as tmp, CaptureGuardian() as guardian: + directory = Path(tmp) + original, replacement = directory / "original.marker", directory / "replacement.marker" + patch, rewritten = patch_adding(original.name), patch_adding(replacement.name) + + def verdict(req): + result = guardian._default_allow(req) + if decision != "allow": + # response-envelope.json requires reasoning on a deny and a modify. + result.update(decision=decision, reasoning="ACS live fixture decision") + if decision == "modify": + result["modifications"] = {"parameter_overrides": {"command": rewritten}} + return result + guardian.handlers["steps/toolCallRequest"] = verdict + + def decide(req): + if outputs(req): + return message("Fixture completed.") + return {"type": "custom_tool_call", "id": "ctc_fixture", "call_id": "call_patch", + "name": "apply_patch", "input": patch} + proc, model = run_codex(codex, directory, hook_for(guardian, directory), decide, multi_agent=False, + name=name, evidence_dir=evidence_dir, model_name=PATCH_MODEL) + check.assertEqual(proc.returncode, 0, proc.stderr) + check.assertEqual(model.errors, []) + check.assertIn("apply_patch", model.advertised(), f"Codex did not offer apply_patch to {PATCH_MODEL}") + check.assertEqual(len(model.tool_outputs), 1, "host must return the tool result to the model") + check.assertEqual(guardian.methods(), ["handshake/hello", "steps/toolCallRequest"]) + guardian.assert_all_valid(check) + payload = guardian.payload_of("steps/toolCallRequest") + check.assertEqual(payload["tool"]["name"], "apply_patch") + check.assertEqual(payload["arguments"]["command"]["value"], patch) + check.assertEqual(original.exists(), decision == "allow") + check.assertEqual(replacement.exists(), decision == "modify") + if decision == "deny": + check.assertIn("ACS live fixture decision", json.dumps(model.tool_outputs)) + save_wire(guardian, name, evidence_dir) + return {"case": name, "original_created": original.exists(), "replacement_created": replacement.exists(), + "signed_envelopes_valid": True, "guardian_methods": guardian.methods(), + "model_requests": len(model.requests), "codex_exit_code": proc.returncode} + + +def verify_spawn(codex: str, decision: str, evidence_dir: Path) -> dict: + """The parent spawns one subagent, whose only step creates the marker its task names, then waits for it.""" + check = unittest.TestCase() + name = f"spawn-{decision}" + with tempfile.TemporaryDirectory(prefix="acs-codex-live-") as tmp, CaptureGuardian() as guardian: + directory = Path(tmp) + marker = directory / "subagent.marker" + replacement = directory / "replacement.marker" + decide_with(guardian, decision, {"parameter_overrides": {"message": REPLACEMENT_TASK}} + if decision == "modify" else None) + + def decide(req): + done = outputs(req) + if is_subagent(req): + target = replacement if REPLACEMENT_TASK in json.dumps(req.get("input")) else marker + return message("Subagent completed.") if done else call( + "exec_command", {"cmd": "touch " + target.name, "workdir": tmp}, "call_sub") + if not done: + return call("spawn_agent", {"message": SUBAGENT_TASK}, "call_spawn", "multi_agent_v1") + ids = UUID.findall(json.dumps(done[-1].get("output"))) if len(done) == 1 else [] + if ids: + return call("wait_agent", {"targets": [ids[0]], "timeout_ms": 30000}, "call_wait", "multi_agent_v1") + return message("Fixture completed.") + proc, model = run_codex(codex, directory, hook_for(guardian, directory), decide, + multi_agent=True, name=name, evidence_dir=evidence_dir) + check.assertEqual(proc.returncode, 0, proc.stderr) + check.assertEqual(model.errors, []) + check.assertIn("multi_agent_v1", model.advertised(), "Codex did not advertise multi-agent tools") + guardian.assert_all_valid(check) + steps = [c.parsed["params"]["payload"] for c in guardian.captures + if c.parsed["method"] == "steps/toolCallRequest"] + check.assertEqual(steps[0]["tool"]["name"], "spawn_agent") + check.assertEqual(steps[0]["arguments"]["message"]["value"], SUBAGENT_TASK) + subagent_ran = any(is_subagent(req) for req in model.requests) + if decision in ("allow", "modify"): + check.assertTrue(subagent_ran, "the allowed subagent never started") + check.assertEqual(marker.exists(), decision == "allow") + check.assertEqual(replacement.exists(), decision == "modify", "the subagent did not get the rewritten task") + # The subagent's own shell step goes through the same hook, and names the subagent. + check.assertEqual([s["tool"]["name"] for s in steps], ["spawn_agent", "Bash"]) + spawn_meta, sub_meta = (c.parsed["params"]["metadata"] for c in guardian.captures + if c.parsed["method"] == "steps/toolCallRequest") + check.assertNotIn("codex_agent_id", spawn_meta) + check.assertTrue(UUID.fullmatch(sub_meta.get("codex_agent_id", "")), sub_meta) + check.assertEqual(sub_meta["session_id"], spawn_meta["session_id"]) + else: + check.assertFalse(subagent_ran, "a denied subagent reached the model") + check.assertFalse(marker.exists()) + check.assertEqual(len(steps), 1) + check.assertIn("ACS live fixture decision", json.dumps(model.tool_outputs)) + save_wire(guardian, name, evidence_dir) + return {"case": name, "subagent_started": subagent_ran, "marker_created": marker.exists(), + "replacement_created": replacement.exists(), + "signed_envelopes_valid": True, "guardian_methods": guardian.methods(), + "model_requests": len(model.requests), "codex_exit_code": proc.returncode} + + +def verify_missing_hook(codex: str, evidence_dir: Path) -> dict: + """A hook command that cannot run must block the call through the example's exit-2 fallback.""" + check = unittest.TestCase() + name = "missing-hook" + with tempfile.TemporaryDirectory(prefix="acs-codex-live-") as tmp: + directory = Path(tmp) + marker = directory / "original.marker" + command = "printf original > " + shlex.quote(str(marker)) + + def decide(req): + if outputs(req): + return message("Fixture completed.") + return call("exec_command", {"cmd": command, "workdir": tmp}, "call_fixture") + proc, model = run_codex(codex, directory, shlex.quote(str(directory / "missing" / "acs-codex-hook")), + decide, multi_agent=False, name=name, evidence_dir=evidence_dir) + check.assertEqual(proc.returncode, 0, proc.stderr) + check.assertEqual(model.errors, []) + check.assertFalse(marker.exists(), "Codex ran the call when the hook command could not run") + check.assertEqual(len(model.tool_outputs), 1, "host must return the block to the model") + return {"case": name, "original_created": marker.exists(), + "model_requests": len(model.requests), "codex_exit_code": proc.returncode} + + +def main(): + parser = argparse.ArgumentParser(description=__doc__) + parser.add_argument("--codex", default="codex") + parser.add_argument("--evidence-dir", type=Path, required=True) + args = parser.parse_args() + executable = shutil.which(args.codex) + if executable is None: + parser.error("Codex CLI is required; this check does not skip") + if not HOOK.is_file(): + parser.error(f"{HOOK} is missing; install the adapters into this interpreter's environment") + args.evidence_dir.mkdir(parents=True, exist_ok=True) + version = subprocess.check_output([executable, "--version"], text=True).strip() + results = [] + for decision in ("allow", "deny", "modify"): + results.append(verify_bash(executable, decision, args.evidence_dir)) + print(json.dumps(results[-1]), flush=True) + for decision in ("allow", "deny", "modify"): + results.append(verify_patch(executable, decision, args.evidence_dir)) + print(json.dumps(results[-1]), flush=True) + for decision in ("allow", "deny", "modify"): + results.append(verify_spawn(executable, decision, args.evidence_dir)) + print(json.dumps(results[-1]), flush=True) + results.append(verify_missing_hook(executable, args.evidence_dir)) + print(json.dumps(results[-1]), flush=True) + matcher, _ = example_hook() + report = {"codex_version": version, "adapter_version": acs.__version__, + "model": "local scripted Responses fixture (no inference)", "matcher": matcher, + "scope": "real CLI PreToolUse enforcement of exec_command/Bash, apply_patch and spawn_agent, " + "and the exit-2 fallback for a hook that cannot run", "cases": results} + (args.evidence_dir / "summary.json").write_text(json.dumps(report, indent=2) + "\n") + return 0 + + +if __name__ == "__main__": + raise SystemExit(main()) diff --git a/adapters/tests/integrations/codex/test_adapter.py b/adapters/tests/integrations/codex/test_adapter.py new file mode 100644 index 00000000..35152ccb --- /dev/null +++ b/adapters/tests/integrations/codex/test_adapter.py @@ -0,0 +1,487 @@ +"""Run the Codex adapter as a subprocess and check its output and the bytes it sends.""" +from __future__ import annotations + +import json +import os +import re +import shlex +import subprocess +import sys +import tempfile +import time +import unittest +from pathlib import Path + +from acs import client, core +from acs.testing.capture_guardian import CaptureGuardian +from acs.testing.harness import TEST_KEY_ID, launch_guardian, signing_env + +ADAPTER = Path(__file__).resolve().parents[3] / "src" / "acs" / "integrations" / "codex" / "adapter.py" +HOOK = Path(sys.executable).parent / "acs-codex-hook" + + +def event(**changes) -> dict: + result = {"hook_event_name": "PreToolUse", "session_id": "thr_test_session", + "turn_id": "turn_7", "tool_use_id": "call_3", "cwd": "/tmp/acs-test", + "tool_name": "Bash", "tool_input": {"command": "printf harmless", "timeout_ms": 1000}} + result.update(changes) + return result + + +class WireGuardian(CaptureGuardian): + """Changes a step's response after it is signed, or replaces it with raw bytes.""" + mutate = None + raw_reply = None + reply_status = 200 + + def _make_handler_cls(self): + base = super()._make_handler_cls() + guardian = self + + class Handler(base): + def _reply(self, reply): + if guardian.received[-1]["method"] == "steps/toolCallRequest": + if guardian.mutate: + guardian.mutate(reply) + if guardian.raw_reply is not None: + self.send_response(guardian.reply_status) + self.send_header("Content-Length", str(len(guardian.raw_reply))) + self.end_headers() + try: + self.wfile.write(guardian.raw_reply) + except (BrokenPipeError, ConnectionResetError): + pass + return + super()._reply(reply) + return Handler + + def resign(self, reply: dict) -> None: + session_id = self.received[-1]["params"]["metadata"]["session_id"] + key = core.derive_session_key(self.hmac_secret.encode(), session_id) + reply.pop("signature", None) + (reply.get("result") or reply.get("error")).pop("signature", None) + client.sign_envelope(reply, key=key, session_id=session_id, key_id=self.key_id) + + +class AdapterTests(unittest.TestCase): + def setUp(self) -> None: + self.tmp = tempfile.TemporaryDirectory() + self.addCleanup(self.tmp.cleanup) + self.guardian = WireGuardian() + self.guardian.start() + self.addCleanup(self.guardian.stop) + # A developer's own ACS settings must not reach the adapter. + self.env = {k: v for k, v in os.environ.items() if not k.startswith("ACS_")} + self.env.update(signing_env(), ACS_GUARDIAN_URL=self.guardian.url(), ACS_DEFAULT_DENY="0", + ACS_HANDSHAKE_CACHE=str(Path(self.tmp.name) / "handshake"), + ACS_SESSION_STATE_DIR=str(Path(self.tmp.name) / "state")) + + def fresh_session_records(self) -> None: + self.env["ACS_HANDSHAKE_CACHE"] = tempfile.mkdtemp(dir=self.tmp.name) + + def invoke(self, input_event: dict | None = None, raw: str | None = None, + command: list[str] | None = None) -> tuple[dict, str]: + proc = subprocess.run(command or [sys.executable, str(ADAPTER)], + input=raw if raw is not None else json.dumps(input_event or event()), + env=self.env, capture_output=True, text=True, timeout=20) + self.assertEqual(proc.returncode, 0, proc.stderr) + output = json.loads(proc.stdout or "{}") + return output.get("hookSpecificOutput", {}), proc.stderr + + def verdict(self, decision, **extra) -> None: + # response-envelope.json requires reasoning on every decision but allow. + self.guardian.handlers["steps/toolCallRequest"] = lambda req: { + **self.guardian._default_allow(req), "decision": decision, "reasoning": "fixture", **extra} + + def hello(self, **changes) -> None: + def answer(req): + result = self.guardian._default_handshake(req) + result.update(changes) + return result + self.guardian.handlers["handshake/hello"] = answer + + def test_allow_prints_nothing_and_the_envelopes_declare_one_method(self) -> None: + out, audit = self.invoke() + # Nothing printed, so Codex's own approval policy and sandbox still decide. + self.assertEqual(out, {}) + self.assertNotIn("ACS_AUDIT", audit) + self.assertEqual(self.guardian.methods(), ["handshake/hello", "steps/toolCallRequest"]) + self.guardian.assert_all_valid(self) + hello = self.guardian.payload_of("handshake/hello") + self.assertEqual(hello["methods_implemented"], ["steps/toolCallRequest"]) + self.assertEqual(hello["profiles_supported"], []) + self.assertEqual(hello["transports_supported"], ["http"]) + self.assertEqual(hello["wrapped_protocols"], []) + request = self.guardian.only("steps/toolCallRequest").parsed + metadata = request["params"]["metadata"] + self.assertEqual(metadata["turn_id"], "turn_7") + self.assertEqual(metadata["codex_session_id"], "thr_test_session") + self.assertEqual(metadata["platform"], "codex") + self.assertEqual(request["params"]["signature"]["key_id"], TEST_KEY_ID) + self.assertEqual(request["params"]["payload"]["arguments"]["command"], {"value": "printf harmless"}) + + def test_deny_carries_the_reasoning(self) -> None: + self.verdict("deny", reasoning="policy") + out, _ = self.invoke() + self.assertEqual(out["permissionDecision"], "deny") + self.assertEqual(out["permissionDecisionReason"], "policy") + + def test_modify_merges_overrides_and_keeps_other_arguments(self) -> None: + self.verdict("modify", modifications={"parameter_overrides": {"command": "printf replacement"}}) + out, _ = self.invoke() + self.assertEqual(out["permissionDecision"], "allow") + self.assertEqual(out["updatedInput"], {"command": "printf replacement", "timeout_ms": 1000}) + + def test_patch_and_mcp_arguments_are_wrapped(self) -> None: + for tool, value in (("apply_patch", {"command": "*** Begin Patch\n*** End Patch"}), + ("mcp__local__read", {"path": "sample", "options": {"limit": 2}})): + with self.subTest(tool=tool): + self.guardian.captures.clear() + out, _ = self.invoke(event(tool_name=tool, tool_input=value)) + self.assertEqual(out, {}) + self.guardian.assert_all_valid(self) + self.assertEqual(self.guardian.payload_of("steps/toolCallRequest")["arguments"], + {k: {"value": v} for k, v in value.items()}) + + def test_mcp_modify_keeps_unmodified_arguments(self) -> None: + self.verdict("modify", modifications={"parameter_overrides": {"path": "safe"}}) + out, _ = self.invoke(event(tool_name="mcp__local__read", tool_input={"path": "old", "limit": 2})) + self.assertEqual(out["updatedInput"], {"path": "safe", "limit": 2}) + + def test_ask_and_defer_deny_because_codex_cannot_hold_a_call(self) -> None: + details = {"ask": {"ask_details": {"approver": {"type": "human", "id": "owner"}, "question": "Run it?", + "timeout_seconds": 60}}, + "defer": {"defer_details": {"reason": "pending_dependency", "resolution_method": "timeout", + "resolution_timeout_ms": 1000, "timeout_decision": "deny"}}} + for decision in ("ask", "defer"): + with self.subTest(decision=decision): + self.verdict(decision, reasoning="needs a person", **details[decision]) + out, audit = self.invoke() + self.assertEqual(out["permissionDecision"], "deny") + self.assertIn("needs a person", out["permissionDecisionReason"]) + self.assertIn("unenforceable_decision", audit) + + def test_unusable_decisions_deny(self) -> None: + for decision in ("future", " DENY ", None, [], 1): + with self.subTest(decision=decision): + self.verdict(decision) + out, _ = self.invoke() + self.assertEqual(out["permissionDecision"], "deny") + + def test_a_modify_codex_cannot_apply_in_full_denies(self) -> None: + for modifications in (None, [], {}, {"modified_content": "text"}, + {"parameter_overrides": []}, + {"parameter_overrides": {"command": None}}, + {"parameter_overrides": {"command": 3}}, + {"parameter_overrides": {"timeout_ms": 20}}, + {"parameter_overrides": {}, "redactions": []}, + {"parameter_overrides": {}, "future_operation": {}}, + {"parameter_overrides": {}, "modified_content": "both"}): + with self.subTest(modifications=modifications): + self.verdict("modify", modifications=modifications) + out, _ = self.invoke() + self.assertEqual(out["permissionDecision"], "deny") + self.assertNotIn("updatedInput", out) + + def test_malformed_input_denies_and_sends_nothing(self) -> None: + for raw in ("", "[]", "null", "{", '{"tool_name":"Bash"}', + '{"hook_event_name":"PreToolUse","tool_input":NaN}', + '{"hook_event_name":"PreToolUse","hook_event_name":"Stop"}'): + with self.subTest(raw=raw): + out, audit = self.invoke(raw=raw) + self.assertEqual(out["permissionDecision"], "deny") + self.assertIn("codex_call_refused", audit) + self.assertEqual(self.guardian.methods(), []) + + def test_missing_identity_or_unusable_input_denies_and_sends_nothing(self) -> None: + for key, value in (("session_id", None), ("turn_id", ""), ("tool_use_id", None), ("tool_name", None), + ("tool_input", None), ("tool_input", {"command": ["printf", "x"]})): + with self.subTest(key=key, value=value): + out, _ = self.invoke(event(**{key: value})) + self.assertEqual(out["permissionDecision"], "deny") + self.assertEqual(self.guardian.methods(), []) + + def test_another_hook_is_audited_and_sends_nothing(self) -> None: + out, audit = self.invoke(event(hook_event_name="Stop")) + self.assertEqual(out, {}) + self.assertIn("unmapped_hook_event", audit) + self.assertEqual(self.guardian.methods(), []) + + def test_signing_material_is_required_under_either_posture(self) -> None: + for missing in ("ACS_HMAC_SECRET", "ACS_HMAC_KEY_ID"): + with self.subTest(missing=missing): + env = dict(self.env) + self.env.pop(missing) + out, audit = self.invoke() + self.env = env + self.assertEqual(out["permissionDecision"], "deny") + self.assertIn("session_refused_fail_closed", audit) + self.assertEqual(self.guardian.methods(), []) + + def test_secret_file_is_read(self) -> None: + secret = Path(self.tmp.name) / "secret" + secret.write_text(self.guardian.hmac_secret) + secret.chmod(0o600) + self.env.pop("ACS_HMAC_SECRET") + self.env["ACS_HMAC_SECRET_FILE"] = str(secret) + out, _ = self.invoke() + self.assertEqual(out, {}) + self.guardian.assert_all_valid(self) + + def test_unsigned_or_tampered_response_denies_under_fail_open(self) -> None: + for mutate in (lambda reply: reply["result"].pop("signature"), + lambda reply: reply["result"].update(reasoning="tampered")): + with self.subTest(mutate=mutate): + self.guardian.mutate = mutate + out, audit = self.invoke() + self.assertEqual(out["permissionDecision"], "deny") + self.assertIn("response_signature_invalid", audit) + + def test_a_signed_response_bound_to_another_request_denies(self) -> None: + for field in ("id", "request_id"): + def mutate(reply, field=field): + (reply if field == "id" else reply["result"])[field] = "other-request" + self.guardian.resign(reply) + with self.subTest(field=field): + self.guardian.mutate = mutate + out, audit = self.invoke() + self.assertEqual(out["permissionDecision"], "deny") + self.assertIn("response_binding_mismatch", audit) + + def test_an_unreadable_response_follows_the_posture(self) -> None: + for raw in (b"no json", b"[]", b'{"jsonrpc":"2.0","result":[]}'): + for posture in ("0", "1"): + with self.subTest(raw=raw, posture=posture): + self.env["ACS_DEFAULT_DENY"] = posture + self.guardian.raw_reply = raw + out, audit = self.invoke() + self.assertEqual(out.get("permissionDecision"), "deny" if posture == "1" else None) + self.assertIn("malformed_response", audit) + + def test_an_http_refusal_denies_under_fail_open(self) -> None: + self.guardian.raw_reply = b"refused" + self.guardian.reply_status = 400 + out, audit = self.invoke() + self.assertEqual(out["permissionDecision"], "deny") + self.assertIn("http_400_refusal", audit) + + def test_an_unanswered_handshake_follows_the_startup_posture(self) -> None: + """§4.1: a session with no negotiated Guardian denies its calls unless the hook sets ACS_STARTUP_POSTURE=proceed.""" + base = {k: v for k, v in self.env.items() if k not in ("ACS_DEFAULT_DENY", "ACS_STARTUP_POSTURE")} + base["ACS_GUARDIAN_URL"] = "http://127.0.0.1:1/acs" + table = [({}, "deny", "session_not_started_fail_closed"), + ({"ACS_DEFAULT_DENY": "0"}, "deny", "session_not_started_fail_closed"), + ({"ACS_DEFAULT_DENY": "1"}, "deny", "session_not_started_fail_closed"), + ({"ACS_STARTUP_POSTURE": "proceed"}, None, "unguarded_session_bypass")] + for overrides, decision, audit_event in table: + with self.subTest(env=overrides): + self.env = {**base, **overrides} + self.fresh_session_records() + out, audit = self.invoke() + self.assertEqual(out.get("permissionDecision"), decision) + self.assertIn(audit_event, audit) + + def test_a_server_hello_that_does_not_answer_the_client_hello_sends_no_step(self) -> None: + self.env["ACS_DEFAULT_DENY"] = "1" + for change in ({"timeout_config": {"default_ms": True}}, {"selected_transport": "stdio"}, + {"methods_evaluated": "steps/toolCallRequest"}, {"methods_evaluated": [None]}, + {"on_decision_failure": "unknown"}, {"negotiated_version": "9.0.0"}): + with self.subTest(change=change): + self.fresh_session_records() + self.hello(**change) + out, audit = self.invoke() + self.assertEqual(out["permissionDecision"], "deny") + self.assertIn("invalid_server_hello", audit) + self.assertNotIn("steps/toolCallRequest", self.guardian.methods()) + + def test_a_decision_timeout_past_the_hook_budget_denies_before_the_step(self) -> None: + self.hello(timeout_config={"default_ms": 30000}) + out, audit = self.invoke() + self.assertEqual(out["permissionDecision"], "deny") + self.assertIn("unsupported_decision_timeout", audit) + self.assertNotIn("steps/toolCallRequest", self.guardian.methods()) + + def test_the_local_deny_posture_holds_against_a_server_proceed(self) -> None: + self.env["ACS_DEFAULT_DENY"] = "1" + self.hello(on_decision_failure="proceed", timeout_config={"default_ms": 50}) + + def slow(req): + time.sleep(0.15) + return self.guardian._default_allow(req) + self.guardian.handlers["steps/toolCallRequest"] = slow + out, audit = self.invoke() + self.assertEqual(out["permissionDecision"], "deny") + self.assertIn("decision_failure_fail_closed", audit) + + def test_the_negotiated_timeout_and_posture_apply(self) -> None: + for posture in ("proceed", "deny"): + with self.subTest(posture=posture): + self.fresh_session_records() + self.hello(on_decision_failure=posture, timeout_config={"default_ms": 50}) + + def slow(req): + time.sleep(0.15) + return self.guardian._default_allow(req) + self.guardian.handlers["steps/toolCallRequest"] = slow + out, audit = self.invoke() + self.assertEqual(out.get("permissionDecision"), "deny" if posture == "deny" else None) + self.assertIn("decision_timeout", audit) + + def test_an_unevaluated_method_leaves_codex_in_charge(self) -> None: + for local in ("0", "1"): + for negotiated in ("proceed", "deny"): + with self.subTest(local=local, negotiated=negotiated): + self.fresh_session_records() + self.env["ACS_DEFAULT_DENY"] = local + self.hello(methods_evaluated=[], on_decision_failure=negotiated) + before = len(self.guardian.received) + out, audit = self.invoke() + self.assertEqual(out, {}) + self.assertIn("method_not_evaluated", audit) + self.assertNotIn("fail_open_bypass", audit) + self.assertNotIn("decision_failure_fail_closed", audit) + self.assertEqual(self.guardian.methods_seen()[before:], ["handshake/hello"]) + # The stored session decision answers the next call with no traffic. + before = len(self.guardian.received) + out, audit = self.invoke(event(tool_use_id="call_cached")) + self.assertEqual(out, {}) + self.assertIn("method_not_evaluated", audit) + self.assertEqual(self.guardian.methods_seen()[before:], []) + self.guardian.assert_all_valid(self) + + def test_an_unevaluated_method_does_not_skip_the_server_hello_check(self) -> None: + self.guardian.sign_responses = False + self.hello(methods_evaluated=[]) + out, audit = self.invoke() + self.assertEqual(out["permissionDecision"], "deny") + self.assertIn("server_hello_signature_invalid", audit) + self.assertNotIn("method_not_evaluated", audit) + self.assertEqual(self.guardian.methods(), ["handshake/hello"]) + + def test_a_signed_guardian_error_follows_the_posture_unless_it_is_a_refusal(self) -> None: + for code in (-32603, -32005): + for posture in ("0", "1"): + with self.subTest(code=code, posture=posture): + self.env["ACS_DEFAULT_DENY"] = posture + + def mutate(reply, code=code): + reply.pop("result") + reply["error"] = {"code": code, "message": "fixture"} + self.guardian.resign(reply) + self.guardian.mutate = mutate + out, _ = self.invoke() + refused = code == -32005 + self.assertEqual(out.get("permissionDecision"), "deny" if refused or posture == "1" else None) + + def test_a_session_handshakes_once_and_request_ids_are_session_and_turn_scoped(self) -> None: + self.invoke() + self.invoke(event(tool_use_id="call_4")) + # tool_use_id is the model's own call id, so a later turn, or a subagent's turn, can reuse it. + out, _ = self.invoke(event(turn_id="turn_8")) + self.assertEqual(out, {}, "a reused call id in another turn is not a replay") + self.assertEqual(self.guardian.methods().count("handshake/hello"), 1) + self.invoke(event(session_id="thr_other")) + self.assertEqual(self.guardian.methods().count("handshake/hello"), 2) + self.assertEqual(self.guardian.duplicate_request_ids(), []) + self.guardian.assert_all_valid(self) + + def test_a_subagent_step_names_its_subagent(self) -> None: + self.invoke(event(agent_id="019e-subagent", agent_type="worker", turn_id="turn_sub")) + self.invoke(event(tool_use_id="call_parent")) + sub, parent = (c.parsed["params"]["metadata"] for c in self.guardian.captures + if c.parsed["method"] == "steps/toolCallRequest") + self.assertEqual((sub["codex_agent_id"], sub["codex_agent_type"]), ("019e-subagent", "worker")) + self.assertEqual(sub["session_id"], parent["session_id"]) + self.assertNotIn("codex_agent_id", parent) + self.guardian.assert_all_valid(self) + + def test_disabled_sends_nothing_and_is_audited(self) -> None: + self.env["ACS_DISABLED"] = "1" + out, audit = self.invoke() + self.assertEqual(out, {}) + self.assertIn("acs_disabled_bypass", audit) + self.assertEqual(self.guardian.methods(), []) + + def test_the_hook_command_follows_the_posture_when_the_adapter_cannot_load(self) -> None: + """No Guardian can decide a call the adapter cannot send, so it proceeds only when the hook opts into both postures.""" + shadow = Path(self.tmp.name) / "shadow" + shadow.mkdir() + (shadow / "rfc8785.py").write_text("raise ImportError('test missing dependency')\n") + base = {k: v for k, v in self.env.items() if k not in ("ACS_DEFAULT_DENY", "ACS_STARTUP_POSTURE")} + base["PYTHONPATH"] = str(shadow) + table = [({}, "deny"), ({"ACS_DEFAULT_DENY": "0"}, "deny"), ({"ACS_DEFAULT_DENY": "1"}, "deny"), + ({"ACS_STARTUP_POSTURE": "proceed"}, "deny"), + ({"ACS_STARTUP_POSTURE": "proceed", "ACS_DEFAULT_DENY": "0"}, None)] + for overrides, decision in table: + with self.subTest(env=overrides): + self.env = {**base, **overrides} + out, audit = self.invoke(command=[str(HOOK)]) + self.assertEqual(out.get("permissionDecision"), decision) + self.assertIn("adapter_unavailable", audit) + self.assertEqual(self.guardian.methods(), []) + + def test_the_hook_command_runs_the_adapter(self) -> None: + out, _ = self.invoke(command=[str(HOOK)]) + self.assertEqual(out, {}) + self.assertEqual(self.guardian.methods(), ["handshake/hello", "steps/toolCallRequest"]) + + def test_a_subagent_spawn_is_gated_as_a_tool_call(self) -> None: + """Codex runs a spawn as the spawn_agent tool, and SubagentStart cannot stop one, so PreToolUse is the gate.""" + spawn = event(tool_name="spawn_agent", tool_input={"message": "review the diff", "agent_type": "worker"}) + self.verdict("deny", reasoning="no subagents") + out, _ = self.invoke(spawn) + self.assertEqual(out["permissionDecision"], "deny") + payload = self.guardian.payload_of("steps/toolCallRequest") + self.assertEqual(payload["tool"]["name"], "spawn_agent") + self.assertEqual(payload["arguments"]["message"], {"value": "review the diff"}) + self.verdict("modify", modifications={"parameter_overrides": {"message": "summarize the diff"}}) + out, _ = self.invoke(event(tool_name="spawn_agent", tool_use_id="call_4", tool_input=spawn["tool_input"])) + self.assertEqual(out["permissionDecision"], "allow") + self.assertEqual(out["updatedInput"], {"message": "summarize the diff", "agent_type": "worker"}) + + def test_the_example_hook_covers_shell_and_spawns_and_blocks_when_it_cannot_run(self) -> None: + example = Path(__file__).resolve().parents[3] / "docs" / "codex" / "hooks.json.example" + entry = json.loads(example.read_text(encoding="utf-8"))["hooks"]["PreToolUse"][0] + for name, matched in (("Bash", True), ("apply_patch", True), ("spawn_agent", True), ("BashTool", False), + ("multi_agent_v1wait_agent", False), ("mcp__local__spawn_agent", False)): + with self.subTest(tool=name): + self.assertEqual(re.search(entry["matcher"], name) is not None, matched) + words = shlex.split(entry["hooks"][0]["command"]) + self.assertEqual(words[:2], ["sh", "-c"]) + self.assertEqual(len(words), 3) + hook, fallback = words[2].split(" || ", 1) + self.assertTrue(hook.endswith("/acs-codex-hook")) + # sh takes the assignments itself, so the hook needs no env process. + self.assertTrue(all("=" in word for word in shlex.split(hook)[:-1]), hook) + # Codex runs the call after any hook failure but exit 2, so a hook that cannot start must exit 2. + missing = Path(self.tmp.name) / "missing" / "acs-codex-hook" + proc = subprocess.run(["sh", "-c", f"{shlex.quote(str(missing))} || {fallback}"], + input=json.dumps(event()), capture_output=True, text=True, timeout=20) + self.assertEqual(proc.returncode, 2) + self.assertEqual(proc.stdout, "") + self.assertIn("blocked", proc.stderr) + # A hook that runs keeps its exit 0 and its decision. + self.verdict("deny", reasoning="policy") + out, _ = self.invoke(command=["sh", "-c", f"{shlex.quote(str(HOOK))} || {fallback}"]) + self.assertEqual(out["permissionDecision"], "deny") + + def test_the_example_guardian_allows_denies_and_rejects_a_replay(self) -> None: + self.env["ACS_GUARDIAN_STATE_DIR"] = str(Path(self.tmp.name) / "guardian-state") + proc, port = launch_guardian(env=self.env) + try: + self.env["ACS_GUARDIAN_URL"] = f"http://127.0.0.1:{port}/acs" + out, _ = self.invoke() + self.assertEqual(out, {}) + out, _ = self.invoke() + self.assertEqual(out["permissionDecision"], "deny", "the same call id is a replay") + # Adapter input only: the test never runs this command. + out, _ = self.invoke(event(tool_use_id="different-call", + tool_input={"command": "rm -rf /acs-test-do-not-execute"})) + self.assertEqual(out["permissionDecision"], "deny") + finally: + proc.terminate() + proc.communicate(timeout=5) + + +if __name__ == "__main__": + unittest.main() diff --git a/adapters/tests/package/test_metadata.py b/adapters/tests/package/test_metadata.py index dd778ad0..2d1d5274 100644 --- a/adapters/tests/package/test_metadata.py +++ b/adapters/tests/package/test_metadata.py @@ -12,6 +12,7 @@ SCRIPTS = { "acs-claude-code-hook": "acs.integrations.claude_code.hook:main", "acs-claude-code-wire": "acs.integrations.claude_code.wire:main", + "acs-codex-hook": "acs.integrations.codex.hook:main", "acs-cursor-hook": "acs.integrations.cursor.hook:main", "acs-cursor-wire": "acs.integrations.cursor.wire:main", "acs-nat-wire": "acs.integrations.nat.wire:main", @@ -22,11 +23,12 @@ class Metadata(unittest.TestCase): def test_one_version_from_package_metadata(self) -> None: from acs.integrations.claude_code import adapter as claude + from acs.integrations.codex import adapter as codex from acs.integrations.cursor import adapter as cursor from acs.integrations.nat import adapter as nat version = metadata.version("acs-adapters") self.assertEqual(acs.__version__, version) - for module in (claude, cursor, nat): + for module in (claude, codex, cursor, nat): with self.subTest(module=module.__name__): self.assertEqual(module.ADAPTER_VERSION, version) @@ -43,7 +45,7 @@ def test_scripts_load_with_only_the_base_dependencies(self) -> None: def test_integrations_entry_point_group_lists_each_integration(self) -> None: group = {e.name: e for e in entry_points(group="acs.integrations")} - self.assertEqual(set(group), {"claude_code", "cursor", "nat"}) + self.assertEqual(set(group), {"claude_code", "codex", "cursor", "nat"}) for name, entry in group.items(): with self.subTest(integration=name): self.assertEqual(entry.load().__name__, f"acs.integrations.{name}") @@ -54,7 +56,8 @@ def test_nat_discovers_the_middleware_module(self) -> None: def test_installed_hook_scripts_run(self) -> None: bindir = Path(sys.executable).parent - for name, label in (("acs-claude-code-hook", "claude-code"), ("acs-cursor-hook", "cursor")): + for name, label in (("acs-claude-code-hook", "claude-code"), ("acs-codex-hook", "codex"), + ("acs-cursor-hook", "cursor")): with self.subTest(script=name): p = subprocess.run([str(bindir / name), "--version"], capture_output=True, text=True, timeout=30)