雷电芽衣
bc6983a62e
Split the 657-line operation service into eor, submit, and verify modules, add the AGENT_REVOKED and KEY_RETIRED codes the Backend already emits, and condense every multi-line comment. Mirror Node, Python, and Go at 2.1.0 with the deep-health 503, non-idempotent retry, rotation grace, and path-encoding fixes. Document which routes and response fields the self-hosted server does not provide. Claude-Session: https://claude.ai/code/session_01Agoc9bRVE9KL7mmWtey9Vq
371 lines
12 KiB
Python
371 lines
12 KiB
Python
"""Elydora CLI: install, uninstall, and inspect agent audit hooks."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import argparse
|
|
import json
|
|
import os
|
|
import sys
|
|
from typing import Dict, NamedTuple, NoReturn, Optional, Type
|
|
|
|
from ._cli_secrets import resolve_install_secrets
|
|
from ._version import __version__
|
|
from ._runtime_paths import (
|
|
ensure_private_directory,
|
|
require_physical_directory,
|
|
require_physical_file,
|
|
resolve_agent_directory,
|
|
runtime_root,
|
|
)
|
|
from .crypto import get_public_key_base64url
|
|
from .plugins.base import AgentPlugin, InstallConfig
|
|
from .plugins.registry import SUPPORTED_AGENTS, get_agent_names
|
|
from .plugins.guard_template import generate_guard_script
|
|
from .plugins.augment import AugmentPlugin
|
|
from .plugins.claudecode import ClaudeCodePlugin
|
|
from .plugins.codex import CodexPlugin
|
|
from .plugins.cline import ClinePlugin
|
|
from .plugins.copilot import CopilotPlugin
|
|
from .plugins.cursor import CursorPlugin
|
|
from .plugins.droid import DroidPlugin
|
|
from .plugins.gemini import GeminiPlugin
|
|
from .plugins.grok import GrokPlugin
|
|
from .plugins.kirocli import KiroCliPlugin
|
|
from .plugins.kiroide import KiroIdePlugin
|
|
from .plugins.kimi import KimiPlugin
|
|
from .plugins.letta import LettaPlugin
|
|
from .plugins.opencode import OpenCodePlugin
|
|
from .plugins.qwen import QwenPlugin
|
|
from .plugins._file_io import write_text_atomic
|
|
from .plugins._runtime_removal import (
|
|
commit_runtime_removal,
|
|
prepare_runtime_removal,
|
|
)
|
|
from .plugins._transaction import recover_pending_transactions
|
|
|
|
|
|
PLUGIN_MAP: Dict[str, Type[AgentPlugin]] = {
|
|
"augment": AugmentPlugin,
|
|
"claudecode": ClaudeCodePlugin,
|
|
"codex": CodexPlugin,
|
|
"cline": ClinePlugin,
|
|
"copilot": CopilotPlugin,
|
|
"cursor": CursorPlugin,
|
|
"droid": DroidPlugin,
|
|
"gemini": GeminiPlugin,
|
|
"grok": GrokPlugin,
|
|
"kirocli": KiroCliPlugin,
|
|
"kiroide": KiroIdePlugin,
|
|
"kimi": KimiPlugin,
|
|
"letta": LettaPlugin,
|
|
"opencode": OpenCodePlugin,
|
|
"qwen": QwenPlugin,
|
|
}
|
|
|
|
LEGACY_SECRET_OPTIONS = {
|
|
"--private_key": "--private_key_file",
|
|
"--token": "--token_file",
|
|
}
|
|
|
|
|
|
def _exit_with_error(message: str) -> NoReturn:
|
|
print(f"Error: {message}", file=sys.stderr)
|
|
raise SystemExit(1)
|
|
|
|
|
|
def _reject_legacy_secret_arguments(arguments: list[str]) -> None:
|
|
for argument in arguments:
|
|
option = argument.split("=", 1)[0]
|
|
replacement = LEGACY_SECRET_OPTIONS.get(option)
|
|
if replacement:
|
|
_exit_with_error(
|
|
f"{option} exposes credentials in process arguments; "
|
|
f"use {replacement} or hidden terminal input"
|
|
)
|
|
|
|
|
|
def _resolve_agent_directory_or_exit(agent_id: str) -> str:
|
|
try:
|
|
return resolve_agent_directory(runtime_root(), agent_id)
|
|
except ValueError as error:
|
|
_exit_with_error(str(error))
|
|
|
|
|
|
class _InstalledAgent(NamedTuple):
|
|
agent_id: str
|
|
agent_name: str
|
|
config_path: str
|
|
|
|
|
|
def _require_physical_directory_or_exit(path: str) -> bool:
|
|
try:
|
|
return require_physical_directory(path)
|
|
except OSError as error:
|
|
_exit_with_error(str(error))
|
|
|
|
|
|
def _require_physical_file_or_exit(path: str) -> bool:
|
|
try:
|
|
return require_physical_file(path)
|
|
except OSError as error:
|
|
_exit_with_error(str(error))
|
|
|
|
|
|
def _read_runtime_config_or_exit(config_path: str) -> Dict[str, object]:
|
|
try:
|
|
with open(config_path, "r", encoding="utf-8") as file:
|
|
config = json.load(file)
|
|
except (OSError, json.JSONDecodeError) as error:
|
|
_exit_with_error(f"Read agent config at {config_path}: {error}")
|
|
if not isinstance(config, dict):
|
|
_exit_with_error(f"Agent config at {config_path} must contain a JSON object")
|
|
return config
|
|
|
|
|
|
def _read_installed_agent_or_exit(agent_id: str) -> Optional[_InstalledAgent]:
|
|
agent_directory = _resolve_agent_directory_or_exit(agent_id)
|
|
if not _require_physical_directory_or_exit(agent_directory):
|
|
return None
|
|
|
|
config_path = os.path.join(agent_directory, "config.json")
|
|
if not _require_physical_file_or_exit(config_path):
|
|
return None
|
|
config = _read_runtime_config_or_exit(config_path)
|
|
stored_agent_id = config.get("agent_id")
|
|
stored_agent_name = config.get("agent_name")
|
|
if not isinstance(stored_agent_id, str) or not isinstance(
|
|
stored_agent_name, str
|
|
):
|
|
_exit_with_error(f"Agent config at {config_path} has an invalid identity")
|
|
if stored_agent_id != agent_id:
|
|
_exit_with_error(
|
|
f"Agent config at {config_path} crosses its runtime directory"
|
|
)
|
|
_resolve_agent_directory_or_exit(stored_agent_id)
|
|
return _InstalledAgent(agent_id, stored_agent_name, config_path)
|
|
|
|
|
|
def _discover_installed_agents_or_exit() -> list[_InstalledAgent]:
|
|
root = runtime_root()
|
|
if not _require_physical_directory_or_exit(root):
|
|
return []
|
|
|
|
installed_agents: list[_InstalledAgent] = []
|
|
try:
|
|
with os.scandir(root) as entries:
|
|
for entry in entries:
|
|
if entry.is_symlink():
|
|
_exit_with_error(
|
|
f"Agent runtime path is not a physical directory: {entry.path}"
|
|
)
|
|
if not entry.is_dir(follow_symlinks=False):
|
|
continue
|
|
installed_agent = _read_installed_agent_or_exit(entry.name)
|
|
if installed_agent is not None:
|
|
installed_agents.append(installed_agent)
|
|
except OSError as error:
|
|
_exit_with_error(f"Scan agent runtime root at {root}: {error}")
|
|
return installed_agents
|
|
|
|
|
|
def _get_plugin(agent_name: str) -> AgentPlugin:
|
|
cls = PLUGIN_MAP.get(agent_name)
|
|
if cls is None:
|
|
_exit_with_error(
|
|
f"Unknown agent {agent_name!r}. Supported agents: {', '.join(get_agent_names())}"
|
|
)
|
|
return cls()
|
|
|
|
|
|
def cmd_install(args: argparse.Namespace) -> None:
|
|
recover_pending_transactions()
|
|
agent_name: str = args.agent
|
|
agent_dir = _resolve_agent_directory_or_exit(args.agent_id)
|
|
plugin = _get_plugin(agent_name)
|
|
|
|
try:
|
|
secrets = resolve_install_secrets(
|
|
private_key_file=args.private_key_file,
|
|
token_file=args.token_file,
|
|
)
|
|
except (OSError, RuntimeError, ValueError) as error:
|
|
_exit_with_error(str(error))
|
|
|
|
try:
|
|
pub = get_public_key_base64url(secrets.private_key)
|
|
except Exception as exc:
|
|
_exit_with_error(f"Invalid private key: {exc}")
|
|
|
|
print(f"Verified key pair (public key: {pub[:16]}...)")
|
|
|
|
guard_script_path = os.path.join(agent_dir, "guard.py")
|
|
config: InstallConfig = {
|
|
"org_id": args.org_id,
|
|
"agent_id": args.agent_id,
|
|
"agent_name": agent_name,
|
|
"private_key": secrets.private_key,
|
|
"kid": args.kid,
|
|
"base_url": args.base_url,
|
|
"guard_script_path": guard_script_path,
|
|
}
|
|
if secrets.token:
|
|
config["token"] = secrets.token
|
|
plugin.preflight_install(config)
|
|
|
|
if not getattr(plugin, "manages_runtime_directories", False):
|
|
ensure_private_directory(runtime_root())
|
|
ensure_private_directory(agent_dir)
|
|
|
|
if not plugin.manages_guard_runtime:
|
|
guard_script = generate_guard_script(agent_name, args.agent_id)
|
|
write_text_atomic(
|
|
guard_script_path,
|
|
guard_script,
|
|
0o700,
|
|
"Elydora guard runtime",
|
|
)
|
|
|
|
plugin.install(config)
|
|
print(f" Guard script: {guard_script_path}")
|
|
|
|
|
|
def cmd_uninstall(args: argparse.Namespace) -> None:
|
|
recover_pending_transactions()
|
|
explicit_agent_id = getattr(args, "agent_id", None)
|
|
elydora_dir = runtime_root()
|
|
agent_id = explicit_agent_id
|
|
if agent_id:
|
|
agent_dir = _resolve_agent_directory_or_exit(agent_id)
|
|
runtime_root_exists = _require_physical_directory_or_exit(elydora_dir)
|
|
agent_directory_exists = (
|
|
_require_physical_directory_or_exit(agent_dir)
|
|
if runtime_root_exists
|
|
else False
|
|
)
|
|
if agent_directory_exists:
|
|
installed_agent = _read_installed_agent_or_exit(agent_id)
|
|
if installed_agent and installed_agent.agent_name != args.agent:
|
|
_exit_with_error(
|
|
f"Agent runtime {agent_id} belongs to "
|
|
f"{installed_agent.agent_name}, not {args.agent}"
|
|
)
|
|
else:
|
|
matches = [
|
|
installed_agent
|
|
for installed_agent in _discover_installed_agents_or_exit()
|
|
if installed_agent.agent_name == args.agent
|
|
]
|
|
if not matches:
|
|
_exit_with_error(
|
|
f"No installed agent found for {args.agent!r}; pass --agent_id explicitly"
|
|
)
|
|
if len(matches) > 1:
|
|
_exit_with_error(
|
|
f"Multiple installed agents found for {args.agent!r}; "
|
|
"pass --agent_id explicitly"
|
|
)
|
|
agent_id = matches[0].agent_id
|
|
agent_dir = _resolve_agent_directory_or_exit(agent_id)
|
|
agent_directory_exists = True
|
|
|
|
plugin = _get_plugin(args.agent)
|
|
manages_runtime_removal = getattr(
|
|
plugin, "manages_runtime_removal", False
|
|
)
|
|
runtime_removal = (
|
|
None
|
|
if not agent_directory_exists or manages_runtime_removal
|
|
else prepare_runtime_removal(agent_id, args.agent)
|
|
)
|
|
plugin.uninstall(agent_id=agent_id)
|
|
|
|
if runtime_removal is not None:
|
|
commit_runtime_removal(runtime_removal)
|
|
print(f" Removed agent directory: {agent_dir}")
|
|
|
|
|
|
def cmd_status(args: argparse.Namespace) -> None:
|
|
recover_pending_transactions()
|
|
print("Elydora Agent Hook Status")
|
|
print("=" * 40)
|
|
for name in get_agent_names():
|
|
plugin = PLUGIN_MAP[name]()
|
|
st = plugin.status()
|
|
marker = "[installed]" if st["installed"] else "[not installed]"
|
|
display = SUPPORTED_AGENTS[name]["name"]
|
|
print(f" {display:20s} {marker:16s} {st['details']}")
|
|
|
|
|
|
def cmd_agents(args: argparse.Namespace) -> None:
|
|
print("Supported agents:")
|
|
for name in get_agent_names():
|
|
info = SUPPORTED_AGENTS[name]
|
|
print(f" {name:15s} {info['name']:20s} (hook: {info['hook_event']})")
|
|
|
|
|
|
def build_parser() -> argparse.ArgumentParser:
|
|
parser = argparse.ArgumentParser(
|
|
prog="elydora",
|
|
description="Elydora — tamper-evident audit trail for AI agents",
|
|
)
|
|
parser.add_argument(
|
|
"--version", action="version", version=f"%(prog)s {__version__}"
|
|
)
|
|
|
|
subparsers = parser.add_subparsers(dest="command", help="Available commands")
|
|
|
|
install_parser = subparsers.add_parser("install", help="Install audit hook for an agent")
|
|
install_parser.add_argument("--agent", required=True, help="Agent name (e.g. claudecode, cursor)")
|
|
install_parser.add_argument("--org_id", required=True, help="Organization ID")
|
|
install_parser.add_argument("--agent_id", required=True, help="Agent ID")
|
|
install_parser.add_argument(
|
|
"--private_key_file",
|
|
default=None,
|
|
help="Owner-only file containing the Ed25519 private key seed",
|
|
)
|
|
install_parser.add_argument("--kid", required=True, help="Key ID")
|
|
install_parser.add_argument(
|
|
"--token_file",
|
|
default=None,
|
|
help="Owner-only file containing the optional API token",
|
|
)
|
|
install_parser.add_argument("--base_url", default="https://api.elydora.com", help="API base URL")
|
|
|
|
uninstall_parser = subparsers.add_parser("uninstall", help="Uninstall audit hook for an agent")
|
|
uninstall_parser.add_argument("--agent", required=True, help="Agent name")
|
|
uninstall_parser.add_argument("--agent_id", default=None, help="Agent ID (if omitted, scans config files for matching agent name)")
|
|
|
|
subparsers.add_parser("status", help="Show installation status of all agents")
|
|
|
|
subparsers.add_parser("agents", help="List supported agents")
|
|
|
|
return parser
|
|
|
|
|
|
def main() -> None:
|
|
_reject_legacy_secret_arguments(sys.argv[1:])
|
|
parser = build_parser()
|
|
args = parser.parse_args()
|
|
|
|
if args.command is None:
|
|
parser.print_help()
|
|
sys.exit(0)
|
|
|
|
handlers = {
|
|
"install": cmd_install,
|
|
"uninstall": cmd_uninstall,
|
|
"status": cmd_status,
|
|
"agents": cmd_agents,
|
|
}
|
|
|
|
handler = handlers.get(args.command)
|
|
if handler is None:
|
|
parser.print_help()
|
|
sys.exit(1)
|
|
|
|
handler(args)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
main()
|