fix: sync Python Kiro CLI hook contracts
This commit is contained in:
ZacharyZhang-NY committed
1 parent
ea903cfac5
commit
12b28f2166
4 files changed
+737
-159
No files matched your search
@@ -1,211 +1,431 @@
|
||||
"""Kiro CLI plugin — merges PreToolUse/PostToolUse hooks into ~/.kiro/settings.json."""
|
||||
"""Kiro CLI v2 custom-agent and v3 standalone-hook integration."""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import json
|
||||
import os
|
||||
import shlex
|
||||
import stat
|
||||
import subprocess
|
||||
import sys
|
||||
import tempfile
|
||||
from typing import Any, Dict, List, Optional, Tuple
|
||||
|
||||
from .base import AgentPlugin, InstallConfig, PluginStatus
|
||||
from .hook_template import generate_hook_script
|
||||
|
||||
|
||||
SETTINGS_PATH = os.path.join(os.path.expanduser("~"), ".kiro", "settings.json")
|
||||
AGENT_KEY = "kirocli"
|
||||
V2_AGENT_NAME = "elydora-audit"
|
||||
V2_DESCRIPTION = "Kiro CLI with Elydora audit and freeze enforcement"
|
||||
V3_GUARD_NAME = "elydora-guard"
|
||||
V3_AUDIT_NAME = "elydora-audit"
|
||||
ELYDORA_DIR = os.path.join(os.path.expanduser("~"), ".elydora")
|
||||
V2_AGENT_PATH = os.path.join(
|
||||
os.path.expanduser("~"), ".kiro", "agents", f"{V2_AGENT_NAME}.json"
|
||||
)
|
||||
V3_HOOKS_PATH = os.path.join(
|
||||
os.path.expanduser("~"), ".kiro", "hooks", "elydora-audit.json"
|
||||
)
|
||||
|
||||
JsonObject = Dict[str, Any]
|
||||
HookContract = Tuple[str, str, str]
|
||||
|
||||
|
||||
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")
|
||||
"""Install Elydora into both supported Kiro CLI hook generations."""
|
||||
|
||||
def install(self, config: InstallConfig) -> None:
|
||||
agent_id = config.get("agent_id", "")
|
||||
agent_name = config.get("agent_name", "")
|
||||
guard_path = config.get("guard_script_path", "")
|
||||
if not agent_id:
|
||||
raise ValueError("agent_id is required")
|
||||
if not guard_path:
|
||||
raise ValueError("guard_script_path is required")
|
||||
if not _regular_file_exists(guard_path, "Elydora guard runtime"):
|
||||
raise FileNotFoundError(f"Elydora guard runtime is missing: {guard_path}")
|
||||
|
||||
v2_settings = _read_json(V2_AGENT_PATH, "Kiro CLI v2 agent config") or {}
|
||||
v3_settings = _read_json(V3_HOOKS_PATH, "Kiro CLI v3 hooks config") or {}
|
||||
v2_hooks = _hooks_object(v2_settings, "Kiro CLI v2 agent config")
|
||||
current_v3_hooks = _v3_hooks(v3_settings)
|
||||
|
||||
v2_hooks["preToolUse"] = [
|
||||
*_without_v2_hooks(
|
||||
_hook_entries(v2_hooks, "preToolUse", "Kiro CLI v2 agent config")
|
||||
),
|
||||
_build_v2_hook(guard_path),
|
||||
]
|
||||
|
||||
# Create per-agent directory
|
||||
agent_dir = os.path.join(ELYDORA_DIR, agent_id)
|
||||
os.makedirs(agent_dir, exist_ok=True)
|
||||
hook_path = os.path.join(agent_dir, "hook.py")
|
||||
v2_hooks["postToolUse"] = [
|
||||
*_without_v2_hooks(
|
||||
_hook_entries(v2_hooks, "postToolUse", "Kiro CLI v2 agent config")
|
||||
),
|
||||
_build_v2_hook(hook_path),
|
||||
]
|
||||
|
||||
# Write config.json
|
||||
config_data = {
|
||||
next_v2_settings: JsonObject = {
|
||||
"name": V2_AGENT_NAME,
|
||||
"description": V2_DESCRIPTION,
|
||||
"tools": ["*"],
|
||||
"includeMcpJson": True,
|
||||
**v2_settings,
|
||||
"hooks": v2_hooks,
|
||||
}
|
||||
next_v3_settings: JsonObject = {
|
||||
**v3_settings,
|
||||
"version": "v1",
|
||||
"hooks": [
|
||||
*[hook for hook in current_v3_hooks if not _is_managed_v3_hook(hook)],
|
||||
_build_v3_hook(
|
||||
V3_GUARD_NAME,
|
||||
"Block tool use when the Elydora agent is frozen",
|
||||
"PreToolUse",
|
||||
guard_path,
|
||||
),
|
||||
_build_v3_hook(
|
||||
V3_AUDIT_NAME,
|
||||
"Record tool use in the Elydora audit trail",
|
||||
"PostToolUse",
|
||||
hook_path,
|
||||
),
|
||||
],
|
||||
}
|
||||
|
||||
runtime_config = {
|
||||
"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,
|
||||
"agent_name": config.get("agent_name", AGENT_KEY),
|
||||
}
|
||||
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(
|
||||
hook_script = generate_hook_script(
|
||||
org_id=config.get("org_id", ""),
|
||||
agent_id=agent_id,
|
||||
private_key=config.get("private_key", ""),
|
||||
kid=config.get("kid", ""),
|
||||
base_url=config.get("base_url", "https://api.elydora.com"),
|
||||
)
|
||||
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
|
||||
_write_json_atomic(os.path.join(agent_dir, "config.json"), runtime_config, 0o600)
|
||||
_write_text_atomic(
|
||||
os.path.join(agent_dir, "private.key"),
|
||||
config.get("private_key", ""),
|
||||
0o600,
|
||||
)
|
||||
_write_text_atomic(hook_path, hook_script, 0o700)
|
||||
_write_json_atomic(V2_AGENT_PATH, next_v2_settings, 0o600)
|
||||
_write_json_atomic(V3_HOOKS_PATH, next_v3_settings, 0o600)
|
||||
|
||||
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}")
|
||||
print('Kiro CLI v2: start with "kiro-cli --agent elydora-audit".')
|
||||
print('Kiro CLI v3: start with "kiro-cli --v3"; global hooks load automatically.')
|
||||
|
||||
def uninstall(self, agent_id: str = "") -> None:
|
||||
if os.path.exists(SETTINGS_PATH):
|
||||
settings = _load_json(SETTINGS_PATH)
|
||||
hooks = settings.get("hooks", {})
|
||||
changed = False
|
||||
v2_settings = _read_json(V2_AGENT_PATH, "Kiro CLI v2 agent config")
|
||||
v3_settings = _read_json(V3_HOOKS_PATH, "Kiro CLI v3 hooks config")
|
||||
|
||||
# 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
|
||||
next_v2: Optional[Tuple[JsonObject, bool]] = None
|
||||
if v2_settings is not None:
|
||||
v2_hooks = _hooks_object(v2_settings, "Kiro CLI v2 agent config")
|
||||
current_pre = _hook_entries(v2_hooks, "preToolUse", "Kiro CLI v2 agent config")
|
||||
current_post = _hook_entries(v2_hooks, "postToolUse", "Kiro CLI v2 agent config")
|
||||
next_pre = _without_v2_hooks(current_pre, agent_id)
|
||||
next_post = _without_v2_hooks(current_post, agent_id)
|
||||
if next_pre != current_pre or next_post != current_post:
|
||||
v2_hooks["preToolUse"] = next_pre
|
||||
v2_hooks["postToolUse"] = next_post
|
||||
next_v2 = (
|
||||
{**v2_settings, "hooks": v2_hooks},
|
||||
_is_owned_v2(v2_settings, v2_hooks),
|
||||
)
|
||||
|
||||
# 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
|
||||
next_v3: Optional[Tuple[JsonObject, bool]] = None
|
||||
if v3_settings is not None:
|
||||
current_hooks = _v3_hooks(v3_settings)
|
||||
v3_hooks = [
|
||||
hook for hook in current_hooks if not _is_managed_v3_hook(hook, agent_id)
|
||||
]
|
||||
if v3_hooks != current_hooks:
|
||||
owned = all(key in {"version", "hooks"} for key in v3_settings)
|
||||
next_v3 = ({**v3_settings, "hooks": v3_hooks}, owned and not v3_hooks)
|
||||
|
||||
if changed:
|
||||
if not hooks:
|
||||
del settings["hooks"]
|
||||
_save_json(SETTINGS_PATH, settings)
|
||||
if next_v2 is not None:
|
||||
settings, remove = next_v2
|
||||
if remove:
|
||||
_remove_file(V2_AGENT_PATH, "Kiro CLI v2 agent config")
|
||||
else:
|
||||
_write_json_atomic(V2_AGENT_PATH, settings, 0o600)
|
||||
|
||||
# Hook script removal is handled by cli.py cmd_uninstall (rmtree of agent dir)
|
||||
print("Elydora hook uninstalled from Kiro CLI.")
|
||||
if next_v3 is not None:
|
||||
settings, remove = next_v3
|
||||
if remove:
|
||||
_remove_file(V3_HOOKS_PATH, "Kiro CLI v3 hooks config")
|
||||
else:
|
||||
_write_json_atomic(V3_HOOKS_PATH, settings, 0o600)
|
||||
|
||||
print("Elydora hooks 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
|
||||
v2_settings = _read_json(V2_AGENT_PATH, "Kiro CLI v2 agent config")
|
||||
v3_settings = _read_json(V3_HOOKS_PATH, "Kiro CLI v3 hooks config")
|
||||
contracts: List[HookContract] = []
|
||||
|
||||
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
|
||||
if v2_settings is not None:
|
||||
v2_hooks = _hooks_object(v2_settings, "Kiro CLI v2 agent config")
|
||||
guard = _find_v2_command(
|
||||
_hook_entries(v2_hooks, "preToolUse", "Kiro CLI v2 agent config"),
|
||||
"guard.py",
|
||||
)
|
||||
audit = _find_v2_command(
|
||||
_hook_entries(v2_hooks, "postToolUse", "Kiro CLI v2 agent config"),
|
||||
"hook.py",
|
||||
)
|
||||
if guard and audit:
|
||||
contracts.append((guard, audit, V2_AGENT_PATH))
|
||||
|
||||
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"
|
||||
if v3_settings is not None:
|
||||
v3_hooks = _v3_hooks(v3_settings)
|
||||
guard = _find_v3_command(v3_hooks, V3_GUARD_NAME, "guard.py")
|
||||
audit = _find_v3_command(v3_hooks, V3_AUDIT_NAME, "hook.py")
|
||||
if guard and audit:
|
||||
contracts.append((guard, audit, V3_HOOKS_PATH))
|
||||
|
||||
return PluginStatus(installed=installed, agent="kirocli", details=details)
|
||||
if not contracts:
|
||||
return PluginStatus(installed=False, agent=AGENT_KEY, details="Not installed")
|
||||
|
||||
|
||||
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)
|
||||
config_path = contracts[-1][2]
|
||||
installed = _runtime_scripts_exist(contracts)
|
||||
details = (
|
||||
f"Config: {config_path}"
|
||||
if installed
|
||||
else f"Configured at {config_path}; runtime scripts missing"
|
||||
)
|
||||
else:
|
||||
commands.append(entry.get("command", ""))
|
||||
return PluginStatus(installed=installed, agent=AGENT_KEY, details=details)
|
||||
|
||||
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
|
||||
|
||||
def _read_json(path: str, label: str) -> Optional[JsonObject]:
|
||||
try:
|
||||
with open(path, "r", encoding="utf-8") as file:
|
||||
raw = file.read()
|
||||
except FileNotFoundError:
|
||||
return None
|
||||
except OSError as error:
|
||||
raise OSError(f"Read {label} at {path}: {error}") from error
|
||||
|
||||
try:
|
||||
value = json.loads(raw)
|
||||
except json.JSONDecodeError as error:
|
||||
raise ValueError(f"Failed to parse {label} at {path}: {error}") from error
|
||||
if not isinstance(value, dict):
|
||||
raise ValueError(f"{label} at {path} must contain a JSON object")
|
||||
return value
|
||||
|
||||
|
||||
def _write_text_atomic(path: str, content: str, mode: int) -> None:
|
||||
directory = os.path.dirname(path) or "."
|
||||
os.makedirs(directory, exist_ok=True)
|
||||
descriptor, temporary_path = tempfile.mkstemp(
|
||||
dir=directory,
|
||||
prefix=f".{os.path.basename(path)}.",
|
||||
suffix=".tmp",
|
||||
text=True,
|
||||
)
|
||||
try:
|
||||
with os.fdopen(descriptor, "w", encoding="utf-8", newline="\n") as file:
|
||||
file.write(content)
|
||||
os.chmod(temporary_path, mode)
|
||||
os.replace(temporary_path, path)
|
||||
except BaseException as write_error:
|
||||
try:
|
||||
os.remove(temporary_path)
|
||||
except FileNotFoundError:
|
||||
pass
|
||||
except OSError as cleanup_error:
|
||||
raise RuntimeError(
|
||||
f"Write {path} failed and temporary file cleanup also failed: {cleanup_error}"
|
||||
) from write_error
|
||||
raise
|
||||
|
||||
|
||||
def _write_json_atomic(path: str, value: JsonObject, mode: int) -> None:
|
||||
_write_text_atomic(path, json.dumps(value, indent=2) + "\n", mode)
|
||||
|
||||
|
||||
def _remove_file(path: str, label: str) -> None:
|
||||
try:
|
||||
os.remove(path)
|
||||
except FileNotFoundError:
|
||||
return
|
||||
except OSError as error:
|
||||
raise OSError(f"Remove {label} at {path}: {error}") from error
|
||||
|
||||
|
||||
def _build_command(script_path: str) -> str:
|
||||
arguments = [sys.executable, script_path]
|
||||
return subprocess.list2cmdline(arguments) if os.name == "nt" else shlex.join(arguments)
|
||||
|
||||
|
||||
def _hooks_object(settings: JsonObject, label: str) -> JsonObject:
|
||||
if "hooks" not in settings:
|
||||
return {}
|
||||
value = settings["hooks"]
|
||||
if not isinstance(value, dict):
|
||||
raise ValueError(f'{label} field "hooks" must be an object')
|
||||
return dict(value)
|
||||
|
||||
|
||||
def _hook_entries(hooks: JsonObject, event: str, label: str) -> List[JsonObject]:
|
||||
value = hooks.get(event)
|
||||
if value is None:
|
||||
return []
|
||||
if not isinstance(value, list) or not all(isinstance(item, dict) for item in value):
|
||||
raise ValueError(f'{label} field "hooks.{event}" must be an array of objects')
|
||||
return value
|
||||
|
||||
|
||||
def _is_managed_command(command: Any, script_name: str, agent_id: str = "") -> bool:
|
||||
if not isinstance(command, str):
|
||||
return False
|
||||
normalized = command.lower()
|
||||
if ".elydora" not in normalized or script_name not in normalized:
|
||||
return False
|
||||
return not agent_id or agent_id in command
|
||||
|
||||
|
||||
def _without_v2_hooks(
|
||||
entries: List[JsonObject], agent_id: str = ""
|
||||
) -> List[JsonObject]:
|
||||
return [
|
||||
hook
|
||||
for hook in entries
|
||||
if not _is_managed_command(hook.get("command"), "guard.py", agent_id)
|
||||
and not _is_managed_command(hook.get("command"), "hook.py", agent_id)
|
||||
]
|
||||
|
||||
|
||||
def _build_v2_hook(script_path: str) -> JsonObject:
|
||||
return {
|
||||
"matcher": "*",
|
||||
"command": _build_command(script_path),
|
||||
"timeout_ms": 5000,
|
||||
}
|
||||
|
||||
|
||||
def _is_owned_v2(settings: JsonObject, hooks: JsonObject) -> bool:
|
||||
config_keys = {"name", "description", "tools", "includeMcpJson", "hooks"}
|
||||
hook_keys = {"preToolUse", "postToolUse"}
|
||||
return (
|
||||
all(key in config_keys for key in settings)
|
||||
and settings.get("name") == V2_AGENT_NAME
|
||||
and settings.get("description") == V2_DESCRIPTION
|
||||
and settings.get("tools") == ["*"]
|
||||
and settings.get("includeMcpJson") is True
|
||||
and all(key in hook_keys for key in hooks)
|
||||
and all(isinstance(value, list) and not value for value in hooks.values())
|
||||
)
|
||||
|
||||
|
||||
def _v3_hooks(settings: JsonObject) -> List[JsonObject]:
|
||||
if "version" in settings and settings["version"] != "v1":
|
||||
raise ValueError('Kiro CLI v3 hooks config field "version" must be "v1"')
|
||||
if "hooks" not in settings:
|
||||
return []
|
||||
hooks = settings["hooks"]
|
||||
if not isinstance(hooks, list) or not all(isinstance(item, dict) for item in hooks):
|
||||
raise ValueError('Kiro CLI v3 hooks config field "hooks" must be an array of objects')
|
||||
return hooks
|
||||
|
||||
|
||||
def _v3_action_command(hook: JsonObject) -> Any:
|
||||
action = hook.get("action")
|
||||
return action.get("command") if isinstance(action, dict) else None
|
||||
|
||||
|
||||
def _is_managed_v3_hook(hook: JsonObject, agent_id: str = "") -> bool:
|
||||
if hook.get("name") == V3_GUARD_NAME:
|
||||
return _is_managed_command(_v3_action_command(hook), "guard.py", agent_id)
|
||||
if hook.get("name") == V3_AUDIT_NAME:
|
||||
return _is_managed_command(_v3_action_command(hook), "hook.py", agent_id)
|
||||
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 _build_v3_hook(
|
||||
name: str,
|
||||
description: str,
|
||||
trigger: str,
|
||||
script_path: str,
|
||||
) -> JsonObject:
|
||||
return {
|
||||
"name": name,
|
||||
"description": description,
|
||||
"trigger": trigger,
|
||||
"matcher": ".*",
|
||||
"action": {"type": "command", "command": _build_command(script_path)},
|
||||
"timeout": 5,
|
||||
"enabled": True,
|
||||
}
|
||||
|
||||
|
||||
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")
|
||||
def _find_v2_command(entries: List[JsonObject], script_name: str) -> Optional[str]:
|
||||
for hook in entries:
|
||||
command = hook.get("command")
|
||||
if _is_managed_command(command, script_name):
|
||||
return command
|
||||
return None
|
||||
|
||||
|
||||
def _find_v3_command(
|
||||
entries: List[JsonObject], name: str, script_name: str
|
||||
) -> Optional[str]:
|
||||
for hook in entries:
|
||||
command = _v3_action_command(hook)
|
||||
if hook.get("name") == name and _is_managed_command(command, script_name):
|
||||
return command
|
||||
return None
|
||||
|
||||
|
||||
def _regular_file_exists(path: str, label: str) -> bool:
|
||||
try:
|
||||
metadata = os.stat(path)
|
||||
except FileNotFoundError:
|
||||
return False
|
||||
except OSError as error:
|
||||
raise OSError(f"Read {label} at {path}: {error}") from error
|
||||
return stat.S_ISREG(metadata.st_mode)
|
||||
|
||||
|
||||
def _runtime_scripts_exist(contracts: List[HookContract]) -> bool:
|
||||
try:
|
||||
with os.scandir(ELYDORA_DIR) as iterator:
|
||||
entries = sorted(iterator, key=lambda entry: entry.name)
|
||||
except FileNotFoundError:
|
||||
return False
|
||||
except OSError as error:
|
||||
raise OSError(f"Read Elydora runtime directory at {ELYDORA_DIR}: {error}") from error
|
||||
|
||||
for entry in entries:
|
||||
try:
|
||||
is_directory = entry.is_dir(follow_symlinks=False)
|
||||
except OSError as error:
|
||||
raise OSError(f"Read Elydora runtime entry at {entry.path}: {error}") from error
|
||||
if not is_directory:
|
||||
continue
|
||||
|
||||
guard_path = os.path.join(entry.path, "guard.py")
|
||||
hook_path = os.path.join(entry.path, "hook.py")
|
||||
references_runtime = any(
|
||||
guard_path in guard and hook_path in audit for guard, audit, _path in contracts
|
||||
)
|
||||
if not references_runtime:
|
||||
continue
|
||||
|
||||
config_path = os.path.join(entry.path, "config.json")
|
||||
config = _read_json(config_path, "Elydora runtime config")
|
||||
if config is None or config.get("agent_name") != AGENT_KEY:
|
||||
continue
|
||||
return _regular_file_exists(
|
||||
guard_path, "Elydora guard runtime"
|
||||
) and _regular_file_exists(hook_path, "Elydora audit runtime")
|
||||
return False
|
||||
@@ -28,8 +28,8 @@ SUPPORTED_AGENTS: Dict[str, Dict[str, str]] = {
|
||||
},
|
||||
"kirocli": {
|
||||
"name": "Kiro CLI",
|
||||
"hook_event": "PostToolUse",
|
||||
"config_path": "~/.kiro/settings.json",
|
||||
"hook_event": "PreToolUse/PostToolUse",
|
||||
"config_path": "~/.kiro/hooks/elydora-audit.json",
|
||||
},
|
||||
"kiroide": {
|
||||
"name": "Kiro IDE",
|
||||
|
||||
Reference in new issue
Block a user