"""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, kid: str, base_url: str, success_output: str = "", ) -> str: """Return a complete, self-contained Python hook script as a string. The script: 1. Reads JSON from stdin (tool-use event) 2. Constructs an Elydora Operation Record (EOR) 3. Signs it with Ed25519 4. POSTs it fire-and-forget to the Elydora API 5. Always exits 0 (never blocks the host agent) Chain state is persisted in ~/.elydora/{agent_id}/chain-state.json. """ shebang = sys.executable return f'''#!{shebang} """Elydora audit hook — auto-generated, do not edit.""" import base64 import hashlib import json import math import os import stat import struct import sys import time # --------------------------------------------------------------------------- # Configuration (baked at install time) # --------------------------------------------------------------------------- ORG_ID = {org_id!r} AGENT_ID = {agent_id!r} KID = {kid!r} BASE_URL = {base_url!r} SUCCESS_OUTPUT = {success_output!r} ELYDORA_DIR = os.path.join(os.path.expanduser("~"), ".elydora") PRIVATE_KEY_PATH = os.path.join(ELYDORA_DIR, AGENT_ID, "private.key") CHAIN_STATE_PATH = os.path.join(ELYDORA_DIR, AGENT_ID, "chain-state.json") ERROR_LOG_PATH = os.path.join(ELYDORA_DIR, AGENT_ID, "error.log") ZERO_CHAIN_HASH = "AAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAA" MAX_PRIVATE_KEY_BYTES = 64 * 1024 # --------------------------------------------------------------------------- # 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)) def read_private_key(): before = os.lstat(PRIVATE_KEY_PATH) if not stat.S_ISREG(before.st_mode): raise ValueError("Private key path is not a regular file: " + PRIVATE_KEY_PATH) if before.st_size > MAX_PRIVATE_KEY_BYTES: raise ValueError("Private key file exceeds the size limit: " + PRIVATE_KEY_PATH) if os.name != "nt" and before.st_mode & (stat.S_IRWXG | stat.S_IRWXO): raise PermissionError("Private key file permissions are too broad: " + PRIVATE_KEY_PATH) flags = os.O_RDONLY flags |= getattr(os, "O_BINARY", 0) flags |= getattr(os, "O_CLOEXEC", 0) flags |= getattr(os, "O_NOFOLLOW", 0) descriptor = os.open(PRIVATE_KEY_PATH, flags) try: file = os.fdopen(descriptor, "rb") descriptor = -1 with file: after = os.fstat(file.fileno()) if not stat.S_ISREG(after.st_mode): raise ValueError("Private key path is not a regular file: " + PRIVATE_KEY_PATH) if after.st_size > MAX_PRIVATE_KEY_BYTES: raise ValueError("Private key file exceeds the size limit: " + PRIVATE_KEY_PATH) if os.name != "nt" and after.st_mode & (stat.S_IRWXG | stat.S_IRWXO): raise PermissionError("Private key file permissions are too broad: " + PRIVATE_KEY_PATH) if (before.st_dev, before.st_ino) != (after.st_dev, after.st_ino): raise OSError("Private key file changed while opening: " + PRIVATE_KEY_PATH) encoded = file.read(MAX_PRIVATE_KEY_BYTES + 1) finally: if descriptor >= 0: os.close(descriptor) if len(encoded) > MAX_PRIVATE_KEY_BYTES: raise ValueError("Private key file exceeds the size limit: " + PRIVATE_KEY_PATH) value = encoded.decode("utf-8") if not value or "\\r" in value or "\\n" in value or "\\0" in value: raise ValueError("Private key file must contain exactly one line") return value # --------------------------------------------------------------------------- # 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): cline_event = event.get("tool_result") if not isinstance(cline_event, dict): cline_event = event.get("tool_call") if not isinstance(cline_event, dict): cline_event = {{}} tool_name = event.get( "tool_name", event.get("toolName", event.get("name", cline_event.get("name", "unknown"))), ) tool_input = event.get( "tool_input", event.get("toolInput", event.get("input", cline_event.get("input", {{}}))), ) session_id = event.get( "session_id", event.get("sessionId", event.get("session", event.get("taskId", "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(read_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() if SUCCESS_OUTPUT: sys.stdout.write(SUCCESS_OUTPUT) ''' def generate_guard_script( agent_name: str, agent_id: str, success_output: str = "", ) -> str: """Return a self-contained Python guard script for PreToolUse freeze enforcement. The script: 1. Reads agent config from ~/.elydora/{agent_id}/config.json 2. Checks cached status from ~/.elydora/{agent_id}/status-cache.json (60s TTL) 3. If cache is stale, fetches agent status from GET {base_url}/v1/agents/{agent_id} 4. If status is "frozen", writes error to stderr and exits with blocking code 2 5. Fail-open: if API unreachable or config missing, exits 0 (allow) 6. Uses only stdlib — no external deps """ shebang = sys.executable return f'''#!{shebang} """Elydora agent guard (PreToolUse) for: {agent_name} Generated by elydora-sdk — DO NOT EDIT.""" import json import os import sys import time AGENT_NAME = {agent_name!r} AGENT_ID = {agent_id!r} SUCCESS_OUTPUT = {success_output!r} ELYDORA_DIR = os.path.join(os.path.expanduser("~"), ".elydora") CONFIG_PATH = os.path.join(ELYDORA_DIR, AGENT_ID, "config.json") STATUS_CACHE_PATH = os.path.join(ELYDORA_DIR, AGENT_ID, "status-cache.json") CACHE_TTL_S = 60 # 60 seconds # Consume stdin so the parent process doesn't block on a full pipe try: sys.stdin.read() except Exception: pass def main(): # Read agent config try: with open(CONFIG_PATH, "r") as f: config = json.load(f) except Exception: return # Can't read config — fail-open (allow) # Check cached status try: with open(STATUS_CACHE_PATH, "r") as f: cache = json.load(f) age = time.time() - cache.get("cached_at", 0) if age < CACHE_TTL_S: if cache.get("status") == "frozen": sys.stderr.write( 'Agent "' + AGENT_NAME + '" is frozen by Elydora. Tool execution blocked.\\n' ) sys.exit(2) 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(2) except SystemExit: raise # Re-raise sys.exit except Exception: pass # API unreachable — fail-open (allow) if __name__ == "__main__": main() if SUCCESS_OUTPUT: sys.stdout.write(SUCCESS_OUTPUT) '''