From b42b670c2ae2ea39ded3952c7a403eccbbb675da Mon Sep 17 00:00:00 2001 From: ZacharyZhang Date: Sat, 28 Feb 2026 18:03:09 -0500 Subject: [PATCH] Initial commit: Elydora Python SDK Python SDK for Elydora agent operations, cryptographic signing, and audit trail integration. Co-Authored-By: Claude Opus 4.6 --- .gitignore | 16 ++ README.md | 55 +++++ elydora/__init__.py | 86 +++++++ elydora/async_client.py | 359 ++++++++++++++++++++++++++++ elydora/cli.py | 182 +++++++++++++++ elydora/client.py | 347 +++++++++++++++++++++++++++ elydora/crypto.py | 167 +++++++++++++ elydora/errors.py | 30 +++ elydora/plugins/__init__.py | 1 + elydora/plugins/augment.py | 160 +++++++++++++ elydora/plugins/base.py | 44 ++++ elydora/plugins/claudecode.py | 167 +++++++++++++ elydora/plugins/cursor.py | 143 ++++++++++++ elydora/plugins/gemini.py | 160 +++++++++++++ elydora/plugins/hook_template.py | 390 +++++++++++++++++++++++++++++++ elydora/plugins/kiro.py | 113 +++++++++ elydora/plugins/opencode.py | 266 +++++++++++++++++++++ elydora/plugins/registry.py | 44 ++++ elydora/types.py | 293 +++++++++++++++++++++++ elydora/utils.py | 59 +++++ pyproject.toml | 42 ++++ setup.py | 3 + 22 files changed, 3127 insertions(+) create mode 100644 .gitignore create mode 100644 README.md create mode 100644 elydora/__init__.py create mode 100644 elydora/async_client.py create mode 100644 elydora/cli.py create mode 100644 elydora/client.py create mode 100644 elydora/crypto.py create mode 100644 elydora/errors.py create mode 100644 elydora/plugins/__init__.py create mode 100644 elydora/plugins/augment.py create mode 100644 elydora/plugins/base.py create mode 100644 elydora/plugins/claudecode.py create mode 100644 elydora/plugins/cursor.py create mode 100644 elydora/plugins/gemini.py create mode 100644 elydora/plugins/hook_template.py create mode 100644 elydora/plugins/kiro.py create mode 100644 elydora/plugins/opencode.py create mode 100644 elydora/plugins/registry.py create mode 100644 elydora/types.py create mode 100644 elydora/utils.py create mode 100644 pyproject.toml create mode 100644 setup.py diff --git a/.gitignore b/.gitignore new file mode 100644 index 0000000..ec50188 --- /dev/null +++ b/.gitignore @@ -0,0 +1,16 @@ +__pycache__/ +*.py[cod] +*$py.class +*.so +dist/ +build/ +*.egg-info/ +*.egg +.eggs/ +.venv/ +venv/ +env/ +.mypy_cache/ +.pytest_cache/ +.tox/ +*.whl diff --git a/README.md b/README.md new file mode 100644 index 0000000..9640eca --- /dev/null +++ b/README.md @@ -0,0 +1,55 @@ +# Elydora Python SDK + +Python SDK for the Elydora tamper-evident audit platform. + +## Installation + +```bash +pip install elydora +``` + +## Quick Start + +```python +from elydora import ElydoraClient + +client = ElydoraClient( + org_id="org_...", + agent_id="agent_...", + private_key="", + base_url="https://api.elydora.com", + token="", +) + +# Create and submit an operation +eor = client.create_operation( + operation_type="data.access", + subject={"user_id": "u123"}, + action={"type": "read", "resource": "documents"}, + payload={"document_id": "doc_456"}, +) +response = client.submit_operation(eor) +print(response["receipt"]["receipt_id"]) +``` + +## Async Usage + +```python +from elydora import AsyncElydoraClient + +async def main(): + client = AsyncElydoraClient( + org_id="org_...", + agent_id="agent_...", + private_key="", + base_url="https://api.elydora.com", + token="", + ) + eor = client.create_operation( + operation_type="data.access", + subject={"user_id": "u123"}, + action={"type": "read", "resource": "documents"}, + ) + response = await client.submit_operation(eor) + await client.close() +``` diff --git a/elydora/__init__.py b/elydora/__init__.py new file mode 100644 index 0000000..3b715fa --- /dev/null +++ b/elydora/__init__.py @@ -0,0 +1,86 @@ +"""Elydora Python SDK — tamper-evident audit trail for AI agents.""" + +from .client import ElydoraClient +from .async_client import AsyncElydoraClient +from .errors import ElydoraError +from .crypto import ( + compute_chain_hash, + compute_payload_hash, + jcs_canonicalize, + sha256_base64url, + sign_ed25519, + sign_eor, + get_public_key_base64url, +) +from .utils import base64url_encode, base64url_decode, generate_nonce, generate_uuidv7 +from .types import ( + EOR, + EAR, + Agent, + AgentKey, + Operation, + Receipt, + Epoch, + Export, + Organization, + User, + RegisterAgentRequest, + RegisterAgentResponse, + GetAgentResponse, + SubmitOperationResponse, + GetOperationResponse, + VerifyOperationResponse, + AuditQueryResponse, + GetEpochResponse, + ListEpochsResponse, + CreateExportResponse, + GetExportResponse, + ListExportsResponse, + JWKSResponse, + AuthRegisterResponse, + AuthLoginResponse, +) + +__version__ = "1.0.0" + +__all__ = [ + "ElydoraClient", + "AsyncElydoraClient", + "ElydoraError", + "compute_chain_hash", + "compute_payload_hash", + "jcs_canonicalize", + "sha256_base64url", + "sign_ed25519", + "sign_eor", + "get_public_key_base64url", + "base64url_encode", + "base64url_decode", + "generate_nonce", + "generate_uuidv7", + "EOR", + "EAR", + "Agent", + "AgentKey", + "Operation", + "Receipt", + "Epoch", + "Export", + "Organization", + "User", + "RegisterAgentRequest", + "RegisterAgentResponse", + "GetAgentResponse", + "SubmitOperationResponse", + "GetOperationResponse", + "VerifyOperationResponse", + "AuditQueryResponse", + "GetEpochResponse", + "ListEpochsResponse", + "CreateExportResponse", + "GetExportResponse", + "ListExportsResponse", + "JWKSResponse", + "AuthRegisterResponse", + "AuthLoginResponse", +] diff --git a/elydora/async_client.py b/elydora/async_client.py new file mode 100644 index 0000000..7bf4e4b --- /dev/null +++ b/elydora/async_client.py @@ -0,0 +1,359 @@ +"""Asynchronous Elydora client using aiohttp.""" + +from __future__ import annotations + +import time +import asyncio +from typing import Any, Dict, List, Optional, Union + +import aiohttp + +from .crypto import compute_chain_hash, compute_payload_hash, sign_eor +from .errors import ElydoraError +from .types import ( + AuditQueryResponse, + AuthLoginResponse, + AuthRegisterResponse, + CreateExportResponse, + EOR, + GetAgentResponse, + GetEpochResponse, + GetExportResponse, + GetOperationResponse, + JWKSResponse, + ListEpochsResponse, + ListExportsResponse, + RegisterAgentRequest, + RegisterAgentResponse, + SubmitOperationResponse, + VerifyOperationResponse, +) +from .utils import generate_nonce, generate_uuidv7 + + +class AsyncElydoraClient: + """Asynchronous client for the Elydora API. + + Args: + org_id: Organization identifier. + agent_id: Agent identifier. + private_key: Base64url-encoded 32-byte Ed25519 private key seed. + base_url: API base URL. + ttl_ms: Time-to-live for operations in milliseconds. + max_retries: Maximum number of retries on transient failures. + token: Optional JWT bearer token for authenticated endpoints. + """ + + def __init__( + self, + org_id: str, + agent_id: str, + private_key: str, + *, + base_url: str = "https://api.elydora.com", + ttl_ms: int = 30000, + max_retries: int = 3, + token: Optional[str] = None, + ) -> None: + self.org_id = org_id + self.agent_id = agent_id + self.private_key = private_key + self.base_url = base_url.rstrip("/") + self.ttl_ms = ttl_ms + self.max_retries = max_retries + self.token = token + + self._prev_chain_hash = "" + self._kid = "" + self._session: Optional[aiohttp.ClientSession] = None + + def set_kid(self, kid: str) -> None: + """Set the key ID used for signing operations.""" + self._kid = kid + + def set_prev_chain_hash(self, prev_chain_hash: str) -> None: + """Set the previous chain hash for the next operation.""" + self._prev_chain_hash = prev_chain_hash + + async def _get_session(self) -> aiohttp.ClientSession: + if self._session is None or self._session.closed: + self._session = aiohttp.ClientSession() + return self._session + + async def close(self) -> None: + """Close the underlying HTTP session.""" + if self._session and not self._session.closed: + await self._session.close() + + # ----------------------------------------------------------------- + # Internal HTTP helpers + # ----------------------------------------------------------------- + + def _headers(self) -> Dict[str, str]: + headers: Dict[str, str] = {"Content-Type": "application/json"} + if self.token: + headers["Authorization"] = f"Bearer {self.token}" + return headers + + async def _request( + self, + method: str, + path: str, + *, + json_body: Any = None, + params: Optional[Dict[str, str]] = None, + headers: Optional[Dict[str, str]] = None, + ) -> Any: + url = f"{self.base_url}{path}" + hdrs = headers or self._headers() + session = await self._get_session() + + last_exc: Optional[Exception] = None + for attempt in range(self.max_retries): + try: + async with session.request( + method, url, json=json_body, params=params, headers=hdrs, timeout=aiohttp.ClientTimeout(total=30) + ) as resp: + return await self._handle_response(resp) + except ElydoraError: + raise + except (aiohttp.ClientConnectionError, asyncio.TimeoutError) as exc: + last_exc = exc + if attempt < self.max_retries - 1: + await asyncio.sleep(min(2 ** attempt, 8)) + continue + raise + + raise last_exc # type: ignore[misc] + + @staticmethod + async def _handle_response(resp: aiohttp.ClientResponse) -> Any: + if resp.status >= 400: + try: + body = await resp.json() + except Exception: + text = await resp.text() + raise ElydoraError( + code="INTERNAL_ERROR", + message=text or "Unknown error", + status_code=resp.status, + ) + err = body.get("error", {}) + raise ElydoraError( + code=err.get("code", "INTERNAL_ERROR"), + message=err.get("message", ""), + request_id=err.get("request_id", ""), + details=err.get("details"), + status_code=resp.status, + ) + if resp.status == 204: + return None + return await resp.json() + + # ----------------------------------------------------------------- + # Auth (static methods) + # ----------------------------------------------------------------- + + @staticmethod + async def register( + base_url: str, + email: str, + password: str, + display_name: Optional[str] = None, + org_name: Optional[str] = None, + ) -> AuthRegisterResponse: + """Register a new user and organization.""" + url = f"{base_url.rstrip('/')}/v1/auth/register" + body: Dict[str, Any] = {"email": email, "password": password} + if display_name is not None: + body["display_name"] = display_name + if org_name is not None: + body["org_name"] = org_name + async with aiohttp.ClientSession() as session: + async with session.post( + url, json=body, headers={"Content-Type": "application/json"}, timeout=aiohttp.ClientTimeout(total=30) + ) as resp: + return await AsyncElydoraClient._handle_response(resp) + + @staticmethod + async def login(base_url: str, email: str, password: str) -> AuthLoginResponse: + """Authenticate and receive a JWT.""" + url = f"{base_url.rstrip('/')}/v1/auth/login" + body = {"email": email, "password": password} + async with aiohttp.ClientSession() as session: + async with session.post( + url, json=body, headers={"Content-Type": "application/json"}, timeout=aiohttp.ClientTimeout(total=30) + ) as resp: + return await AsyncElydoraClient._handle_response(resp) + + # ----------------------------------------------------------------- + # Agent management + # ----------------------------------------------------------------- + + async def register_agent(self, request: RegisterAgentRequest) -> RegisterAgentResponse: + """Register a new agent with the organization.""" + return await self._request("POST", "/v1/agents/register", json_body=request) + + async def get_agent(self, agent_id: str) -> GetAgentResponse: + """Retrieve agent details and keys.""" + return await self._request("GET", f"/v1/agents/{agent_id}") + + async def freeze_agent(self, agent_id: str, reason: str) -> None: + """Freeze an agent.""" + await self._request("POST", f"/v1/agents/{agent_id}/freeze", json_body={"reason": reason}) + + async def revoke_key(self, agent_id: str, kid: str, reason: str) -> None: + """Revoke an agent's key.""" + await self._request( + "POST", + f"/v1/agents/{agent_id}/revoke", + json_body={"kid": kid, "reason": reason}, + ) + + # ----------------------------------------------------------------- + # Operations (CORE) + # ----------------------------------------------------------------- + + def create_operation( + self, + operation_type: str, + subject: Dict[str, Any], + action: Dict[str, Any], + payload: Union[Dict[str, Any], str, None] = None, + ) -> EOR: + """Build and sign an Elydora Operation Record (EOR). + + This is synchronous because it only performs local crypto operations. + Call submit_operation() with the returned EOR to submit it. + """ + operation_id = generate_uuidv7() + issued_at = int(time.time() * 1000) + nonce = generate_nonce() + payload_hash = compute_payload_hash(payload) + chain_hash = compute_chain_hash( + self._prev_chain_hash, payload_hash, operation_id, issued_at + ) + + eor: Dict[str, Any] = { + "op_version": "1.0", + "operation_id": operation_id, + "org_id": self.org_id, + "agent_id": self.agent_id, + "issued_at": issued_at, + "ttl_ms": self.ttl_ms, + "nonce": nonce, + "operation_type": operation_type, + "subject": subject, + "action": action, + "payload": payload, + "payload_hash": payload_hash, + "prev_chain_hash": self._prev_chain_hash, + "agent_pubkey_kid": self._kid, + "signature": "", + } + + eor["signature"] = sign_eor(eor, self.private_key) + + self._prev_chain_hash = chain_hash + + return eor # type: ignore[return-value] + + async def submit_operation(self, eor: EOR) -> SubmitOperationResponse: + """Submit a signed EOR to the server.""" + return await self._request("POST", "/v1/operations", json_body=eor) + + async def get_operation(self, operation_id: str) -> GetOperationResponse: + """Retrieve an operation by ID.""" + return await self._request("GET", f"/v1/operations/{operation_id}") + + async def verify_operation(self, operation_id: str) -> VerifyOperationResponse: + """Verify an operation's integrity.""" + return await self._request("POST", f"/v1/operations/{operation_id}/verify") + + # ----------------------------------------------------------------- + # Audit + # ----------------------------------------------------------------- + + async def query_audit( + self, + *, + org_id: Optional[str] = None, + agent_id: Optional[str] = None, + operation_type: Optional[str] = None, + start_time: Optional[int] = None, + end_time: Optional[int] = None, + cursor: Optional[str] = None, + limit: Optional[int] = None, + ) -> AuditQueryResponse: + """Query the tamper-evident audit log.""" + body: Dict[str, Any] = {} + if org_id is not None: + body["org_id"] = org_id + if agent_id is not None: + body["agent_id"] = agent_id + if operation_type is not None: + body["operation_type"] = operation_type + if start_time is not None: + body["start_time"] = start_time + if end_time is not None: + body["end_time"] = end_time + if cursor is not None: + body["cursor"] = cursor + if limit is not None: + body["limit"] = limit + return await self._request("POST", "/v1/audit/query", json_body=body) + + # ----------------------------------------------------------------- + # Epochs + # ----------------------------------------------------------------- + + async def list_epochs(self) -> ListEpochsResponse: + """List all epochs for the organization.""" + return await self._request("GET", "/v1/epochs") + + async def get_epoch(self, epoch_id: str) -> GetEpochResponse: + """Retrieve an epoch root record.""" + return await self._request("GET", f"/v1/epochs/{epoch_id}") + + # ----------------------------------------------------------------- + # Exports + # ----------------------------------------------------------------- + + async def create_export( + self, + start_time: int, + end_time: int, + format: str = "json", + agent_id: Optional[str] = None, + operation_type: Optional[str] = None, + ) -> CreateExportResponse: + """Create a compliance export job.""" + body: Dict[str, Any] = { + "start_time": start_time, + "end_time": end_time, + "format": format, + } + if agent_id is not None: + body["agent_id"] = agent_id + if operation_type is not None: + body["operation_type"] = operation_type + return await self._request("POST", "/v1/exports", json_body=body) + + async def list_exports(self) -> ListExportsResponse: + """List all exports for the organization.""" + return await self._request("GET", "/v1/exports") + + async def get_export(self, export_id: str) -> GetExportResponse: + """Retrieve export status and download URL.""" + return await self._request("GET", f"/v1/exports/{export_id}") + + # ----------------------------------------------------------------- + # JWKS + # ----------------------------------------------------------------- + + async def get_jwks(self) -> JWKSResponse: + """Retrieve the platform JWKS (public, no auth required).""" + url = f"{self.base_url}/.well-known/elydora/jwks.json" + session = await self._get_session() + async with session.get(url, timeout=aiohttp.ClientTimeout(total=30)) as resp: + return await self._handle_response(resp) diff --git a/elydora/cli.py b/elydora/cli.py new file mode 100644 index 0000000..a061204 --- /dev/null +++ b/elydora/cli.py @@ -0,0 +1,182 @@ +"""Elydora CLI — install, uninstall, and manage agent audit hooks. + +Entry point: ``elydora`` console script (see pyproject.toml). +Uses only stdlib argparse — zero external dependencies for the CLI itself. +""" + +from __future__ import annotations + +import argparse +import os +import sys +from typing import Dict, Type + +from .crypto import get_public_key_base64url +from .plugins.base import AgentPlugin, InstallConfig +from .plugins.registry import SUPPORTED_AGENTS, get_agent_names +from .plugins.hook_template import generate_guard_script +from .plugins.claudecode import ClaudeCodePlugin +from .plugins.cursor import CursorPlugin +from .plugins.gemini import GeminiPlugin +from .plugins.augment import AugmentPlugin +from .plugins.kiro import KiroPlugin +from .plugins.opencode import OpenCodePlugin + + +PLUGIN_MAP: Dict[str, Type[AgentPlugin]] = { + "claudecode": ClaudeCodePlugin, + "cursor": CursorPlugin, + "gemini": GeminiPlugin, + "augment": AugmentPlugin, + "kiro": KiroPlugin, + "opencode": OpenCodePlugin, +} + + +def _get_plugin(agent_name: str) -> AgentPlugin: + """Instantiate the plugin for the given agent name.""" + cls = PLUGIN_MAP.get(agent_name) + if cls is None: + print(f"Error: Unknown agent '{agent_name}'.", file=sys.stderr) + print(f"Supported agents: {', '.join(get_agent_names())}", file=sys.stderr) + sys.exit(1) + return cls() + + +def cmd_install(args: argparse.Namespace) -> None: + """Handle the 'install' subcommand.""" + agent_name: str = args.agent + + # Derive public key to verify the private key is valid + try: + pub = get_public_key_base64url(args.private_key) + except Exception as exc: + print(f"Error: Invalid private key — {exc}", file=sys.stderr) + sys.exit(1) + + print(f"Verified key pair (public key: {pub[:16]}...)") + + # Generate and write the guard script + guard_dir = os.path.join(os.path.expanduser("~"), ".elydora", "hooks") + os.makedirs(guard_dir, exist_ok=True) + guard_script_path = os.path.join(guard_dir, f"{agent_name}-guard.py") + guard_script = generate_guard_script(agent_name) + with open(guard_script_path, "w", encoding="utf-8") as f: + f.write(guard_script) + try: + import stat + os.chmod(guard_script_path, os.stat(guard_script_path).st_mode | stat.S_IEXEC) + except Exception: + pass # chmod may fail on Windows + print(f" Guard script: {guard_script_path}") + + config: InstallConfig = { + "org_id": args.org_id, + "agent_id": args.agent_id, + "private_key": args.private_key, + "kid": args.kid, + "base_url": args.base_url, + "guard_script_path": guard_script_path, + } + if args.token: + config["token"] = args.token + + plugin = _get_plugin(agent_name) + plugin.install(config) + + +def cmd_uninstall(args: argparse.Namespace) -> None: + """Handle the 'uninstall' subcommand.""" + plugin = _get_plugin(args.agent) + plugin.uninstall() + + # Remove the guard script + guard_script_path = os.path.join( + os.path.expanduser("~"), ".elydora", "hooks", f"{args.agent}-guard.py" + ) + if os.path.exists(guard_script_path): + os.remove(guard_script_path) + print(f" Removed guard script: {guard_script_path}") + + +def cmd_status(args: argparse.Namespace) -> None: + """Handle the 'status' subcommand.""" + 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: + """Handle the 'agents' subcommand.""" + 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: + """Build the argparse parser.""" + parser = argparse.ArgumentParser( + prog="elydora", + description="Elydora — tamper-evident audit trail for AI agents", + ) + parser.add_argument( + "--version", action="version", version="%(prog)s 1.0.0" + ) + + subparsers = parser.add_subparsers(dest="command", help="Available commands") + + # install + 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", required=True, help="Base64url-encoded Ed25519 private key seed") + install_parser.add_argument("--kid", required=True, help="Key ID") + install_parser.add_argument("--token", default="", help="JWT bearer token (optional)") + install_parser.add_argument("--base_url", default="https://api.elydora.com", help="API base URL") + + # uninstall + uninstall_parser = subparsers.add_parser("uninstall", help="Uninstall audit hook for an agent") + uninstall_parser.add_argument("--agent", required=True, help="Agent name") + + # status + subparsers.add_parser("status", help="Show installation status of all agents") + + # agents + subparsers.add_parser("agents", help="List supported agents") + + return parser + + +def main() -> None: + """CLI entry point.""" + 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() diff --git a/elydora/client.py b/elydora/client.py new file mode 100644 index 0000000..f4bd6be --- /dev/null +++ b/elydora/client.py @@ -0,0 +1,347 @@ +"""Synchronous Elydora client using the requests library.""" + +from __future__ import annotations + +import time +from typing import Any, Dict, List, Optional, Union + +import requests + +from .crypto import compute_chain_hash, compute_payload_hash, sign_eor +from .errors import ElydoraError +from .types import ( + AuditQueryResponse, + AuthLoginResponse, + AuthRegisterResponse, + CreateExportResponse, + EOR, + GetAgentResponse, + GetEpochResponse, + GetExportResponse, + GetOperationResponse, + JWKSResponse, + ListEpochsResponse, + ListExportsResponse, + RegisterAgentRequest, + RegisterAgentResponse, + SubmitOperationResponse, + VerifyOperationResponse, +) +from .utils import generate_nonce, generate_uuidv7 + + +class ElydoraClient: + """Synchronous client for the Elydora API. + + Args: + org_id: Organization identifier. + agent_id: Agent identifier. + private_key: Base64url-encoded 32-byte Ed25519 private key seed. + base_url: API base URL. + ttl_ms: Time-to-live for operations in milliseconds. + max_retries: Maximum number of retries on transient failures. + token: Optional JWT bearer token for authenticated endpoints. + """ + + def __init__( + self, + org_id: str, + agent_id: str, + private_key: str, + *, + base_url: str = "https://api.elydora.com", + ttl_ms: int = 30000, + max_retries: int = 3, + token: Optional[str] = None, + ) -> None: + self.org_id = org_id + self.agent_id = agent_id + self.private_key = private_key + self.base_url = base_url.rstrip("/") + self.ttl_ms = ttl_ms + self.max_retries = max_retries + self.token = token + + self._prev_chain_hash = "" + self._kid = "" + self._session = requests.Session() + + def set_kid(self, kid: str) -> None: + """Set the key ID used for signing operations.""" + self._kid = kid + + def set_prev_chain_hash(self, prev_chain_hash: str) -> None: + """Set the previous chain hash for the next operation.""" + self._prev_chain_hash = prev_chain_hash + + # ----------------------------------------------------------------- + # Internal HTTP helpers + # ----------------------------------------------------------------- + + def _headers(self) -> Dict[str, str]: + headers: Dict[str, str] = {"Content-Type": "application/json"} + if self.token: + headers["Authorization"] = f"Bearer {self.token}" + return headers + + def _request( + self, + method: str, + path: str, + *, + json_body: Any = None, + params: Optional[Dict[str, str]] = None, + headers: Optional[Dict[str, str]] = None, + ) -> Any: + url = f"{self.base_url}{path}" + hdrs = headers or self._headers() + + last_exc: Optional[Exception] = None + for attempt in range(self.max_retries): + try: + resp = self._session.request( + method, url, json=json_body, params=params, headers=hdrs, timeout=30 + ) + return self._handle_response(resp) + except ElydoraError: + raise + except requests.exceptions.ConnectionError as exc: + last_exc = exc + if attempt < self.max_retries - 1: + time.sleep(min(2 ** attempt, 8)) + continue + raise + except requests.exceptions.Timeout as exc: + last_exc = exc + if attempt < self.max_retries - 1: + time.sleep(min(2 ** attempt, 8)) + continue + raise + + raise last_exc # type: ignore[misc] + + @staticmethod + def _handle_response(resp: requests.Response) -> Any: + if resp.status_code >= 400: + try: + body = resp.json() + except Exception: + raise ElydoraError( + code="INTERNAL_ERROR", + message=resp.text or "Unknown error", + status_code=resp.status_code, + ) + err = body.get("error", {}) + raise ElydoraError( + code=err.get("code", "INTERNAL_ERROR"), + message=err.get("message", resp.text), + request_id=err.get("request_id", ""), + details=err.get("details"), + status_code=resp.status_code, + ) + if resp.status_code == 204: + return None + return resp.json() + + # ----------------------------------------------------------------- + # Auth (static methods) + # ----------------------------------------------------------------- + + @staticmethod + def register( + base_url: str, + email: str, + password: str, + display_name: Optional[str] = None, + org_name: Optional[str] = None, + ) -> AuthRegisterResponse: + """Register a new user and organization.""" + url = f"{base_url.rstrip('/')}/v1/auth/register" + body: Dict[str, Any] = {"email": email, "password": password} + if display_name is not None: + body["display_name"] = display_name + if org_name is not None: + body["org_name"] = org_name + resp = requests.post(url, json=body, headers={"Content-Type": "application/json"}, timeout=30) + return ElydoraClient._handle_response(resp) + + @staticmethod + def login(base_url: str, email: str, password: str) -> AuthLoginResponse: + """Authenticate and receive a JWT.""" + url = f"{base_url.rstrip('/')}/v1/auth/login" + body = {"email": email, "password": password} + resp = requests.post(url, json=body, headers={"Content-Type": "application/json"}, timeout=30) + return ElydoraClient._handle_response(resp) + + # ----------------------------------------------------------------- + # Agent management + # ----------------------------------------------------------------- + + def register_agent(self, request: RegisterAgentRequest) -> RegisterAgentResponse: + """Register a new agent with the organization.""" + return self._request("POST", "/v1/agents/register", json_body=request) + + def get_agent(self, agent_id: str) -> GetAgentResponse: + """Retrieve agent details and keys.""" + return self._request("GET", f"/v1/agents/{agent_id}") + + def freeze_agent(self, agent_id: str, reason: str) -> None: + """Freeze an agent.""" + self._request("POST", f"/v1/agents/{agent_id}/freeze", json_body={"reason": reason}) + + def revoke_key(self, agent_id: str, kid: str, reason: str) -> None: + """Revoke an agent's key.""" + self._request( + "POST", + f"/v1/agents/{agent_id}/revoke", + json_body={"kid": kid, "reason": reason}, + ) + + # ----------------------------------------------------------------- + # Operations (CORE) + # ----------------------------------------------------------------- + + def create_operation( + self, + operation_type: str, + subject: Dict[str, Any], + action: Dict[str, Any], + payload: Union[Dict[str, Any], str, None] = None, + ) -> EOR: + """Build and sign an Elydora Operation Record (EOR). + + This does NOT submit the operation to the server. Call submit_operation() + with the returned EOR to submit it. + """ + operation_id = generate_uuidv7() + issued_at = int(time.time() * 1000) + nonce = generate_nonce() + payload_hash = compute_payload_hash(payload) + chain_hash = compute_chain_hash( + self._prev_chain_hash, payload_hash, operation_id, issued_at + ) + + eor: Dict[str, Any] = { + "op_version": "1.0", + "operation_id": operation_id, + "org_id": self.org_id, + "agent_id": self.agent_id, + "issued_at": issued_at, + "ttl_ms": self.ttl_ms, + "nonce": nonce, + "operation_type": operation_type, + "subject": subject, + "action": action, + "payload": payload, + "payload_hash": payload_hash, + "prev_chain_hash": self._prev_chain_hash, + "agent_pubkey_kid": self._kid, + "signature": "", + } + + # Sign the EOR (signature field excluded from canonical form) + eor["signature"] = sign_eor(eor, self.private_key) + + # Update chain state + self._prev_chain_hash = chain_hash + + return eor # type: ignore[return-value] + + def submit_operation(self, eor: EOR) -> SubmitOperationResponse: + """Submit a signed EOR to the server.""" + return self._request("POST", "/v1/operations", json_body=eor) + + def get_operation(self, operation_id: str) -> GetOperationResponse: + """Retrieve an operation by ID.""" + return self._request("GET", f"/v1/operations/{operation_id}") + + def verify_operation(self, operation_id: str) -> VerifyOperationResponse: + """Verify an operation's integrity.""" + return self._request("POST", f"/v1/operations/{operation_id}/verify") + + # ----------------------------------------------------------------- + # Audit + # ----------------------------------------------------------------- + + def query_audit( + self, + *, + org_id: Optional[str] = None, + agent_id: Optional[str] = None, + operation_type: Optional[str] = None, + start_time: Optional[int] = None, + end_time: Optional[int] = None, + cursor: Optional[str] = None, + limit: Optional[int] = None, + ) -> AuditQueryResponse: + """Query the tamper-evident audit log.""" + body: Dict[str, Any] = {} + if org_id is not None: + body["org_id"] = org_id + if agent_id is not None: + body["agent_id"] = agent_id + if operation_type is not None: + body["operation_type"] = operation_type + if start_time is not None: + body["start_time"] = start_time + if end_time is not None: + body["end_time"] = end_time + if cursor is not None: + body["cursor"] = cursor + if limit is not None: + body["limit"] = limit + return self._request("POST", "/v1/audit/query", json_body=body) + + # ----------------------------------------------------------------- + # Epochs + # ----------------------------------------------------------------- + + def list_epochs(self) -> ListEpochsResponse: + """List all epochs for the organization.""" + return self._request("GET", "/v1/epochs") + + def get_epoch(self, epoch_id: str) -> GetEpochResponse: + """Retrieve an epoch root record.""" + return self._request("GET", f"/v1/epochs/{epoch_id}") + + # ----------------------------------------------------------------- + # Exports + # ----------------------------------------------------------------- + + def create_export( + self, + start_time: int, + end_time: int, + format: str = "json", + agent_id: Optional[str] = None, + operation_type: Optional[str] = None, + ) -> CreateExportResponse: + """Create a compliance export job.""" + body: Dict[str, Any] = { + "start_time": start_time, + "end_time": end_time, + "format": format, + } + if agent_id is not None: + body["agent_id"] = agent_id + if operation_type is not None: + body["operation_type"] = operation_type + return self._request("POST", "/v1/exports", json_body=body) + + def list_exports(self) -> ListExportsResponse: + """List all exports for the organization.""" + return self._request("GET", "/v1/exports") + + def get_export(self, export_id: str) -> GetExportResponse: + """Retrieve export status and download URL.""" + return self._request("GET", f"/v1/exports/{export_id}") + + # ----------------------------------------------------------------- + # JWKS + # ----------------------------------------------------------------- + + def get_jwks(self) -> JWKSResponse: + """Retrieve the platform JWKS (public, no auth required).""" + url = f"{self.base_url}/.well-known/elydora/jwks.json" + resp = self._session.get(url, timeout=30) + return self._handle_response(resp) diff --git a/elydora/crypto.py b/elydora/crypto.py new file mode 100644 index 0000000..aabd5e5 --- /dev/null +++ b/elydora/crypto.py @@ -0,0 +1,167 @@ +"""Cryptographic utilities: Ed25519 signing, SHA-256, chain hash, JCS canonicalization. + +Mirrors the backend implementation in ElydoraBackend/src/utils/crypto.ts exactly. +""" + +from __future__ import annotations + +import hashlib +import json +import math +from typing import Any, Dict, Optional, Union + +from cryptography.hazmat.primitives.asymmetric.ed25519 import Ed25519PrivateKey + +from .utils import base64url_decode, base64url_encode + + +# --------------------------------------------------------------------------- +# SHA-256 +# --------------------------------------------------------------------------- + + +def sha256_base64url(data: Union[str, bytes]) -> str: + """Compute SHA-256 of data and return base64url-encoded hash.""" + if isinstance(data, str): + data = data.encode("utf-8") + digest = hashlib.sha256(data).digest() + return base64url_encode(digest) + + +# --------------------------------------------------------------------------- +# JCS Canonicalization (RFC 8785) +# --------------------------------------------------------------------------- + + +def _jcs_serialize_number(value: Union[int, float]) -> str: + """Serialize a number per JCS / ES2015 Number serialization rules.""" + if isinstance(value, bool): + # bool is a subclass of int in Python; handle before int check + return "true" if value else "false" + if isinstance(value, int): + return str(value) + # float + if math.isnan(value) or math.isinf(value): + return "null" + if value == 0.0: + # Handle -0.0 -> "0" + return "0" + # Use Python's repr which matches ES2015 for normal floats, + # but we need to ensure no trailing zeros beyond what's needed. + # json.dumps handles this correctly per the JSON spec. + return json.dumps(value) + + +def jcs_canonicalize(value: Any) -> str: + """Canonicalize a value according to JCS (RFC 8785). + + - Object keys sorted lexicographically + - No whitespace + - Numbers serialized using ES2015 rules + - Strings serialized with minimal JSON escaping + - undefined values in objects are omitted + """ + if value is None: + return "null" + + if isinstance(value, bool): + return "true" if value else "false" + + if isinstance(value, (int, float)): + return _jcs_serialize_number(value) + + if isinstance(value, str): + return json.dumps(value, ensure_ascii=False) + + if isinstance(value, list): + elements = [jcs_canonicalize(v) for v in value] + return "[" + ",".join(elements) + "]" + + if isinstance(value, dict): + keys = sorted(value.keys()) + pairs = [] + for key in keys: + v = value[key] + if v is not None or key in value: + # Include keys with None values (maps to JSON null), + # but skip keys that don't exist (which won't happen with dict iteration) + pairs.append(json.dumps(key, ensure_ascii=False) + ":" + jcs_canonicalize(v)) + return "{" + ",".join(pairs) + "}" + + return json.dumps(value) + + +# --------------------------------------------------------------------------- +# Payload hash +# --------------------------------------------------------------------------- + + +def compute_payload_hash(payload: Any) -> str: + """Compute SHA-256 hash of JCS-canonicalized payload, base64url encoded.""" + canonical = jcs_canonicalize(payload) + return sha256_base64url(canonical) + + +# --------------------------------------------------------------------------- +# Chain hash +# --------------------------------------------------------------------------- + + +def compute_chain_hash( + prev_chain_hash: str, + payload_hash: str, + operation_id: str, + issued_at: int, +) -> str: + """Compute chain hash: SHA-256(prev|payload_hash|op_id|issued_at) as base64url. + + Matches the backend formula exactly: + chain_hash = SHA-256(prev_chain_hash + "|" + payload_hash + "|" + operation_id + "|" + str(issued_at)) + """ + input_str = f"{prev_chain_hash}|{payload_hash}|{operation_id}|{issued_at}" + return sha256_base64url(input_str) + + +# --------------------------------------------------------------------------- +# Ed25519 signing +# --------------------------------------------------------------------------- + + +def sign_ed25519(private_key_base64url: str, data: bytes) -> str: + """Sign data with Ed25519 private key (32-byte seed, base64url encoded). + + Returns a base64url-encoded 64-byte signature. + """ + seed = base64url_decode(private_key_base64url) + key = Ed25519PrivateKey.from_private_bytes(seed) + signature = key.sign(data) + return base64url_encode(signature) + + +def get_public_key_base64url(private_key_base64url: str) -> str: + """Derive the Ed25519 public key from the private key seed. + + Returns base64url-encoded 32-byte public key. + """ + seed = base64url_decode(private_key_base64url) + key = Ed25519PrivateKey.from_private_bytes(seed) + pub = key.public_key() + from cryptography.hazmat.primitives.serialization import Encoding, PublicFormat + raw_bytes = pub.public_bytes(Encoding.Raw, PublicFormat.Raw) + return base64url_encode(raw_bytes) + + +# --------------------------------------------------------------------------- +# EOR signing +# --------------------------------------------------------------------------- + + +def sign_eor(eor_dict: Dict[str, Any], private_key_base64url: str) -> str: + """Sign an EOR by canonicalizing all fields except 'signature', then signing. + + Returns the base64url-encoded Ed25519 signature. + """ + # Build the signable object: all EOR fields except 'signature' + signable = {k: v for k, v in eor_dict.items() if k != "signature"} + canonical = jcs_canonicalize(signable) + return sign_ed25519(private_key_base64url, canonical.encode("utf-8")) diff --git a/elydora/errors.py b/elydora/errors.py new file mode 100644 index 0000000..763c5d0 --- /dev/null +++ b/elydora/errors.py @@ -0,0 +1,30 @@ +"""Elydora SDK error types.""" + +from __future__ import annotations + +from typing import Any, Dict, Optional + + +class ElydoraError(Exception): + """Error returned by the Elydora API.""" + + def __init__( + self, + code: str, + message: str, + request_id: str = "", + details: Optional[Dict[str, Any]] = None, + status_code: int = 0, + ) -> None: + super().__init__(message) + self.code = code + self.message = message + self.request_id = request_id + self.details = details or {} + self.status_code = status_code + + def __repr__(self) -> str: + return ( + f"ElydoraError(code={self.code!r}, message={self.message!r}, " + f"request_id={self.request_id!r})" + ) diff --git a/elydora/plugins/__init__.py b/elydora/plugins/__init__.py new file mode 100644 index 0000000..3c605b0 --- /dev/null +++ b/elydora/plugins/__init__.py @@ -0,0 +1 @@ +"""Elydora agent plugins for AI coding tools.""" diff --git a/elydora/plugins/augment.py b/elydora/plugins/augment.py new file mode 100644 index 0000000..cfb0e23 --- /dev/null +++ b/elydora/plugins/augment.py @@ -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") diff --git a/elydora/plugins/base.py b/elydora/plugins/base.py new file mode 100644 index 0000000..eda6186 --- /dev/null +++ b/elydora/plugins/base.py @@ -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.""" diff --git a/elydora/plugins/claudecode.py b/elydora/plugins/claudecode.py new file mode 100644 index 0000000..1e11d69 --- /dev/null +++ b/elydora/plugins/claudecode.py @@ -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") diff --git a/elydora/plugins/cursor.py b/elydora/plugins/cursor.py new file mode 100644 index 0000000..b5f18ad --- /dev/null +++ b/elydora/plugins/cursor.py @@ -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") diff --git a/elydora/plugins/gemini.py b/elydora/plugins/gemini.py new file mode 100644 index 0000000..4aa1c30 --- /dev/null +++ b/elydora/plugins/gemini.py @@ -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") diff --git a/elydora/plugins/hook_template.py b/elydora/plugins/hook_template.py new file mode 100644 index 0000000..8931b3d --- /dev/null +++ b/elydora/plugins/hook_template.py @@ -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() +''' diff --git a/elydora/plugins/kiro.py b/elydora/plugins/kiro.py new file mode 100644 index 0000000..40f8a99 --- /dev/null +++ b/elydora/plugins/kiro.py @@ -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) diff --git a/elydora/plugins/opencode.py b/elydora/plugins/opencode.py new file mode 100644 index 0000000..d51fff9 --- /dev/null +++ b/elydora/plugins/opencode.py @@ -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) diff --git a/elydora/plugins/registry.py b/elydora/plugins/registry.py new file mode 100644 index 0000000..2665bff --- /dev/null +++ b/elydora/plugins/registry.py @@ -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()) diff --git a/elydora/types.py b/elydora/types.py new file mode 100644 index 0000000..b3cd065 --- /dev/null +++ b/elydora/types.py @@ -0,0 +1,293 @@ +"""Type definitions matching the Elydora backend API contract. + +All types use TypedDict for structural typing, compatible with Python 3.9+. +""" + +from __future__ import annotations + +import sys +from typing import Any, Dict, List, Optional + +if sys.version_info >= (3, 11): + from typing import Literal, TypedDict +else: + from typing_extensions import Literal, TypedDict + + +# --------------------------------------------------------------------------- +# Enums (as Literal types) +# --------------------------------------------------------------------------- + +AgentStatus = Literal["active", "frozen", "revoked"] +KeyStatus = Literal["active", "retired", "revoked"] +ExportStatus = Literal["queued", "running", "done", "failed"] +RbacRole = Literal[ + "org_owner", + "security_admin", + "compliance_auditor", + "readonly_investigator", + "integration_engineer", +] +ErrorCode = Literal[ + "INVALID_SIGNATURE", + "UNKNOWN_AGENT", + "KEY_REVOKED", + "AGENT_FROZEN", + "TTL_EXPIRED", + "REPLAY_DETECTED", + "PREV_HASH_MISMATCH", + "PAYLOAD_TOO_LARGE", + "RATE_LIMITED", + "INTERNAL_ERROR", + "UNAUTHORIZED", + "FORBIDDEN", + "NOT_FOUND", + "VALIDATION_ERROR", +] + +# --------------------------------------------------------------------------- +# Entity types +# --------------------------------------------------------------------------- + + +class Agent(TypedDict): + agent_id: str + org_id: str + display_name: str + responsible_entity: str + status: AgentStatus + created_at: int + updated_at: int + + +class AgentKey(TypedDict): + kid: str + agent_id: str + public_key: str + algorithm: Literal["ed25519"] + status: KeyStatus + created_at: int + retired_at: Optional[int] + + +class Operation(TypedDict): + operation_id: str + org_id: str + agent_id: str + seq_no: int + operation_type: str + issued_at: int + ttl_ms: int + nonce: str + subject: str + action: str + payload_hash: str + prev_chain_hash: str + chain_hash: str + agent_pubkey_kid: str + signature: str + r2_payload_key: Optional[str] + created_at: int + + +class Receipt(TypedDict): + receipt_id: str + operation_id: str + r2_receipt_key: str + created_at: int + + +class Epoch(TypedDict): + epoch_id: str + org_id: str + start_time: int + end_time: int + root_hash: str + leaf_count: int + r2_epoch_key: str + created_at: int + + +class Organization(TypedDict): + org_id: str + name: str + created_at: int + updated_at: int + + +class User(TypedDict): + user_id: str + org_id: str + email: str + display_name: str + role: RbacRole + status: Literal["active", "suspended"] + created_at: int + updated_at: int + + +class Export(TypedDict): + export_id: str + org_id: str + status: ExportStatus + query_params: str + r2_export_key: Optional[str] + created_at: int + completed_at: Optional[int] + + +# --------------------------------------------------------------------------- +# Protocol types +# --------------------------------------------------------------------------- + + +class EOR(TypedDict): + """Elydora Operation Record.""" + op_version: Literal["1.0"] + operation_id: str + org_id: str + agent_id: str + issued_at: int + ttl_ms: int + nonce: str + operation_type: str + subject: Dict[str, Any] + action: Dict[str, Any] + payload: Any # Dict[str, Any] | str | None + payload_hash: str + prev_chain_hash: str + agent_pubkey_kid: str + signature: str + + +class EAR(TypedDict): + """Elydora Acknowledgment Receipt.""" + receipt_version: str + receipt_id: str + operation_id: str + org_id: str + agent_id: str + server_received_at: int + seq_no: int + chain_hash: str + queue_message_id: str + receipt_hash: str + elydora_kid: str + elydora_signature: str + + +# --------------------------------------------------------------------------- +# API request types +# --------------------------------------------------------------------------- + + +class RegisterAgentKeyParam(TypedDict): + kid: str + public_key: str + algorithm: Literal["ed25519"] + + +class RegisterAgentRequest(TypedDict, total=False): + agent_id: str + display_name: str + responsible_entity: str + keys: List[RegisterAgentKeyParam] + + +class AuditQueryParams(TypedDict, total=False): + org_id: str + agent_id: str + operation_type: str + start_time: int + end_time: int + cursor: str + limit: int + + +# --------------------------------------------------------------------------- +# API response types +# --------------------------------------------------------------------------- + + +class RegisterAgentResponse(TypedDict): + agent: Agent + keys: List[AgentKey] + + +class GetAgentResponse(TypedDict): + agent: Agent + keys: List[AgentKey] + + +class SubmitOperationResponse(TypedDict): + receipt: EAR + + +class GetOperationResponse(TypedDict, total=False): + operation: Operation + receipt: Receipt + + +class VerifyChecks(TypedDict, total=False): + signature: bool + chain: bool + receipt: bool + merkle: bool + + +class VerifyOperationResponse(TypedDict, total=False): + valid: bool + checks: VerifyChecks + errors: List[str] + + +class AuditQueryResponse(TypedDict): + operations: List[Operation] + cursor: Optional[str] + total_count: int + + +class GetEpochResponse(TypedDict, total=False): + epoch: Epoch + anchor: Dict[str, Any] + + +class ListEpochsResponse(TypedDict): + epochs: List[Epoch] + + +class CreateExportResponse(TypedDict): + export: Export + + +class GetExportResponse(TypedDict, total=False): + export: Export + download_url: str + + +class ListExportsResponse(TypedDict): + exports: List[Export] + + +class JWK(TypedDict, total=False): + kty: str + crv: str + x: str + kid: str + use: str + alg: str + + +class JWKSResponse(TypedDict): + keys: List[JWK] + + +class AuthRegisterResponse(TypedDict): + user: User + organization: Organization + token: str + + +class AuthLoginResponse(TypedDict): + user: User + token: str diff --git a/elydora/utils.py b/elydora/utils.py new file mode 100644 index 0000000..ba90e41 --- /dev/null +++ b/elydora/utils.py @@ -0,0 +1,59 @@ +"""Utility helpers: UUIDv7, nonce generation, base64url encoding.""" + +from __future__ import annotations + +import base64 +import os +import struct +import time + + +def base64url_encode(data: bytes) -> str: + """Encode bytes to base64url string with no padding.""" + return base64.urlsafe_b64encode(data).rstrip(b"=").decode("ascii") + + +def base64url_decode(s: str) -> bytes: + """Decode a base64url string (with or without padding) to bytes.""" + s = s.replace("-", "+").replace("_", "/") + padding = 4 - len(s) % 4 + if padding != 4: + s += "=" * padding + return base64.b64decode(s) + + +def generate_nonce() -> str: + """Generate a 16-byte random nonce, base64url encoded.""" + return base64url_encode(os.urandom(16)) + + +def generate_uuidv7() -> str: + """Generate a UUIDv7 (time-ordered, random) as a string. + + Layout (RFC 9562): + - 48-bit Unix timestamp in milliseconds + - 4-bit version (0b0111) + - 12-bit random + - 2-bit variant (0b10) + - 62-bit random + """ + timestamp_ms = int(time.time() * 1000) + + # 48-bit timestamp + ts_bytes = struct.pack(">Q", timestamp_ms)[2:] # last 6 bytes of 8-byte big-endian + + # 2 bytes: version (4 bits = 0x7) + 12 random bits + rand_a = struct.unpack(">H", os.urandom(2))[0] + rand_a = (rand_a & 0x0FFF) | 0x7000 # set version nibble + + # 8 bytes: variant (2 bits = 0b10) + 62 random bits + rand_b = bytearray(os.urandom(8)) + rand_b[0] = (rand_b[0] & 0x3F) | 0x80 # set variant bits + + uuid_bytes = ts_bytes + struct.pack(">H", rand_a) + bytes(rand_b) + + hex_str = uuid_bytes.hex() + return ( + f"{hex_str[0:8]}-{hex_str[8:12]}-{hex_str[12:16]}-" + f"{hex_str[16:20]}-{hex_str[20:32]}" + ) diff --git a/pyproject.toml b/pyproject.toml new file mode 100644 index 0000000..a8966b4 --- /dev/null +++ b/pyproject.toml @@ -0,0 +1,42 @@ +[build-system] +requires = ["setuptools>=68.0", "wheel"] +build-backend = "setuptools.build_meta" + +[project] +name = "elydora" +version = "1.0.0" +description = "Python SDK for the Elydora tamper-evident audit platform" +readme = "README.md" +license = {text = "MIT"} +requires-python = ">=3.9" +classifiers = [ + "Development Status :: 5 - Production/Stable", + "Intended Audience :: Developers", + "License :: OSI Approved :: MIT License", + "Programming Language :: Python :: 3", + "Programming Language :: Python :: 3.9", + "Programming Language :: Python :: 3.10", + "Programming Language :: Python :: 3.11", + "Programming Language :: Python :: 3.12", + "Programming Language :: Python :: 3.13", + "Typing :: Typed", +] +dependencies = [ + "requests>=2.28.0", + "aiohttp>=3.8.0", + "cryptography>=41.0.0", + "typing_extensions>=4.0.0;python_version<'3.11'", +] + +[project.scripts] +elydora = "elydora.cli:main" + +[project.optional-dependencies] +dev = [ + "pytest>=7.0", + "pytest-asyncio>=0.21", + "mypy>=1.0", +] + +[tool.setuptools.packages.find] +include = ["elydora*"] diff --git a/setup.py b/setup.py new file mode 100644 index 0000000..6068493 --- /dev/null +++ b/setup.py @@ -0,0 +1,3 @@ +from setuptools import setup + +setup()