Initial commit: Elydora Open Source — self-hosted audit platform for AI agents

Full self-hosted port of the Elydora tamper-evident audit protocol:
- Hono API server with PostgreSQL, Redis, MinIO adapters
- Ed25519 signed operations, chain hashing, Merkle epoch rollups
- PBKDF2 password hashing, HMAC-SHA256 JWT, 5-role RBAC
- Next.js management console
- Node.js, Python, and Go SDKs
- Docker Compose for local deployment
- Helm chart for Kubernetes
- Comprehensive security hardening and audit fixes applied

Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
This commit is contained in:
ZacharyZhang-NYandClaude Opus 4.6 committed 2026-03-02 13:41:42 -05:00
1 parent 15547a8c35
commit e390e68594
207 files changed
+32790 -63

No files matched your search

+1
View File
@@ -0,0 +1 @@
"""Elydora agent plugins for AI coding tools."""
+45
View File
@@ -0,0 +1,45 @@
"""Abstract base class for agent plugins."""
from __future__ import annotations
import sys
from abc import ABC, abstractmethod
from typing import Any, Dict
if sys.version_info >= (3, 11):
from typing import TypedDict
else:
from typing_extensions import TypedDict
class InstallConfig(TypedDict, total=False):
org_id: str
agent_id: str
agent_name: str
private_key: str
kid: str
token: str
base_url: str
guard_script_path: str
class PluginStatus(TypedDict):
installed: bool
agent: str
details: str
class AgentPlugin(ABC):
"""Base class for all agent plugins."""
@abstractmethod
def install(self, config: InstallConfig) -> None:
"""Install the Elydora hook for this agent."""
@abstractmethod
def uninstall(self, agent_id: str = "") -> None:
"""Remove the Elydora hook for this agent."""
@abstractmethod
def status(self) -> PluginStatus:
"""Check whether the Elydora hook is installed for this agent."""
+211
View File
@@ -0,0 +1,211 @@
"""Claude Code plugin — merges PostToolUse hook into ~/.claude/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("~"), ".claude", "settings.json")
ELYDORA_DIR = os.path.join(os.path.expanduser("~"), ".elydora")
class ClaudeCodePlugin(AgentPlugin):
"""Install/uninstall Elydora audit hook for Claude Code."""
@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 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"),
)
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
# Merge into Claude Code settings
settings = _load_json(SETTINGS_PATH)
hooks = settings.setdefault("hooks", {})
# --- PreToolUse (guard — freeze enforcement) ---
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({
"hooks": [
{
"type": "command",
"command": f'"{python_exe}" {guard_script_path}',
}
],
})
# --- PostToolUse (audit logging) ---
post_tool_use = hooks.setdefault("PostToolUse", [])
# Remove any existing Elydora hook entry
post_tool_use[:] = [h for h in post_tool_use if not _is_elydora_hook(h)]
post_tool_use.append({
"hooks": [
{
"type": "command",
"command": hook_path,
}
],
})
_save_json(SETTINGS_PATH, settings)
print(f"Elydora hook installed for Claude Code.")
print(f" Hook script: {hook_path}")
print(f" Settings: {SETTINGS_PATH}")
def uninstall(self, agent_id: str = "") -> None:
# Remove hook entries from settings
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 Claude Code.")
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="claudecode", 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")
+200
View File
@@ -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")
+182
View File
@@ -0,0 +1,182 @@
"""Cursor plugin — writes/merges postToolUse hook into ~/.cursor/hooks.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("~"), ".cursor", "hooks.json")
ELYDORA_DIR = os.path.join(os.path.expanduser("~"), ".elydora")
class CursorPlugin(AgentPlugin):
"""Install/uninstall Elydora audit hook for Cursor."""
@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, 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({
"command": f'"{python_exe}" {guard_script_path}',
})
# --- 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({
"command": hook_path,
})
_save_json(SETTINGS_PATH, settings)
print(f"Elydora hook installed for Cursor.")
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 Cursor.")
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 hooks.json"
elif settings_configured:
details = "Configured in hooks.json but hook script missing"
else:
details = "Not installed"
return PluginStatus(installed=installed, agent="cursor", details=details)
def _is_elydora_hook(hook: dict, agent_id: str = "") -> bool:
cmd = hook.get("command", "")
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:
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")
+207
View File
@@ -0,0 +1,207 @@
"""Gemini CLI plugin — merges AfterTool hook into ~/.gemini/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("~"), ".gemini", "settings.json")
ELYDORA_DIR = os.path.join(os.path.expanduser("~"), ".elydora")
class GeminiPlugin(AgentPlugin):
"""Install/uninstall Elydora audit hook for Gemini 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", {})
# --- BeforeTool (guard — freeze enforcement) ---
before_tool = hooks.setdefault("BeforeTool", [])
before_tool[:] = [h for h in before_tool if not _is_elydora_hook(h)]
if guard_script_path:
before_tool.append({
"hooks": [
{
"type": "command",
"command": f'"{python_exe}" {guard_script_path}',
}
],
})
# --- AfterTool (audit logging) ---
after_tool = hooks.setdefault("AfterTool", [])
after_tool[:] = [h for h in after_tool if not _is_elydora_hook(h)]
after_tool.append({
"hooks": [
{
"type": "command",
"command": hook_path,
}
],
})
_save_json(SETTINGS_PATH, settings)
print(f"Elydora hook installed for Gemini 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 BeforeTool entries
before_tool = hooks.get("BeforeTool", [])
before_filtered = [h for h in before_tool if not _is_elydora_hook(h, agent_id)]
if len(before_filtered) != len(before_tool):
hooks["BeforeTool"] = before_filtered
if not before_filtered:
del hooks["BeforeTool"]
changed = True
# Remove AfterTool entries
after_tool = hooks.get("AfterTool", [])
after_filtered = [h for h in after_tool if not _is_elydora_hook(h, agent_id)]
if len(after_filtered) != len(after_tool):
hooks["AfterTool"] = after_filtered
if not after_filtered:
del hooks["AfterTool"]
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 Gemini 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", {})
before_tool = hooks.get("BeforeTool", [])
after_tool = hooks.get("AfterTool", [])
before_configured = any(_is_elydora_hook(h) for h in before_tool)
after_configured = any(_is_elydora_hook(h) for h in after_tool)
settings_configured = before_configured and after_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="gemini", 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,393 @@
"""Generates self-contained Python hook scripts for AI agent integrations.
The generated script contains all crypto and networking inline so it has zero
import dependencies beyond the Python stdlib + cryptography (already an
Elydora SDK dependency).
"""
from __future__ import annotations
import sys
def generate_hook_script(
*,
org_id: str,
agent_id: str,
private_key: str,
kid: str,
base_url: str,
) -> str:
"""Return a complete, self-contained Python hook script as a string.
The script:
1. Reads JSON from stdin (tool-use event)
2. Constructs an Elydora Operation Record (EOR)
3. Signs it with Ed25519
4. POSTs it fire-and-forget to the Elydora API
5. Always exits 0 (never blocks the host agent)
Chain state is persisted in ~/.elydora/{agent_id}/chain-state.json.
"""
shebang = sys.executable
return f'''#!{shebang}
"""Elydora audit hook — auto-generated, do not edit."""
import base64
import hashlib
import json
import math
import os
import struct
import sys
import time
# ---------------------------------------------------------------------------
# Configuration (baked at install time)
# ---------------------------------------------------------------------------
ORG_ID = {org_id!r}
AGENT_ID = {agent_id!r}
PRIVATE_KEY = {private_key!r}
KID = {kid!r}
BASE_URL = {base_url!r}
ELYDORA_DIR = os.path.join(os.path.expanduser("~"), ".elydora")
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"
# ---------------------------------------------------------------------------
# Base64url helpers
# ---------------------------------------------------------------------------
def base64url_encode(data: bytes) -> str:
return base64.urlsafe_b64encode(data).rstrip(b"=").decode("ascii")
def base64url_decode(s: str) -> bytes:
s = s.replace("-", "+").replace("_", "/")
padding = 4 - len(s) % 4
if padding != 4:
s += "=" * padding
return base64.b64decode(s)
# ---------------------------------------------------------------------------
# SHA-256
# ---------------------------------------------------------------------------
def sha256_base64url(data):
if isinstance(data, str):
data = data.encode("utf-8")
return base64url_encode(hashlib.sha256(data).digest())
# ---------------------------------------------------------------------------
# JCS canonicalization (RFC 8785)
# ---------------------------------------------------------------------------
def _jcs_number(value):
if isinstance(value, bool):
return "true" if value else "false"
if isinstance(value, int):
return str(value)
if math.isnan(value) or math.isinf(value):
return "null"
if value == 0.0:
return "0"
return json.dumps(value)
def jcs_canonicalize(value):
if value is None:
return "null"
if isinstance(value, bool):
return "true" if value else "false"
if isinstance(value, (int, float)):
return _jcs_number(value)
if isinstance(value, str):
return json.dumps(value, ensure_ascii=False)
if isinstance(value, list):
return "[" + ",".join(jcs_canonicalize(v) for v in value) + "]"
if isinstance(value, dict):
keys = sorted(value.keys())
pairs = []
for k in keys:
pairs.append(json.dumps(k, ensure_ascii=False) + ":" + jcs_canonicalize(value[k]))
return "{{" + ",".join(pairs) + "}}"
return json.dumps(value)
# ---------------------------------------------------------------------------
# Hash helpers
# ---------------------------------------------------------------------------
def compute_payload_hash(payload):
return sha256_base64url(jcs_canonicalize(payload))
def compute_chain_hash(prev_chain_hash, payload_hash, operation_id, issued_at):
input_str = f"{{prev_chain_hash}}|{{payload_hash}}|{{operation_id}}|{{issued_at}}"
return sha256_base64url(input_str)
# ---------------------------------------------------------------------------
# Ed25519 signing (uses cryptography library)
# ---------------------------------------------------------------------------
def sign_ed25519(private_key_b64url, data):
from cryptography.hazmat.primitives.asymmetric.ed25519 import Ed25519PrivateKey
seed = base64url_decode(private_key_b64url)
key = Ed25519PrivateKey.from_private_bytes(seed)
return base64url_encode(key.sign(data))
# ---------------------------------------------------------------------------
# UUIDv7
# ---------------------------------------------------------------------------
def generate_uuidv7():
timestamp_ms = int(time.time() * 1000)
ts_bytes = struct.pack(">Q", timestamp_ms)[2:]
rand_a = struct.unpack(">H", os.urandom(2))[0]
rand_a = (rand_a & 0x0FFF) | 0x7000
rand_b = bytearray(os.urandom(8))
rand_b[0] = (rand_b[0] & 0x3F) | 0x80
uuid_bytes = ts_bytes + struct.pack(">H", rand_a) + bytes(rand_b)
h = uuid_bytes.hex()
return f"{{h[0:8]}}-{{h[8:12]}}-{{h[12:16]}}-{{h[16:20]}}-{{h[20:32]}}"
def generate_nonce():
return base64url_encode(os.urandom(16))
# ---------------------------------------------------------------------------
# Chain state persistence
# ---------------------------------------------------------------------------
def read_chain_state():
try:
with open(CHAIN_STATE_PATH, "r") as f:
state = json.load(f)
return state.get("prev_chain_hash", ZERO_CHAIN_HASH) or ZERO_CHAIN_HASH
except (FileNotFoundError, json.JSONDecodeError):
return ZERO_CHAIN_HASH
def write_chain_state(chain_hash):
os.makedirs(os.path.dirname(CHAIN_STATE_PATH), exist_ok=True)
tmp_path = CHAIN_STATE_PATH + ".tmp"
with open(tmp_path, "w") as f:
json.dump({{"prev_chain_hash": chain_hash}}, f)
try:
os.replace(tmp_path, CHAIN_STATE_PATH)
except OSError:
import shutil
shutil.move(tmp_path, CHAIN_STATE_PATH)
def log_error(err):
try:
ts = time.strftime("%Y-%m-%dT%H:%M:%SZ", time.gmtime())
msg = f"{{ts}} [elydora-hook] {{err}}\\n"
os.makedirs(os.path.dirname(ERROR_LOG_PATH), exist_ok=True)
with open(ERROR_LOG_PATH, "a") as f:
f.write(msg)
except Exception:
pass
# ---------------------------------------------------------------------------
# Main hook logic
# ---------------------------------------------------------------------------
def main():
try:
raw = sys.stdin.read()
if not raw.strip():
return
try:
event = json.loads(raw)
except (json.JSONDecodeError, ValueError):
return
# Extract tool event data
tool_name = "unknown"
tool_input = {{}}
session_id = "unknown"
if isinstance(event, dict):
tool_name = event.get("tool_name", event.get("toolName", event.get("name", "unknown")))
tool_input = event.get("tool_input", event.get("toolInput", event.get("input", {{}})))
session_id = event.get("session_id", event.get("sessionId", event.get("session", "unknown")))
# Read chain state
prev_chain_hash = read_chain_state()
# Build the EOR
operation_id = generate_uuidv7()
issued_at = int(time.time() * 1000)
nonce = generate_nonce()
payload = {{
"tool_name": tool_name,
"tool_input": tool_input,
"session_id": session_id,
}}
payload_hash = compute_payload_hash(payload)
chain_hash = compute_chain_hash(prev_chain_hash, payload_hash, operation_id, issued_at)
eor_without_sig = {{
"op_version": "1.0",
"operation_id": operation_id,
"org_id": ORG_ID,
"agent_id": AGENT_ID,
"issued_at": issued_at,
"ttl_ms": 30000,
"nonce": nonce,
"operation_type": "ai.tool_use",
"subject": {{"session_id": session_id}},
"action": {{"tool": tool_name}},
"payload": payload,
"payload_hash": payload_hash,
"prev_chain_hash": prev_chain_hash,
"agent_pubkey_kid": KID,
}}
canonical = jcs_canonicalize(eor_without_sig)
signature = sign_ed25519(PRIVATE_KEY, canonical.encode("utf-8"))
eor = dict(eor_without_sig)
eor["chain_hash"] = chain_hash
eor["signature"] = signature
# POST to /v1/operations (5s timeout)
# Only update local chain state on success to prevent desync
try:
import urllib.request
import urllib.error
url = BASE_URL.rstrip("/") + "/v1/operations"
body = json.dumps(eor).encode("utf-8")
headers = {{"Content-Type": "application/json", "Accept": "application/json"}}
req = urllib.request.Request(url, data=body, headers=headers, method="POST")
resp = urllib.request.urlopen(req, timeout=5)
# Server accepted (2xx) — safe to advance local chain state
write_chain_state(chain_hash)
except urllib.error.HTTPError as http_err:
if http_err.code == 400:
# Possible chain hash mismatch — try to resync from server response
try:
import re
err_body = json.loads(http_err.read().decode("utf-8"))
if isinstance(err_body, dict) and isinstance(err_body.get("error"), dict):
if err_body["error"].get("code") == "PREV_HASH_MISMATCH":
match = re.search(r'Expected prev_chain_hash "([^"]+)"', err_body["error"].get("message", ""))
if match:
write_chain_state(match.group(1))
log_error("Chain hash resynced to server: " + match.group(1))
except Exception:
pass # ignore parse errors
else:
log_error("API returned HTTP " + str(http_err.code))
except Exception:
pass # Network error — don't advance chain state, will retry with same prev_chain_hash
except Exception as e:
log_error(e)
if __name__ == "__main__":
main()
'''
def generate_guard_script(agent_name: str, agent_id: str) -> str:
"""Return a self-contained Python guard script for PreToolUse freeze enforcement.
The script:
1. Reads agent config from ~/.elydora/{agent_id}/config.json
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}
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)
6. Uses only stdlib — no external deps
"""
shebang = sys.executable
return f'''#!{shebang}
"""Elydora agent guard (PreToolUse) for: {agent_name}
Generated by elydora-sdk — DO NOT EDIT."""
import json
import os
import sys
import time
AGENT_NAME = {agent_name!r}
AGENT_ID = {agent_id!r}
ELYDORA_DIR = os.path.join(os.path.expanduser("~"), ".elydora")
CONFIG_PATH = os.path.join(ELYDORA_DIR, AGENT_ID, "config.json")
STATUS_CACHE_PATH = os.path.join(ELYDORA_DIR, AGENT_ID, "status-cache.json")
CACHE_TTL_S = 60 # 60 seconds
# Consume stdin so the parent process doesn't block on a full pipe
try:
sys.stdin.read()
except Exception:
pass
def main():
# Read agent config
try:
with open(CONFIG_PATH, "r") as f:
config = json.load(f)
except Exception:
return # Can't read config — fail-open (allow)
# Check cached status
try:
with open(STATUS_CACHE_PATH, "r") as f:
cache = json.load(f)
age = time.time() - cache.get("cached_at", 0)
if age < CACHE_TTL_S:
if cache.get("status") == "frozen":
sys.stderr.write(
'Agent "' + AGENT_NAME + '" is frozen by Elydora. Tool execution blocked.\\n'
)
sys.exit(1)
return # Cache is fresh — use cached result
except Exception:
pass # No cache or invalid — need to check API
# Fetch agent status from API
base_url = config.get("base_url", "https://api.elydora.com").rstrip("/")
agent_id = config.get("agent_id", "")
try:
import urllib.request
import urllib.error
url = base_url + "/v1/agents/" + urllib.request.quote(agent_id, safe="")
req = urllib.request.Request(url, method="GET")
req.add_header("Accept", "application/json")
token = config.get("token", "")
if token:
req.add_header("Authorization", "Bearer " + token)
resp = urllib.request.urlopen(req, timeout=3)
data = json.loads(resp.read().decode("utf-8"))
agent_status = "active"
if isinstance(data, dict) and isinstance(data.get("agent"), dict):
agent_status = data["agent"].get("status", "active")
# Update cache
try:
os.makedirs(os.path.dirname(STATUS_CACHE_PATH), exist_ok=True)
with open(STATUS_CACHE_PATH, "w") as f:
json.dump({{"status": agent_status, "cached_at": time.time()}}, f)
except Exception:
pass # Ignore cache write failures
if agent_status == "frozen":
sys.stderr.write(
'Agent "' + AGENT_NAME + '" is frozen by Elydora. Tool execution blocked.\\n'
)
sys.exit(1)
except SystemExit:
raise # Re-raise sys.exit
except Exception:
pass # API unreachable — fail-open (allow)
if __name__ == "__main__":
main()
'''
+211
View File
@@ -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")
+146
View File
@@ -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)
+209
View File
@@ -0,0 +1,209 @@
"""Letta Code plugin — merges PreToolUse/PostToolUse hooks into ~/.letta/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("~"), ".letta", "settings.json")
ELYDORA_DIR = os.path.join(os.path.expanduser("~"), ".elydora")
class LettaPlugin(AgentPlugin):
"""Install/uninstall Elydora audit hook for Letta Code."""
@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, no timeout_ms) ---
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}',
}
],
})
# --- PostToolUse (audit logging, PascalCase with matcher, no timeout_ms) ---
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,
}
],
})
_save_json(SETTINGS_PATH, settings)
print(f"Elydora hook installed for Letta Code.")
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 Letta Code.")
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="letta", 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")
+308
View File
@@ -0,0 +1,308 @@
"""OpenCode plugin — writes .opencode/plugins/elydora-audit.js (JS plugin)."""
from __future__ import annotations
import json
import os
import stat
import sys
from .base import AgentPlugin, InstallConfig, PluginStatus
PLUGIN_DIR = os.path.join(os.path.expanduser("~"), ".opencode", "plugins")
PLUGIN_FILENAME = "elydora-audit.js"
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.
"""
def _plugin_path(self) -> str:
return os.path.join(PLUGIN_DIR, PLUGIN_FILENAME)
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", "")
# 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(
org_id=org_id,
agent_id=agent_id,
private_key=private_key,
kid=kid,
base_url=base_url,
guard_script_path=guard_script_path,
)
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
print(f"Elydora hook installed for OpenCode.")
print(f" Plugin: {plugin_path}")
def uninstall(self, agent_id: str = "") -> None:
plugin_path = self._plugin_path()
if os.path.exists(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.")
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"
return PluginStatus(installed=installed, agent="opencode", details=details)
def _generate_js_plugin(
*,
org_id: str,
agent_id: str,
private_key: str,
kid: str,
base_url: 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";
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 ORG_ID = {_js_str(org_id)};
const AGENT_ID = {_js_str(agent_id)};
const PRIVATE_KEY = {_js_str(private_key)};
const KID = {_js_str(kid)};
const BASE_URL = {_js_str(base_url)};
const CHAIN_STATE_PATH = path.join(os.homedir(), ".elydora", AGENT_ID, "chain-state.json");
function base64urlEncode(buf) {{
return Buffer.from(buf).toString("base64url");
}}
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: "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);
}}
}},
}},
}};
'''
def _js_str(s: str) -> str:
"""Escape a Python string for embedding as a JS string literal."""
import json
return json.dumps(s)
+54
View File
@@ -0,0 +1,54 @@
"""Supported agent registry."""
from __future__ import annotations
from typing import Dict
SUPPORTED_AGENTS: Dict[str, Dict[str, str]] = {
"claudecode": {
"name": "Claude Code",
"hook_event": "PostToolUse",
"config_path": "~/.claude/settings.json",
},
"copilot": {
"name": "Copilot CLI",
"hook_event": "PostToolUse",
"config_path": ".github/hooks/hooks.json",
},
"cursor": {
"name": "Cursor",
"hook_event": "PostToolUse",
"config_path": "~/.cursor/hooks.json",
},
"gemini": {
"name": "Gemini CLI",
"hook_event": "AfterTool",
"config_path": "~/.gemini/settings.json",
},
"kirocli": {
"name": "Kiro CLI",
"hook_event": "PostToolUse",
"config_path": "~/.kiro/settings.json",
},
"kiroide": {
"name": "Kiro IDE",
"hook_event": "PostToolUse",
"config_path": "~/.kiro/hooks/",
},
"letta": {
"name": "Letta Code",
"hook_event": "PostToolUse",
"config_path": "~/.letta/settings.json",
},
"opencode": {
"name": "OpenCode",
"hook_event": "PostToolUse",
"config_path": "~/.opencode/plugins/",
},
}
def get_agent_names() -> list[str]:
"""Return sorted list of supported agent names."""
return sorted(SUPPORTED_AGENTS.keys())