Migrate to per-agent subdirectories, uniform 8-agent support, cross-platform fixes
- All paths use ~/.elydora/{agentId}/ per-agent isolation
- Add copilot, letta plugins; split kiro into kirocli + kiroide
- Remove augment (SDK-only agent, not one-click)
- Fix guard command: use sys.executable instead of hardcoded python3
- Fix chmod wrapped in try/except for Windows compatibility
- Fix OpenCode chain state: save after HTTP 2xx response
- Fix guard script shebang consistency (sys.executable)
- Uniform 8 agents: claudecode, cursor, gemini, kirocli, kiroide, opencode, copilot, letta
Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
This commit is contained in:
1 parent
eef18a8520
commit
5b92a1dfa3
14 files changed
+973
-259
No files matched your search
@@ -87,10 +87,12 @@ elydora install \
|
|||||||
| Agent | Key |
|
| Agent | Key |
|
||||||
|-------|-----|
|
|-------|-----|
|
||||||
| Claude Code | `claudecode` |
|
| Claude Code | `claudecode` |
|
||||||
|
| Copilot CLI | `copilot` |
|
||||||
| Cursor | `cursor` |
|
| Cursor | `cursor` |
|
||||||
| Gemini CLI | `gemini` |
|
| Gemini CLI | `gemini` |
|
||||||
| Augment Code | `augment` |
|
| Kiro CLI | `kirocli` |
|
||||||
| Kiro | `kiro` |
|
| Kiro IDE | `kiroide` |
|
||||||
|
| Letta Code | `letta` |
|
||||||
| OpenCode | `opencode` |
|
| OpenCode | `opencode` |
|
||||||
|
|
||||||
## API Reference
|
## API Reference
|
||||||
|
|||||||
+51
-16
@@ -7,7 +7,10 @@ Uses only stdlib argparse — zero external dependencies for the CLI itself.
|
|||||||
from __future__ import annotations
|
from __future__ import annotations
|
||||||
|
|
||||||
import argparse
|
import argparse
|
||||||
|
import glob
|
||||||
|
import json
|
||||||
import os
|
import os
|
||||||
|
import shutil
|
||||||
import sys
|
import sys
|
||||||
from typing import Dict, Type
|
from typing import Dict, Type
|
||||||
|
|
||||||
@@ -16,19 +19,23 @@ from .plugins.base import AgentPlugin, InstallConfig
|
|||||||
from .plugins.registry import SUPPORTED_AGENTS, get_agent_names
|
from .plugins.registry import SUPPORTED_AGENTS, get_agent_names
|
||||||
from .plugins.hook_template import generate_guard_script
|
from .plugins.hook_template import generate_guard_script
|
||||||
from .plugins.claudecode import ClaudeCodePlugin
|
from .plugins.claudecode import ClaudeCodePlugin
|
||||||
|
from .plugins.copilot import CopilotPlugin
|
||||||
from .plugins.cursor import CursorPlugin
|
from .plugins.cursor import CursorPlugin
|
||||||
from .plugins.gemini import GeminiPlugin
|
from .plugins.gemini import GeminiPlugin
|
||||||
from .plugins.augment import AugmentPlugin
|
from .plugins.kirocli import KiroCliPlugin
|
||||||
from .plugins.kiro import KiroPlugin
|
from .plugins.kiroide import KiroIdePlugin
|
||||||
|
from .plugins.letta import LettaPlugin
|
||||||
from .plugins.opencode import OpenCodePlugin
|
from .plugins.opencode import OpenCodePlugin
|
||||||
|
|
||||||
|
|
||||||
PLUGIN_MAP: Dict[str, Type[AgentPlugin]] = {
|
PLUGIN_MAP: Dict[str, Type[AgentPlugin]] = {
|
||||||
"claudecode": ClaudeCodePlugin,
|
"claudecode": ClaudeCodePlugin,
|
||||||
|
"copilot": CopilotPlugin,
|
||||||
"cursor": CursorPlugin,
|
"cursor": CursorPlugin,
|
||||||
"gemini": GeminiPlugin,
|
"gemini": GeminiPlugin,
|
||||||
"augment": AugmentPlugin,
|
"kirocli": KiroCliPlugin,
|
||||||
"kiro": KiroPlugin,
|
"kiroide": KiroIdePlugin,
|
||||||
|
"letta": LettaPlugin,
|
||||||
"opencode": OpenCodePlugin,
|
"opencode": OpenCodePlugin,
|
||||||
}
|
}
|
||||||
|
|
||||||
@@ -56,11 +63,13 @@ def cmd_install(args: argparse.Namespace) -> None:
|
|||||||
|
|
||||||
print(f"Verified key pair (public key: {pub[:16]}...)")
|
print(f"Verified key pair (public key: {pub[:16]}...)")
|
||||||
|
|
||||||
|
# Create per-agent directory under ~/.elydora/{agent_id}/
|
||||||
|
agent_dir = os.path.join(os.path.expanduser("~"), ".elydora", args.agent_id)
|
||||||
|
os.makedirs(agent_dir, exist_ok=True)
|
||||||
|
|
||||||
# Generate and write the guard script
|
# Generate and write the guard script
|
||||||
guard_dir = os.path.join(os.path.expanduser("~"), ".elydora", "hooks")
|
guard_script_path = os.path.join(agent_dir, "guard.py")
|
||||||
os.makedirs(guard_dir, exist_ok=True)
|
guard_script = generate_guard_script(agent_name, args.agent_id)
|
||||||
guard_script_path = os.path.join(guard_dir, f"{agent_name}-guard.py")
|
|
||||||
guard_script = generate_guard_script(agent_name)
|
|
||||||
with open(guard_script_path, "w", encoding="utf-8") as f:
|
with open(guard_script_path, "w", encoding="utf-8") as f:
|
||||||
f.write(guard_script)
|
f.write(guard_script)
|
||||||
try:
|
try:
|
||||||
@@ -73,6 +82,7 @@ def cmd_install(args: argparse.Namespace) -> None:
|
|||||||
config: InstallConfig = {
|
config: InstallConfig = {
|
||||||
"org_id": args.org_id,
|
"org_id": args.org_id,
|
||||||
"agent_id": args.agent_id,
|
"agent_id": args.agent_id,
|
||||||
|
"agent_name": agent_name,
|
||||||
"private_key": args.private_key,
|
"private_key": args.private_key,
|
||||||
"kid": args.kid,
|
"kid": args.kid,
|
||||||
"base_url": args.base_url,
|
"base_url": args.base_url,
|
||||||
@@ -88,15 +98,39 @@ def cmd_install(args: argparse.Namespace) -> None:
|
|||||||
def cmd_uninstall(args: argparse.Namespace) -> None:
|
def cmd_uninstall(args: argparse.Namespace) -> None:
|
||||||
"""Handle the 'uninstall' subcommand."""
|
"""Handle the 'uninstall' subcommand."""
|
||||||
plugin = _get_plugin(args.agent)
|
plugin = _get_plugin(args.agent)
|
||||||
plugin.uninstall()
|
agent_id = getattr(args, "agent_id", None) or ""
|
||||||
|
plugin.uninstall(agent_id=agent_id)
|
||||||
|
|
||||||
# Remove the guard script
|
# Determine which agent directory to remove
|
||||||
guard_script_path = os.path.join(
|
elydora_dir = os.path.join(os.path.expanduser("~"), ".elydora")
|
||||||
os.path.expanduser("~"), ".elydora", "hooks", f"{args.agent}-guard.py"
|
agent_id = getattr(args, "agent_id", None)
|
||||||
)
|
|
||||||
if os.path.exists(guard_script_path):
|
if not agent_id:
|
||||||
os.remove(guard_script_path)
|
# Scan ~/.elydora/*/config.json for matching agent_name
|
||||||
print(f" Removed guard script: {guard_script_path}")
|
pattern = os.path.join(elydora_dir, "*", "config.json")
|
||||||
|
for config_path in glob.glob(pattern):
|
||||||
|
try:
|
||||||
|
with open(config_path, "r", encoding="utf-8") as f:
|
||||||
|
cfg = json.load(f)
|
||||||
|
if cfg.get("agent_name") == args.agent:
|
||||||
|
agent_id = cfg.get("agent_id")
|
||||||
|
break
|
||||||
|
except Exception:
|
||||||
|
continue
|
||||||
|
|
||||||
|
if agent_id:
|
||||||
|
agent_dir = os.path.join(elydora_dir, agent_id)
|
||||||
|
if os.path.isdir(agent_dir):
|
||||||
|
shutil.rmtree(agent_dir)
|
||||||
|
print(f" Removed agent directory: {agent_dir}")
|
||||||
|
else:
|
||||||
|
# Fallback: try to remove legacy guard script location
|
||||||
|
guard_script_path = os.path.join(
|
||||||
|
elydora_dir, "hooks", f"{args.agent}-guard.py"
|
||||||
|
)
|
||||||
|
if os.path.exists(guard_script_path):
|
||||||
|
os.remove(guard_script_path)
|
||||||
|
print(f" Removed guard script: {guard_script_path}")
|
||||||
|
|
||||||
|
|
||||||
def cmd_status(args: argparse.Namespace) -> None:
|
def cmd_status(args: argparse.Namespace) -> None:
|
||||||
@@ -144,6 +178,7 @@ def build_parser() -> argparse.ArgumentParser:
|
|||||||
# uninstall
|
# uninstall
|
||||||
uninstall_parser = subparsers.add_parser("uninstall", help="Uninstall audit hook for an agent")
|
uninstall_parser = subparsers.add_parser("uninstall", help="Uninstall audit hook for an agent")
|
||||||
uninstall_parser.add_argument("--agent", required=True, help="Agent name")
|
uninstall_parser.add_argument("--agent", required=True, help="Agent name")
|
||||||
|
uninstall_parser.add_argument("--agent_id", default=None, help="Agent ID (if omitted, scans config files for matching agent name)")
|
||||||
|
|
||||||
# status
|
# status
|
||||||
subparsers.add_parser("status", help="Show installation status of all agents")
|
subparsers.add_parser("status", help="Show installation status of all agents")
|
||||||
|
|||||||
@@ -15,6 +15,7 @@ else:
|
|||||||
class InstallConfig(TypedDict, total=False):
|
class InstallConfig(TypedDict, total=False):
|
||||||
org_id: str
|
org_id: str
|
||||||
agent_id: str
|
agent_id: str
|
||||||
|
agent_name: str
|
||||||
private_key: str
|
private_key: str
|
||||||
kid: str
|
kid: str
|
||||||
token: str
|
token: str
|
||||||
@@ -36,7 +37,7 @@ class AgentPlugin(ABC):
|
|||||||
"""Install the Elydora hook for this agent."""
|
"""Install the Elydora hook for this agent."""
|
||||||
|
|
||||||
@abstractmethod
|
@abstractmethod
|
||||||
def uninstall(self) -> None:
|
def uninstall(self, agent_id: str = "") -> None:
|
||||||
"""Remove the Elydora hook for this agent."""
|
"""Remove the Elydora hook for this agent."""
|
||||||
|
|
||||||
@abstractmethod
|
@abstractmethod
|
||||||
|
|||||||
@@ -5,38 +5,72 @@ from __future__ import annotations
|
|||||||
import json
|
import json
|
||||||
import os
|
import os
|
||||||
import stat
|
import stat
|
||||||
|
import sys
|
||||||
|
|
||||||
from .base import AgentPlugin, InstallConfig, PluginStatus
|
from .base import AgentPlugin, InstallConfig, PluginStatus
|
||||||
from .hook_template import generate_hook_script
|
from .hook_template import generate_hook_script
|
||||||
|
|
||||||
|
|
||||||
SETTINGS_PATH = os.path.join(os.path.expanduser("~"), ".claude", "settings.json")
|
SETTINGS_PATH = os.path.join(os.path.expanduser("~"), ".claude", "settings.json")
|
||||||
HOOK_DIR = os.path.join(os.path.expanduser("~"), ".elydora", "hooks")
|
ELYDORA_DIR = os.path.join(os.path.expanduser("~"), ".elydora")
|
||||||
HOOK_FILENAME = "elydora-audit-hook.py"
|
|
||||||
|
|
||||||
|
|
||||||
class ClaudeCodePlugin(AgentPlugin):
|
class ClaudeCodePlugin(AgentPlugin):
|
||||||
"""Install/uninstall Elydora audit hook for Claude Code."""
|
"""Install/uninstall Elydora audit hook for Claude Code."""
|
||||||
|
|
||||||
def _hook_path(self) -> str:
|
@staticmethod
|
||||||
return os.path.join(HOOK_DIR, HOOK_FILENAME)
|
def _hook_path_for(agent_id: str) -> str:
|
||||||
|
return os.path.join(ELYDORA_DIR, agent_id, "hook.py")
|
||||||
|
|
||||||
def install(self, config: InstallConfig) -> None:
|
def install(self, config: InstallConfig) -> None:
|
||||||
|
agent_id = config.get("agent_id", "")
|
||||||
|
agent_name = config.get("agent_name", "")
|
||||||
|
|
||||||
|
# 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": config.get("org_id", ""),
|
||||||
|
"agent_id": agent_id,
|
||||||
|
"kid": config.get("kid", ""),
|
||||||
|
"base_url": config.get("base_url", "https://api.elydora.com"),
|
||||||
|
"token": config.get("token", ""),
|
||||||
|
"agent_name": agent_name,
|
||||||
|
}
|
||||||
|
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(config.get("private_key", ""))
|
||||||
|
try:
|
||||||
|
os.chmod(private_key_path, stat.S_IRUSR | stat.S_IWUSR)
|
||||||
|
except Exception:
|
||||||
|
pass # chmod may fail on Windows
|
||||||
|
|
||||||
# Write the hook script
|
# Write the hook script
|
||||||
script = generate_hook_script(
|
script = generate_hook_script(
|
||||||
org_id=config.get("org_id", ""),
|
org_id=config.get("org_id", ""),
|
||||||
agent_id=config.get("agent_id", ""),
|
agent_id=agent_id,
|
||||||
private_key=config.get("private_key", ""),
|
private_key=config.get("private_key", ""),
|
||||||
kid=config.get("kid", ""),
|
kid=config.get("kid", ""),
|
||||||
base_url=config.get("base_url", "https://api.elydora.com"),
|
base_url=config.get("base_url", "https://api.elydora.com"),
|
||||||
)
|
)
|
||||||
os.makedirs(HOOK_DIR, exist_ok=True)
|
hook_path = self._hook_path_for(agent_id)
|
||||||
hook_path = self._hook_path()
|
|
||||||
with open(hook_path, "w", encoding="utf-8") as f:
|
with open(hook_path, "w", encoding="utf-8") as f:
|
||||||
f.write(script)
|
f.write(script)
|
||||||
os.chmod(hook_path, stat.S_IRWXU)
|
try:
|
||||||
|
os.chmod(hook_path, stat.S_IRWXU)
|
||||||
|
except Exception:
|
||||||
|
pass # chmod may fail on Windows
|
||||||
|
|
||||||
guard_script_path = config.get("guard_script_path", "")
|
guard_script_path = config.get("guard_script_path", "")
|
||||||
|
python_exe = sys.executable
|
||||||
|
|
||||||
# Merge into Claude Code settings
|
# Merge into Claude Code settings
|
||||||
settings = _load_json(SETTINGS_PATH)
|
settings = _load_json(SETTINGS_PATH)
|
||||||
@@ -50,7 +84,7 @@ class ClaudeCodePlugin(AgentPlugin):
|
|||||||
"hooks": [
|
"hooks": [
|
||||||
{
|
{
|
||||||
"type": "command",
|
"type": "command",
|
||||||
"command": f"python3 {guard_script_path}",
|
"command": f'"{python_exe}" {guard_script_path}',
|
||||||
}
|
}
|
||||||
],
|
],
|
||||||
})
|
})
|
||||||
@@ -75,7 +109,7 @@ class ClaudeCodePlugin(AgentPlugin):
|
|||||||
print(f" Hook script: {hook_path}")
|
print(f" Hook script: {hook_path}")
|
||||||
print(f" Settings: {SETTINGS_PATH}")
|
print(f" Settings: {SETTINGS_PATH}")
|
||||||
|
|
||||||
def uninstall(self) -> None:
|
def uninstall(self, agent_id: str = "") -> None:
|
||||||
# Remove hook entries from settings
|
# Remove hook entries from settings
|
||||||
if os.path.exists(SETTINGS_PATH):
|
if os.path.exists(SETTINGS_PATH):
|
||||||
settings = _load_json(SETTINGS_PATH)
|
settings = _load_json(SETTINGS_PATH)
|
||||||
@@ -84,7 +118,7 @@ class ClaudeCodePlugin(AgentPlugin):
|
|||||||
|
|
||||||
# Remove PreToolUse entries
|
# Remove PreToolUse entries
|
||||||
pre_tool_use = hooks.get("PreToolUse", [])
|
pre_tool_use = hooks.get("PreToolUse", [])
|
||||||
pre_filtered = [h for h in pre_tool_use if not _is_elydora_hook(h)]
|
pre_filtered = [h for h in pre_tool_use if not _is_elydora_hook(h, agent_id)]
|
||||||
if len(pre_filtered) != len(pre_tool_use):
|
if len(pre_filtered) != len(pre_tool_use):
|
||||||
hooks["PreToolUse"] = pre_filtered
|
hooks["PreToolUse"] = pre_filtered
|
||||||
if not pre_filtered:
|
if not pre_filtered:
|
||||||
@@ -93,7 +127,7 @@ class ClaudeCodePlugin(AgentPlugin):
|
|||||||
|
|
||||||
# Remove PostToolUse entries
|
# Remove PostToolUse entries
|
||||||
post_tool_use = hooks.get("PostToolUse", [])
|
post_tool_use = hooks.get("PostToolUse", [])
|
||||||
post_filtered = [h for h in post_tool_use if not _is_elydora_hook(h)]
|
post_filtered = [h for h in post_tool_use if not _is_elydora_hook(h, agent_id)]
|
||||||
if len(post_filtered) != len(post_tool_use):
|
if len(post_filtered) != len(post_tool_use):
|
||||||
hooks["PostToolUse"] = post_filtered
|
hooks["PostToolUse"] = post_filtered
|
||||||
if not post_filtered:
|
if not post_filtered:
|
||||||
@@ -105,16 +139,15 @@ class ClaudeCodePlugin(AgentPlugin):
|
|||||||
del settings["hooks"]
|
del settings["hooks"]
|
||||||
_save_json(SETTINGS_PATH, settings)
|
_save_json(SETTINGS_PATH, settings)
|
||||||
|
|
||||||
# Remove hook script
|
# Hook script removal is handled by cli.py cmd_uninstall (rmtree of agent dir)
|
||||||
hook_path = self._hook_path()
|
|
||||||
if os.path.exists(hook_path):
|
|
||||||
os.remove(hook_path)
|
|
||||||
|
|
||||||
print("Elydora hook uninstalled from Claude Code.")
|
print("Elydora hook uninstalled from Claude Code.")
|
||||||
|
|
||||||
def status(self) -> PluginStatus:
|
def status(self) -> PluginStatus:
|
||||||
hook_path = self._hook_path()
|
# Scan ~/.elydora/*/hook.py for any installed hook
|
||||||
hook_exists = os.path.exists(hook_path)
|
import glob as _glob
|
||||||
|
hook_pattern = os.path.join(ELYDORA_DIR, "*", "hook.py")
|
||||||
|
hook_files = _glob.glob(hook_pattern)
|
||||||
|
hook_exists = len(hook_files) > 0
|
||||||
|
|
||||||
settings_configured = False
|
settings_configured = False
|
||||||
if os.path.exists(SETTINGS_PATH):
|
if os.path.exists(SETTINGS_PATH):
|
||||||
@@ -128,7 +161,7 @@ class ClaudeCodePlugin(AgentPlugin):
|
|||||||
|
|
||||||
installed = hook_exists and settings_configured
|
installed = hook_exists and settings_configured
|
||||||
if installed:
|
if installed:
|
||||||
details = f"Hook: {hook_path}"
|
details = f"Found {len(hook_files)} agent(s): {', '.join(hook_files)}"
|
||||||
elif hook_exists:
|
elif hook_exists:
|
||||||
details = "Hook script exists but not configured in settings"
|
details = "Hook script exists but not configured in settings"
|
||||||
elif settings_configured:
|
elif settings_configured:
|
||||||
@@ -139,18 +172,29 @@ class ClaudeCodePlugin(AgentPlugin):
|
|||||||
return PluginStatus(installed=installed, agent="claudecode", details=details)
|
return PluginStatus(installed=installed, agent="claudecode", details=details)
|
||||||
|
|
||||||
|
|
||||||
def _is_elydora_hook(entry: dict) -> bool:
|
def _is_elydora_hook(entry: dict, agent_id: str = "") -> bool:
|
||||||
# New format: { "matcher": {}, "hooks": [{ "type": "command", "command": "..." }] }
|
# Collect all command strings from the entry
|
||||||
|
commands = []
|
||||||
inner_hooks = entry.get("hooks")
|
inner_hooks = entry.get("hooks")
|
||||||
if isinstance(inner_hooks, list):
|
if isinstance(inner_hooks, list):
|
||||||
return any(
|
commands.extend(
|
||||||
"elydora" in h.get("command", "").lower()
|
h.get("command", "")
|
||||||
for h in inner_hooks
|
for h in inner_hooks
|
||||||
if isinstance(h, dict)
|
if isinstance(h, dict)
|
||||||
)
|
)
|
||||||
# Old format: { "command": "..." }
|
else:
|
||||||
cmd = entry.get("command", "")
|
commands.append(entry.get("command", ""))
|
||||||
return "elydora" in cmd.lower()
|
|
||||||
|
for cmd in commands:
|
||||||
|
cmd_lower = cmd.lower()
|
||||||
|
if "elydora" not in cmd_lower:
|
||||||
|
continue
|
||||||
|
# If agent_id is specified, only match hooks for that specific agent
|
||||||
|
if agent_id and agent_id in cmd:
|
||||||
|
return True
|
||||||
|
if not agent_id:
|
||||||
|
return True
|
||||||
|
return False
|
||||||
|
|
||||||
|
|
||||||
def _load_json(path: str) -> dict:
|
def _load_json(path: str) -> dict:
|
||||||
|
|||||||
@@ -0,0 +1,200 @@
|
|||||||
|
"""Copilot CLI plugin — merges preToolUse/postToolUse hooks into .github/hooks/hooks.json (project-relative)."""
|
||||||
|
|
||||||
|
from __future__ import annotations
|
||||||
|
|
||||||
|
import json
|
||||||
|
import os
|
||||||
|
import stat
|
||||||
|
import sys
|
||||||
|
|
||||||
|
from .base import AgentPlugin, InstallConfig, PluginStatus
|
||||||
|
from .hook_template import generate_hook_script
|
||||||
|
|
||||||
|
|
||||||
|
ELYDORA_DIR = os.path.join(os.path.expanduser("~"), ".elydora")
|
||||||
|
|
||||||
|
|
||||||
|
def _settings_path() -> str:
|
||||||
|
return os.path.join(os.getcwd(), ".github", "hooks", "hooks.json")
|
||||||
|
|
||||||
|
|
||||||
|
class CopilotPlugin(AgentPlugin):
|
||||||
|
"""Install/uninstall Elydora audit hook for Copilot CLI."""
|
||||||
|
|
||||||
|
@staticmethod
|
||||||
|
def _hook_path_for(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", "")
|
||||||
|
|
||||||
|
# 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": config.get("org_id", ""),
|
||||||
|
"agent_id": agent_id,
|
||||||
|
"kid": config.get("kid", ""),
|
||||||
|
"base_url": config.get("base_url", "https://api.elydora.com"),
|
||||||
|
"token": config.get("token", ""),
|
||||||
|
"agent_name": agent_name,
|
||||||
|
}
|
||||||
|
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(config.get("private_key", ""))
|
||||||
|
try:
|
||||||
|
os.chmod(private_key_path, stat.S_IRUSR | stat.S_IWUSR)
|
||||||
|
except Exception:
|
||||||
|
pass # chmod may fail on Windows
|
||||||
|
|
||||||
|
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"),
|
||||||
|
)
|
||||||
|
hook_path = self._hook_path_for(agent_id)
|
||||||
|
with open(hook_path, "w", encoding="utf-8") as f:
|
||||||
|
f.write(script)
|
||||||
|
try:
|
||||||
|
os.chmod(hook_path, stat.S_IRWXU)
|
||||||
|
except Exception:
|
||||||
|
pass # chmod may fail on Windows
|
||||||
|
|
||||||
|
guard_script_path = config.get("guard_script_path", "")
|
||||||
|
python_exe = sys.executable
|
||||||
|
|
||||||
|
settings_path = _settings_path()
|
||||||
|
settings = _load_json(settings_path)
|
||||||
|
settings["version"] = 1
|
||||||
|
hooks = settings.setdefault("hooks", {})
|
||||||
|
|
||||||
|
# --- preToolUse (guard — freeze enforcement, camelCase) ---
|
||||||
|
pre_tool_use = hooks.setdefault("preToolUse", [])
|
||||||
|
pre_tool_use[:] = [h for h in pre_tool_use if not _is_elydora_hook(h)]
|
||||||
|
if guard_script_path:
|
||||||
|
pre_tool_use.append({
|
||||||
|
"type": "command",
|
||||||
|
"bash": f'"{python_exe}" {guard_script_path}',
|
||||||
|
"powershell": f'"{python_exe}" {guard_script_path}',
|
||||||
|
"timeoutSec": 5,
|
||||||
|
})
|
||||||
|
|
||||||
|
# --- postToolUse (audit logging, camelCase) ---
|
||||||
|
post_tool_use = hooks.setdefault("postToolUse", [])
|
||||||
|
|
||||||
|
post_tool_use[:] = [h for h in post_tool_use if not _is_elydora_hook(h)]
|
||||||
|
|
||||||
|
post_tool_use.append({
|
||||||
|
"type": "command",
|
||||||
|
"bash": hook_path,
|
||||||
|
"powershell": hook_path,
|
||||||
|
"timeoutSec": 5,
|
||||||
|
})
|
||||||
|
|
||||||
|
_save_json(settings_path, settings)
|
||||||
|
print(f"Elydora hook installed for Copilot CLI.")
|
||||||
|
print(f" Hook script: {hook_path}")
|
||||||
|
print(f" Settings: {settings_path}")
|
||||||
|
|
||||||
|
def uninstall(self, agent_id: str = "") -> None:
|
||||||
|
settings_path = _settings_path()
|
||||||
|
if os.path.exists(settings_path):
|
||||||
|
settings = _load_json(settings_path)
|
||||||
|
hooks = settings.get("hooks", {})
|
||||||
|
changed = False
|
||||||
|
|
||||||
|
# Remove preToolUse entries
|
||||||
|
pre_tool_use = hooks.get("preToolUse", [])
|
||||||
|
pre_filtered = [h for h in pre_tool_use if not _is_elydora_hook(h, agent_id)]
|
||||||
|
if len(pre_filtered) != len(pre_tool_use):
|
||||||
|
hooks["preToolUse"] = pre_filtered
|
||||||
|
if not pre_filtered:
|
||||||
|
del hooks["preToolUse"]
|
||||||
|
changed = True
|
||||||
|
|
||||||
|
# Remove postToolUse entries
|
||||||
|
post_tool_use = hooks.get("postToolUse", [])
|
||||||
|
post_filtered = [h for h in post_tool_use if not _is_elydora_hook(h, agent_id)]
|
||||||
|
if len(post_filtered) != len(post_tool_use):
|
||||||
|
hooks["postToolUse"] = post_filtered
|
||||||
|
if not post_filtered:
|
||||||
|
del hooks["postToolUse"]
|
||||||
|
changed = True
|
||||||
|
|
||||||
|
if changed:
|
||||||
|
if not hooks:
|
||||||
|
del settings["hooks"]
|
||||||
|
_save_json(settings_path, settings)
|
||||||
|
|
||||||
|
# Hook script removal is handled by cli.py cmd_uninstall (rmtree of agent dir)
|
||||||
|
print("Elydora hook uninstalled from Copilot CLI.")
|
||||||
|
|
||||||
|
def status(self) -> PluginStatus:
|
||||||
|
# Scan ~/.elydora/*/hook.py for any installed hook
|
||||||
|
import glob as _glob
|
||||||
|
hook_pattern = os.path.join(ELYDORA_DIR, "*", "hook.py")
|
||||||
|
hook_files = _glob.glob(hook_pattern)
|
||||||
|
hook_exists = len(hook_files) > 0
|
||||||
|
|
||||||
|
settings_configured = False
|
||||||
|
settings_path = _settings_path()
|
||||||
|
if os.path.exists(settings_path):
|
||||||
|
settings = _load_json(settings_path)
|
||||||
|
hooks = settings.get("hooks", {})
|
||||||
|
pre_tool_use = hooks.get("preToolUse", [])
|
||||||
|
post_tool_use = hooks.get("postToolUse", [])
|
||||||
|
pre_configured = any(_is_elydora_hook(h) for h in pre_tool_use)
|
||||||
|
post_configured = any(_is_elydora_hook(h) for h in post_tool_use)
|
||||||
|
settings_configured = pre_configured and post_configured
|
||||||
|
|
||||||
|
installed = hook_exists and settings_configured
|
||||||
|
if installed:
|
||||||
|
details = f"Found {len(hook_files)} agent(s): {', '.join(hook_files)}"
|
||||||
|
elif hook_exists:
|
||||||
|
details = "Hook script exists but not configured in hooks.json"
|
||||||
|
elif settings_configured:
|
||||||
|
details = "Configured in hooks.json but hook script missing"
|
||||||
|
else:
|
||||||
|
details = "Not installed"
|
||||||
|
|
||||||
|
return PluginStatus(installed=installed, agent="copilot", details=details)
|
||||||
|
|
||||||
|
|
||||||
|
def _is_elydora_hook(hook: dict, agent_id: str = "") -> bool:
|
||||||
|
# Check both bash and powershell fields for Copilot's hook format
|
||||||
|
for field in ("bash", "powershell", "command"):
|
||||||
|
cmd = hook.get(field, "")
|
||||||
|
if not isinstance(cmd, str):
|
||||||
|
continue
|
||||||
|
if "elydora" not in cmd.lower():
|
||||||
|
continue
|
||||||
|
if agent_id and agent_id in cmd:
|
||||||
|
return True
|
||||||
|
if not agent_id:
|
||||||
|
return True
|
||||||
|
return False
|
||||||
|
|
||||||
|
|
||||||
|
def _load_json(path: str) -> dict:
|
||||||
|
if not os.path.exists(path):
|
||||||
|
return {}
|
||||||
|
with open(path, "r", encoding="utf-8") as f:
|
||||||
|
return json.load(f)
|
||||||
|
|
||||||
|
|
||||||
|
def _save_json(path: str, data: dict) -> None:
|
||||||
|
os.makedirs(os.path.dirname(path), exist_ok=True)
|
||||||
|
with open(path, "w", encoding="utf-8") as f:
|
||||||
|
json.dump(data, f, indent=2)
|
||||||
|
f.write("\n")
|
||||||
+60
-21
@@ -5,37 +5,71 @@ from __future__ import annotations
|
|||||||
import json
|
import json
|
||||||
import os
|
import os
|
||||||
import stat
|
import stat
|
||||||
|
import sys
|
||||||
|
|
||||||
from .base import AgentPlugin, InstallConfig, PluginStatus
|
from .base import AgentPlugin, InstallConfig, PluginStatus
|
||||||
from .hook_template import generate_hook_script
|
from .hook_template import generate_hook_script
|
||||||
|
|
||||||
|
|
||||||
SETTINGS_PATH = os.path.join(os.path.expanduser("~"), ".cursor", "hooks.json")
|
SETTINGS_PATH = os.path.join(os.path.expanduser("~"), ".cursor", "hooks.json")
|
||||||
HOOK_DIR = os.path.join(os.path.expanduser("~"), ".elydora", "hooks")
|
ELYDORA_DIR = os.path.join(os.path.expanduser("~"), ".elydora")
|
||||||
HOOK_FILENAME = "elydora-audit-hook-cursor.py"
|
|
||||||
|
|
||||||
|
|
||||||
class CursorPlugin(AgentPlugin):
|
class CursorPlugin(AgentPlugin):
|
||||||
"""Install/uninstall Elydora audit hook for Cursor."""
|
"""Install/uninstall Elydora audit hook for Cursor."""
|
||||||
|
|
||||||
def _hook_path(self) -> str:
|
@staticmethod
|
||||||
return os.path.join(HOOK_DIR, HOOK_FILENAME)
|
def _hook_path_for(agent_id: str) -> str:
|
||||||
|
return os.path.join(ELYDORA_DIR, agent_id, "hook.py")
|
||||||
|
|
||||||
def install(self, config: InstallConfig) -> None:
|
def install(self, config: InstallConfig) -> None:
|
||||||
|
agent_id = config.get("agent_id", "")
|
||||||
|
agent_name = config.get("agent_name", "")
|
||||||
|
|
||||||
|
# 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": config.get("org_id", ""),
|
||||||
|
"agent_id": agent_id,
|
||||||
|
"kid": config.get("kid", ""),
|
||||||
|
"base_url": config.get("base_url", "https://api.elydora.com"),
|
||||||
|
"token": config.get("token", ""),
|
||||||
|
"agent_name": agent_name,
|
||||||
|
}
|
||||||
|
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(config.get("private_key", ""))
|
||||||
|
try:
|
||||||
|
os.chmod(private_key_path, stat.S_IRUSR | stat.S_IWUSR)
|
||||||
|
except Exception:
|
||||||
|
pass # chmod may fail on Windows
|
||||||
|
|
||||||
script = generate_hook_script(
|
script = generate_hook_script(
|
||||||
org_id=config.get("org_id", ""),
|
org_id=config.get("org_id", ""),
|
||||||
agent_id=config.get("agent_id", ""),
|
agent_id=agent_id,
|
||||||
private_key=config.get("private_key", ""),
|
private_key=config.get("private_key", ""),
|
||||||
kid=config.get("kid", ""),
|
kid=config.get("kid", ""),
|
||||||
base_url=config.get("base_url", "https://api.elydora.com"),
|
base_url=config.get("base_url", "https://api.elydora.com"),
|
||||||
)
|
)
|
||||||
os.makedirs(HOOK_DIR, exist_ok=True)
|
hook_path = self._hook_path_for(agent_id)
|
||||||
hook_path = self._hook_path()
|
|
||||||
with open(hook_path, "w", encoding="utf-8") as f:
|
with open(hook_path, "w", encoding="utf-8") as f:
|
||||||
f.write(script)
|
f.write(script)
|
||||||
os.chmod(hook_path, stat.S_IRWXU)
|
try:
|
||||||
|
os.chmod(hook_path, stat.S_IRWXU)
|
||||||
|
except Exception:
|
||||||
|
pass # chmod may fail on Windows
|
||||||
|
|
||||||
guard_script_path = config.get("guard_script_path", "")
|
guard_script_path = config.get("guard_script_path", "")
|
||||||
|
python_exe = sys.executable
|
||||||
|
|
||||||
settings = _load_json(SETTINGS_PATH)
|
settings = _load_json(SETTINGS_PATH)
|
||||||
hooks = settings.setdefault("hooks", {})
|
hooks = settings.setdefault("hooks", {})
|
||||||
@@ -45,7 +79,7 @@ class CursorPlugin(AgentPlugin):
|
|||||||
pre_tool_use[:] = [h for h in pre_tool_use if not _is_elydora_hook(h)]
|
pre_tool_use[:] = [h for h in pre_tool_use if not _is_elydora_hook(h)]
|
||||||
if guard_script_path:
|
if guard_script_path:
|
||||||
pre_tool_use.append({
|
pre_tool_use.append({
|
||||||
"command": f"python3 {guard_script_path}",
|
"command": f'"{python_exe}" {guard_script_path}',
|
||||||
})
|
})
|
||||||
|
|
||||||
# --- postToolUse (audit logging, camelCase) ---
|
# --- postToolUse (audit logging, camelCase) ---
|
||||||
@@ -62,7 +96,7 @@ class CursorPlugin(AgentPlugin):
|
|||||||
print(f" Hook script: {hook_path}")
|
print(f" Hook script: {hook_path}")
|
||||||
print(f" Settings: {SETTINGS_PATH}")
|
print(f" Settings: {SETTINGS_PATH}")
|
||||||
|
|
||||||
def uninstall(self) -> None:
|
def uninstall(self, agent_id: str = "") -> None:
|
||||||
if os.path.exists(SETTINGS_PATH):
|
if os.path.exists(SETTINGS_PATH):
|
||||||
settings = _load_json(SETTINGS_PATH)
|
settings = _load_json(SETTINGS_PATH)
|
||||||
hooks = settings.get("hooks", {})
|
hooks = settings.get("hooks", {})
|
||||||
@@ -70,7 +104,7 @@ class CursorPlugin(AgentPlugin):
|
|||||||
|
|
||||||
# Remove preToolUse entries
|
# Remove preToolUse entries
|
||||||
pre_tool_use = hooks.get("preToolUse", [])
|
pre_tool_use = hooks.get("preToolUse", [])
|
||||||
pre_filtered = [h for h in pre_tool_use if not _is_elydora_hook(h)]
|
pre_filtered = [h for h in pre_tool_use if not _is_elydora_hook(h, agent_id)]
|
||||||
if len(pre_filtered) != len(pre_tool_use):
|
if len(pre_filtered) != len(pre_tool_use):
|
||||||
hooks["preToolUse"] = pre_filtered
|
hooks["preToolUse"] = pre_filtered
|
||||||
if not pre_filtered:
|
if not pre_filtered:
|
||||||
@@ -79,7 +113,7 @@ class CursorPlugin(AgentPlugin):
|
|||||||
|
|
||||||
# Remove postToolUse entries
|
# Remove postToolUse entries
|
||||||
post_tool_use = hooks.get("postToolUse", [])
|
post_tool_use = hooks.get("postToolUse", [])
|
||||||
post_filtered = [h for h in post_tool_use if not _is_elydora_hook(h)]
|
post_filtered = [h for h in post_tool_use if not _is_elydora_hook(h, agent_id)]
|
||||||
if len(post_filtered) != len(post_tool_use):
|
if len(post_filtered) != len(post_tool_use):
|
||||||
hooks["postToolUse"] = post_filtered
|
hooks["postToolUse"] = post_filtered
|
||||||
if not post_filtered:
|
if not post_filtered:
|
||||||
@@ -91,15 +125,15 @@ class CursorPlugin(AgentPlugin):
|
|||||||
del settings["hooks"]
|
del settings["hooks"]
|
||||||
_save_json(SETTINGS_PATH, settings)
|
_save_json(SETTINGS_PATH, settings)
|
||||||
|
|
||||||
hook_path = self._hook_path()
|
# Hook script removal is handled by cli.py cmd_uninstall (rmtree of agent dir)
|
||||||
if os.path.exists(hook_path):
|
|
||||||
os.remove(hook_path)
|
|
||||||
|
|
||||||
print("Elydora hook uninstalled from Cursor.")
|
print("Elydora hook uninstalled from Cursor.")
|
||||||
|
|
||||||
def status(self) -> PluginStatus:
|
def status(self) -> PluginStatus:
|
||||||
hook_path = self._hook_path()
|
# Scan ~/.elydora/*/hook.py for any installed hook
|
||||||
hook_exists = os.path.exists(hook_path)
|
import glob as _glob
|
||||||
|
hook_pattern = os.path.join(ELYDORA_DIR, "*", "hook.py")
|
||||||
|
hook_files = _glob.glob(hook_pattern)
|
||||||
|
hook_exists = len(hook_files) > 0
|
||||||
|
|
||||||
settings_configured = False
|
settings_configured = False
|
||||||
if os.path.exists(SETTINGS_PATH):
|
if os.path.exists(SETTINGS_PATH):
|
||||||
@@ -113,7 +147,7 @@ class CursorPlugin(AgentPlugin):
|
|||||||
|
|
||||||
installed = hook_exists and settings_configured
|
installed = hook_exists and settings_configured
|
||||||
if installed:
|
if installed:
|
||||||
details = f"Hook: {hook_path}"
|
details = f"Found {len(hook_files)} agent(s): {', '.join(hook_files)}"
|
||||||
elif hook_exists:
|
elif hook_exists:
|
||||||
details = "Hook script exists but not configured in hooks.json"
|
details = "Hook script exists but not configured in hooks.json"
|
||||||
elif settings_configured:
|
elif settings_configured:
|
||||||
@@ -124,9 +158,14 @@ class CursorPlugin(AgentPlugin):
|
|||||||
return PluginStatus(installed=installed, agent="cursor", details=details)
|
return PluginStatus(installed=installed, agent="cursor", details=details)
|
||||||
|
|
||||||
|
|
||||||
def _is_elydora_hook(hook: dict) -> bool:
|
def _is_elydora_hook(hook: dict, agent_id: str = "") -> bool:
|
||||||
cmd = hook.get("command", "")
|
cmd = hook.get("command", "")
|
||||||
return "elydora" in cmd.lower()
|
if "elydora" not in cmd.lower():
|
||||||
|
return False
|
||||||
|
# If agent_id is specified, only match hooks for that specific agent
|
||||||
|
if agent_id:
|
||||||
|
return agent_id in cmd
|
||||||
|
return True
|
||||||
|
|
||||||
|
|
||||||
def _load_json(path: str) -> dict:
|
def _load_json(path: str) -> dict:
|
||||||
|
|||||||
+71
-24
@@ -5,37 +5,71 @@ from __future__ import annotations
|
|||||||
import json
|
import json
|
||||||
import os
|
import os
|
||||||
import stat
|
import stat
|
||||||
|
import sys
|
||||||
|
|
||||||
from .base import AgentPlugin, InstallConfig, PluginStatus
|
from .base import AgentPlugin, InstallConfig, PluginStatus
|
||||||
from .hook_template import generate_hook_script
|
from .hook_template import generate_hook_script
|
||||||
|
|
||||||
|
|
||||||
SETTINGS_PATH = os.path.join(os.path.expanduser("~"), ".gemini", "settings.json")
|
SETTINGS_PATH = os.path.join(os.path.expanduser("~"), ".gemini", "settings.json")
|
||||||
HOOK_DIR = os.path.join(os.path.expanduser("~"), ".elydora", "hooks")
|
ELYDORA_DIR = os.path.join(os.path.expanduser("~"), ".elydora")
|
||||||
HOOK_FILENAME = "elydora-audit-hook-gemini.py"
|
|
||||||
|
|
||||||
|
|
||||||
class GeminiPlugin(AgentPlugin):
|
class GeminiPlugin(AgentPlugin):
|
||||||
"""Install/uninstall Elydora audit hook for Gemini CLI."""
|
"""Install/uninstall Elydora audit hook for Gemini CLI."""
|
||||||
|
|
||||||
def _hook_path(self) -> str:
|
@staticmethod
|
||||||
return os.path.join(HOOK_DIR, HOOK_FILENAME)
|
def _hook_path_for(agent_id: str) -> str:
|
||||||
|
return os.path.join(ELYDORA_DIR, agent_id, "hook.py")
|
||||||
|
|
||||||
def install(self, config: InstallConfig) -> None:
|
def install(self, config: InstallConfig) -> None:
|
||||||
|
agent_id = config.get("agent_id", "")
|
||||||
|
agent_name = config.get("agent_name", "")
|
||||||
|
|
||||||
|
# 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": config.get("org_id", ""),
|
||||||
|
"agent_id": agent_id,
|
||||||
|
"kid": config.get("kid", ""),
|
||||||
|
"base_url": config.get("base_url", "https://api.elydora.com"),
|
||||||
|
"token": config.get("token", ""),
|
||||||
|
"agent_name": agent_name,
|
||||||
|
}
|
||||||
|
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(config.get("private_key", ""))
|
||||||
|
try:
|
||||||
|
os.chmod(private_key_path, stat.S_IRUSR | stat.S_IWUSR)
|
||||||
|
except Exception:
|
||||||
|
pass # chmod may fail on Windows
|
||||||
|
|
||||||
script = generate_hook_script(
|
script = generate_hook_script(
|
||||||
org_id=config.get("org_id", ""),
|
org_id=config.get("org_id", ""),
|
||||||
agent_id=config.get("agent_id", ""),
|
agent_id=agent_id,
|
||||||
private_key=config.get("private_key", ""),
|
private_key=config.get("private_key", ""),
|
||||||
kid=config.get("kid", ""),
|
kid=config.get("kid", ""),
|
||||||
base_url=config.get("base_url", "https://api.elydora.com"),
|
base_url=config.get("base_url", "https://api.elydora.com"),
|
||||||
)
|
)
|
||||||
os.makedirs(HOOK_DIR, exist_ok=True)
|
hook_path = self._hook_path_for(agent_id)
|
||||||
hook_path = self._hook_path()
|
|
||||||
with open(hook_path, "w", encoding="utf-8") as f:
|
with open(hook_path, "w", encoding="utf-8") as f:
|
||||||
f.write(script)
|
f.write(script)
|
||||||
os.chmod(hook_path, stat.S_IRWXU)
|
try:
|
||||||
|
os.chmod(hook_path, stat.S_IRWXU)
|
||||||
|
except Exception:
|
||||||
|
pass # chmod may fail on Windows
|
||||||
|
|
||||||
guard_script_path = config.get("guard_script_path", "")
|
guard_script_path = config.get("guard_script_path", "")
|
||||||
|
python_exe = sys.executable
|
||||||
|
|
||||||
settings = _load_json(SETTINGS_PATH)
|
settings = _load_json(SETTINGS_PATH)
|
||||||
hooks = settings.setdefault("hooks", {})
|
hooks = settings.setdefault("hooks", {})
|
||||||
@@ -48,7 +82,7 @@ class GeminiPlugin(AgentPlugin):
|
|||||||
"hooks": [
|
"hooks": [
|
||||||
{
|
{
|
||||||
"type": "command",
|
"type": "command",
|
||||||
"command": f"python3 {guard_script_path}",
|
"command": f'"{python_exe}" {guard_script_path}',
|
||||||
}
|
}
|
||||||
],
|
],
|
||||||
})
|
})
|
||||||
@@ -72,7 +106,7 @@ class GeminiPlugin(AgentPlugin):
|
|||||||
print(f" Hook script: {hook_path}")
|
print(f" Hook script: {hook_path}")
|
||||||
print(f" Settings: {SETTINGS_PATH}")
|
print(f" Settings: {SETTINGS_PATH}")
|
||||||
|
|
||||||
def uninstall(self) -> None:
|
def uninstall(self, agent_id: str = "") -> None:
|
||||||
if os.path.exists(SETTINGS_PATH):
|
if os.path.exists(SETTINGS_PATH):
|
||||||
settings = _load_json(SETTINGS_PATH)
|
settings = _load_json(SETTINGS_PATH)
|
||||||
hooks = settings.get("hooks", {})
|
hooks = settings.get("hooks", {})
|
||||||
@@ -80,7 +114,7 @@ class GeminiPlugin(AgentPlugin):
|
|||||||
|
|
||||||
# Remove BeforeTool entries
|
# Remove BeforeTool entries
|
||||||
before_tool = hooks.get("BeforeTool", [])
|
before_tool = hooks.get("BeforeTool", [])
|
||||||
before_filtered = [h for h in before_tool if not _is_elydora_hook(h)]
|
before_filtered = [h for h in before_tool if not _is_elydora_hook(h, agent_id)]
|
||||||
if len(before_filtered) != len(before_tool):
|
if len(before_filtered) != len(before_tool):
|
||||||
hooks["BeforeTool"] = before_filtered
|
hooks["BeforeTool"] = before_filtered
|
||||||
if not before_filtered:
|
if not before_filtered:
|
||||||
@@ -89,7 +123,7 @@ class GeminiPlugin(AgentPlugin):
|
|||||||
|
|
||||||
# Remove AfterTool entries
|
# Remove AfterTool entries
|
||||||
after_tool = hooks.get("AfterTool", [])
|
after_tool = hooks.get("AfterTool", [])
|
||||||
after_filtered = [h for h in after_tool if not _is_elydora_hook(h)]
|
after_filtered = [h for h in after_tool if not _is_elydora_hook(h, agent_id)]
|
||||||
if len(after_filtered) != len(after_tool):
|
if len(after_filtered) != len(after_tool):
|
||||||
hooks["AfterTool"] = after_filtered
|
hooks["AfterTool"] = after_filtered
|
||||||
if not after_filtered:
|
if not after_filtered:
|
||||||
@@ -101,15 +135,15 @@ class GeminiPlugin(AgentPlugin):
|
|||||||
del settings["hooks"]
|
del settings["hooks"]
|
||||||
_save_json(SETTINGS_PATH, settings)
|
_save_json(SETTINGS_PATH, settings)
|
||||||
|
|
||||||
hook_path = self._hook_path()
|
# Hook script removal is handled by cli.py cmd_uninstall (rmtree of agent dir)
|
||||||
if os.path.exists(hook_path):
|
|
||||||
os.remove(hook_path)
|
|
||||||
|
|
||||||
print("Elydora hook uninstalled from Gemini CLI.")
|
print("Elydora hook uninstalled from Gemini CLI.")
|
||||||
|
|
||||||
def status(self) -> PluginStatus:
|
def status(self) -> PluginStatus:
|
||||||
hook_path = self._hook_path()
|
# Scan ~/.elydora/*/hook.py for any installed hook
|
||||||
hook_exists = os.path.exists(hook_path)
|
import glob as _glob
|
||||||
|
hook_pattern = os.path.join(ELYDORA_DIR, "*", "hook.py")
|
||||||
|
hook_files = _glob.glob(hook_pattern)
|
||||||
|
hook_exists = len(hook_files) > 0
|
||||||
|
|
||||||
settings_configured = False
|
settings_configured = False
|
||||||
if os.path.exists(SETTINGS_PATH):
|
if os.path.exists(SETTINGS_PATH):
|
||||||
@@ -123,7 +157,7 @@ class GeminiPlugin(AgentPlugin):
|
|||||||
|
|
||||||
installed = hook_exists and settings_configured
|
installed = hook_exists and settings_configured
|
||||||
if installed:
|
if installed:
|
||||||
details = f"Hook: {hook_path}"
|
details = f"Found {len(hook_files)} agent(s): {', '.join(hook_files)}"
|
||||||
elif hook_exists:
|
elif hook_exists:
|
||||||
details = "Hook script exists but not configured in settings"
|
details = "Hook script exists but not configured in settings"
|
||||||
elif settings_configured:
|
elif settings_configured:
|
||||||
@@ -134,16 +168,29 @@ class GeminiPlugin(AgentPlugin):
|
|||||||
return PluginStatus(installed=installed, agent="gemini", details=details)
|
return PluginStatus(installed=installed, agent="gemini", details=details)
|
||||||
|
|
||||||
|
|
||||||
def _is_elydora_hook(entry: dict) -> bool:
|
def _is_elydora_hook(entry: dict, agent_id: str = "") -> bool:
|
||||||
|
# Collect all command strings from the entry
|
||||||
|
commands = []
|
||||||
inner_hooks = entry.get("hooks")
|
inner_hooks = entry.get("hooks")
|
||||||
if isinstance(inner_hooks, list):
|
if isinstance(inner_hooks, list):
|
||||||
return any(
|
commands.extend(
|
||||||
"elydora" in h.get("command", "").lower()
|
h.get("command", "")
|
||||||
for h in inner_hooks
|
for h in inner_hooks
|
||||||
if isinstance(h, dict)
|
if isinstance(h, dict)
|
||||||
)
|
)
|
||||||
cmd = entry.get("command", "")
|
else:
|
||||||
return "elydora" in cmd.lower()
|
commands.append(entry.get("command", ""))
|
||||||
|
|
||||||
|
for cmd in commands:
|
||||||
|
cmd_lower = cmd.lower()
|
||||||
|
if "elydora" not in cmd_lower:
|
||||||
|
continue
|
||||||
|
# If agent_id is specified, only match hooks for that specific agent
|
||||||
|
if agent_id and agent_id in cmd:
|
||||||
|
return True
|
||||||
|
if not agent_id:
|
||||||
|
return True
|
||||||
|
return False
|
||||||
|
|
||||||
|
|
||||||
def _load_json(path: str) -> dict:
|
def _load_json(path: str) -> dict:
|
||||||
|
|||||||
@@ -27,7 +27,7 @@ def generate_hook_script(
|
|||||||
4. POSTs it fire-and-forget to the Elydora API
|
4. POSTs it fire-and-forget to the Elydora API
|
||||||
5. Always exits 0 (never blocks the host agent)
|
5. Always exits 0 (never blocks the host agent)
|
||||||
|
|
||||||
Chain state is persisted in ~/.elydora/chain-state.json.
|
Chain state is persisted in ~/.elydora/{agent_id}/chain-state.json.
|
||||||
"""
|
"""
|
||||||
shebang = sys.executable
|
shebang = sys.executable
|
||||||
return f'''#!{shebang}
|
return f'''#!{shebang}
|
||||||
@@ -51,8 +51,9 @@ PRIVATE_KEY = {private_key!r}
|
|||||||
KID = {kid!r}
|
KID = {kid!r}
|
||||||
BASE_URL = {base_url!r}
|
BASE_URL = {base_url!r}
|
||||||
|
|
||||||
CHAIN_STATE_PATH = os.path.join(os.path.expanduser("~"), ".elydora", "chain-state.json")
|
ELYDORA_DIR = os.path.join(os.path.expanduser("~"), ".elydora")
|
||||||
ERROR_LOG_PATH = os.path.join(os.path.expanduser("~"), ".elydora", "error.log")
|
CHAIN_STATE_PATH = os.path.join(ELYDORA_DIR, AGENT_ID, "chain-state.json")
|
||||||
|
ERROR_LOG_PATH = os.path.join(ELYDORA_DIR, AGENT_ID, "error.log")
|
||||||
ZERO_CHAIN_HASH = "AAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAA"
|
ZERO_CHAIN_HASH = "AAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAA"
|
||||||
|
|
||||||
# ---------------------------------------------------------------------------
|
# ---------------------------------------------------------------------------
|
||||||
@@ -289,18 +290,19 @@ if __name__ == "__main__":
|
|||||||
'''
|
'''
|
||||||
|
|
||||||
|
|
||||||
def generate_guard_script(agent_name: str) -> str:
|
def generate_guard_script(agent_name: str, agent_id: str) -> str:
|
||||||
"""Return a self-contained Python guard script for PreToolUse freeze enforcement.
|
"""Return a self-contained Python guard script for PreToolUse freeze enforcement.
|
||||||
|
|
||||||
The script:
|
The script:
|
||||||
1. Reads agent config from ~/.elydora/agents/{agent_name}.json
|
1. Reads agent config from ~/.elydora/{agent_id}/config.json
|
||||||
2. Checks cached status from ~/.elydora/agents/{agent_name}.status.json (60s TTL)
|
2. Checks cached status from ~/.elydora/{agent_id}/status-cache.json (60s TTL)
|
||||||
3. If cache is stale, fetches agent status from GET {base_url}/v1/agents/{agent_id}
|
3. If cache is stale, fetches agent status from GET {base_url}/v1/agents/{agent_id}
|
||||||
4. If status is "frozen", writes error to stderr and exits with code 1 (blocks tool)
|
4. If status is "frozen", writes error to stderr and exits with code 1 (blocks tool)
|
||||||
5. Fail-open: if API unreachable or config missing, exits 0 (allow)
|
5. Fail-open: if API unreachable or config missing, exits 0 (allow)
|
||||||
6. Uses only stdlib — no external deps
|
6. Uses only stdlib — no external deps
|
||||||
"""
|
"""
|
||||||
return f'''#!/usr/bin/env python3
|
shebang = sys.executable
|
||||||
|
return f'''#!{shebang}
|
||||||
"""Elydora agent guard (PreToolUse) for: {agent_name}
|
"""Elydora agent guard (PreToolUse) for: {agent_name}
|
||||||
Generated by elydora-sdk — DO NOT EDIT."""
|
Generated by elydora-sdk — DO NOT EDIT."""
|
||||||
|
|
||||||
@@ -310,9 +312,10 @@ import sys
|
|||||||
import time
|
import time
|
||||||
|
|
||||||
AGENT_NAME = {agent_name!r}
|
AGENT_NAME = {agent_name!r}
|
||||||
|
AGENT_ID = {agent_id!r}
|
||||||
ELYDORA_DIR = os.path.join(os.path.expanduser("~"), ".elydora")
|
ELYDORA_DIR = os.path.join(os.path.expanduser("~"), ".elydora")
|
||||||
CONFIG_PATH = os.path.join(ELYDORA_DIR, "agents", AGENT_NAME + ".json")
|
CONFIG_PATH = os.path.join(ELYDORA_DIR, AGENT_ID, "config.json")
|
||||||
STATUS_CACHE_PATH = os.path.join(ELYDORA_DIR, "agents", AGENT_NAME + ".status.json")
|
STATUS_CACHE_PATH = os.path.join(ELYDORA_DIR, AGENT_ID, "status-cache.json")
|
||||||
CACHE_TTL_S = 60 # 60 seconds
|
CACHE_TTL_S = 60 # 60 seconds
|
||||||
|
|
||||||
# Consume stdin so the parent process doesn't block on a full pipe
|
# Consume stdin so the parent process doesn't block on a full pipe
|
||||||
|
|||||||
@@ -1,113 +0,0 @@
|
|||||||
"""Kiro plugin — writes .kiro/hooks/elydora-audit.kiro.hook file."""
|
|
||||||
|
|
||||||
from __future__ import annotations
|
|
||||||
|
|
||||||
import os
|
|
||||||
import stat
|
|
||||||
|
|
||||||
from .base import AgentPlugin, InstallConfig, PluginStatus
|
|
||||||
from .hook_template import generate_hook_script
|
|
||||||
|
|
||||||
|
|
||||||
HOOK_DIR = os.path.join(os.path.expanduser("~"), ".kiro", "hooks")
|
|
||||||
HOOK_FILENAME = "elydora-audit.kiro.hook"
|
|
||||||
SCRIPT_DIR = os.path.join(os.path.expanduser("~"), ".elydora", "hooks")
|
|
||||||
SCRIPT_FILENAME = "elydora-audit-hook-kiro.py"
|
|
||||||
|
|
||||||
|
|
||||||
class KiroPlugin(AgentPlugin):
|
|
||||||
"""Install/uninstall Elydora audit hook for Kiro."""
|
|
||||||
|
|
||||||
def _hook_path(self) -> str:
|
|
||||||
return os.path.join(HOOK_DIR, HOOK_FILENAME)
|
|
||||||
|
|
||||||
def _script_path(self) -> str:
|
|
||||||
return os.path.join(SCRIPT_DIR, SCRIPT_FILENAME)
|
|
||||||
|
|
||||||
def install(self, config: InstallConfig) -> None:
|
|
||||||
# Write the Python hook script
|
|
||||||
script = generate_hook_script(
|
|
||||||
org_id=config.get("org_id", ""),
|
|
||||||
agent_id=config.get("agent_id", ""),
|
|
||||||
private_key=config.get("private_key", ""),
|
|
||||||
kid=config.get("kid", ""),
|
|
||||||
base_url=config.get("base_url", "https://api.elydora.com"),
|
|
||||||
)
|
|
||||||
os.makedirs(SCRIPT_DIR, exist_ok=True)
|
|
||||||
script_path = self._script_path()
|
|
||||||
with open(script_path, "w", encoding="utf-8") as f:
|
|
||||||
f.write(script)
|
|
||||||
os.chmod(script_path, stat.S_IRWXU)
|
|
||||||
|
|
||||||
guard_script_path = config.get("guard_script_path", "")
|
|
||||||
|
|
||||||
# Write the Kiro hook definition file (JSON format matching Node SDK)
|
|
||||||
import json
|
|
||||||
|
|
||||||
hook_config = {
|
|
||||||
"name": "Elydora Audit",
|
|
||||||
"description": "Sends tool-use events to the Elydora tamper-evident audit platform",
|
|
||||||
"version": "1.0.0",
|
|
||||||
"hooks": {
|
|
||||||
"pre_tool_use": {
|
|
||||||
"command": f"python3 {guard_script_path}" if guard_script_path else "",
|
|
||||||
"timeout_ms": 5000,
|
|
||||||
},
|
|
||||||
"post_tool_use": {
|
|
||||||
"command": script_path,
|
|
||||||
"timeout_ms": 5000,
|
|
||||||
},
|
|
||||||
},
|
|
||||||
}
|
|
||||||
|
|
||||||
os.makedirs(HOOK_DIR, exist_ok=True)
|
|
||||||
hook_path = self._hook_path()
|
|
||||||
with open(hook_path, "w", encoding="utf-8") as f:
|
|
||||||
json.dump(hook_config, f, indent=2)
|
|
||||||
f.write("\n")
|
|
||||||
|
|
||||||
print(f"Elydora hook installed for Kiro.")
|
|
||||||
print(f" Hook script: {script_path}")
|
|
||||||
print(f" Hook definition: {hook_path}")
|
|
||||||
|
|
||||||
def uninstall(self) -> None:
|
|
||||||
hook_path = self._hook_path()
|
|
||||||
if os.path.exists(hook_path):
|
|
||||||
os.remove(hook_path)
|
|
||||||
|
|
||||||
script_path = self._script_path()
|
|
||||||
if os.path.exists(script_path):
|
|
||||||
os.remove(script_path)
|
|
||||||
|
|
||||||
print("Elydora hook uninstalled from Kiro.")
|
|
||||||
|
|
||||||
def status(self) -> PluginStatus:
|
|
||||||
hook_path = self._hook_path()
|
|
||||||
script_path = self._script_path()
|
|
||||||
hook_exists = os.path.exists(hook_path)
|
|
||||||
script_exists = os.path.exists(script_path)
|
|
||||||
|
|
||||||
hook_configured = False
|
|
||||||
if hook_exists:
|
|
||||||
try:
|
|
||||||
import json
|
|
||||||
with open(hook_path, "r", encoding="utf-8") as f:
|
|
||||||
cfg = json.load(f)
|
|
||||||
hooks = cfg.get("hooks", {})
|
|
||||||
hook_configured = bool(
|
|
||||||
hooks.get("pre_tool_use") and hooks.get("post_tool_use")
|
|
||||||
)
|
|
||||||
except Exception:
|
|
||||||
hook_configured = hook_exists # Fallback
|
|
||||||
|
|
||||||
installed = hook_exists and script_exists and hook_configured
|
|
||||||
if installed:
|
|
||||||
details = f"Hook: {hook_path}"
|
|
||||||
elif hook_exists:
|
|
||||||
details = "Hook definition exists but script missing"
|
|
||||||
elif script_exists:
|
|
||||||
details = "Script exists but hook definition missing"
|
|
||||||
else:
|
|
||||||
details = "Not installed"
|
|
||||||
|
|
||||||
return PluginStatus(installed=installed, agent="kiro", details=details)
|
|
||||||
@@ -0,0 +1,211 @@
|
|||||||
|
"""Kiro CLI plugin — merges PreToolUse/PostToolUse hooks into ~/.kiro/settings.json."""
|
||||||
|
|
||||||
|
from __future__ import annotations
|
||||||
|
|
||||||
|
import json
|
||||||
|
import os
|
||||||
|
import stat
|
||||||
|
import sys
|
||||||
|
|
||||||
|
from .base import AgentPlugin, InstallConfig, PluginStatus
|
||||||
|
from .hook_template import generate_hook_script
|
||||||
|
|
||||||
|
|
||||||
|
SETTINGS_PATH = os.path.join(os.path.expanduser("~"), ".kiro", "settings.json")
|
||||||
|
ELYDORA_DIR = os.path.join(os.path.expanduser("~"), ".elydora")
|
||||||
|
|
||||||
|
|
||||||
|
class KiroCliPlugin(AgentPlugin):
|
||||||
|
"""Install/uninstall Elydora audit hook for Kiro CLI."""
|
||||||
|
|
||||||
|
@staticmethod
|
||||||
|
def _hook_path_for(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", "")
|
||||||
|
|
||||||
|
# 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": config.get("org_id", ""),
|
||||||
|
"agent_id": agent_id,
|
||||||
|
"kid": config.get("kid", ""),
|
||||||
|
"base_url": config.get("base_url", "https://api.elydora.com"),
|
||||||
|
"token": config.get("token", ""),
|
||||||
|
"agent_name": agent_name,
|
||||||
|
}
|
||||||
|
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(config.get("private_key", ""))
|
||||||
|
try:
|
||||||
|
os.chmod(private_key_path, stat.S_IRUSR | stat.S_IWUSR)
|
||||||
|
except Exception:
|
||||||
|
pass # chmod may fail on Windows
|
||||||
|
|
||||||
|
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"),
|
||||||
|
)
|
||||||
|
hook_path = self._hook_path_for(agent_id)
|
||||||
|
with open(hook_path, "w", encoding="utf-8") as f:
|
||||||
|
f.write(script)
|
||||||
|
try:
|
||||||
|
os.chmod(hook_path, stat.S_IRWXU)
|
||||||
|
except Exception:
|
||||||
|
pass # chmod may fail on Windows
|
||||||
|
|
||||||
|
guard_script_path = config.get("guard_script_path", "")
|
||||||
|
python_exe = sys.executable
|
||||||
|
|
||||||
|
settings = _load_json(SETTINGS_PATH)
|
||||||
|
hooks = settings.setdefault("hooks", {})
|
||||||
|
|
||||||
|
# --- PreToolUse (guard — freeze enforcement, PascalCase with matcher) ---
|
||||||
|
pre_tool_use = hooks.setdefault("PreToolUse", [])
|
||||||
|
pre_tool_use[:] = [h for h in pre_tool_use if not _is_elydora_hook(h)]
|
||||||
|
if guard_script_path:
|
||||||
|
pre_tool_use.append({
|
||||||
|
"matcher": "*",
|
||||||
|
"hooks": [
|
||||||
|
{
|
||||||
|
"type": "command",
|
||||||
|
"command": f'"{python_exe}" {guard_script_path}',
|
||||||
|
"timeout_ms": 5000,
|
||||||
|
}
|
||||||
|
],
|
||||||
|
})
|
||||||
|
|
||||||
|
# --- PostToolUse (audit logging, PascalCase with matcher) ---
|
||||||
|
post_tool_use = hooks.setdefault("PostToolUse", [])
|
||||||
|
|
||||||
|
post_tool_use[:] = [h for h in post_tool_use if not _is_elydora_hook(h)]
|
||||||
|
|
||||||
|
post_tool_use.append({
|
||||||
|
"matcher": "*",
|
||||||
|
"hooks": [
|
||||||
|
{
|
||||||
|
"type": "command",
|
||||||
|
"command": hook_path,
|
||||||
|
"timeout_ms": 5000,
|
||||||
|
}
|
||||||
|
],
|
||||||
|
})
|
||||||
|
|
||||||
|
_save_json(SETTINGS_PATH, settings)
|
||||||
|
print(f"Elydora hook installed for Kiro CLI.")
|
||||||
|
print(f" Hook script: {hook_path}")
|
||||||
|
print(f" Settings: {SETTINGS_PATH}")
|
||||||
|
|
||||||
|
def uninstall(self, agent_id: str = "") -> None:
|
||||||
|
if os.path.exists(SETTINGS_PATH):
|
||||||
|
settings = _load_json(SETTINGS_PATH)
|
||||||
|
hooks = settings.get("hooks", {})
|
||||||
|
changed = False
|
||||||
|
|
||||||
|
# Remove PreToolUse entries
|
||||||
|
pre_tool_use = hooks.get("PreToolUse", [])
|
||||||
|
pre_filtered = [h for h in pre_tool_use if not _is_elydora_hook(h, agent_id)]
|
||||||
|
if len(pre_filtered) != len(pre_tool_use):
|
||||||
|
hooks["PreToolUse"] = pre_filtered
|
||||||
|
if not pre_filtered:
|
||||||
|
del hooks["PreToolUse"]
|
||||||
|
changed = True
|
||||||
|
|
||||||
|
# Remove PostToolUse entries
|
||||||
|
post_tool_use = hooks.get("PostToolUse", [])
|
||||||
|
post_filtered = [h for h in post_tool_use if not _is_elydora_hook(h, agent_id)]
|
||||||
|
if len(post_filtered) != len(post_tool_use):
|
||||||
|
hooks["PostToolUse"] = post_filtered
|
||||||
|
if not post_filtered:
|
||||||
|
del hooks["PostToolUse"]
|
||||||
|
changed = True
|
||||||
|
|
||||||
|
if changed:
|
||||||
|
if not hooks:
|
||||||
|
del settings["hooks"]
|
||||||
|
_save_json(SETTINGS_PATH, settings)
|
||||||
|
|
||||||
|
# Hook script removal is handled by cli.py cmd_uninstall (rmtree of agent dir)
|
||||||
|
print("Elydora hook uninstalled from Kiro CLI.")
|
||||||
|
|
||||||
|
def status(self) -> PluginStatus:
|
||||||
|
# Scan ~/.elydora/*/hook.py for any installed hook
|
||||||
|
import glob as _glob
|
||||||
|
hook_pattern = os.path.join(ELYDORA_DIR, "*", "hook.py")
|
||||||
|
hook_files = _glob.glob(hook_pattern)
|
||||||
|
hook_exists = len(hook_files) > 0
|
||||||
|
|
||||||
|
settings_configured = False
|
||||||
|
if os.path.exists(SETTINGS_PATH):
|
||||||
|
settings = _load_json(SETTINGS_PATH)
|
||||||
|
hooks = settings.get("hooks", {})
|
||||||
|
pre_tool_use = hooks.get("PreToolUse", [])
|
||||||
|
post_tool_use = hooks.get("PostToolUse", [])
|
||||||
|
pre_configured = any(_is_elydora_hook(h) for h in pre_tool_use)
|
||||||
|
post_configured = any(_is_elydora_hook(h) for h in post_tool_use)
|
||||||
|
settings_configured = pre_configured and post_configured
|
||||||
|
|
||||||
|
installed = hook_exists and settings_configured
|
||||||
|
if installed:
|
||||||
|
details = f"Found {len(hook_files)} agent(s): {', '.join(hook_files)}"
|
||||||
|
elif hook_exists:
|
||||||
|
details = "Hook script exists but not configured in settings"
|
||||||
|
elif settings_configured:
|
||||||
|
details = "Configured in settings but hook script missing"
|
||||||
|
else:
|
||||||
|
details = "Not installed"
|
||||||
|
|
||||||
|
return PluginStatus(installed=installed, agent="kirocli", details=details)
|
||||||
|
|
||||||
|
|
||||||
|
def _is_elydora_hook(entry: dict, agent_id: str = "") -> bool:
|
||||||
|
# Collect all command strings from the entry
|
||||||
|
commands = []
|
||||||
|
inner_hooks = entry.get("hooks")
|
||||||
|
if isinstance(inner_hooks, list):
|
||||||
|
commands.extend(
|
||||||
|
h.get("command", "")
|
||||||
|
for h in inner_hooks
|
||||||
|
if isinstance(h, dict)
|
||||||
|
)
|
||||||
|
else:
|
||||||
|
commands.append(entry.get("command", ""))
|
||||||
|
|
||||||
|
for cmd in commands:
|
||||||
|
cmd_lower = cmd.lower()
|
||||||
|
if "elydora" not in cmd_lower:
|
||||||
|
continue
|
||||||
|
# If agent_id is specified, only match hooks for that specific agent
|
||||||
|
if agent_id and agent_id in cmd:
|
||||||
|
return True
|
||||||
|
if not agent_id:
|
||||||
|
return True
|
||||||
|
return False
|
||||||
|
|
||||||
|
|
||||||
|
def _load_json(path: str) -> dict:
|
||||||
|
if not os.path.exists(path):
|
||||||
|
return {}
|
||||||
|
with open(path, "r", encoding="utf-8") as f:
|
||||||
|
return json.load(f)
|
||||||
|
|
||||||
|
|
||||||
|
def _save_json(path: str, data: dict) -> None:
|
||||||
|
os.makedirs(os.path.dirname(path), exist_ok=True)
|
||||||
|
with open(path, "w", encoding="utf-8") as f:
|
||||||
|
json.dump(data, f, indent=2)
|
||||||
|
f.write("\n")
|
||||||
@@ -0,0 +1,146 @@
|
|||||||
|
"""Kiro IDE plugin — writes .kiro/hooks/elydora-audit.kiro.hook file."""
|
||||||
|
|
||||||
|
from __future__ import annotations
|
||||||
|
|
||||||
|
import json
|
||||||
|
import os
|
||||||
|
import stat
|
||||||
|
import sys
|
||||||
|
|
||||||
|
from .base import AgentPlugin, InstallConfig, PluginStatus
|
||||||
|
from .hook_template import generate_hook_script
|
||||||
|
|
||||||
|
|
||||||
|
KIRO_HOOK_DIR = os.path.join(os.path.expanduser("~"), ".kiro", "hooks")
|
||||||
|
KIRO_HOOK_FILENAME = "elydora-audit.kiro.hook"
|
||||||
|
ELYDORA_DIR = os.path.join(os.path.expanduser("~"), ".elydora")
|
||||||
|
|
||||||
|
|
||||||
|
class KiroIdePlugin(AgentPlugin):
|
||||||
|
"""Install/uninstall Elydora audit hook for Kiro IDE."""
|
||||||
|
|
||||||
|
def _kiro_hook_path(self) -> str:
|
||||||
|
return os.path.join(KIRO_HOOK_DIR, KIRO_HOOK_FILENAME)
|
||||||
|
|
||||||
|
@staticmethod
|
||||||
|
def _hook_path_for(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", "")
|
||||||
|
|
||||||
|
# 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": config.get("org_id", ""),
|
||||||
|
"agent_id": agent_id,
|
||||||
|
"kid": config.get("kid", ""),
|
||||||
|
"base_url": config.get("base_url", "https://api.elydora.com"),
|
||||||
|
"token": config.get("token", ""),
|
||||||
|
"agent_name": agent_name,
|
||||||
|
}
|
||||||
|
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(config.get("private_key", ""))
|
||||||
|
try:
|
||||||
|
os.chmod(private_key_path, stat.S_IRUSR | stat.S_IWUSR)
|
||||||
|
except Exception:
|
||||||
|
pass # chmod may fail on Windows
|
||||||
|
|
||||||
|
# Write the Python hook script
|
||||||
|
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"),
|
||||||
|
)
|
||||||
|
script_path = self._hook_path_for(agent_id)
|
||||||
|
with open(script_path, "w", encoding="utf-8") as f:
|
||||||
|
f.write(script)
|
||||||
|
try:
|
||||||
|
os.chmod(script_path, stat.S_IRWXU)
|
||||||
|
except Exception:
|
||||||
|
pass # chmod may fail on Windows
|
||||||
|
|
||||||
|
guard_script_path = config.get("guard_script_path", "")
|
||||||
|
python_exe = sys.executable
|
||||||
|
|
||||||
|
# Write the Kiro hook definition file (JSON format matching Node SDK)
|
||||||
|
hook_config = {
|
||||||
|
"name": "Elydora Audit",
|
||||||
|
"description": "Sends tool-use events to the Elydora tamper-evident audit platform",
|
||||||
|
"version": "1.0.0",
|
||||||
|
"hooks": {
|
||||||
|
"pre_tool_use": {
|
||||||
|
"command": f'"{python_exe}" {guard_script_path}' if guard_script_path else "",
|
||||||
|
"timeout_ms": 5000,
|
||||||
|
},
|
||||||
|
"post_tool_use": {
|
||||||
|
"command": script_path,
|
||||||
|
"timeout_ms": 5000,
|
||||||
|
},
|
||||||
|
},
|
||||||
|
}
|
||||||
|
|
||||||
|
os.makedirs(KIRO_HOOK_DIR, exist_ok=True)
|
||||||
|
hook_path = self._kiro_hook_path()
|
||||||
|
with open(hook_path, "w", encoding="utf-8") as f:
|
||||||
|
json.dump(hook_config, f, indent=2)
|
||||||
|
f.write("\n")
|
||||||
|
|
||||||
|
print(f"Elydora hook installed for Kiro IDE.")
|
||||||
|
print(f" Hook script: {script_path}")
|
||||||
|
print(f" Hook definition: {hook_path}")
|
||||||
|
|
||||||
|
def uninstall(self, agent_id: str = "") -> None:
|
||||||
|
hook_path = self._kiro_hook_path()
|
||||||
|
if os.path.exists(hook_path):
|
||||||
|
os.remove(hook_path)
|
||||||
|
|
||||||
|
# Hook script removal is handled by cli.py cmd_uninstall (rmtree of agent dir)
|
||||||
|
print("Elydora hook uninstalled from Kiro IDE.")
|
||||||
|
|
||||||
|
def status(self) -> PluginStatus:
|
||||||
|
hook_path = self._kiro_hook_path()
|
||||||
|
hook_exists = os.path.exists(hook_path)
|
||||||
|
|
||||||
|
# Scan ~/.elydora/*/hook.py for any installed hook script
|
||||||
|
import glob as _glob
|
||||||
|
hook_pattern = os.path.join(ELYDORA_DIR, "*", "hook.py")
|
||||||
|
hook_files = _glob.glob(hook_pattern)
|
||||||
|
script_exists = len(hook_files) > 0
|
||||||
|
|
||||||
|
hook_configured = False
|
||||||
|
if hook_exists:
|
||||||
|
try:
|
||||||
|
with open(hook_path, "r", encoding="utf-8") as f:
|
||||||
|
cfg = json.load(f)
|
||||||
|
hooks = cfg.get("hooks", {})
|
||||||
|
hook_configured = bool(
|
||||||
|
hooks.get("pre_tool_use") and hooks.get("post_tool_use")
|
||||||
|
)
|
||||||
|
except Exception:
|
||||||
|
hook_configured = hook_exists # Fallback
|
||||||
|
|
||||||
|
installed = hook_exists and script_exists and hook_configured
|
||||||
|
if installed:
|
||||||
|
details = f"Found {len(hook_files)} agent(s): {', '.join(hook_files)}"
|
||||||
|
elif hook_exists:
|
||||||
|
details = "Hook definition exists but script missing"
|
||||||
|
elif script_exists:
|
||||||
|
details = "Script exists but hook definition missing"
|
||||||
|
else:
|
||||||
|
details = "Not installed"
|
||||||
|
|
||||||
|
return PluginStatus(installed=installed, agent="kiroide", details=details)
|
||||||
@@ -1,64 +1,100 @@
|
|||||||
"""Augment Code plugin — merges PostToolUse hook into ~/.augment/settings.json."""
|
"""Letta Code plugin — merges PreToolUse/PostToolUse hooks into ~/.letta/settings.json."""
|
||||||
|
|
||||||
from __future__ import annotations
|
from __future__ import annotations
|
||||||
|
|
||||||
import json
|
import json
|
||||||
import os
|
import os
|
||||||
import stat
|
import stat
|
||||||
|
import sys
|
||||||
|
|
||||||
from .base import AgentPlugin, InstallConfig, PluginStatus
|
from .base import AgentPlugin, InstallConfig, PluginStatus
|
||||||
from .hook_template import generate_hook_script
|
from .hook_template import generate_hook_script
|
||||||
|
|
||||||
|
|
||||||
SETTINGS_PATH = os.path.join(os.path.expanduser("~"), ".augment", "settings.json")
|
SETTINGS_PATH = os.path.join(os.path.expanduser("~"), ".letta", "settings.json")
|
||||||
HOOK_DIR = os.path.join(os.path.expanduser("~"), ".elydora", "hooks")
|
ELYDORA_DIR = os.path.join(os.path.expanduser("~"), ".elydora")
|
||||||
HOOK_FILENAME = "elydora-audit-hook-augment.py"
|
|
||||||
|
|
||||||
|
|
||||||
class AugmentPlugin(AgentPlugin):
|
class LettaPlugin(AgentPlugin):
|
||||||
"""Install/uninstall Elydora audit hook for Augment Code."""
|
"""Install/uninstall Elydora audit hook for Letta Code."""
|
||||||
|
|
||||||
def _hook_path(self) -> str:
|
@staticmethod
|
||||||
return os.path.join(HOOK_DIR, HOOK_FILENAME)
|
def _hook_path_for(agent_id: str) -> str:
|
||||||
|
return os.path.join(ELYDORA_DIR, agent_id, "hook.py")
|
||||||
|
|
||||||
def install(self, config: InstallConfig) -> None:
|
def install(self, config: InstallConfig) -> None:
|
||||||
|
agent_id = config.get("agent_id", "")
|
||||||
|
agent_name = config.get("agent_name", "")
|
||||||
|
|
||||||
|
# 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": config.get("org_id", ""),
|
||||||
|
"agent_id": agent_id,
|
||||||
|
"kid": config.get("kid", ""),
|
||||||
|
"base_url": config.get("base_url", "https://api.elydora.com"),
|
||||||
|
"token": config.get("token", ""),
|
||||||
|
"agent_name": agent_name,
|
||||||
|
}
|
||||||
|
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(config.get("private_key", ""))
|
||||||
|
try:
|
||||||
|
os.chmod(private_key_path, stat.S_IRUSR | stat.S_IWUSR)
|
||||||
|
except Exception:
|
||||||
|
pass # chmod may fail on Windows
|
||||||
|
|
||||||
script = generate_hook_script(
|
script = generate_hook_script(
|
||||||
org_id=config.get("org_id", ""),
|
org_id=config.get("org_id", ""),
|
||||||
agent_id=config.get("agent_id", ""),
|
agent_id=agent_id,
|
||||||
private_key=config.get("private_key", ""),
|
private_key=config.get("private_key", ""),
|
||||||
kid=config.get("kid", ""),
|
kid=config.get("kid", ""),
|
||||||
base_url=config.get("base_url", "https://api.elydora.com"),
|
base_url=config.get("base_url", "https://api.elydora.com"),
|
||||||
)
|
)
|
||||||
os.makedirs(HOOK_DIR, exist_ok=True)
|
hook_path = self._hook_path_for(agent_id)
|
||||||
hook_path = self._hook_path()
|
|
||||||
with open(hook_path, "w", encoding="utf-8") as f:
|
with open(hook_path, "w", encoding="utf-8") as f:
|
||||||
f.write(script)
|
f.write(script)
|
||||||
os.chmod(hook_path, stat.S_IRWXU)
|
try:
|
||||||
|
os.chmod(hook_path, stat.S_IRWXU)
|
||||||
|
except Exception:
|
||||||
|
pass # chmod may fail on Windows
|
||||||
|
|
||||||
guard_script_path = config.get("guard_script_path", "")
|
guard_script_path = config.get("guard_script_path", "")
|
||||||
|
python_exe = sys.executable
|
||||||
|
|
||||||
settings = _load_json(SETTINGS_PATH)
|
settings = _load_json(SETTINGS_PATH)
|
||||||
hooks = settings.setdefault("hooks", {})
|
hooks = settings.setdefault("hooks", {})
|
||||||
|
|
||||||
# --- PreToolUse (guard — freeze enforcement) ---
|
# --- PreToolUse (guard — freeze enforcement, PascalCase with matcher, no timeout_ms) ---
|
||||||
pre_tool_use = hooks.setdefault("PreToolUse", [])
|
pre_tool_use = hooks.setdefault("PreToolUse", [])
|
||||||
pre_tool_use[:] = [h for h in pre_tool_use if not _is_elydora_hook(h)]
|
pre_tool_use[:] = [h for h in pre_tool_use if not _is_elydora_hook(h)]
|
||||||
if guard_script_path:
|
if guard_script_path:
|
||||||
pre_tool_use.append({
|
pre_tool_use.append({
|
||||||
|
"matcher": "*",
|
||||||
"hooks": [
|
"hooks": [
|
||||||
{
|
{
|
||||||
"type": "command",
|
"type": "command",
|
||||||
"command": f"python3 {guard_script_path}",
|
"command": f'"{python_exe}" {guard_script_path}',
|
||||||
}
|
}
|
||||||
],
|
],
|
||||||
})
|
})
|
||||||
|
|
||||||
# --- PostToolUse (audit logging) ---
|
# --- PostToolUse (audit logging, PascalCase with matcher, no timeout_ms) ---
|
||||||
post_tool_use = hooks.setdefault("PostToolUse", [])
|
post_tool_use = hooks.setdefault("PostToolUse", [])
|
||||||
|
|
||||||
post_tool_use[:] = [h for h in post_tool_use if not _is_elydora_hook(h)]
|
post_tool_use[:] = [h for h in post_tool_use if not _is_elydora_hook(h)]
|
||||||
|
|
||||||
post_tool_use.append({
|
post_tool_use.append({
|
||||||
|
"matcher": "*",
|
||||||
"hooks": [
|
"hooks": [
|
||||||
{
|
{
|
||||||
"type": "command",
|
"type": "command",
|
||||||
@@ -68,11 +104,11 @@ class AugmentPlugin(AgentPlugin):
|
|||||||
})
|
})
|
||||||
|
|
||||||
_save_json(SETTINGS_PATH, settings)
|
_save_json(SETTINGS_PATH, settings)
|
||||||
print(f"Elydora hook installed for Augment Code.")
|
print(f"Elydora hook installed for Letta Code.")
|
||||||
print(f" Hook script: {hook_path}")
|
print(f" Hook script: {hook_path}")
|
||||||
print(f" Settings: {SETTINGS_PATH}")
|
print(f" Settings: {SETTINGS_PATH}")
|
||||||
|
|
||||||
def uninstall(self) -> None:
|
def uninstall(self, agent_id: str = "") -> None:
|
||||||
if os.path.exists(SETTINGS_PATH):
|
if os.path.exists(SETTINGS_PATH):
|
||||||
settings = _load_json(SETTINGS_PATH)
|
settings = _load_json(SETTINGS_PATH)
|
||||||
hooks = settings.get("hooks", {})
|
hooks = settings.get("hooks", {})
|
||||||
@@ -80,7 +116,7 @@ class AugmentPlugin(AgentPlugin):
|
|||||||
|
|
||||||
# Remove PreToolUse entries
|
# Remove PreToolUse entries
|
||||||
pre_tool_use = hooks.get("PreToolUse", [])
|
pre_tool_use = hooks.get("PreToolUse", [])
|
||||||
pre_filtered = [h for h in pre_tool_use if not _is_elydora_hook(h)]
|
pre_filtered = [h for h in pre_tool_use if not _is_elydora_hook(h, agent_id)]
|
||||||
if len(pre_filtered) != len(pre_tool_use):
|
if len(pre_filtered) != len(pre_tool_use):
|
||||||
hooks["PreToolUse"] = pre_filtered
|
hooks["PreToolUse"] = pre_filtered
|
||||||
if not pre_filtered:
|
if not pre_filtered:
|
||||||
@@ -89,7 +125,7 @@ class AugmentPlugin(AgentPlugin):
|
|||||||
|
|
||||||
# Remove PostToolUse entries
|
# Remove PostToolUse entries
|
||||||
post_tool_use = hooks.get("PostToolUse", [])
|
post_tool_use = hooks.get("PostToolUse", [])
|
||||||
post_filtered = [h for h in post_tool_use if not _is_elydora_hook(h)]
|
post_filtered = [h for h in post_tool_use if not _is_elydora_hook(h, agent_id)]
|
||||||
if len(post_filtered) != len(post_tool_use):
|
if len(post_filtered) != len(post_tool_use):
|
||||||
hooks["PostToolUse"] = post_filtered
|
hooks["PostToolUse"] = post_filtered
|
||||||
if not post_filtered:
|
if not post_filtered:
|
||||||
@@ -101,15 +137,15 @@ class AugmentPlugin(AgentPlugin):
|
|||||||
del settings["hooks"]
|
del settings["hooks"]
|
||||||
_save_json(SETTINGS_PATH, settings)
|
_save_json(SETTINGS_PATH, settings)
|
||||||
|
|
||||||
hook_path = self._hook_path()
|
# Hook script removal is handled by cli.py cmd_uninstall (rmtree of agent dir)
|
||||||
if os.path.exists(hook_path):
|
print("Elydora hook uninstalled from Letta Code.")
|
||||||
os.remove(hook_path)
|
|
||||||
|
|
||||||
print("Elydora hook uninstalled from Augment Code.")
|
|
||||||
|
|
||||||
def status(self) -> PluginStatus:
|
def status(self) -> PluginStatus:
|
||||||
hook_path = self._hook_path()
|
# Scan ~/.elydora/*/hook.py for any installed hook
|
||||||
hook_exists = os.path.exists(hook_path)
|
import glob as _glob
|
||||||
|
hook_pattern = os.path.join(ELYDORA_DIR, "*", "hook.py")
|
||||||
|
hook_files = _glob.glob(hook_pattern)
|
||||||
|
hook_exists = len(hook_files) > 0
|
||||||
|
|
||||||
settings_configured = False
|
settings_configured = False
|
||||||
if os.path.exists(SETTINGS_PATH):
|
if os.path.exists(SETTINGS_PATH):
|
||||||
@@ -123,7 +159,7 @@ class AugmentPlugin(AgentPlugin):
|
|||||||
|
|
||||||
installed = hook_exists and settings_configured
|
installed = hook_exists and settings_configured
|
||||||
if installed:
|
if installed:
|
||||||
details = f"Hook: {hook_path}"
|
details = f"Found {len(hook_files)} agent(s): {', '.join(hook_files)}"
|
||||||
elif hook_exists:
|
elif hook_exists:
|
||||||
details = "Hook script exists but not configured in settings"
|
details = "Hook script exists but not configured in settings"
|
||||||
elif settings_configured:
|
elif settings_configured:
|
||||||
@@ -131,19 +167,32 @@ class AugmentPlugin(AgentPlugin):
|
|||||||
else:
|
else:
|
||||||
details = "Not installed"
|
details = "Not installed"
|
||||||
|
|
||||||
return PluginStatus(installed=installed, agent="augment", details=details)
|
return PluginStatus(installed=installed, agent="letta", details=details)
|
||||||
|
|
||||||
|
|
||||||
def _is_elydora_hook(entry: dict) -> bool:
|
def _is_elydora_hook(entry: dict, agent_id: str = "") -> bool:
|
||||||
|
# Collect all command strings from the entry
|
||||||
|
commands = []
|
||||||
inner_hooks = entry.get("hooks")
|
inner_hooks = entry.get("hooks")
|
||||||
if isinstance(inner_hooks, list):
|
if isinstance(inner_hooks, list):
|
||||||
return any(
|
commands.extend(
|
||||||
"elydora" in h.get("command", "").lower()
|
h.get("command", "")
|
||||||
for h in inner_hooks
|
for h in inner_hooks
|
||||||
if isinstance(h, dict)
|
if isinstance(h, dict)
|
||||||
)
|
)
|
||||||
cmd = entry.get("command", "")
|
else:
|
||||||
return "elydora" in cmd.lower()
|
commands.append(entry.get("command", ""))
|
||||||
|
|
||||||
|
for cmd in commands:
|
||||||
|
cmd_lower = cmd.lower()
|
||||||
|
if "elydora" not in cmd_lower:
|
||||||
|
continue
|
||||||
|
# If agent_id is specified, only match hooks for that specific agent
|
||||||
|
if agent_id and agent_id in cmd:
|
||||||
|
return True
|
||||||
|
if not agent_id:
|
||||||
|
return True
|
||||||
|
return False
|
||||||
|
|
||||||
|
|
||||||
def _load_json(path: str) -> dict:
|
def _load_json(path: str) -> dict:
|
||||||
@@ -2,14 +2,17 @@
|
|||||||
|
|
||||||
from __future__ import annotations
|
from __future__ import annotations
|
||||||
|
|
||||||
|
import json
|
||||||
import os
|
import os
|
||||||
import stat
|
import stat
|
||||||
|
import sys
|
||||||
|
|
||||||
from .base import AgentPlugin, InstallConfig, PluginStatus
|
from .base import AgentPlugin, InstallConfig, PluginStatus
|
||||||
|
|
||||||
|
|
||||||
PLUGIN_DIR = os.path.join(os.path.expanduser("~"), ".opencode", "plugins")
|
PLUGIN_DIR = os.path.join(os.path.expanduser("~"), ".opencode", "plugins")
|
||||||
PLUGIN_FILENAME = "elydora-audit.js"
|
PLUGIN_FILENAME = "elydora-audit.js"
|
||||||
|
ELYDORA_DIR = os.path.join(os.path.expanduser("~"), ".elydora")
|
||||||
|
|
||||||
|
|
||||||
class OpenCodePlugin(AgentPlugin):
|
class OpenCodePlugin(AgentPlugin):
|
||||||
@@ -23,13 +26,41 @@ class OpenCodePlugin(AgentPlugin):
|
|||||||
return os.path.join(PLUGIN_DIR, PLUGIN_FILENAME)
|
return os.path.join(PLUGIN_DIR, PLUGIN_FILENAME)
|
||||||
|
|
||||||
def install(self, config: InstallConfig) -> None:
|
def install(self, config: InstallConfig) -> None:
|
||||||
org_id = config.get("org_id", "")
|
|
||||||
agent_id = config.get("agent_id", "")
|
agent_id = config.get("agent_id", "")
|
||||||
|
agent_name = config.get("agent_name", "")
|
||||||
|
org_id = config.get("org_id", "")
|
||||||
private_key = config.get("private_key", "")
|
private_key = config.get("private_key", "")
|
||||||
kid = config.get("kid", "")
|
kid = config.get("kid", "")
|
||||||
base_url = config.get("base_url", "https://api.elydora.com")
|
base_url = config.get("base_url", "https://api.elydora.com")
|
||||||
guard_script_path = config.get("guard_script_path", "")
|
guard_script_path = config.get("guard_script_path", "")
|
||||||
|
|
||||||
|
# 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,
|
||||||
|
"agent_id": agent_id,
|
||||||
|
"kid": kid,
|
||||||
|
"base_url": base_url,
|
||||||
|
"token": config.get("token", ""),
|
||||||
|
"agent_name": agent_name,
|
||||||
|
}
|
||||||
|
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(
|
js_content = _generate_js_plugin(
|
||||||
org_id=org_id,
|
org_id=org_id,
|
||||||
agent_id=agent_id,
|
agent_id=agent_id,
|
||||||
@@ -43,15 +74,19 @@ class OpenCodePlugin(AgentPlugin):
|
|||||||
plugin_path = self._plugin_path()
|
plugin_path = self._plugin_path()
|
||||||
with open(plugin_path, "w", encoding="utf-8") as f:
|
with open(plugin_path, "w", encoding="utf-8") as f:
|
||||||
f.write(js_content)
|
f.write(js_content)
|
||||||
os.chmod(plugin_path, stat.S_IRWXU)
|
try:
|
||||||
|
os.chmod(plugin_path, stat.S_IRWXU)
|
||||||
|
except Exception:
|
||||||
|
pass # chmod may fail on Windows
|
||||||
|
|
||||||
print(f"Elydora hook installed for OpenCode.")
|
print(f"Elydora hook installed for OpenCode.")
|
||||||
print(f" Plugin: {plugin_path}")
|
print(f" Plugin: {plugin_path}")
|
||||||
|
|
||||||
def uninstall(self) -> None:
|
def uninstall(self, agent_id: str = "") -> None:
|
||||||
plugin_path = self._plugin_path()
|
plugin_path = self._plugin_path()
|
||||||
if os.path.exists(plugin_path):
|
if os.path.exists(plugin_path):
|
||||||
os.remove(plugin_path)
|
os.remove(plugin_path)
|
||||||
|
# Hook script removal is handled by cli.py cmd_uninstall (rmtree of agent dir)
|
||||||
print("Elydora hook uninstalled from OpenCode.")
|
print("Elydora hook uninstalled from OpenCode.")
|
||||||
|
|
||||||
def status(self) -> PluginStatus:
|
def status(self) -> PluginStatus:
|
||||||
@@ -74,6 +109,7 @@ def _generate_js_plugin(
|
|||||||
guard_script_path: str = "",
|
guard_script_path: str = "",
|
||||||
) -> str:
|
) -> str:
|
||||||
guard_path_escaped = guard_script_path.replace("\\", "\\\\") if guard_script_path else ""
|
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.
|
return f'''// Elydora audit plugin for OpenCode — auto-generated, do not edit.
|
||||||
"use strict";
|
"use strict";
|
||||||
|
|
||||||
@@ -91,7 +127,7 @@ const PRIVATE_KEY = {_js_str(private_key)};
|
|||||||
const KID = {_js_str(kid)};
|
const KID = {_js_str(kid)};
|
||||||
const BASE_URL = {_js_str(base_url)};
|
const BASE_URL = {_js_str(base_url)};
|
||||||
|
|
||||||
const CHAIN_STATE_PATH = path.join(os.homedir(), ".elydora", "chain-state.json");
|
const CHAIN_STATE_PATH = path.join(os.homedir(), ".elydora", AGENT_ID, "chain-state.json");
|
||||||
|
|
||||||
function base64urlEncode(buf) {{
|
function base64urlEncode(buf) {{
|
||||||
return Buffer.from(buf).toString("base64url");
|
return Buffer.from(buf).toString("base64url");
|
||||||
@@ -174,6 +210,7 @@ function saveChainState(state) {{
|
|||||||
fs.renameSync(tmp, CHAIN_STATE_PATH);
|
fs.renameSync(tmp, CHAIN_STATE_PATH);
|
||||||
}}
|
}}
|
||||||
|
|
||||||
|
const PYTHON_EXE = "{python_exe_escaped}";
|
||||||
const GUARD_SCRIPT_PATH = "{guard_path_escaped}";
|
const GUARD_SCRIPT_PATH = "{guard_path_escaped}";
|
||||||
|
|
||||||
module.exports = {{
|
module.exports = {{
|
||||||
@@ -186,7 +223,7 @@ module.exports = {{
|
|||||||
// Synchronous guard — blocks tool if agent is frozen
|
// Synchronous guard — blocks tool if agent is frozen
|
||||||
if (GUARD_SCRIPT_PATH) {{
|
if (GUARD_SCRIPT_PATH) {{
|
||||||
try {{
|
try {{
|
||||||
const result = spawnSync("python3", [GUARD_SCRIPT_PATH], {{
|
const result = spawnSync(PYTHON_EXE, [GUARD_SCRIPT_PATH], {{
|
||||||
timeout: 5000,
|
timeout: 5000,
|
||||||
stdio: ["pipe", "ignore", "pipe"],
|
stdio: ["pipe", "ignore", "pipe"],
|
||||||
}});
|
}});
|
||||||
@@ -238,9 +275,7 @@ module.exports = {{
|
|||||||
const signable = Object.fromEntries(Object.entries(eor).filter(([k]) => k !== "signature"));
|
const signable = Object.fromEntries(Object.entries(eor).filter(([k]) => k !== "signature"));
|
||||||
eor.signature = signEd25519(PRIVATE_KEY, jcsCanonicalize(signable));
|
eor.signature = signEd25519(PRIVATE_KEY, jcsCanonicalize(signable));
|
||||||
|
|
||||||
saveChainState({{ prev_chain_hash: chainHash }});
|
// POST to API — only save chain state on 2xx response
|
||||||
|
|
||||||
// Fire-and-forget POST
|
|
||||||
const url = new URL(BASE_URL.replace(/\\/$/, "") + "/v1/operations");
|
const url = new URL(BASE_URL.replace(/\\/$/, "") + "/v1/operations");
|
||||||
const body = JSON.stringify(eor);
|
const body = JSON.stringify(eor);
|
||||||
const mod = url.protocol === "https:" ? https : http;
|
const mod = url.protocol === "https:" ? https : http;
|
||||||
@@ -248,6 +283,11 @@ module.exports = {{
|
|||||||
method: "POST",
|
method: "POST",
|
||||||
headers: {{ "Content-Type": "application/json", "Content-Length": Buffer.byteLength(body) }},
|
headers: {{ "Content-Type": "application/json", "Content-Length": Buffer.byteLength(body) }},
|
||||||
timeout: 5000,
|
timeout: 5000,
|
||||||
|
}}, (res) => {{
|
||||||
|
if (res.statusCode >= 200 && res.statusCode < 300) {{
|
||||||
|
saveChainState({{ prev_chain_hash: chainHash }});
|
||||||
|
}}
|
||||||
|
res.resume();
|
||||||
}});
|
}});
|
||||||
req.on("error", () => {{}});
|
req.on("error", () => {{}});
|
||||||
req.write(body);
|
req.write(body);
|
||||||
|
|||||||
@@ -11,6 +11,11 @@ SUPPORTED_AGENTS: Dict[str, Dict[str, str]] = {
|
|||||||
"hook_event": "PostToolUse",
|
"hook_event": "PostToolUse",
|
||||||
"config_path": "~/.claude/settings.json",
|
"config_path": "~/.claude/settings.json",
|
||||||
},
|
},
|
||||||
|
"copilot": {
|
||||||
|
"name": "Copilot CLI",
|
||||||
|
"hook_event": "PostToolUse",
|
||||||
|
"config_path": ".github/hooks/hooks.json",
|
||||||
|
},
|
||||||
"cursor": {
|
"cursor": {
|
||||||
"name": "Cursor",
|
"name": "Cursor",
|
||||||
"hook_event": "PostToolUse",
|
"hook_event": "PostToolUse",
|
||||||
@@ -21,16 +26,21 @@ SUPPORTED_AGENTS: Dict[str, Dict[str, str]] = {
|
|||||||
"hook_event": "AfterTool",
|
"hook_event": "AfterTool",
|
||||||
"config_path": "~/.gemini/settings.json",
|
"config_path": "~/.gemini/settings.json",
|
||||||
},
|
},
|
||||||
"augment": {
|
"kirocli": {
|
||||||
"name": "Augment Code",
|
"name": "Kiro CLI",
|
||||||
"hook_event": "PostToolUse",
|
"hook_event": "PostToolUse",
|
||||||
"config_path": "~/.augment/settings.json",
|
"config_path": "~/.kiro/settings.json",
|
||||||
},
|
},
|
||||||
"kiro": {
|
"kiroide": {
|
||||||
"name": "Kiro",
|
"name": "Kiro IDE",
|
||||||
"hook_event": "PostToolUse",
|
"hook_event": "PostToolUse",
|
||||||
"config_path": "~/.kiro/hooks/",
|
"config_path": "~/.kiro/hooks/",
|
||||||
},
|
},
|
||||||
|
"letta": {
|
||||||
|
"name": "Letta Code",
|
||||||
|
"hook_event": "PostToolUse",
|
||||||
|
"config_path": "~/.letta/settings.json",
|
||||||
|
},
|
||||||
"opencode": {
|
"opencode": {
|
||||||
"name": "OpenCode",
|
"name": "OpenCode",
|
||||||
"hook_event": "PostToolUse",
|
"hook_event": "PostToolUse",
|
||||||
|
|||||||
Reference in new issue
Block a user