diff --git a/elydora/plugins/opencode.py b/elydora/plugins/opencode.py index df4334e..5a4a77c 100644 --- a/elydora/plugins/opencode.py +++ b/elydora/plugins/opencode.py @@ -1,13 +1,14 @@ -"""OpenCode plugin — writes ~/.config/opencode/plugins/elydora-audit.mjs (JS plugin).""" +"""OpenCode integration using its current ESM plugin API.""" from __future__ import annotations +import glob import json import os -import stat import sys from .base import AgentPlugin, InstallConfig, PluginStatus +from .hook_template import generate_hook_script PLUGIN_DIR = os.path.join(os.path.expanduser("~"), ".config", "opencode", "plugins") @@ -16,288 +17,148 @@ ELYDORA_DIR = os.path.join(os.path.expanduser("~"), ".elydora") class OpenCodePlugin(AgentPlugin): - """Install/uninstall Elydora audit hook for OpenCode. - - OpenCode expects JavaScript plugins, so this generates a self-contained - JS file that performs the same audit flow using Node.js built-ins. - """ + """Install the Elydora guard and audit hooks for OpenCode.""" def _plugin_path(self) -> str: return os.path.join(PLUGIN_DIR, PLUGIN_FILENAME) + @staticmethod + def _hook_path(agent_id: str) -> str: + return os.path.join(ELYDORA_DIR, agent_id, "hook.py") + def install(self, config: InstallConfig) -> None: agent_id = config.get("agent_id", "") - agent_name = config.get("agent_name", "") - org_id = config.get("org_id", "") - private_key = config.get("private_key", "") - kid = config.get("kid", "") - base_url = config.get("base_url", "https://api.elydora.com") guard_script_path = config.get("guard_script_path", "") + if not agent_id: + raise ValueError("agent_id is required") + if not guard_script_path: + raise ValueError("guard_script_path is required") - # Create per-agent directory agent_dir = os.path.join(ELYDORA_DIR, agent_id) os.makedirs(agent_dir, exist_ok=True) - - # Write config.json config_data = { - "org_id": org_id, + "org_id": config.get("org_id", ""), "agent_id": agent_id, - "kid": kid, - "base_url": base_url, + "kid": config.get("kid", ""), + "base_url": config.get("base_url", "https://api.elydora.com"), "token": config.get("token", ""), - "agent_name": agent_name, + "agent_name": config.get("agent_name", "opencode"), } - config_path = os.path.join(agent_dir, "config.json") - with open(config_path, "w", encoding="utf-8") as f: - json.dump(config_data, f, indent=2) - f.write("\n") - - # Write private key - private_key_path = os.path.join(agent_dir, "private.key") - with open(private_key_path, "w", encoding="utf-8") as f: - f.write(private_key) - try: - os.chmod(private_key_path, stat.S_IRUSR | stat.S_IWUSR) - except Exception: - pass # chmod may fail on Windows - - js_content = _generate_js_plugin( - agent_id=agent_id, - guard_script_path=guard_script_path, + _write_text( + os.path.join(agent_dir, "config.json"), + json.dumps(config_data, indent=2) + "\n", + 0o600, + ) + _write_text( + os.path.join(agent_dir, "private.key"), + config.get("private_key", ""), + 0o600, ) - os.makedirs(PLUGIN_DIR, exist_ok=True) - plugin_path = self._plugin_path() - with open(plugin_path, "w", encoding="utf-8") as f: - f.write(js_content) - try: - os.chmod(plugin_path, stat.S_IRWXU) - except Exception: - pass # chmod may fail on Windows + hook_path = self._hook_path(agent_id) + hook_script = generate_hook_script( + org_id=config.get("org_id", ""), + agent_id=agent_id, + private_key=config.get("private_key", ""), + kid=config.get("kid", ""), + base_url=config.get("base_url", "https://api.elydora.com"), + ) + _write_text(hook_path, hook_script, 0o700) - print(f"Elydora hook installed for OpenCode.") - print(f" Plugin: {plugin_path}") + os.makedirs(PLUGIN_DIR, exist_ok=True) + _write_text( + self._plugin_path(), + _generate_js_plugin( + hook_script_path=hook_path, + guard_script_path=guard_script_path, + ), + 0o700, + ) + print("Elydora hook installed for OpenCode.") + print(f" Plugin: {self._plugin_path()}") def uninstall(self, agent_id: str = "") -> None: - plugin_path = self._plugin_path() try: - os.remove(plugin_path) + os.remove(self._plugin_path()) except FileNotFoundError: pass - # Hook script removal is handled by cli.py cmd_uninstall (rmtree of agent dir) print("Elydora hook uninstalled from OpenCode.") def status(self) -> PluginStatus: plugin_path = self._plugin_path() - installed = os.path.exists(plugin_path) - if installed: - details = f"Plugin: {plugin_path}" - else: - details = "Not installed" + runtime_ready = False + for config_path in glob.glob(os.path.join(ELYDORA_DIR, "*", "config.json")): + try: + with open(config_path, "r", encoding="utf-8") as file: + config = json.load(file) + except (OSError, json.JSONDecodeError): + continue + agent_dir = os.path.dirname(config_path) + if config.get("agent_name") == "opencode": + runtime_ready = os.path.isfile(os.path.join(agent_dir, "hook.py")) and os.path.isfile( + os.path.join(agent_dir, "guard.py") + ) + if runtime_ready: + break + + installed = os.path.isfile(plugin_path) and runtime_ready + details = f"Plugin: {plugin_path}" if installed else "Not installed" return PluginStatus(installed=installed, agent="opencode", details=details) -def _generate_js_plugin( - *, - agent_id: str, - guard_script_path: str = "", -) -> str: - guard_path_escaped = guard_script_path.replace("\\", "\\\\") if guard_script_path else "" - python_exe_escaped = sys.executable.replace("\\", "\\\\") - return f'''// Elydora audit plugin for OpenCode — auto-generated, do not edit. -"use strict"; +def _write_text(path: str, content: str, mode: int) -> None: + with open(path, "w", encoding="utf-8") as file: + file.write(content) + try: + os.chmod(path, mode) + except OSError: + pass -const crypto = require("crypto"); -const https = require("https"); -const http = require("http"); -const fs = require("fs"); -const path = require("path"); -const os = require("os"); -const {{ spawnSync }} = require("child_process"); -const CONFIG_DIR = path.join(os.homedir(), ".elydora", {_js_str(agent_id)}); -const CONFIG = JSON.parse(fs.readFileSync(path.join(CONFIG_DIR, "config.json"), "utf-8")); -const PRIVATE_KEY = fs.readFileSync(path.join(CONFIG_DIR, "private.key"), "utf-8").trim(); -const ORG_ID = CONFIG.org_id; -const AGENT_ID = CONFIG.agent_id; -const KID = CONFIG.kid; -const BASE_URL = CONFIG.base_url; +def _generate_js_plugin(*, hook_script_path: str, guard_script_path: str) -> str: + return f'''// Elydora Audit Plugin for OpenCode +// Generated by elydora-sdk — DO NOT EDIT -const CHAIN_STATE_PATH = path.join(CONFIG_DIR, "chain-state.json"); +import {{ spawn, spawnSync }} from 'node:child_process'; -function base64urlEncode(buf) {{ - return Buffer.from(buf).toString("base64url"); -}} +const PYTHON_EXE = {_js_str(sys.executable)}; +const HOOK_SCRIPT_PATH = {_js_str(hook_script_path)}; +const GUARD_SCRIPT_PATH = {_js_str(guard_script_path)}; -function base64urlDecode(s) {{ - return Buffer.from(s, "base64url"); -}} - -function sha256Base64url(data) {{ - if (typeof data === "string") data = Buffer.from(data, "utf-8"); - return base64urlEncode(crypto.createHash("sha256").update(data).digest()); -}} - -function jcsCanonicalize(value) {{ - if (value === null || value === undefined) return "null"; - if (typeof value === "boolean") return value ? "true" : "false"; - if (typeof value === "number") {{ - if (!isFinite(value)) return "null"; - if (Object.is(value, -0)) return "0"; - return JSON.stringify(value); - }} - if (typeof value === "string") return JSON.stringify(value); - if (Array.isArray(value)) return "[" + value.map(jcsCanonicalize).join(",") + "]"; - if (typeof value === "object") {{ - const keys = Object.keys(value).sort(); - const pairs = keys.map(k => JSON.stringify(k) + ":" + jcsCanonicalize(value[k])); - return "{{" + pairs.join(",") + "}}"; - }} - return JSON.stringify(value); -}} - -function computePayloadHash(payload) {{ - return sha256Base64url(jcsCanonicalize(payload)); -}} - -function computeChainHash(prev, payloadHash, opId, issuedAt) {{ - return sha256Base64url(`${{prev}}|${{payloadHash}}|${{opId}}|${{issuedAt}}`); -}} - -function signEd25519(privKeyB64url, data) {{ - const seed = base64urlDecode(privKeyB64url); - const key = crypto.createPrivateKey({{ - key: Buffer.concat([Buffer.from("302e020100300506032b657004220420", "hex"), seed]), - format: "der", - type: "pkcs8", - }}); - return base64urlEncode(crypto.sign(null, Buffer.from(data, "utf-8"), key)); -}} - -function generateUuidv7() {{ - const ts = Date.now(); - const buf = Buffer.alloc(16); - buf.writeUIntBE(ts, 0, 6); - const rand = crypto.randomBytes(10); - rand.copy(buf, 6); - buf[6] = (buf[6] & 0x0f) | 0x70; - buf[8] = (buf[8] & 0x3f) | 0x80; - const h = buf.toString("hex"); - return `${{h.slice(0,8)}}-${{h.slice(8,12)}}-${{h.slice(12,16)}}-${{h.slice(16,20)}}-${{h.slice(20)}}`; -}} - -function generateNonce() {{ - return base64urlEncode(crypto.randomBytes(16)); -}} - -function loadChainState() {{ - try {{ - return JSON.parse(fs.readFileSync(CHAIN_STATE_PATH, "utf-8")); - }} catch {{ - return {{ prev_chain_hash: "" }}; - }} -}} - -function saveChainState(state) {{ - const dir = path.dirname(CHAIN_STATE_PATH); - fs.mkdirSync(dir, {{ recursive: true }}); - const tmp = CHAIN_STATE_PATH + ".tmp"; - fs.writeFileSync(tmp, JSON.stringify(state)); - fs.renameSync(tmp, CHAIN_STATE_PATH); -}} - -const PYTHON_EXE = "{python_exe_escaped}"; -const GUARD_SCRIPT_PATH = "{guard_path_escaped}"; - -module.exports = {{ - name: "elydora-audit", - version: "1.0.0", - description: "Sends tool-use events to the Elydora tamper-evident audit platform", - - hooks: {{ - preToolUse(context) {{ - // Synchronous guard — blocks tool if agent is frozen - if (GUARD_SCRIPT_PATH) {{ - try {{ - const result = spawnSync(PYTHON_EXE, [GUARD_SCRIPT_PATH], {{ - timeout: 5000, - stdio: ["pipe", "ignore", "pipe"], - }}); - if (result.status !== 0) {{ - const msg = result.stderr ? result.stderr.toString().trim() : "Agent is frozen by Elydora."; - return {{ blocked: true, reason: msg }}; - }} - }} catch {{ - // Fail-open — allow if guard can't run - }} - }} - }}, - - postToolUse(context) {{ - const event = context || {{}}; - try {{ - const toolName = event.tool_name || event.toolName || event.name || ""; - const toolInput = event.tool_input || event.toolInput || event.input || {{}}; - const toolOutput = event.tool_output || event.toolOutput || event.output || {{}}; - - const operationId = generateUuidv7(); - const issuedAt = Date.now(); - const nonce = generateNonce(); - const payload = {{ tool_name: toolName, tool_input: toolInput, tool_output: toolOutput }}; - const payloadHash = computePayloadHash(payload); - - const chainState = loadChainState(); - const prevChainHash = chainState.prev_chain_hash || ""; - const chainHash = computeChainHash(prevChainHash, payloadHash, operationId, issuedAt); - - const eor = {{ - op_version: "1.0", - operation_id: operationId, - org_id: ORG_ID, - agent_id: AGENT_ID, - issued_at: issuedAt, - ttl_ms: 30000, - nonce: nonce, - operation_type: "ai.tool_use", - subject: {{ type: "tool", id: toolName }}, - action: {{ type: "execute", tool: toolName }}, - payload: payload, - payload_hash: payloadHash, - prev_chain_hash: prevChainHash, - agent_pubkey_kid: KID, - signature: "", - }}; - - const signable = Object.fromEntries(Object.entries(eor).filter(([k]) => k !== "signature")); - eor.signature = signEd25519(PRIVATE_KEY, jcsCanonicalize(signable)); - - // POST to API — only save chain state on 2xx response - const url = new URL(BASE_URL.replace(/\\/$/, "") + "/v1/operations"); - const body = JSON.stringify(eor); - const mod = url.protocol === "https:" ? https : http; - const req = mod.request(url, {{ - method: "POST", - headers: {{ "Content-Type": "application/json", "Content-Length": Buffer.byteLength(body) }}, - timeout: 5000, - }}, (res) => {{ - if (res.statusCode >= 200 && res.statusCode < 300) {{ - saveChainState({{ prev_chain_hash: chainHash }}); - }} - res.resume(); - }}); - req.on("error", () => {{}}); - req.write(body); - req.end(); - }} catch (e) {{ - // Never block the host agent - console.error("[elydora]", e.message); - }} - }}, +export const ElydoraAuditPlugin = async () => ({{ + "tool.execute.before": async () => {{ + const result = spawnSync(PYTHON_EXE, [GUARD_SCRIPT_PATH], {{ + timeout: 5000, + stdio: ['pipe', 'ignore', 'pipe'], + }}); + if (result.error) {{ + throw new Error('Elydora guard failed: ' + result.error.message); + }} + if (result.status !== 0) {{ + const message = result.stderr?.toString().trim() || 'Elydora guard blocked tool execution.'; + throw new Error(message); + }} }}, -}}; + + "tool.execute.after": async (input) => {{ + const event = JSON.stringify({{ + tool_name: input?.tool || 'unknown', + tool_input: input?.args || {{}}, + session_id: input?.sessionID || 'unknown', + }}); + const child = spawn(PYTHON_EXE, [HOOK_SCRIPT_PATH], {{ + detached: true, + stdio: ['pipe', 'ignore', 'ignore'], + }}); + child.on('error', (error) => {{ + console.error('[elydora] audit hook failed:', error.message); + }}); + child.stdin.end(event); + child.unref(); + }}, +}}); ''' -def _js_str(s: str) -> str: - """Escape a Python string for embedding as a JS string literal.""" - return json.dumps(s) +def _js_str(value: str) -> str: + return json.dumps(value) diff --git a/elydora/plugins/registry.py b/elydora/plugins/registry.py index 2346f23..54de588 100644 --- a/elydora/plugins/registry.py +++ b/elydora/plugins/registry.py @@ -43,8 +43,8 @@ SUPPORTED_AGENTS: Dict[str, Dict[str, str]] = { }, "opencode": { "name": "OpenCode", - "hook_event": "PostToolUse", - "config_path": "~/.opencode/plugins/", + "hook_event": "tool.execute.after", + "config_path": "~/.config/opencode/plugins/", }, } diff --git a/tests/test_opencode_plugin.py b/tests/test_opencode_plugin.py new file mode 100644 index 0000000..eb427db --- /dev/null +++ b/tests/test_opencode_plugin.py @@ -0,0 +1,135 @@ +from __future__ import annotations + +import json +import os +from pathlib import Path +import shutil +import subprocess +import time + +import pytest + +from elydora.plugins import opencode +from elydora.plugins.registry import SUPPORTED_AGENTS + + +AGENT_ID = "agent-1" + + +def install_plugin( + monkeypatch: pytest.MonkeyPatch, + tmp_path: Path, +) -> tuple[Path, Path, Path]: + plugin_dir = tmp_path / ".config" / "opencode" / "plugins" + elydora_dir = tmp_path / ".elydora" + monkeypatch.setattr(opencode, "PLUGIN_DIR", str(plugin_dir)) + monkeypatch.setattr(opencode, "ELYDORA_DIR", str(elydora_dir)) + + agent_dir = elydora_dir / AGENT_ID + agent_dir.mkdir(parents=True) + guard_path = agent_dir / "guard.py" + guard_path.write_text( + "import sys\nsys.stderr.write('Agent is frozen by Elydora.')\nraise SystemExit(2)\n", + encoding="utf-8", + ) + + plugin = opencode.OpenCodePlugin() + plugin.install({ + "agent_id": AGENT_ID, + "agent_name": "opencode", + "org_id": "org-1", + "private_key": "test-key", + "kid": "kid-1", + "base_url": "https://api.elydora.test", + "guard_script_path": str(guard_path), + }) + return plugin_dir / "elydora-audit.mjs", agent_dir / "hook.py", guard_path + + +def run_node(script: str, home_dir: Path) -> subprocess.CompletedProcess[str]: + node = shutil.which("node") + if node is None: + pytest.skip("Node.js is required to execute generated OpenCode plugins") + env = { + **os.environ, + "HOME": str(home_dir), + "USERPROFILE": str(home_dir), + } + return subprocess.run( + [node, "--input-type=module", "--eval", script], + capture_output=True, + check=False, + env=env, + text=True, + ) + + +def test_generated_opencode_plugin_uses_current_api_and_blocks_frozen_agent( + monkeypatch: pytest.MonkeyPatch, + tmp_path: Path, +) -> None: + plugin_path, hook_path, _guard_path = install_plugin(monkeypatch, tmp_path) + + assert hook_path.is_file() + module_url = plugin_path.as_uri() + script = f""" + const pluginModule = await import({json.dumps(module_url)}); + const hooks = await pluginModule.ElydoraAuditPlugin({{ project: {{ name: 'project' }} }}); + if (typeof hooks['tool.execute.before'] !== 'function') process.exit(10); + if (typeof hooks['tool.execute.after'] !== 'function') process.exit(11); + try {{ + await hooks['tool.execute.before']( + {{ tool: 'bash', sessionID: 'session-1', callID: 'call-1' }}, + {{ args: {{ command: 'echo test' }} }}, + ); + process.exit(12); + }} catch (error) {{ + if (!String(error.message).includes('Agent is frozen by Elydora')) process.exit(13); + }} + """ + + result = run_node(script, tmp_path) + + assert result.returncode == 0, result.stderr + assert opencode.OpenCodePlugin().status()["installed"] is True + assert SUPPORTED_AGENTS["opencode"]["config_path"] == "~/.config/opencode/plugins/" + + +def test_generated_opencode_plugin_forwards_tool_event( + monkeypatch: pytest.MonkeyPatch, + tmp_path: Path, +) -> None: + plugin_path, hook_path, _guard_path = install_plugin(monkeypatch, tmp_path) + capture_path = tmp_path / "captured-event.json" + hook_path.write_text( + "import pathlib, sys\n" + f"pathlib.Path({str(capture_path)!r}).write_text(sys.stdin.read(), encoding='utf-8')\n", + encoding="utf-8", + ) + script = f""" + const pluginModule = await import({json.dumps(plugin_path.as_uri())}); + const hooks = await pluginModule.ElydoraAuditPlugin({{ project: {{ name: 'project' }} }}); + await hooks['tool.execute.after']( + {{ + tool: 'bash', + sessionID: 'session-1', + callID: 'call-1', + args: {{ command: 'echo test' }}, + }}, + {{ title: 'Shell', output: 'test' }}, + ); + """ + + result = run_node(script, tmp_path) + assert result.returncode == 0, result.stderr + + deadline = time.monotonic() + 3 + while not capture_path.exists() and time.monotonic() < deadline: + time.sleep(0.02) + assert capture_path.is_file() + event = json.loads(capture_path.read_text(encoding="utf-8")) + assert event == { + "tool_name": "bash", + "tool_input": {"command": "echo test"}, + "session_id": "session-1", + }