From f0a8f963e30dda69a7f115200f2d27add27970d4 Mon Sep 17 00:00:00 2001 From: Diego Marinho <4332052+dmoliveira@users.noreply.github.com> Date: Tue, 10 Nov 2026 16:12:40 +1100 Subject: [PATCH] test: measure V2 direct shell plugin activation --- .github/workflows/ci.yml | 1 + docs/plan/doing/opencode-v2-migration.md | 15 + .../v2/direct_shell_activation_live_smoke.py | 388 ++++++++++++++++++ ...test_gateway_v2_direct_shell_activation.py | 141 +++++++ 4 files changed, 545 insertions(+) create mode 100644 plugin/gateway-core/test/v2/direct_shell_activation_live_smoke.py create mode 100644 tests/test_gateway_v2_direct_shell_activation.py diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index 6cae9f42..3f6f23fe 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -149,6 +149,7 @@ jobs: python scripts/gateway_v2_tool_error_live_smoke.py python scripts/gateway_v2_shell_live_smoke.py python plugin/gateway-core/test/v2/execution_status_live_smoke.py + python plugin/gateway-core/test/v2/direct_shell_activation_live_smoke.py windows-bounded-subprocess: runs-on: windows-latest diff --git a/docs/plan/doing/opencode-v2-migration.md b/docs/plan/doing/opencode-v2-migration.md index 25e90650..cf8f5e11 100644 --- a/docs/plan/doing/opencode-v2-migration.md +++ b/docs/plan/doing/opencode-v2-migration.md @@ -69,6 +69,21 @@ trusted later hook from changing the shell and `BASH_ENV` so that an owned startup file ran the same denied command. This is a process-binding limitation, not a claim that policies sandbox plugin code. +`python plugin/gateway-core/test/v2/direct_shell_activation_live_smoke.py` +measures a separate, authenticated direct-shell path in disposable local +servers on 2.2.5 and 2.2.6. With a configured plugin deliberately paused in +setup, a benign direct API command completes before the shell hook is +registered. After setup, the hook sees and can reject a direct command, but +the host does not supply the session-ID update observed for model-driven shell +calls. `tool.use` is an agent-tool policy; these controls do not claim that it +governs direct user API calls or demonstrate external access. MEASURED means +the known host behavior was reproduced, not that a guard is active. CI measures +both pinned versions. A plugin-only universal admission boundary cannot rely +on this hook being active from server startup; the host's +[direct-shell handler](https://github.com/anomalyco/opencode/blob/dc2ffcf8bf26f9e669b6aa22b95f50e12be15e06/packages/server/src/handlers/shell.ts) +and [shell creation](https://github.com/anomalyco/opencode/blob/dc2ffcf8bf26f9e669b6aa22b95f50e12be15e06/packages/core/src/shell.ts) +remain relevant to a host-level solution. + Experimental, unconfigured adapters cover V2 model-context and native transport redaction, tool-output and error redaction, two shell guards, and write/protected-worktree guards. `make gateway-v2-slice-smoke` builds and exercises them with isolated OpenCode processes, synthetic data, disposable Git worktrees and a localhost provider. The live checks cover a late context mutation before the HTTP hook, rejection of unsupported HTTP content, allowed and blocked shell commands, tool-output redaction, sanitized tool errors and redactor failures, explicit permission denial, interactive rejection with or without feedback, one running-tool interruption, and linked versus primary worktree writes. The error adapter accepts only OpenCode 2.2.5 or 2.2.6. It replaces the error with a host `Tool.Error` containing a redacted or generic message and no original metadata. For a plain rejection, it builds a clean cause from a separately supplied trusted host-decline factory and never takes a prototype from the tool's error. Registration calls the factory once and refuses to install the hook if it fails or produces a structurally invalid sample. The structural check cannot authenticate a host type; the caller must establish that the sample and its prototype are trusted. The permission smoke first captures a genuine decline from its own controlled `ctx.permission.assert` call, then registers the adapter with that independently obtained sample. On both tested versions, a forged matching prototype exposed a synthetic canary through custom inspection without the adapter; with the adapter, inspection was clean and the rejection still stopped the step. This fixture is not a production source of trusted types. The 2.2.6 public plugin API does not expose the host's decline constructor, and a factory that derives its prototype from an arbitrary tool error would reintroduce the vulnerability. A trusted production capability is still required; this adapter does not protect the trusted prototype against same-privilege mutation by other plugins. Other AFTER-hook failures can reach the model as tool errors rather than stop the step, so registration alone does not establish an independent fail-closed boundary. diff --git a/plugin/gateway-core/test/v2/direct_shell_activation_live_smoke.py b/plugin/gateway-core/test/v2/direct_shell_activation_live_smoke.py new file mode 100644 index 00000000..9b3ea8fa --- /dev/null +++ b/plugin/gateway-core/test/v2/direct_shell_activation_live_smoke.py @@ -0,0 +1,388 @@ +"""Measure whether the V2 direct shell API waits for a configured plugin. + +MEASURED means the pinned hosts reproduced the observed behavior, not that +shell execution is guarded. All commands run in owned disposable fixtures. +""" + +from __future__ import annotations + +import argparse +import http.client +import json +import os +import re +import shutil +import stat +import subprocess +import sys +import tempfile +import threading +import time +from collections.abc import Callable +from pathlib import Path +from urllib.parse import urlencode + +ROOT = Path(__file__).resolve().parents[4] +sys.path.insert(0, str(ROOT / "scripts")) + +from gateway_v2_auxiliary_redaction_probe import write_fixture +from gateway_v2_dependency_contract_probe import ( + MANAGERS, + ProbeFailure, + child_env, + preflight, + run_child, +) +from gateway_v2_dependency_profile_probe import ( + PrivateRuntime, + activated_inventory, + owned_fixture, +) + +VERSIONS = {"opencode v2.2.5", "opencode v2.2.6"} +CASE_SECONDS = 40 +CLEANUP_SECONDS = 10 + + +def read_owned(path: Path, limit: int = 4096) -> str | None: + try: + descriptor = os.open(path, os.O_RDONLY | os.O_NOFOLLOW | os.O_NONBLOCK) + except FileNotFoundError: + return None + try: + info = os.fstat(descriptor) + if ( + not stat.S_ISREG(info.st_mode) + or info.st_uid != os.getuid() + or info.st_nlink != 1 + or info.st_size > limit + ): + raise ProbeFailure("fixture_observation_invalid") + content = os.read(descriptor, limit + 1) + if len(content) > limit: + raise ProbeFailure("fixture_observation_invalid") + return content.decode("utf-8") + finally: + os.close(descriptor) + + +def wait_for( + predicate: Callable[[], bool], runtime: PrivateRuntime, reason: str +) -> None: + deadline = min(runtime.deadline, time.monotonic() + 8) + while time.monotonic() < deadline: + # Keep the process-group leader unreaped until PrivateRuntime.close() + # signals its group; poll() would release that identity too early. + if runtime.process is None: + raise ProbeFailure("runtime_not_started") + if predicate(): + return + threading.Event().wait(0.025) + raise ProbeFailure(reason) + + +def wait_for_shell_exit( + runtime: PrivateRuntime, response: object, command: str, cwd: Path +) -> None: + """Confirm an owned direct shell exited; a timeout status is not settlement.""" + if not isinstance(response, dict) or not isinstance(response.get("data"), dict): + raise ProbeFailure("shell_response_invalid") + shell_id = response["data"].get("id") + if not isinstance(shell_id, str) or not re.fullmatch( + r"[A-Za-z0-9_-]{1,160}", shell_id + ): + raise ProbeFailure("shell_identity_invalid") + for _attempt in range(6): + result = runtime.request("GET", f"/api/shell/{shell_id}") + if not isinstance(result, dict) or not isinstance(result.get("data"), dict): + raise ProbeFailure("shell_status_invalid") + info = result["data"] + if ( + info.get("id") != shell_id + or info.get("command") != command + or info.get("cwd") != str(cwd) + or info.get("shell") != "/bin/sh" + ): + raise ProbeFailure("shell_identity_mismatch") + if info.get("status") == "exited" and info.get("exit") == 0: + return + if info.get("status") != "running": + raise ProbeFailure("shell_exit_unsuccessful") + threading.Event().wait(0.25) + raise ProbeFailure("shell_settlement_unconfirmed") + + +def prepare(root: Path) -> None: + for name in ("home", "project", "tmp", "bin"): + (root / name).mkdir(mode=0o700) + trace = root / "trace" + gate = root / "release" + plugin = f"""import {{ Plugin }} from "@opencode/plugin" +import {{ appendFileSync, existsSync, writeFileSync }} from "node:fs" +const trace = {json.dumps(str(trace))}, gate = {json.dumps(str(gate))} +const record = value => appendFileSync(trace, value + "\\n", {{ mode: 0o600 }}) +export default Plugin.define({{ id: "profile-owned", async setup(ctx) {{ + writeFileSync(trace, "setup\\n", {{ mode: 0o600 }}) + const deadline = Date.now() + 15000 + while (!existsSync(gate)) {{ + if (Date.now() >= deadline) throw new Error("PROBE_SETUP_TIMEOUT") + await new Promise(resolve => setTimeout(resolve, 20)) + }} + await ctx.shell.hook("create.before", event => {{ + record("hook") + if (event.command === "printf BLOCKED > blocked-witness") {{ + record("blocked") + throw new Error("PROBE_BLOCKED") + }} + let id = event.env.OPENCODE_SESSION_ID + Object.defineProperty(event.env, "OPENCODE_SESSION_ID", {{ + enumerable: true, configurable: false, get() {{ return id }}, + set(value) {{ record("session-set"); id = value }}, + }}) + }}) + record("ready") +}} }}) +""" + write_fixture(root, "project/owned/index.ts", plugin) + config = { + "plugins": [str(root / "project/owned")], + "shell": "/bin/sh", + "update": "disable", + "snapshots": False, + "formatter": False, + "permissions": [{"action": "*", "resource": "*", "effect": "deny"}], + "policies": [{"action": "tool.use", "resource": "shell:*", "effect": "deny"}], + } + write_fixture(root, "project/opencode.json", json.dumps(config)) + write_fixture(root, "home/.config/opencode/opencode.json", '{"update":"disable"}') + for manager in (*MANAGERS, "corepack", "pnpx", "yarnpkg", "pipx", "uvx"): + write_fixture(root, f"bin/{manager}", "#!/bin/sh\nexit 97\n", executable=True) + + +def run_probe(binary: str, parent: Path) -> dict[str, object]: + preflight(parent) + root = Path(tempfile.mkdtemp(prefix="v2-direct-shell-", dir=parent)) + identity = root.lstat() + runtime: PrivateRuntime | None = None + warm_thread: threading.Thread | None = None + warm_started = False + warm_complete = threading.Event() + cleanup_errors: list[str] = [] + shell_unsettled = False + report: dict[str, object] = { + "result": "FAIL", + "meaning": "measurement only; not protection", + } + try: + prepare(root) + code, stdout, _ = run_child( + [binary, "--version"], root / "project", child_env(root), timeout=10 + ) + version = stdout.decode("utf-8", errors="replace").strip() + if code != 0 or version not in VERSIONS: + raise ProbeFailure("unsupported_host_version") + report["host"] = version + runtime = PrivateRuntime(binary, root, time.monotonic() + CASE_SECONDS) + runtime.start() + runtime.ready() + + def warm() -> None: + try: + runtime.request( + "GET", + "/api/plugin?" + + urlencode({"location[directory]": str(root / "project")}), + ) + except (ProbeFailure, OSError, ValueError, http.client.HTTPException): + return + warm_complete.set() + + warm_thread = threading.Thread(target=warm, daemon=True) + warm_thread.start() + warm_started = True + trace = root / "trace" + wait_for( + lambda: read_owned(trace) == "setup\n", runtime, "plugin_setup_not_observed" + ) + shell_unsettled = True + first = runtime.mutate( + "POST", + "/api/shell", + { + "command": "printf FIRST > first-witness", + "cwd": str(root / "project"), + "timeout": 5000, + }, + ) + first_witness = root / "project/first-witness" + wait_for( + lambda: read_owned(first_witness) == "FIRST", + runtime, + "first_witness_missing", + ) + wait_for_shell_exit( + runtime, first, "printf FIRST > first-witness", root / "project" + ) + shell_unsettled = False + before_release = read_owned(trace) + write_fixture(root, "release", "release\n") + wait_for( + lambda: "ready\n" in (read_owned(trace) or ""), + runtime, + "plugin_ready_not_observed", + ) + warm_thread.join(timeout=5) + if warm_thread.is_alive() or not warm_complete.is_set(): + raise ProbeFailure("plugin_list_did_not_settle") + inventory = activated_inventory(runtime, root) + if not any( + item.get("source") == "owned" and item.get("status") == "active" + for item in inventory + ): + raise ProbeFailure("fixture_plugin_not_active") + shell_unsettled = True + second = runtime.mutate( + "POST", + "/api/shell", + { + "command": "printf SECOND > second-witness", + "cwd": str(root / "project"), + "timeout": 5000, + }, + ) + second_witness = root / "project/second-witness" + wait_for( + lambda: read_owned(second_witness) == "SECOND", + runtime, + "second_witness_missing", + ) + wait_for_shell_exit( + runtime, second, "printf SECOND > second-witness", root / "project" + ) + shell_unsettled = False + blocked_status = None + shell_unsettled = True + try: + blocked = runtime.mutate( + "POST", + "/api/shell", + { + "command": "printf BLOCKED > blocked-witness", + "cwd": str(root / "project"), + "timeout": 5000, + }, + ) + wait_for_shell_exit( + runtime, blocked, "printf BLOCKED > blocked-witness", root / "project" + ) + shell_unsettled = False + except ProbeFailure as error: + if str(error) != "native_api_status_500": + raise + blocked_status = str(error) + lines = (read_owned(trace) or "").splitlines() + if blocked_status == "native_api_status_500" and lines[-2:] == [ + "hook", + "blocked", + ]: + shell_unsettled = False + report.update( + { + "first_response_created": isinstance(first, dict) + and isinstance(first.get("data"), dict), + "first_witness_before_plugin_ready": read_owned(first_witness) + == "FIRST" + and before_release == "setup\n", + "configured_plugin_active": True, + "second_response_created": isinstance(second, dict) + and isinstance(second.get("data"), dict), + "second_witness_after_plugin_ready": read_owned(second_witness) + == "SECOND", + "post_activation_hook_seen": lines + == ["setup", "ready", "hook", "hook", "blocked"], + "post_activation_host_session_set": "session-set" in lines, + "post_activation_block_refused": blocked_status + == "native_api_status_500" + and "blocked" in lines, + } + ) + except ( + ProbeFailure, + OSError, + ValueError, + KeyError, + TypeError, + RuntimeError, + http.client.HTTPException, + subprocess.SubprocessError, + ) as error: + report["reason"] = ( + str(error) if isinstance(error, ProbeFailure) else "probe_failed" + ) + finally: + if shell_unsettled: + cleanup_errors.append("shell_settlement_unconfirmed") + if runtime is not None: + try: + runtime.close(time.monotonic() + CLEANUP_SECONDS) + except (ProbeFailure, OSError, RuntimeError, subprocess.SubprocessError): + cleanup_errors.append("runtime_cleanup_failed") + if warm_thread is not None and warm_started: + warm_thread.join(timeout=5) + if warm_thread.is_alive(): + cleanup_errors.append("plugin_list_thread_alive") + try: + if read_owned(root / "project/blocked-witness") is not None: + cleanup_errors.append("blocked_witness_present") + except (ProbeFailure, OSError, UnicodeError): + cleanup_errors.append("blocked_witness_invalid") + if not owned_fixture(root, identity): + cleanup_errors.append("fixture_identity_changed") + if not cleanup_errors: + try: + shutil.rmtree(root) + except OSError: + cleanup_errors.append("fixture_removal_failed") + report["cleanup_complete"] = not cleanup_errors + if cleanup_errors: + report["reason"] = "cleanup_failed" + if measured(report): + report["result"] = "MEASURED" + return report + + +def measured(report: dict[str, object]) -> bool: + """Success reproduces a measured limitation; it never grants authority.""" + return ( + "reason" not in report + and report.get("host") in VERSIONS + and report.get("first_response_created") is True + and report.get("first_witness_before_plugin_ready") is True + and report.get("configured_plugin_active") is True + and report.get("second_response_created") is True + and report.get("second_witness_after_plugin_ready") is True + and report.get("post_activation_hook_seen") is True + and report.get("post_activation_host_session_set") is False + and report.get("post_activation_block_refused") is True + and report.get("cleanup_complete") is True + ) + + +def main() -> int: + parser = argparse.ArgumentParser(description=__doc__) + parser.add_argument("--opencode-bin", default="opencode") + parser.add_argument("--temp-parent", type=Path, default=Path(tempfile.gettempdir())) + args = parser.parse_args() + binary = shutil.which(args.opencode_bin) + if not binary or os.name != "posix": + parser.error("an installed POSIX OpenCode 2.2.5 or 2.2.6 is required") + try: + report = run_probe(str(Path(binary).resolve()), args.temp_parent.resolve()) + except (ProbeFailure, OSError, ValueError, RuntimeError): + report = {"result": "FAIL", "reason": "probe_preflight_failed"} + print(json.dumps(report, sort_keys=True)) + return 0 if report.get("result") == "MEASURED" else 1 + + +if __name__ == "__main__": + raise SystemExit(main()) diff --git a/tests/test_gateway_v2_direct_shell_activation.py b/tests/test_gateway_v2_direct_shell_activation.py new file mode 100644 index 00000000..c4362f30 --- /dev/null +++ b/tests/test_gateway_v2_direct_shell_activation.py @@ -0,0 +1,141 @@ +from __future__ import annotations + +import os +import sys +import tempfile +import time +import unittest +from pathlib import Path +from types import SimpleNamespace +from unittest.mock import Mock, patch + +if os.name == "posix": + sys.path.insert( + 0, str(Path(__file__).resolve().parents[1] / "plugin/gateway-core/test/v2") + ) + import direct_shell_activation_live_smoke as probe + + +def complete_measurement() -> dict[str, object]: + return { + "host": "opencode v2.2.6", + "first_response_created": True, + "first_witness_before_plugin_ready": True, + "configured_plugin_active": True, + "second_response_created": True, + "second_witness_after_plugin_ready": True, + "post_activation_hook_seen": True, + "post_activation_host_session_set": False, + "post_activation_block_refused": True, + "cleanup_complete": True, + } + + +@unittest.skipUnless(os.name == "posix", "native shell probe is POSIX-only") +class DirectShellActivationTests(unittest.TestCase): + def test_wait_keeps_runtime_leader_unreaped_until_group_cleanup(self) -> None: + process = Mock() + runtime = SimpleNamespace(process=process, deadline=time.monotonic() + 1) + probe.wait_for(lambda: True, runtime, "not_observed") + process.poll.assert_not_called() + + def test_direct_shell_requires_matching_successful_exit_not_timeout(self) -> None: + command = "printf FIRST > first-witness" + cwd = Path("/private-owned-fixture") + created = {"data": {"id": "shell_owned"}} + info = { + "id": "shell_owned", + "command": command, + "cwd": str(cwd), + "shell": "/bin/sh", + } + runtime = SimpleNamespace( + request=Mock(return_value={"data": {**info, "status": "exited", "exit": 0}}) + ) + probe.wait_for_shell_exit(runtime, created, command, cwd) + for update in ( + {"status": "timeout"}, + {"status": "exited", "exit": 1}, + {"status": "exited", "exit": 0, "id": "shell_other"}, + {"status": "exited", "exit": 0, "shell": "/other"}, + ): + with self.subTest(update=update): + runtime.request.return_value = {"data": {**info, **update}} + with self.assertRaises(probe.ProbeFailure): + probe.wait_for_shell_exit(runtime, created, command, cwd) + + def test_measurement_requires_all_observations_and_successful_cleanup(self) -> None: + evidence = complete_measurement() + self.assertTrue(probe.measured(evidence)) + for name in evidence: + with self.subTest(missing=name): + incomplete = dict(evidence) + del incomplete[name] + self.assertFalse(probe.measured(incomplete)) + for change in ( + {"host": "opencode v2.2.7"}, + {"post_activation_host_session_set": True}, + {"cleanup_complete": False}, + {"reason": "probe_failed"}, + ): + with self.subTest(change=change): + self.assertFalse(probe.measured({**evidence, **change})) + + def test_observations_do_not_follow_links_or_accept_hardlinks(self) -> None: + with tempfile.TemporaryDirectory() as temporary: + root = Path(temporary) + regular = root / "regular" + regular.write_text("ready", encoding="utf-8") + self.assertEqual(probe.read_owned(regular), "ready") + self.assertIsNone(probe.read_owned(root / "absent")) + symlink = root / "symlink" + symlink.symlink_to(regular) + with self.assertRaises(OSError): + probe.read_owned(symlink) + os.link(regular, root / "hardlink") + with self.assertRaises(probe.ProbeFailure): + probe.read_owned(regular) + + def test_failed_fixture_removal_cannot_report_a_successful_measurement( + self, + ) -> None: + with tempfile.TemporaryDirectory() as temporary: + parent = Path(temporary) + with ( + patch.object(probe, "preflight"), + patch.object( + probe, "prepare", side_effect=probe.ProbeFailure("setup_failed") + ), + patch.object( + probe.shutil, "rmtree", side_effect=OSError("cleanup failed") + ), + ): + report = probe.run_probe("unused", parent) + self.assertFalse(report["cleanup_complete"]) + self.assertEqual(report["reason"], "cleanup_failed") + self.assertEqual(report["result"], "FAIL") + self.assertEqual(len(list(parent.iterdir())), 1) + + def test_unknown_shell_settlement_preserves_the_fixture(self) -> None: + with tempfile.TemporaryDirectory() as temporary: + parent = Path(temporary) + runtime = Mock() + runtime.mutate.side_effect = probe.ProbeFailure("response_lost") + with ( + patch.object(probe, "preflight"), + patch.object(probe, "prepare"), + patch.object( + probe, "run_child", return_value=(0, b"opencode v2.2.6", b"") + ), + patch.object(probe, "PrivateRuntime", return_value=runtime), + patch.object(probe, "wait_for"), + ): + report = probe.run_probe("unused", parent) + self.assertFalse(report["cleanup_complete"]) + self.assertEqual(report["reason"], "cleanup_failed") + self.assertEqual(report["result"], "FAIL") + self.assertEqual(len(list(parent.iterdir())), 1) + + +if __name__ == "__main__": + unittest.main()