diff --git a/Makefile b/Makefile index 405ddb12..094d535a 100644 --- a/Makefile +++ b/Makefile @@ -13,7 +13,7 @@ OPENCODE_RESUME_E2E_VERSION := 1.18.18 help: ## Show available targets @awk 'BEGIN {FS = ":.*##"} /^[a-zA-Z0-9_-]+:.*##/ {printf "%-14s %s\n", $$1, $$2}' $(MAKEFILE_LIST) -PYTHON_TARGETS := data-lifecycle-help validate build-agents build-agents-check build-agents-prune release-index-update docs-automation-summary-update docs-automation-check pages-readiness-check release-note-validation-check release-note-quality-check plan-hygiene-check wave-linkage-check wave-handoff-summary wave-completion-update quality-fast quality-strict quality-off quality-status tasker-production-live-smoke rolling-decomposition-live gateway-status gateway-enable gateway-disable gateway-doctor gateway-secret-redaction-smoke gateway-v2-dependency-probe gateway-v2-dependency-profile-probe gateway-v2-slice-smoke gateway-resume-redaction-e2e gateway-resume-redaction-e2e-prebuilt gateway-execution-status-live-smoke gateway-turn-watch gateway-turn-watch-webhook harness-wave2-task4-smoke notify-icons-generate notify-icons-select reservation-status task-lease-status selftest doctor doctor-json sqlite-doctor sqlite-doctor-json devtools-status hooks-install install-test install-test-full release-check release +PYTHON_TARGETS := data-lifecycle-help validate build-agents build-agents-check build-agents-prune release-index-update docs-automation-summary-update docs-automation-check pages-readiness-check release-note-validation-check release-note-quality-check plan-hygiene-check wave-linkage-check wave-handoff-summary wave-completion-update quality-fast quality-strict quality-off quality-status tasker-production-live-smoke rolling-decomposition-live gateway-status gateway-enable gateway-disable gateway-doctor gateway-secret-redaction-smoke gateway-v2-dependency-probe gateway-v2-dependency-profile-probe gateway-v2-auxiliary-redaction-probe gateway-v2-slice-smoke gateway-resume-redaction-e2e gateway-resume-redaction-e2e-prebuilt gateway-execution-status-live-smoke gateway-turn-watch gateway-turn-watch-webhook harness-wave2-task4-smoke notify-icons-generate notify-icons-select reservation-status task-lease-status selftest doctor doctor-json sqlite-doctor sqlite-doctor-json devtools-status hooks-install install-test install-test-full release-check release $(PYTHON_TARGETS): python-check @@ -113,6 +113,11 @@ gateway-v2-dependency-probe: python-check ## Measure native V2 shell boundaries gateway-v2-dependency-profile-probe: ## Qualify isolated V2 profile boundaries without dependency authority $(PYTHON) scripts/gateway_v2_dependency_profile_probe.py --opencode-bin "$(OPENCODE_BIN)" +.PHONY: gateway-v2-auxiliary-redaction-probe +gateway-v2-auxiliary-redaction-probe: python-check ## Measure four V2 request kinds without gateway authority + npm --prefix plugin/gateway-core run build + $(PYTHON) scripts/gateway_v2_auxiliary_redaction_probe.py --opencode-bin "$(OPENCODE_BIN)" + gateway-v2-slice-smoke: ## Test experimental V2 security adapters against localhost (not full gateway parity) npm --prefix plugin/gateway-core run build $(PYTHON) scripts/gateway_v2_context_live_smoke.py diff --git a/docs/index.md b/docs/index.md index eca39049..9dd23b43 100644 --- a/docs/index.md +++ b/docs/index.md @@ -35,6 +35,7 @@ This repo-local index exists because `AGENTS.md` and related startup docs refere - `plan/opencode-reliability-review-runbook.md`: reusable reliability/E2E review campaign runbook - `specs/e8-plan-handoff-continuity-mapping.md`: continuity and handoff semantics - `specs/opencode-v2-owner-admission-proposal.md`: non-operational owner-bound execution requirements +- `specs/opencode-v2-auxiliary-redaction-probe.md`: bounded synthetic measurement of four V2 request kinds - `specs/background-artifact-authority.md`: design-only no-clobber and provenance requirements for background state - `plugin-gateway-plan.md`: gateway/plugin architecture and follow-up planning - `readme-deep-notes.md`: deeper architecture and release-history notes diff --git a/docs/plan/doing/opencode-v2-migration.md b/docs/plan/doing/opencode-v2-migration.md index 7d2001a0..023dd420 100644 --- a/docs/plan/doing/opencode-v2-migration.md +++ b/docs/plan/doing/opencode-v2-migration.md @@ -63,4 +63,6 @@ The legacy `/mcp` helper refuses an observed native or ambiguous `mcp.servers` l The legacy `/plugin` helper similarly refuses an observed native or ambiguous `plugins` key, including during locked mutations. Its results describe only helper-resolved legacy configuration, not loaded plugins or active gateway hooks; it does not port V2 plugin management. +The [auxiliary redaction probe](../../specs/opencode-v2-auxiliary-redaction-probe.md) exercises native context, title, generate and compaction request hooks against an owned localhost provider, with unprotected controls and separate system/message canaries. It measures those synthetic requests only; it does not verify compaction completion, other transports, later-hook ordering or gateway protection. + Reference: and . diff --git a/docs/specs/opencode-v2-auxiliary-redaction-probe.md b/docs/specs/opencode-v2-auxiliary-redaction-probe.md new file mode 100644 index 00000000..98e34db4 --- /dev/null +++ b/docs/specs/opencode-v2-auxiliary-redaction-probe.md @@ -0,0 +1,75 @@ +# V2 auxiliary request redaction probe + +Status: **synthetic measurement only**. This probe does not install or configure +the gateway, establish a security boundary, or grant authority. The V2 gateway +and dependency profile remain **unqualified**. + +## Run + +With Python 3.11+, an existing gateway build environment, and OpenCode 2.2.5: + +```sh +make gateway-v2-auxiliary-redaction-probe OPENCODE_BIN=/path/to/opencode +``` + +The target builds `plugin/gateway-core` and runs the probe. It does not install +dependencies. To select a private fixture parent, invoke the script directly +after the build with `--temp-parent `. A disposable home is not an OS +sandbox; choose an isolated environment when preserving local data is critical. + +## What it measures + +Two separate private authenticated runtimes use an owned local plugin and a +synthetic OpenAI-compatible provider listening on loopback. Each case uses +different random provider and trace credentials. The control does not register the +redactor; the protected case registers only the existing experimental context +redactor. For each native `context`, `title`, `generate`, and `compaction` hook, +the fixture introduces a synthetic system canary and supplies a separate +synthetic message canary. It checks the hook inputs and outputs, then independently +checks the provider's received request for both canaries and positive markers. + +The control must deliver both canaries for every kind. The protected case must +remove both and retain redaction and positive markers for every observed request. +Unknown, duplicated hook, missing, malformed, or ambiguous evidence is a failure. +More than one provider request per hook is allowed; each is checked. +The report contains counts, finite reason codes, fixture basenames, and the +selected executable's SHA-256, not request bodies, session IDs, or canary values. + +The probe awaits location activation and confirms its local plugin before creating +a synthetic session. It distinguishes admitting compaction from observing a +compaction request: receipt of that request does **not** establish that compaction +completed. It uses fresh HOME/XDG/DB paths, an environment allowlist, a private +server password, PID checks before API mutations, bounded output/requests/time, +package-manager stubs that exit without writing, and owned-process teardown. The +stubs only cover the listed commands found through the child `PATH`; they do not +detect every attempted installation or replace network isolation. System or +ancestor OpenCode configuration causes preflight refusal. All fixture roots, +including version probes, are retained under the selected temporary parent; +the probe's unit tests also retain their temporary fixture roots. Teardown reports +whether owned processes and sockets settled. Retained fixtures contain synthetic +prompts and may contain local runtime metadata; inspect them before sharing or +cleaning them up. The selected executable is version-checked +and its identity is compared before and after; that does not prevent a transient +swap. Hook witnesses travel over an authenticated localhost channel, not a trace +file; its key is present in the retained synthetic plugin, and the provider key +is present in its retained configuration. Fixture paths, channel authentication, +and metadata checks do not establish protection against a hostile process with +the same user identity. + +## Observed result and limits + +On 2026-11-06, one local executable reporting `opencode v2.2.5` produced one +context, one title, one generate, and two compaction provider requests in each +case. Its SHA-256 was +`abe4d88594bffb12b364375b8a6732d2d1f7a977195ed1a35c5c76f8e89c9e0e`. +Both canaries reached the control provider and were absent from every protected +observation. These results describe that executable and synthetic run, not every +binary reporting the same version or other versions. + +The probe does not test a later hook changing content after redaction, transport +hook ordering, WebSocket, streaming semantics, other providers or encodings, +images, large or resumed histories, or the complete secret detector. It does not +prove event completeness or production isolation. The separate +[owner-bound admission proposal](opencode-v2-owner-admission-proposal.md) and +[dependency-profile measurement](opencode-v2-dependency-profile.md) retain their +unresolved activation gates. diff --git a/scripts/config_writer_sink_manifest.json b/scripts/config_writer_sink_manifest.json index 8db5e61a..42cf63a8 100644 --- a/scripts/config_writer_sink_manifest.json +++ b/scripts/config_writer_sink_manifest.json @@ -1,4 +1,49 @@ [ + { + "language": "python", + "file": "gateway_v2_auxiliary_redaction_probe.py", + "function": "prepare_fixture", + "kind": "path.mkdir", + "destination": "root / name", + "count": 1, + "classification": "isolated_fixture_exemption" + }, + { + "language": "python", + "file": "gateway_v2_auxiliary_redaction_probe.py", + "function": "write_fixture", + "kind": "os.mkdir", + "destination": "part", + "count": 1, + "classification": "isolated_fixture_exemption" + }, + { + "language": "python", + "file": "gateway_v2_auxiliary_redaction_probe.py", + "function": "write_fixture", + "kind": "os.open", + "destination": "part", + "count": 1, + "classification": "isolated_fixture_exemption" + }, + { + "language": "python", + "file": "gateway_v2_auxiliary_redaction_probe.py", + "function": "write_fixture", + "kind": "os.open", + "destination": "path.parts[-1]", + "count": 1, + "classification": "isolated_fixture_exemption" + }, + { + "language": "python", + "file": "gateway_v2_auxiliary_redaction_probe.py", + "function": "write_fixture", + "kind": "os.open", + "destination": "root", + "count": 1, + "classification": "isolated_fixture_exemption" + }, { "language": "python", "file": "gateway_v2_dependency_profile_probe.py", diff --git a/scripts/gateway_v2_auxiliary_redaction_probe.py b/scripts/gateway_v2_auxiliary_redaction_probe.py new file mode 100644 index 00000000..e4543b18 --- /dev/null +++ b/scripts/gateway_v2_auxiliary_redaction_probe.py @@ -0,0 +1,676 @@ +"""Measure experimental V2 redaction with an owned synthetic provider. + +This is evidence from private child runtimes, not gateway qualification. +""" + +from __future__ import annotations + +import argparse +import hashlib +import http.client +import json +import os +import secrets +import shutil +import stat +import subprocess +import tempfile +import threading +import time +from dataclasses import dataclass, field +from pathlib import Path +from typing import Any + +from gateway_v2_dependency_contract_probe import ( + MANAGERS, + ProbeFailure, + child_env, + preflight, + run_child, +) +from gateway_v2_dependency_profile_probe import ( + CLEANUP_SECONDS, + MAX_CONNECTIONS, + PRIMARY_SECONDS, + PrivateRuntime, + ProfileHandler, + ProfileProvider, + ProfileState, + activated_inventory, + native_id, + owned_fixture, + wait_until, +) + +ROOT = Path(__file__).resolve().parent.parent +KINDS = ("context", "generate", "compaction", "title") +MODES = ("control", "protected") +SYSTEM_CANARY = "V2AuxSystemSecret_123456789" +MESSAGE_CANARY = "V2AuxMessageSecret_987654321" +MESSAGE_MARKER = "V2_AUX_MESSAGE_MARKER" +SYSTEM_MARKER = "V2_AUX_MARKER_" +REDACTION = "[REDACTED]" +MAX_TRACE_BYTES = 4096 +MAX_TRACE_ROWS = 32 +MAX_FIXTURE_BYTES = 1024 * 1024 +INSTALLER_STUB = "#!/bin/sh\nexit 97\n" + + +def write_fixture( + root: Path, relative: str, text: str, *, executable: bool = False +) -> None: + """Create a new fixture file without truncating or following an existing entry.""" + path = Path(relative) + if ( + path.is_absolute() + or not path.parts + or any(part in ("", ".", "..") for part in path.parts) + ): + raise ProbeFailure("fixture_path_invalid") + if ( + not hasattr(os, "O_NOFOLLOW") + or not hasattr(os, "O_DIRECTORY") + or os.open not in os.supports_dir_fd + or os.mkdir not in os.supports_dir_fd + ): + raise ProbeFailure("fixture_primitives_unavailable") + content = text.encode("utf-8") + if len(content) > MAX_FIXTURE_BYTES: + raise ProbeFailure("fixture_size_limit") + before = root.lstat() + if ( + not stat.S_ISDIR(before.st_mode) + or before.st_uid != os.getuid() + or stat.S_IMODE(before.st_mode) != 0o700 + ): + raise ProbeFailure("fixture_root_invalid") + directories: list[int] = [] + try: + current = os.open(root, os.O_RDONLY | os.O_DIRECTORY | os.O_NOFOLLOW) + directories.append(current) + opened = os.fstat(current) + if (opened.st_dev, opened.st_ino) != (before.st_dev, before.st_ino): + raise ProbeFailure("fixture_root_changed") + for part in path.parts[:-1]: + try: + os.mkdir(part, mode=0o700, dir_fd=current) + except FileExistsError: + pass + current = os.open( + part, os.O_RDONLY | os.O_DIRECTORY | os.O_NOFOLLOW, dir_fd=current + ) + directories.append(current) + metadata = os.fstat(current) + if ( + not stat.S_ISDIR(metadata.st_mode) + or metadata.st_uid != os.getuid() + or stat.S_IMODE(metadata.st_mode) != 0o700 + ): + raise ProbeFailure("fixture_parent_invalid") + mode = 0o700 if executable else 0o600 + descriptor = os.open( + path.parts[-1], + os.O_WRONLY | os.O_CREAT | os.O_EXCL | os.O_NOFOLLOW, + mode, + dir_fd=current, + ) + try: + metadata = os.fstat(descriptor) + if ( + not stat.S_ISREG(metadata.st_mode) + or metadata.st_uid != os.getuid() + or metadata.st_nlink != 1 + or metadata.st_size != 0 + ): + raise ProbeFailure("fixture_file_invalid") + os.fchmod(descriptor, mode) + remaining = memoryview(content) + while remaining: + written = os.write(descriptor, remaining) + if written <= 0: + raise ProbeFailure("fixture_write_incomplete") + remaining = remaining[written:] + finally: + os.close(descriptor) + finally: + for directory in reversed(directories): + os.close(directory) + + +@dataclass +class CaptureState(ProfileState): + observations: list[dict[str, Any]] = field(default_factory=list) + trace_rows: list[dict[str, Any]] = field(default_factory=list) + provider_key: str = field(default_factory=lambda: secrets.token_hex(24)) + responses: int = 0 + + def answer(self, payload: dict[str, Any]) -> tuple[dict[str, Any], str]: + text = json.dumps(payload, sort_keys=True) + with self.lock: + self.requests += 1 + if self.requests > MAX_CONNECTIONS: + raise ProbeFailure("provider_request_limit") + self.observations.append( + { + "markers": [kind for kind in KINDS if SYSTEM_MARKER + kind in text], + "system_canary": SYSTEM_CANARY in text, + "message_canary": MESSAGE_CANARY in text, + "message_marker": MESSAGE_MARKER in text, + "redacted": REDACTION in text, + } + ) + return {"role": "assistant", "content": "SYNTHETIC_OK"}, "stop" + + +class CaptureHandler(ProfileHandler): + def do_POST(self) -> None: + if self.path != "/trace": + state: CaptureState = self.server.state # type: ignore[attr-defined] + if self.headers.get("Authorization") != f"Bearer {state.provider_key}": + self.started() + state.failure = state.failure or "provider_authentication_failed" + self.send_error(401) + return + # The reused synthetic provider handler expects its fixed fixture + # token. Only adapt the header after our per-case check succeeds. + self.headers.replace_header("Authorization", "Bearer synthetic-only") + super().do_POST() + return + state: CaptureState = self.started() # type: ignore[assignment] + try: + if self.headers.get("Authorization") != f"Bearer {state.control_key}": + raise ProbeFailure("trace_authentication_failed") + length = int(self.headers.get("Content-Length", "0")) + if not 0 < length <= MAX_TRACE_BYTES: + raise ProbeFailure("trace_request_size") + raw = self.rfile.read(length) + if len(raw) != length: + raise ProbeFailure("trace_request_incomplete") + row = json.loads(raw) + if not isinstance(row, dict): + raise ProbeFailure("trace_schema_invalid") + with state.lock: + if len(state.trace_rows) >= MAX_TRACE_ROWS: + raise ProbeFailure("trace_row_limit") + state.trace_rows.append(row) + self.reply(b'{"ok":true}') + except (ProbeFailure, ValueError, TypeError, OSError) as error: + if not state.stopped.is_set(): + state.failure = state.failure or ( + str(error) + if isinstance(error, ProbeFailure) + else "trace_protocol_error" + ) + self.send_error(400) + + def do_GET(self) -> None: + state = self.started() + state.failure = "unexpected_provider_method" + self.send_error(405) + + def reply(self, body: bytes, content_type: str = "application/json") -> None: + super().reply(body, content_type) + if self.path == "/v1/chat/completions": + state: CaptureState = self.server.state # type: ignore[attr-defined] + with state.lock: + state.responses += 1 + + +class CaptureProvider(ProfileProvider): + def __init__(self, state: CaptureState): + super().__init__(state) + self.RequestHandlerClass = CaptureHandler + + +def fixture_plugin(root: Path, port: int, key: str, *, protected: bool) -> str: + dist = ROOT / "plugin" / "gateway-core" / "dist" + values: dict[str, Any] = { + "__BOUNDARY__": (dist / "v2" / "provider-boundary.js").as_uri(), + "__REDACTOR__": (dist / "hooks" / "shared" / "secret-redaction.js").as_uri(), + "__TRACE_URL__": f"http://127.0.0.1:{port}/trace", + "__TRACE_KEY__": key, + "__SYSTEM_CANARY__": SYSTEM_CANARY, + "__MESSAGE_CANARY__": MESSAGE_CANARY, + "__PROTECTED__": protected, + } + source = r""" +import { Plugin } from "@opencode/plugin" +import { registerV2ContextRedaction } from __BOUNDARY__ +import { createSecretRedactor } from __REDACTOR__ +const traceUrl = __TRACE_URL__ +const traceKey = __TRACE_KEY__ +const systemCanary = __SYSTEM_CANARY__ +const messageCanary = __MESSAGE_CANARY__ +const protectedMode = __PROTECTED__ +const kinds = ["context", "generate", "compaction", "title"] +let traceRows = 0 +const record = async row => { + if (++traceRows > 32) throw new Error("AUX_TRACE_LIMIT") + const response = await fetch(traceUrl, {method:"POST", headers:{"Authorization":`Bearer ${traceKey}`, "Content-Type":"application/json"}, body:JSON.stringify(row)}) + if (!response.ok) throw new Error("AUX_TRACE_FAILED") +} +export default Plugin.define({ id: "profile-owned", async setup(ctx) { + for (const kind of kinds) await ctx.session.hook(kind, async event => { + const valid = Array.isArray(event.system) && Array.isArray(event.messages) + const messages = valid ? JSON.stringify(event.messages) : "" + await record({ phase:"before", kind, valid, messageMarker:messages.includes("V2_AUX_MESSAGE_MARKER"), messageCanary:messages.includes(messageCanary) }) + if (valid) event.system.push({type:"text", text:`V2_AUX_MARKER_${kind} ${systemCanary}`}) + }) + const redactor = createSecretRedactor({ patterns:[systemCanary,messageCanary], redactionToken:"[REDACTED]", limits:{maxDepth:64,maxNodes:100000,maxChars:5000000} }) + const dispose = protectedMode ? await registerV2ContextRedaction(ctx.session, redactor) : async () => {} + for (const kind of kinds) await ctx.session.hook(kind, async event => { + const system = JSON.stringify(event.system), messages = JSON.stringify(event.messages) + await record({ phase:"after", kind, systemMarker:system.includes(`V2_AUX_MARKER_${kind}`), messageMarker:messages.includes("V2_AUX_MESSAGE_MARKER"), systemCanary:system.includes(systemCanary), messageCanary:messages.includes(messageCanary), systemRedacted:system.includes("[REDACTED]"), messageRedacted:messages.includes("[REDACTED]") }) + }) + await record({phase:"ready", version:ctx.app.version}) + return dispose +} }) +""" + for placeholder, value in values.items(): + source = source.replace(placeholder, json.dumps(value)) + return source + + +def prepare_fixture( + root: Path, port: int, key: str, provider_key: str, *, protected: bool +) -> dict[str, Any]: + for name in ("home", "tmp", "bin"): + (root / name).mkdir(mode=0o700) + write_fixture( + root, + "project/owned/index.ts", + fixture_plugin(root, port, key, protected=protected), + ) + rules = [{"action": "*", "resource": "*", "effect": "deny"}] + config = { + "update": "disable", + "model": "probe/mock", + "default_agent": "build", + "snapshots": False, + "formatter": False, + "permissions": rules, + "agents": { + "build": {"permissions": rules}, + "title": {"model": "probe/mock"}, + "compaction": {"model": "probe/mock"}, + }, + "plugins": [str(root / "project" / "owned")], + "providers": { + "probe": { + "name": "Synthetic local provider", + "package": "@opencode/ai/providers/openai-compatible", + "settings": { + "baseURL": f"http://127.0.0.1:{port}/v1", + "apiKey": provider_key, + }, + "models": {"mock": {"name": "mock", "capabilities": {"tools": False}}}, + } + }, + } + 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}", + INSTALLER_STUB, + executable=True, + ) + return config + + +def assess( + mode: str, rows: list[dict[str, Any]], requests: list[dict[str, Any]] +) -> dict[str, int]: + """Require both pre-redaction controls and matching provider observations.""" + if mode not in MODES: + raise ProbeFailure("invalid_case") + if ( + not isinstance(rows, list) + or not isinstance(requests, list) + or any(not isinstance(row, dict) for row in rows) + or any(not isinstance(request, dict) for request in requests) + ): + raise ProbeFailure("observation_schema_invalid") + if len(rows) > MAX_TRACE_ROWS or len(requests) > MAX_CONNECTIONS: + raise ProbeFailure("observation_limit") + ready = [row for row in rows if row.get("phase") == "ready"] + if ready != [{"phase": "ready", "version": "2.2.5"}]: + raise ProbeFailure("plugin_ready_missing") + before_keys = {"phase", "kind", "valid", "messageMarker", "messageCanary"} + after_keys = { + "phase", + "kind", + "systemMarker", + "messageMarker", + "systemCanary", + "messageCanary", + "systemRedacted", + "messageRedacted", + } + for row in rows: + phase = row.get("phase") + if phase == "ready": + continue + keys = ( + before_keys + if phase == "before" + else after_keys + if phase == "after" + else set() + ) + if set(row) != keys or row.get("kind") not in KINDS: + raise ProbeFailure("trace_schema_invalid") + if any(type(row[key]) is not bool for key in keys - {"phase", "kind"}): + raise ProbeFailure("trace_schema_invalid") + expected_canary = mode == "control" + counts: dict[str, int] = {} + for kind in KINDS: + before = [ + row + for row in rows + if row.get("phase") == "before" and row.get("kind") == kind + ] + after = [ + row + for row in rows + if row.get("phase") == "after" and row.get("kind") == kind + ] + if ( + len(before) != 1 + or len(after) != 1 + or rows.index(before[0]) >= rows.index(after[0]) + ): + raise ProbeFailure("hook_observation_incomplete") + first, last = before[0], after[0] + if not (first["valid"] and first["messageMarker"] and first["messageCanary"]): + raise ProbeFailure("hook_input_control_missing") + if not (last["systemMarker"] and last["messageMarker"]): + raise ProbeFailure("hook_marker_missing") + if ( + last["systemCanary"] is not expected_canary + or last["messageCanary"] is not expected_canary + ): + raise ProbeFailure("hook_redaction_mismatch") + if mode == "protected" and not ( + last["systemRedacted"] and last["messageRedacted"] + ): + raise ProbeFailure("hook_redaction_token_missing") + counts[kind] = 0 + request_keys = { + "markers", + "system_canary", + "message_canary", + "message_marker", + "redacted", + } + for request in requests: + if set(request) != request_keys or not isinstance(request["markers"], list): + raise ProbeFailure("provider_observation_invalid") + if any(type(request[key]) is not bool for key in request_keys - {"markers"}): + raise ProbeFailure("provider_observation_invalid") + markers = request["markers"] + if len(markers) != 1 or markers[0] not in KINDS: + raise ProbeFailure("provider_request_ambiguous") + if not request["message_marker"]: + raise ProbeFailure("provider_message_marker_missing") + if ( + request["system_canary"] is not expected_canary + or request["message_canary"] is not expected_canary + ): + raise ProbeFailure("provider_redaction_mismatch") + if mode == "protected" and not request["redacted"]: + raise ProbeFailure("provider_redaction_token_missing") + counts[markers[0]] += 1 + if any(count == 0 for count in counts.values()): + raise ProbeFailure("provider_observation_incomplete") + return counts + + +def exercise( + runtime: PrivateRuntime, root: Path, config: dict[str, Any], state: CaptureState +) -> None: + runtime.ready() + inventory = activated_inventory(runtime, root) + active = [ + item + for item in inventory + if item.get("source") == "owned" and item.get("status") == "active" + ] + if len(active) != 1 or config.get("plugins") != [str(root / "project" / "owned")]: + raise ProbeFailure("fixture_plugin_not_active") + if any( + item.get("source") not in ("builtin", "owned") + and item.get("status") == "active" + for item in inventory + ): + raise ProbeFailure("unexpected_plugin_topology") + created = runtime.mutate( + "POST", + "/api/session", + { + "location": {"directory": str(root / "project")}, + "model": {"providerID": "probe", "id": "mock"}, + "agent": "build", + }, + ) + session = native_id(created["data"]["id"]) + runtime.mutate( + "POST", + f"/api/session/{session}/prompt", + {"text": f"{MESSAGE_MARKER} {MESSAGE_CANARY}"}, + ) + runtime.mutate("POST", f"/api/experimental/session/{session}/wait") + runtime.mutate( + "POST", + f"/api/session/{session}/generate", + {"prompt": f"{MESSAGE_MARKER} {MESSAGE_CANARY}"}, + ) + runtime.mutate("POST", f"/api/session/{session}/compact", {}) + runtime.mutate("POST", f"/api/experimental/session/{session}/wait") + + def all_kinds_seen() -> bool: + with state.lock: + return { + kind for item in state.observations for kind in item["markers"] + } == set(KINDS) and state.responses == state.requests + + wait_until(all_kinds_seen, state, "provider_observation_incomplete") + + +def run_case(binary: str, parent: Path, mode: str) -> dict[str, Any]: + root = Path(tempfile.mkdtemp(prefix="v2-aux-redaction-", dir=parent)) + identity = root.lstat() + state = CaptureState(mode, root, time.monotonic() + PRIMARY_SECONDS) + provider: CaptureProvider | None = None + provider_thread: threading.Thread | None = None + provider_started = False + runtime: PrivateRuntime | None = None + completed = False + report: dict[str, Any] = {"case": mode, "measurement": "FAIL"} + + def expire() -> None: + state.failure = state.failure or "case_timeout" + state.stop() + if provider is not None: + provider.close_connections() + if runtime is not None: + runtime.abort_connections() + + watchdog = threading.Timer(max(0, state.deadline - time.monotonic()), expire) + watchdog.daemon = True + try: + watchdog.start() + provider = CaptureProvider(state) + provider_thread = threading.Thread( + target=provider.serve_forever, kwargs={"poll_interval": 0.05}, daemon=True + ) + provider_thread.start() + provider_started = True + config = prepare_fixture( + root, + provider.server_port, + state.control_key, + state.provider_key, + protected=mode == "protected", + ) + runtime = PrivateRuntime(binary, root, state.deadline) + runtime.start() + exercise(runtime, root, config, state) + completed = True + except ( + ProbeFailure, + OSError, + ValueError, + KeyError, + TypeError, + RuntimeError, + http.client.HTTPException, + ) as error: + report["reason"] = ( + str(error) if isinstance(error, ProbeFailure) else "probe_protocol_error" + ) + finally: + deadline = time.monotonic() + CLEANUP_SECONDS + watchdog.cancel() + if watchdog.ident is not None: + watchdog.join(max(0, deadline - time.monotonic())) + state.stop() + cleanup_errors: list[str] = [] + if provider is not None: + try: + if provider_started: + provider.shutdown() + provider.close_connections() + except OSError: + cleanup_errors.append("provider-stop") + if runtime is not None: + try: + runtime.close(deadline) + except (ProbeFailure, OSError, subprocess.SubprocessError): + cleanup_errors.append("runtime-stop") + if provider is not None: + try: + provider.server_close() + except OSError: + cleanup_errors.append("provider-close") + for worker in provider.workers: + worker.join(max(0, deadline - time.monotonic())) + if worker.is_alive(): + cleanup_errors.append("provider-worker") + if provider.admissions != state.handlers: + state.failure = state.failure or "unhandled_provider_connection" + if provider_thread is not None and provider_started: + provider_thread.join(max(0, deadline - time.monotonic())) + if provider_thread.is_alive(): + cleanup_errors.append("provider-thread") + if watchdog.is_alive() or time.monotonic() > deadline: + cleanup_errors.append("deadline-teardown") + if not owned_fixture(root, identity): + cleanup_errors.append("fixture-identity") + if completed and not cleanup_errors and not state.failure: + try: + if state.responses != state.requests or state.requests != len( + state.observations + ): + raise ProbeFailure("provider_response_incomplete") + counts = assess(mode, state.trace_rows, state.observations) + report.update( + {"measurement": "PASS", "provider_requests_by_kind": counts} + ) + except ProbeFailure as error: + report["reason"] = str(error) + except (OSError, ValueError, TypeError): + report["reason"] = "evidence_incomplete_or_mismatched" + if cleanup_errors or state.failure: + report.update( + { + "measurement": "FAIL", + "reason": state.failure or "cleanup_failed", + "teardown_failures": cleanup_errors, + } + ) + # A same-UID actor can replace pathnames or move unrelated files into a + # fixture after inspection. Retain it rather than recursively deleting + # data whose provenance cannot be established by these checks. + report["retained_fixture"] = root.name + report["teardown_complete"] = not cleanup_errors + return report + + +def probe(binary: str, parent: Path) -> dict[str, Any]: + report: dict[str, Any] = { + "measurement": "FAIL", + "scope": "owned-synthetic-child-only", + "authority": "disabled-unqualified", + "cases": [], + } + try: + preflight(parent) + for relative in ("v2/provider-boundary.js", "hooks/shared/secret-redaction.js"): + if not (ROOT / "plugin" / "gateway-core" / "dist" / relative).is_file(): + raise ProbeFailure("gateway_build_missing") + path = Path(binary) + initial = path.stat() + if not stat.S_ISREG(initial.st_mode) or initial.st_size > 1024 * 1024 * 1024: + raise ProbeFailure("binary_identity_invalid") + with path.open("rb") as stream: + digest = hashlib.file_digest(stream, "sha256").hexdigest() + report["binary_sha256"] = digest + home = Path(tempfile.mkdtemp(prefix="v2-aux-version-", dir=parent)) + report["version_fixture"] = home.name + code, output, _ = run_child( + [binary, "--version"], home, child_env(home), timeout=10 + ) + if code != 0 or output.decode().strip() != "opencode v2.2.5": + raise ProbeFailure("unsupported_host_version") + for mode in MODES: + result = run_case(binary, parent, mode) + report["cases"].append(result) + counts = result.get("provider_requests_by_kind") + if ( + result.get("case") != mode + or result.get("measurement") != "PASS" + or result.get("teardown_complete") is not True + or not isinstance(counts, dict) + or set(counts) != set(KINDS) + or any(type(count) is not int or count < 1 for count in counts.values()) + or sum(counts.values()) > MAX_CONNECTIONS + ): + raise ProbeFailure("case_failed") + current = path.stat() + with path.open("rb") as stream: + unchanged = hashlib.file_digest(stream, "sha256").hexdigest() == digest + if not unchanged or (initial.st_dev, initial.st_ino, initial.st_size) != ( + current.st_dev, + current.st_ino, + current.st_size, + ): + raise ProbeFailure("binary_identity_changed") + report["measurement"] = "PASS" + except (ProbeFailure, OSError, ValueError, subprocess.SubprocessError) as error: + report["reason"] = ( + str(error) if isinstance(error, ProbeFailure) else "probe_internal_error" + ) + return report + + +def main() -> int: + parser = argparse.ArgumentParser(description=__doc__) + parser.add_argument("--opencode-bin", default="opencode") + parent = Path(tempfile.gettempdir()) + preferred = parent / "opencode" + parser.add_argument( + "--temp-parent", type=Path, default=preferred if preferred.is_dir() else parent + ) + 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 host is required") + result = probe(str(Path(binary).resolve()), args.temp_parent) + print(json.dumps(result, sort_keys=True)) + return 0 if result["measurement"] == "PASS" else 1 + + +if __name__ == "__main__": + raise SystemExit(main()) diff --git a/tests/test_gateway_v2_auxiliary_redaction_probe.py b/tests/test_gateway_v2_auxiliary_redaction_probe.py new file mode 100644 index 00000000..02eb132e --- /dev/null +++ b/tests/test_gateway_v2_auxiliary_redaction_probe.py @@ -0,0 +1,526 @@ +from __future__ import annotations + +import copy +import http.client +import json +import os +import subprocess +import sys +import tempfile +import threading +import time +import unittest +from collections.abc import Iterator +from contextlib import contextmanager +from pathlib import Path +from unittest.mock import Mock, patch + +ROOT = Path(__file__).resolve().parents[1] +sys.path.insert(0, str(ROOT / "scripts")) + +import gateway_v2_auxiliary_redaction_probe as probe + + +@contextmanager +def retained_test_directory() -> Iterator[str]: + """Do not recursively remove version or runtime fixture paths after tests.""" + yield tempfile.mkdtemp(prefix="v2-aux-unit-") + + +def evidence(mode: str) -> tuple[list[dict], list[dict]]: + unprotected = mode == "control" + rows: list[dict] = [{"phase": "ready", "version": "2.2.5"}] + requests: list[dict] = [] + for kind in probe.KINDS: + rows.append( + { + "phase": "before", + "kind": kind, + "valid": True, + "messageMarker": True, + "messageCanary": True, + } + ) + rows.append( + { + "phase": "after", + "kind": kind, + "systemMarker": True, + "messageMarker": True, + "systemCanary": unprotected, + "messageCanary": unprotected, + "systemRedacted": not unprotected, + "messageRedacted": not unprotected, + } + ) + requests.append( + { + "markers": [kind], + "system_canary": unprotected, + "message_canary": unprotected, + "message_marker": True, + "redacted": not unprotected, + } + ) + return rows, requests + + +class AuxiliaryRedactionEvidenceTest(unittest.TestCase): + def test_both_modes_require_all_four_native_kinds(self) -> None: + for mode in probe.MODES: + with self.subTest(mode=mode): + rows, requests = evidence(mode) + requests.append(copy.deepcopy(requests[2])) + counts = probe.assess(mode, rows, requests) + self.assertEqual(set(probe.KINDS), set(counts)) + self.assertEqual(2, counts["compaction"]) + + def test_missing_duplicate_or_out_of_order_hook_evidence_fails(self) -> None: + rows, requests = evidence("protected") + invalid = [ + rows[:-1], + rows + [copy.deepcopy(rows[-1])], + [rows[0], rows[2], rows[1], *rows[3:]], + [{"phase": "ready", "version": "2.2.4"}, *rows[1:]], + [*rows, {"phase": "unexpected", "raw": "synthetic"}], + ] + for candidate in invalid: + with ( + self.subTest(candidate=candidate), + self.assertRaises(probe.ProbeFailure), + ): + probe.assess("protected", candidate, requests) + + def test_requires_actual_input_canaries_and_distinct_controls(self) -> None: + for mode in probe.MODES: + rows, requests = evidence(mode) + for key in ("valid", "messageMarker", "messageCanary"): + changed = copy.deepcopy(rows) + changed[1][key] = False + with ( + self.subTest(mode=mode, key=key), + self.assertRaises(probe.ProbeFailure), + ): + probe.assess(mode, changed, requests) + altered = copy.deepcopy(requests) + altered[0]["system_canary"] = mode != "control" + with self.subTest(mode=mode), self.assertRaises(probe.ProbeFailure): + probe.assess(mode, rows, altered) + + def test_protected_case_rejects_leaked_or_unredacted_values(self) -> None: + rows, requests = evidence("protected") + for key in ("systemCanary", "messageCanary"): + changed = copy.deepcopy(rows) + changed[2][key] = True + with self.subTest(key=key), self.assertRaises(probe.ProbeFailure): + probe.assess("protected", changed, requests) + for key in ( + "systemRedacted", + "messageRedacted", + "messageMarker", + "systemMarker", + ): + changed = copy.deepcopy(rows) + changed[2][key] = False + with self.subTest(key=key), self.assertRaises(probe.ProbeFailure): + probe.assess("protected", changed, requests) + for key in ("system_canary", "message_canary"): + changed = copy.deepcopy(requests) + changed[0][key] = True + with self.subTest(key=key), self.assertRaises(probe.ProbeFailure): + probe.assess("protected", rows, changed) + + def test_missing_ambiguous_or_non_boolean_provider_evidence_fails(self) -> None: + rows, requests = evidence("protected") + cases = [ + requests[:-1], + [], + [*requests, {**requests[0], "markers": []}], + [*requests, {**requests[0], "markers": ["context", "title"]}], + [*requests, {**requests[0], "message_marker": 1}], + [*requests, {**requests[0], "redacted": False}], + ] + for candidate in cases: + with ( + self.subTest(candidate=candidate), + self.assertRaises(probe.ProbeFailure), + ): + probe.assess("protected", rows, candidate) + + def test_malformed_evidence_never_becomes_a_pass(self) -> None: + rows, requests = evidence("protected") + for invalid_rows, invalid_requests in ( + ([None], requests), + (rows, [None]), + (rows, None), + ): + with ( + self.subTest(rows=invalid_rows, requests=invalid_requests), + self.assertRaises(probe.ProbeFailure), + ): + probe.assess("protected", invalid_rows, invalid_requests) + with self.assertRaises(probe.ProbeFailure): + probe.assess("unknown", rows, requests) + + def test_provider_capture_retains_only_boolean_witnesses(self) -> None: + state = probe.CaptureState("control", Path("/synthetic"), 0) + response, finish = state.answer( + { + "system": f"{probe.SYSTEM_MARKER}context {probe.SYSTEM_CANARY}", + "message": f"{probe.MESSAGE_MARKER} {probe.MESSAGE_CANARY}", + "private": "SYNTHETIC_UNRELATED_PRIVATE_DATA", + } + ) + self.assertEqual("stop", finish) + self.assertEqual({"role": "assistant", "content": "SYNTHETIC_OK"}, response) + encoded = json.dumps(state.observations) + self.assertNotIn(probe.SYSTEM_CANARY, encoded) + self.assertNotIn(probe.MESSAGE_CANARY, encoded) + self.assertNotIn("SYNTHETIC_UNRELATED_PRIVATE_DATA", encoded) + self.assertEqual(["context"], state.observations[0]["markers"]) + + @unittest.skipUnless(os.name == "posix", "POSIX shell required") + def test_installer_stub_exits_without_writing_a_marker(self) -> None: + result = subprocess.run( + ["/bin/sh", "-c", probe.INSTALLER_STUB], + stdin=subprocess.DEVNULL, + capture_output=True, + timeout=2, + check=False, + ) + self.assertEqual(97, result.returncode) + self.assertEqual(b"", result.stdout) + self.assertEqual(b"", result.stderr) + + def test_trace_channel_rejects_oversized_and_excess_rows(self) -> None: + for full in (False, True): + with self.subTest(full=full): + state = probe.CaptureState( + "control", Path("/synthetic"), time.monotonic() + 10 + ) + if full: + state.trace_rows = [{"phase": "ready"}] * probe.MAX_TRACE_ROWS + provider = probe.CaptureProvider(state) + thread = threading.Thread( + target=provider.serve_forever, + kwargs={"poll_interval": 0.05}, + daemon=True, + ) + thread.start() + try: + body = b"{}" if full else b"x" * (probe.MAX_TRACE_BYTES + 1) + connection = http.client.HTTPConnection( + "127.0.0.1", provider.server_port, timeout=2 + ) + connection.request( + "POST", + "/trace", + body, + {"Authorization": f"Bearer {state.control_key}"}, + ) + response = connection.getresponse() + self.assertEqual(400, response.status) + response.read() + connection.close() + expected = "trace_row_limit" if full else "trace_request_size" + self.assertEqual(expected, state.failure) + self.assertEqual( + probe.MAX_TRACE_ROWS if full else 0, len(state.trace_rows) + ) + finally: + state.stop() + provider.shutdown() + provider.close_connections() + provider.server_close() + for worker in provider.workers: + worker.join(timeout=2) + thread.join(timeout=2) + self.assertFalse(thread.is_alive()) + + def test_provider_connection_limit_includes_trace_connections(self) -> None: + state = probe.CaptureState("control", Path("/synthetic"), time.monotonic() + 10) + provider = probe.CaptureProvider(state) + try: + for _ in range(probe.MAX_CONNECTIONS): + self.assertTrue(provider.verify_request(Mock(), ("127.0.0.1", 1))) + self.assertFalse(provider.verify_request(Mock(), ("127.0.0.1", 1))) + self.assertEqual("provider_connection_limit", state.failure) + self.assertEqual(probe.MAX_CONNECTIONS, len(provider.admitted)) + finally: + state.stop() + provider.close_connections() + provider.server_close() + + def test_provider_requires_its_own_per_case_token(self) -> None: + for authorized in (False, True): + with self.subTest(authorized=authorized): + state = probe.CaptureState( + "control", Path("/synthetic"), time.monotonic() + 10 + ) + self.assertNotEqual(state.control_key, state.provider_key) + provider = probe.CaptureProvider(state) + thread = threading.Thread( + target=provider.serve_forever, + kwargs={"poll_interval": 0.05}, + daemon=True, + ) + thread.start() + try: + connection = http.client.HTTPConnection( + "127.0.0.1", provider.server_port, timeout=2 + ) + credential = state.provider_key if authorized else "synthetic-only" + payload = { + "model": "mock", + "messages": [ + { + "role": "user", + "content": f"{probe.MESSAGE_MARKER} {probe.MESSAGE_CANARY}", + } + ], + } + connection.request( + "POST", + "/v1/chat/completions", + json.dumps(payload), + {"Authorization": f"Bearer {credential}"}, + ) + response = connection.getresponse() + self.assertEqual(200 if authorized else 401, response.status) + response.read() + connection.close() + for worker in provider.workers: + worker.join(timeout=2) + if authorized: + self.assertEqual(1, state.requests) + self.assertEqual(1, state.responses) + self.assertIsNone(state.failure) + else: + self.assertEqual(0, state.requests) + self.assertEqual( + "provider_authentication_failed", state.failure + ) + finally: + state.stop() + provider.shutdown() + provider.close_connections() + provider.server_close() + for worker in provider.workers: + worker.join(timeout=2) + thread.join(timeout=2) + self.assertFalse(thread.is_alive()) + + @unittest.skipUnless(os.name == "posix", "POSIX fixture ownership required") + def test_setup_failure_settles_owned_provider_without_touching_other_files( + self, + ) -> None: + with retained_test_directory() as raw: + parent = Path(raw) + sentinel = parent / "unrelated" + sentinel.write_text("preserve", encoding="utf-8") + with patch.object( + probe, + "prepare_fixture", + side_effect=probe.ProbeFailure("synthetic_setup_failure"), + ): + report = probe.run_case("/nonexistent", parent, "control") + self.assertEqual("FAIL", report["measurement"]) + self.assertEqual("synthetic_setup_failure", report["reason"]) + self.assertIs(report["teardown_complete"], True) + self.assertTrue((parent / report["retained_fixture"]).is_dir()) + self.assertEqual("preserve", sentinel.read_text(encoding="utf-8")) + + @unittest.skipUnless(os.name == "posix", "POSIX fixture ownership required") + def test_unsettled_runtime_retains_only_its_owned_fixture(self) -> None: + with retained_test_directory() as raw: + parent = Path(raw) + sentinel = parent / "unrelated" + sentinel.write_text("preserve", encoding="utf-8") + runtime = Mock() + runtime.close.side_effect = probe.ProbeFailure( + "synthetic_unsettled_runtime" + ) + with ( + patch.object(probe, "prepare_fixture", return_value={}), + patch.object(probe, "PrivateRuntime", return_value=runtime), + patch.object(probe, "exercise"), + ): + report = probe.run_case("/nonexistent", parent, "control") + self.assertEqual("FAIL", report["measurement"]) + self.assertIs(report["teardown_complete"], False) + self.assertIn("runtime-stop", report["teardown_failures"]) + retained = parent / report["retained_fixture"] + self.assertTrue(retained.is_dir()) + self.assertEqual("preserve", sentinel.read_text(encoding="utf-8")) + + @unittest.skipUnless(os.name == "posix", "POSIX fixture creation required") + def test_fixture_creation_is_exclusive_and_does_not_follow_links(self) -> None: + with retained_test_directory() as raw: + root = Path(raw) + probe.write_fixture(root, "owned/config.json", "synthetic") + target = root / "owned" / "config.json" + self.assertEqual("synthetic", target.read_text(encoding="utf-8")) + self.assertEqual(0o600, target.stat().st_mode & 0o777) + with self.assertRaises(FileExistsError): + probe.write_fixture(root, "owned/config.json", "overwrite") + self.assertEqual("synthetic", target.read_text(encoding="utf-8")) + + (root / "link").symlink_to(root / "owned", target_is_directory=True) + with self.assertRaises((OSError, probe.ProbeFailure)): + probe.write_fixture(root, "link/other", "blocked") + self.assertFalse((root / "owned" / "other").exists()) + (root / "dangling").symlink_to(root / "missing") + with self.assertRaises((OSError, probe.ProbeFailure)): + probe.write_fixture(root, "dangling", "blocked") + for relative in ("../escape", "/absolute/path"): + with ( + self.subTest(relative=relative), + self.assertRaises(probe.ProbeFailure), + ): + probe.write_fixture(root, relative, "blocked") + + def test_trace_channel_authenticates_and_keeps_provider_counts_separate( + self, + ) -> None: + state = probe.CaptureState("control", Path("/synthetic"), time.monotonic() + 10) + provider = probe.CaptureProvider(state) + thread = threading.Thread( + target=provider.serve_forever, kwargs={"poll_interval": 0.05}, daemon=True + ) + thread.start() + try: + row = {"phase": "ready", "version": "2.2.5"} + connection = http.client.HTTPConnection( + "127.0.0.1", provider.server_port, timeout=2 + ) + connection.request( + "POST", + "/trace", + json.dumps(row), + { + "Authorization": f"Bearer {state.control_key}", + "Content-Type": "application/json", + }, + ) + response = connection.getresponse() + self.assertEqual(200, response.status) + response.read() + connection.close() + self.assertEqual([row], state.trace_rows) + self.assertEqual(0, state.responses) + self.assertEqual(0, state.requests) + + connection = http.client.HTTPConnection( + "127.0.0.1", provider.server_port, timeout=2 + ) + connection.request( + "POST", "/trace", "{}", {"Authorization": "Bearer wrong"} + ) + response = connection.getresponse() + self.assertEqual(400, response.status) + response.read() + connection.close() + self.assertEqual("trace_authentication_failed", state.failure) + self.assertEqual([row], state.trace_rows) + finally: + state.stop() + provider.shutdown() + provider.close_connections() + provider.server_close() + for worker in provider.workers: + worker.join(timeout=2) + thread.join(timeout=2) + self.assertFalse(thread.is_alive()) + + def test_probe_never_qualifies_incomplete_cases_or_changed_binary(self) -> None: + with retained_test_directory() as raw: + root = Path(raw) + binary = root / "opencode" + binary.write_bytes(b"synthetic-binary") + dist = root / "plugin" / "gateway-core" / "dist" + for relative in ( + "v2/provider-boundary.js", + "hooks/shared/secret-redaction.js", + ): + path = dist / relative + path.parent.mkdir(parents=True, exist_ok=True) + path.write_text("synthetic", encoding="utf-8") + + def success(_binary: str, _parent: Path, mode: str) -> dict: + return { + "case": mode, + "measurement": "PASS", + "teardown_complete": True, + "provider_requests_by_kind": {kind: 1 for kind in probe.KINDS}, + } + + with ( + patch.object(probe, "ROOT", root), + patch.object(probe, "preflight"), + patch.object( + probe, "run_child", return_value=(0, b"opencode v2.2.5\n", b"") + ), + patch.object(probe, "run_case", side_effect=success), + ): + result = probe.probe(str(binary), root) + self.assertEqual("PASS", result["measurement"]) + self.assertEqual("disabled-unqualified", result["authority"]) + self.assertTrue((root / result["version_fixture"]).is_dir()) + self.assertEqual( + list(probe.MODES), [case["case"] for case in result["cases"]] + ) + + with ( + patch.object( + probe, "run_child", return_value=(0, b"other version\n", b"") + ), + patch.object(probe, "run_case") as skipped, + ): + unsupported = probe.probe(str(binary), root) + self.assertEqual("FAIL", unsupported["measurement"]) + self.assertEqual("unsupported_host_version", unsupported["reason"]) + self.assertTrue((root / unsupported["version_fixture"]).is_dir()) + skipped.assert_not_called() + + def fail_control(_binary: str, _parent: Path, mode: str) -> dict: + return {"case": mode, "measurement": "FAIL"} + + with patch.object(probe, "run_case", side_effect=fail_control): + failed = probe.probe(str(binary), root) + self.assertEqual("FAIL", failed["measurement"]) + self.assertEqual(1, len(failed["cases"])) + + def unsettled(_binary: str, _parent: Path, mode: str) -> dict: + return { + **success(_binary, _parent, mode), + "teardown_complete": False, + } + + with patch.object(probe, "run_case", side_effect=unsettled): + self.assertEqual( + "FAIL", probe.probe(str(binary), root)["measurement"] + ) + + def missing_kind(_binary: str, _parent: Path, mode: str) -> dict: + return { + **success(_binary, _parent, mode), + "provider_requests_by_kind": {"context": 1}, + } + + with patch.object(probe, "run_case", side_effect=missing_kind): + self.assertEqual( + "FAIL", probe.probe(str(binary), root)["measurement"] + ) + + def replace_binary(_binary: str, _parent: Path, mode: str) -> dict: + binary.write_bytes(b"changed-synthetic-binary") + return success(_binary, _parent, mode) + + with patch.object(probe, "run_case", side_effect=replace_binary): + changed = probe.probe(str(binary), root) + self.assertEqual("FAIL", changed["measurement"]) + self.assertEqual("binary_identity_changed", changed["reason"]) + + +if __name__ == "__main__": + unittest.main()