fix: make Cursor hooks fail closed

elydora/plugins/guard_template.py

@@ -0,0 +1,216 @@
+"""Generate self-contained agent status guards."""
+
+from __future__ import annotations
+
+import sys
+
+from .runtime_reader_template import PROTECTED_RUNTIME_READER
+
+
+def generate_guard_script(
+    agent_name: str,
+    agent_id: str,
+    success_output: str = "",
+    *,
+    fail_closed: bool = False,
+) -> str:
+    """Return a self-contained status guard for a command-hook provider."""
+    return f'''#!{sys.executable}
+"""Elydora agent guard for {agent_name}. Generated by elydora-sdk."""
+
+import json
+import math
+import os
+import stat
+import sys
+import time
+import urllib.error
+import urllib.parse
+import urllib.request
+
+AGENT_NAME = {agent_name!r}
+AGENT_ID = {agent_id!r}
+SUCCESS_OUTPUT = {success_output!r}
+FAIL_CLOSED = {fail_closed!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_SECONDS = 60
+
+{PROTECTED_RUNTIME_READER}
+
+
+def validate_status(value, source):
+    if value not in ("active", "frozen", "revoked"):
+        raise ValueError(source + " contains an invalid agent status")
+    return value
+
+
+def block_agent(status):
+    state = "frozen" if status == "frozen" else "revoked"
+    message = 'Agent "' + AGENT_NAME + '" is ' + state + " in Elydora."
+    if SUCCESS_OUTPUT:
+        sys.stdout.write(json.dumps({{
+            "permission": "deny",
+            "userMessage": message,
+            "agentMessage": message + " Tool execution is blocked.",
+        }}) + "\\n")
+        sys.stdout.flush()
+    sys.stderr.write("[Elydora guard] " + message + " Tool execution blocked.\\n")
+    raise SystemExit(2)
+
+
+def read_input():
+    raw = sys.stdin.read()
+    if not raw.strip():
+        if FAIL_CLOSED:
+            raise ValueError("Hook input is empty")
+        return
+    try:
+        value = json.loads(raw)
+    except (json.JSONDecodeError, ValueError) as error:
+        if FAIL_CLOSED:
+            raise ValueError("Hook input is invalid JSON: " + str(error)) from error
+        return
+    if FAIL_CLOSED and not isinstance(value, dict):
+        raise ValueError("Hook input must contain a JSON object")
+
+
+def read_config():
+    try:
+        raw = read_protected_file(
+            CONFIG_PATH,
+            "Agent config",
+            MAX_PROTECTED_CONFIG_BYTES,
+        ).decode("utf-8")
+        config = json.loads(raw)
+    except Exception as error:
+        raise RuntimeError("Failed to read agent config: " + str(error)) from error
+    if not isinstance(config, dict):
+        raise ValueError("Agent config must contain a JSON object")
+    if config.get("agent_id") != AGENT_ID:
+        raise ValueError("Agent config identity does not match the generated guard")
+    configured_name = config.get("agent_name")
+    if configured_name is not None and configured_name != AGENT_NAME:
+        raise ValueError("Agent config identity does not match the generated guard")
+    return config
+
+
+def read_status_cache():
+    try:
+        raw = read_protected_file(
+            STATUS_CACHE_PATH,
+            "Status cache",
+            MAX_PROTECTED_CONFIG_BYTES,
+            require_private_permissions=False,
+        ).decode("utf-8")
+    except FileNotFoundError:
+        return None
+    try:
+        cache = json.loads(raw)
+    except (json.JSONDecodeError, ValueError) as error:
+        raise ValueError("Status cache is invalid JSON: " + str(error)) from error
+    if not isinstance(cache, dict):
+        raise ValueError("Status cache must contain a JSON object")
+    cached_at = cache.get("cached_at")
+    if isinstance(cached_at, bool) or not isinstance(cached_at, (int, float)):
+        raise ValueError("Status cache has an invalid timestamp")
+    if not math.isfinite(cached_at):
+        raise ValueError("Status cache has an invalid timestamp")
+    age = time.time() - cached_at
+    if age < 0:
+        raise ValueError("Status cache timestamp is in the future")
+    return validate_status(cache.get("status"), "Status cache"), age
+
+
+def write_status_cache(status):
+    token = os.urandom(8).hex()
+    temporary_path = STATUS_CACHE_PATH + "." + str(os.getpid()) + "." + token + ".tmp"
+    descriptor = -1
+    try:
+        flags = os.O_WRONLY | os.O_CREAT | os.O_EXCL | getattr(os, "O_BINARY", 0)
+        descriptor = os.open(temporary_path, flags, 0o600)
+        with os.fdopen(descriptor, "w", encoding="utf-8") as file:
+            descriptor = -1
+            json.dump({{"status": status, "cached_at": time.time()}}, file)
+            file.flush()
+            os.fsync(file.fileno())
+        os.replace(temporary_path, STATUS_CACHE_PATH)
+    finally:
+        if descriptor >= 0:
+            os.close(descriptor)
+        try:
+            os.remove(temporary_path)
+        except FileNotFoundError:
+            pass
+
+
+def resolve_remote_status(config):
+    base_url = config.get("base_url", "https://api.elydora.com")
+    if not isinstance(base_url, str):
+        raise ValueError("Agent config base_url must be a string")
+    parsed = urllib.parse.urlsplit(base_url)
+    if parsed.scheme not in ("http", "https"):
+        raise ValueError("Agent config base_url must use HTTP or HTTPS")
+    token = config.get("token", "")
+    if not isinstance(token, str):
+        raise ValueError("Agent config token must be a string")
+    url = base_url.rstrip("/") + "/v1/agents/" + urllib.parse.quote(AGENT_ID, safe="")
+    request = urllib.request.Request(url, method="GET")
+    request.add_header("Accept", "application/json")
+    if token:
+        request.add_header("Authorization", "Bearer " + token)
+    try:
+        with urllib.request.urlopen(request, timeout=3) as response:
+            value = json.loads(response.read().decode("utf-8"))
+    except urllib.error.HTTPError as error:
+        raise RuntimeError(
+            "Agent status API returned HTTP " + str(error.code)
+        ) from error
+    status = value.get("agent", {{}}).get("status") if isinstance(value, dict) else None
+    return validate_status(status, "Agent status API")
+
+
+def main():
+    read_input()
+    try:
+        config = read_config()
+    except Exception as error:
+        if FAIL_CLOSED:
+            raise
+        sys.stderr.write("[Elydora guard] " + str(error) + "\\n")
+        return
+    try:
+        cache = read_status_cache()
+        if cache is not None and cache[1] < CACHE_TTL_SECONDS:
+            if cache[0] != "active":
+                block_agent(cache[0])
+            return
+    except Exception as error:
+        if FAIL_CLOSED:
+            raise
+        sys.stderr.write("[Elydora guard] " + str(error) + "\\n")
+    try:
+        status = resolve_remote_status(config)
+        write_status_cache(status)
+        if status != "active":
+            block_agent(status)
+    except SystemExit:
+        raise
+    except Exception as error:
+        if FAIL_CLOSED:
+            raise RuntimeError("Failed to resolve agent status: " + str(error)) from error
+        sys.stderr.write("[Elydora guard] Failed to resolve agent status: " + str(error) + "\\n")
+
+
+if __name__ == "__main__":
+    try:
+        main()
+    except SystemExit:
+        raise
+    except Exception as error:
+        sys.stderr.write("[Elydora guard] " + repr(error) + "\\n")
+        raise SystemExit(1)
+    if SUCCESS_OUTPUT:
+        sys.stdout.write(SUCCESS_OUTPUT)
+'''