Initial commit: Elydora Python SDK
Python SDK for Elydora agent operations, cryptographic signing, and audit trail integration. Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
This commit is contained in:
commit
b42b670c2a
22 files changed
+3127
No files matched your search
@@ -0,0 +1 @@
|
||||
"""Elydora agent plugins for AI coding tools."""
|
||||
@@ -0,0 +1,160 @@
|
||||
"""Augment Code plugin — merges PostToolUse hook into ~/.augment/settings.json."""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import json
|
||||
import os
|
||||
import stat
|
||||
|
||||
from .base import AgentPlugin, InstallConfig, PluginStatus
|
||||
from .hook_template import generate_hook_script
|
||||
|
||||
|
||||
SETTINGS_PATH = os.path.join(os.path.expanduser("~"), ".augment", "settings.json")
|
||||
HOOK_DIR = os.path.join(os.path.expanduser("~"), ".elydora", "hooks")
|
||||
HOOK_FILENAME = "elydora-audit-hook-augment.py"
|
||||
|
||||
|
||||
class AugmentPlugin(AgentPlugin):
|
||||
"""Install/uninstall Elydora audit hook for Augment Code."""
|
||||
|
||||
def _hook_path(self) -> str:
|
||||
return os.path.join(HOOK_DIR, HOOK_FILENAME)
|
||||
|
||||
def install(self, config: InstallConfig) -> None:
|
||||
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(HOOK_DIR, exist_ok=True)
|
||||
hook_path = self._hook_path()
|
||||
with open(hook_path, "w", encoding="utf-8") as f:
|
||||
f.write(script)
|
||||
os.chmod(hook_path, os.stat(hook_path).st_mode | stat.S_IEXEC)
|
||||
|
||||
guard_script_path = config.get("guard_script_path", "")
|
||||
|
||||
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"python3 {guard_script_path}",
|
||||
}
|
||||
],
|
||||
})
|
||||
|
||||
# --- PostToolUse (audit logging) ---
|
||||
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({
|
||||
"hooks": [
|
||||
{
|
||||
"type": "command",
|
||||
"command": hook_path,
|
||||
}
|
||||
],
|
||||
})
|
||||
|
||||
_save_json(SETTINGS_PATH, settings)
|
||||
print(f"Elydora hook installed for Augment Code.")
|
||||
print(f" Hook script: {hook_path}")
|
||||
print(f" Settings: {SETTINGS_PATH}")
|
||||
|
||||
def uninstall(self) -> 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)]
|
||||
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)]
|
||||
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_path = self._hook_path()
|
||||
if os.path.exists(hook_path):
|
||||
os.remove(hook_path)
|
||||
|
||||
print("Elydora hook uninstalled from Augment Code.")
|
||||
|
||||
def status(self) -> PluginStatus:
|
||||
hook_path = self._hook_path()
|
||||
hook_exists = os.path.exists(hook_path)
|
||||
|
||||
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"Hook: {hook_path}"
|
||||
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="augment", details=details)
|
||||
|
||||
|
||||
def _is_elydora_hook(entry: dict) -> bool:
|
||||
inner_hooks = entry.get("hooks")
|
||||
if isinstance(inner_hooks, list):
|
||||
return any(
|
||||
"elydora" in h.get("command", "").lower()
|
||||
for h in inner_hooks
|
||||
if isinstance(h, dict)
|
||||
)
|
||||
cmd = entry.get("command", "")
|
||||
return "elydora" in cmd.lower()
|
||||
|
||||
|
||||
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,44 @@
|
||||
"""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
|
||||
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) -> None:
|
||||
"""Remove the Elydora hook for this agent."""
|
||||
|
||||
@abstractmethod
|
||||
def status(self) -> PluginStatus:
|
||||
"""Check whether the Elydora hook is installed for this agent."""
|
||||
@@ -0,0 +1,167 @@
|
||||
"""Claude Code plugin — merges PostToolUse hook into ~/.claude/settings.json."""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import json
|
||||
import os
|
||||
import stat
|
||||
|
||||
from .base import AgentPlugin, InstallConfig, PluginStatus
|
||||
from .hook_template import generate_hook_script
|
||||
|
||||
|
||||
SETTINGS_PATH = os.path.join(os.path.expanduser("~"), ".claude", "settings.json")
|
||||
HOOK_DIR = os.path.join(os.path.expanduser("~"), ".elydora", "hooks")
|
||||
HOOK_FILENAME = "elydora-audit-hook.py"
|
||||
|
||||
|
||||
class ClaudeCodePlugin(AgentPlugin):
|
||||
"""Install/uninstall Elydora audit hook for Claude Code."""
|
||||
|
||||
def _hook_path(self) -> str:
|
||||
return os.path.join(HOOK_DIR, HOOK_FILENAME)
|
||||
|
||||
def install(self, config: InstallConfig) -> None:
|
||||
# Write the 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(HOOK_DIR, exist_ok=True)
|
||||
hook_path = self._hook_path()
|
||||
with open(hook_path, "w", encoding="utf-8") as f:
|
||||
f.write(script)
|
||||
os.chmod(hook_path, os.stat(hook_path).st_mode | stat.S_IEXEC)
|
||||
|
||||
guard_script_path = config.get("guard_script_path", "")
|
||||
|
||||
# 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"python3 {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) -> 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)]
|
||||
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)]
|
||||
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)
|
||||
|
||||
# Remove hook script
|
||||
hook_path = self._hook_path()
|
||||
if os.path.exists(hook_path):
|
||||
os.remove(hook_path)
|
||||
|
||||
print("Elydora hook uninstalled from Claude Code.")
|
||||
|
||||
def status(self) -> PluginStatus:
|
||||
hook_path = self._hook_path()
|
||||
hook_exists = os.path.exists(hook_path)
|
||||
|
||||
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"Hook: {hook_path}"
|
||||
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) -> bool:
|
||||
# New format: { "matcher": {}, "hooks": [{ "type": "command", "command": "..." }] }
|
||||
inner_hooks = entry.get("hooks")
|
||||
if isinstance(inner_hooks, list):
|
||||
return any(
|
||||
"elydora" in h.get("command", "").lower()
|
||||
for h in inner_hooks
|
||||
if isinstance(h, dict)
|
||||
)
|
||||
# Old format: { "command": "..." }
|
||||
cmd = entry.get("command", "")
|
||||
return "elydora" in cmd.lower()
|
||||
|
||||
|
||||
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,143 @@
|
||||
"""Cursor plugin — writes/merges postToolUse hook into ~/.cursor/hooks.json."""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import json
|
||||
import os
|
||||
import stat
|
||||
|
||||
from .base import AgentPlugin, InstallConfig, PluginStatus
|
||||
from .hook_template import generate_hook_script
|
||||
|
||||
|
||||
SETTINGS_PATH = os.path.join(os.path.expanduser("~"), ".cursor", "hooks.json")
|
||||
HOOK_DIR = os.path.join(os.path.expanduser("~"), ".elydora", "hooks")
|
||||
HOOK_FILENAME = "elydora-audit-hook-cursor.py"
|
||||
|
||||
|
||||
class CursorPlugin(AgentPlugin):
|
||||
"""Install/uninstall Elydora audit hook for Cursor."""
|
||||
|
||||
def _hook_path(self) -> str:
|
||||
return os.path.join(HOOK_DIR, HOOK_FILENAME)
|
||||
|
||||
def install(self, config: InstallConfig) -> None:
|
||||
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(HOOK_DIR, exist_ok=True)
|
||||
hook_path = self._hook_path()
|
||||
with open(hook_path, "w", encoding="utf-8") as f:
|
||||
f.write(script)
|
||||
os.chmod(hook_path, os.stat(hook_path).st_mode | stat.S_IEXEC)
|
||||
|
||||
guard_script_path = config.get("guard_script_path", "")
|
||||
|
||||
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"python3 {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) -> 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)]
|
||||
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)]
|
||||
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_path = self._hook_path()
|
||||
if os.path.exists(hook_path):
|
||||
os.remove(hook_path)
|
||||
|
||||
print("Elydora hook uninstalled from Cursor.")
|
||||
|
||||
def status(self) -> PluginStatus:
|
||||
hook_path = self._hook_path()
|
||||
hook_exists = os.path.exists(hook_path)
|
||||
|
||||
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"Hook: {hook_path}"
|
||||
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) -> bool:
|
||||
cmd = hook.get("command", "")
|
||||
return "elydora" in cmd.lower()
|
||||
|
||||
|
||||
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,160 @@
|
||||
"""Gemini CLI plugin — merges AfterTool hook into ~/.gemini/settings.json."""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import json
|
||||
import os
|
||||
import stat
|
||||
|
||||
from .base import AgentPlugin, InstallConfig, PluginStatus
|
||||
from .hook_template import generate_hook_script
|
||||
|
||||
|
||||
SETTINGS_PATH = os.path.join(os.path.expanduser("~"), ".gemini", "settings.json")
|
||||
HOOK_DIR = os.path.join(os.path.expanduser("~"), ".elydora", "hooks")
|
||||
HOOK_FILENAME = "elydora-audit-hook-gemini.py"
|
||||
|
||||
|
||||
class GeminiPlugin(AgentPlugin):
|
||||
"""Install/uninstall Elydora audit hook for Gemini CLI."""
|
||||
|
||||
def _hook_path(self) -> str:
|
||||
return os.path.join(HOOK_DIR, HOOK_FILENAME)
|
||||
|
||||
def install(self, config: InstallConfig) -> None:
|
||||
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(HOOK_DIR, exist_ok=True)
|
||||
hook_path = self._hook_path()
|
||||
with open(hook_path, "w", encoding="utf-8") as f:
|
||||
f.write(script)
|
||||
os.chmod(hook_path, os.stat(hook_path).st_mode | stat.S_IEXEC)
|
||||
|
||||
guard_script_path = config.get("guard_script_path", "")
|
||||
|
||||
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"python3 {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) -> 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)]
|
||||
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)]
|
||||
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_path = self._hook_path()
|
||||
if os.path.exists(hook_path):
|
||||
os.remove(hook_path)
|
||||
|
||||
print("Elydora hook uninstalled from Gemini CLI.")
|
||||
|
||||
def status(self) -> PluginStatus:
|
||||
hook_path = self._hook_path()
|
||||
hook_exists = os.path.exists(hook_path)
|
||||
|
||||
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"Hook: {hook_path}"
|
||||
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) -> bool:
|
||||
inner_hooks = entry.get("hooks")
|
||||
if isinstance(inner_hooks, list):
|
||||
return any(
|
||||
"elydora" in h.get("command", "").lower()
|
||||
for h in inner_hooks
|
||||
if isinstance(h, dict)
|
||||
)
|
||||
cmd = entry.get("command", "")
|
||||
return "elydora" in cmd.lower()
|
||||
|
||||
|
||||
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,390 @@
|
||||
"""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/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}
|
||||
|
||||
CHAIN_STATE_PATH = os.path.join(os.path.expanduser("~"), ".elydora", "chain-state.json")
|
||||
ERROR_LOG_PATH = os.path.join(os.path.expanduser("~"), ".elydora", "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) -> str:
|
||||
"""Return a self-contained Python guard script for PreToolUse freeze enforcement.
|
||||
|
||||
The script:
|
||||
1. Reads agent config from ~/.elydora/agents/{agent_name}.json
|
||||
2. Checks cached status from ~/.elydora/agents/{agent_name}.status.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
|
||||
"""
|
||||
return f'''#!/usr/bin/env python3
|
||||
"""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}
|
||||
ELYDORA_DIR = os.path.join(os.path.expanduser("~"), ".elydora")
|
||||
CONFIG_PATH = os.path.join(ELYDORA_DIR, "agents", AGENT_NAME + ".json")
|
||||
STATUS_CACHE_PATH = os.path.join(ELYDORA_DIR, "agents", AGENT_NAME + ".status.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()
|
||||
'''
|
||||
@@ -0,0 +1,113 @@
|
||||
"""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, os.stat(script_path).st_mode | stat.S_IEXEC)
|
||||
|
||||
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,266 @@
|
||||
"""OpenCode plugin — writes .opencode/plugins/elydora-audit.js (JS plugin)."""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import os
|
||||
|
||||
from .base import AgentPlugin, InstallConfig, PluginStatus
|
||||
|
||||
|
||||
PLUGIN_DIR = os.path.join(os.path.expanduser("~"), ".opencode", "plugins")
|
||||
PLUGIN_FILENAME = "elydora-audit.js"
|
||||
|
||||
|
||||
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:
|
||||
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")
|
||||
guard_script_path = config.get("guard_script_path", "")
|
||||
|
||||
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)
|
||||
|
||||
print(f"Elydora hook installed for OpenCode.")
|
||||
print(f" Plugin: {plugin_path}")
|
||||
|
||||
def uninstall(self) -> None:
|
||||
plugin_path = self._plugin_path()
|
||||
if os.path.exists(plugin_path):
|
||||
os.remove(plugin_path)
|
||||
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 ""
|
||||
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", "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 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("python3", [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));
|
||||
|
||||
saveChainState({{ prev_chain_hash: chainHash }});
|
||||
|
||||
// Fire-and-forget POST
|
||||
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,
|
||||
}});
|
||||
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)
|
||||
@@ -0,0 +1,44 @@
|
||||
"""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",
|
||||
},
|
||||
"cursor": {
|
||||
"name": "Cursor",
|
||||
"hook_event": "PostToolUse",
|
||||
"config_path": "~/.cursor/hooks.json",
|
||||
},
|
||||
"gemini": {
|
||||
"name": "Gemini CLI",
|
||||
"hook_event": "AfterTool",
|
||||
"config_path": "~/.gemini/settings.json",
|
||||
},
|
||||
"augment": {
|
||||
"name": "Augment Code",
|
||||
"hook_event": "PostToolUse",
|
||||
"config_path": "~/.augment/settings.json",
|
||||
},
|
||||
"kiro": {
|
||||
"name": "Kiro",
|
||||
"hook_event": "PostToolUse",
|
||||
"config_path": "~/.kiro/hooks/",
|
||||
},
|
||||
"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())
|
||||
Reference in new issue
Block a user