Elydora-Python-SDK/elydora/plugins/hook_template.py

"""Generate self-contained audit hook scripts."""

from __future__ import annotations

import sys

from .error_log_template import ERROR_LOG_RUNTIME
from .guard_template import generate_guard_script
from .runtime_reader_template import PROTECTED_RUNTIME_READER

__all__ = ["generate_guard_script", "generate_hook_script"]


def generate_hook_script(
    *,
    org_id: str,
    agent_id: str,
    kid: str,
    base_url: str,
    success_output: str = "",
    fail_closed: bool = False,
    native_payload: bool = False,
    agent_name: str = "",
) -> str:
    """Return a self-contained signed audit hook."""
    return f'''#!{sys.executable}
"""Elydora audit hook. Generated by elydora-sdk."""

import base64
import hashlib
import json
import math
import os
import re
import stat
import struct
import sys
import time
import urllib.error
import urllib.parse
import urllib.request

AGENT_ID = {agent_id!r}
AGENT_NAME = {agent_name!r}
SUCCESS_OUTPUT = {success_output!r}
FAIL_CLOSED = {fail_closed!r}
NATIVE_PAYLOAD = {native_payload!r}
ELYDORA_DIR = os.path.join(os.path.expanduser("~"), ".elydora")
CONFIG_PATH = os.path.join(ELYDORA_DIR, AGENT_ID, "config.json")
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"

{PROTECTED_RUNTIME_READER}

{ERROR_LOG_RUNTIME}


def base64url_encode(data):
    return base64.urlsafe_b64encode(data).rstrip(b"=").decode("ascii")


def base64url_decode(value):
    encoded = value.replace("-", "+").replace("_", "/")
    encoded += "=" * ((4 - len(encoded) % 4) % 4)
    return base64.b64decode(encoded, validate=True)


def sha256_base64url(data):
    if isinstance(data, str):
        data = data.encode("utf-8")
    return base64url_encode(hashlib.sha256(data).digest())


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(item) for item in value) + "]"
    if isinstance(value, dict):
        pairs = [
            json.dumps(key, ensure_ascii=False) + ":" + jcs_canonicalize(value[key])
            for key in sorted(value)
        ]
        return "{{" + ",".join(pairs) + "}}"
    return json.dumps(value)


def compute_payload_hash(payload):
    return sha256_base64url(jcs_canonicalize(payload))


def compute_chain_hash(previous, payload_hash, operation_id, issued_at):
    value = f"{{previous}}|{{payload_hash}}|{{operation_id}}|{{issued_at}}"
    return sha256_base64url(value)


def read_private_key():
    encoded = read_protected_file(PRIVATE_KEY_PATH, "Private key").decode("utf-8")
    if not encoded or "\\r" in encoded or "\\n" in encoded or "\\0" in encoded:
        raise ValueError("Private key file must contain exactly one line")
    seed = base64url_decode(encoded)
    if len(seed) != 32 or base64url_encode(seed) != encoded:
        raise ValueError("Private key must be a canonical 32-byte base64url value")
    return seed


def sign_ed25519(seed, data):
    from cryptography.hazmat.primitives.asymmetric.ed25519 import Ed25519PrivateKey
    return base64url_encode(Ed25519PrivateKey.from_private_bytes(seed).sign(data))


def generate_uuidv7():
    timestamp = struct.pack(">Q", int(time.time() * 1000))[2:]
    random_a = (struct.unpack(">H", os.urandom(2))[0] & 0x0FFF) | 0x7000
    random_b = bytearray(os.urandom(8))
    random_b[0] = (random_b[0] & 0x3F) | 0x80
    value = (timestamp + struct.pack(">H", random_a) + bytes(random_b)).hex()
    return f"{{value[0:8]}}-{{value[8:12]}}-{{value[12:16]}}-{{value[16:20]}}-{{value[20:32]}}"


def log_error(error):
    try:
        append_bounded_error_log(
            ERROR_LOG_PATH,
            "Error log",
            bounded_error_message(error),
            MAX_PROTECTED_LOG_BYTES,
        )
    except Exception as log_failure:
        sys.stderr.write(
            "[Elydora audit] Failed to write error log: "
            + str(log_failure)
            + "\\n"
        )


def read_chain_state():
    try:
        raw = read_protected_file(
            CHAIN_STATE_PATH,
            "Chain state",
            MAX_PROTECTED_CONFIG_BYTES,
        ).decode("utf-8")
    except FileNotFoundError:
        return ZERO_CHAIN_HASH
    except Exception as error:
        raise RuntimeError("Failed to read chain state: " + str(error)) from error
    try:
        state = json.loads(raw)
    except (json.JSONDecodeError, ValueError) as error:
        raise RuntimeError("Chain state is invalid JSON: " + str(error)) from error
    previous = state.get("prev_chain_hash") if isinstance(state, dict) else None
    if not isinstance(previous, str) or re.fullmatch(r"[A-Za-z0-9_-]{{43}}", previous) is None:
        raise ValueError("Chain state contains an invalid prev_chain_hash")
    return previous


def write_chain_state(chain_hash):
    write_protected_json(
        CHAIN_STATE_PATH,
        "Chain state",
        {{"prev_chain_hash": chain_hash}},
    )


def read_runtime_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 audit hook")
    if AGENT_NAME and config.get("agent_name") != AGENT_NAME:
        raise ValueError("Agent config identity does not match the generated audit hook")
    for field in ("org_id", "agent_id", "kid"):
        if not isinstance(config.get(field), str) or not config[field]:
            raise ValueError("Agent config is missing " + field)
    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")
    try:
        parsed = urllib.parse.urlsplit(base_url)
        hostname = parsed.hostname
        parsed.port
    except ValueError as error:
        raise ValueError(
            "Agent config base_url must be an absolute HTTP or HTTPS URL"
        ) from error
    invalid_character = "\\\\" in base_url or any(
        character.isspace() or ord(character) < 32 for character in base_url
    )
    if (
        parsed.scheme not in ("http", "https")
        or not parsed.netloc
        or hostname is None
        or invalid_character
    ):
        raise ValueError("Agent config base_url must be an absolute HTTP or HTTPS URL")
    if parsed.username is not None or parsed.password is not None or parsed.query or parsed.fragment:
        raise ValueError(
            "Agent config base_url must exclude credentials, query parameters, and fragments"
        )
    if not isinstance(config.get("token", ""), str):
        raise ValueError("Agent config token must be a string")
    return config


def read_event():
    raw = sys.stdin.read()
    if not raw.strip():
        raise ValueError("Hook input is empty")
    try:
        event = json.loads(raw)
    except (json.JSONDecodeError, ValueError) as error:
        raise ValueError("Hook input is invalid JSON: " + str(error)) from error
    if not isinstance(event, dict):
        raise ValueError("Hook input must contain a JSON object")
    return event


def submit_operation(event, runtime):
    nested = event.get("tool_result")
    if not isinstance(nested, dict):
        nested = event.get("tool_call")
    if not isinstance(nested, dict):
        nested = {{}}
    tool_name = event.get("tool_name", event.get("toolName", event.get("name", nested.get("name", "unknown"))))
    tool_input = event.get("tool_input", event.get("toolInput", event.get("input", nested.get("input", {{}}))))
    session_id = event.get(
        "conversation_id",
        event.get("session_id", event.get("sessionId", event.get("session", event.get("taskId", "unknown")))),
    )
    previous = read_chain_state()
    operation_id = generate_uuidv7()
    issued_at = int(time.time() * 1000)
    payload = event if NATIVE_PAYLOAD else {{
        "tool_name": tool_name,
        "tool_input": tool_input,
        "session_id": session_id,
    }}
    payload_hash = compute_payload_hash(payload)
    chain_hash = compute_chain_hash(previous, payload_hash, operation_id, issued_at)
    unsigned = {{
        "op_version": "1.0",
        "operation_id": operation_id,
        "org_id": runtime["org_id"],
        "agent_id": runtime["agent_id"],
        "issued_at": issued_at,
        "ttl_ms": 30000,
        "nonce": base64url_encode(os.urandom(16)),
        "operation_type": "ai.tool_use",
        "subject": {{"session_id": session_id}},
        "action": {{"tool": tool_name}},
        "payload": payload,
        "payload_hash": payload_hash,
        "prev_chain_hash": previous,
        "agent_pubkey_kid": runtime["kid"],
    }}
    signature = sign_ed25519(
        read_private_key(),
        jcs_canonicalize(unsigned).encode("utf-8"),
    )
    operation = dict(unsigned)
    operation.update({{"chain_hash": chain_hash, "signature": signature}})
    url = runtime.get("base_url", "https://api.elydora.com").rstrip("/") + "/v1/operations"
    headers = {{"Content-Type": "application/json", "Accept": "application/json"}}
    if runtime.get("token"):
        headers["Authorization"] = "Bearer " + runtime["token"]
    request = urllib.request.Request(
        url,
        data=json.dumps(operation).encode("utf-8"),
        headers=headers,
        method="POST",
    )
    try:
        with urllib.request.urlopen(request, timeout=5):
            write_chain_state(chain_hash)
    except urllib.error.HTTPError as error:
        body = error.read().decode("utf-8")
        if error.code == 400 and body:
            try:
                failure = json.loads(body)
            except (json.JSONDecodeError, ValueError) as parse_error:
                raise RuntimeError("Audit API returned invalid error JSON: " + str(parse_error)) from parse_error
            details = failure.get("error", {{}}) if isinstance(failure, dict) else {{}}
            if details.get("code") == "PREV_HASH_MISMATCH":
                match = re.search(
                    r'Expected prev_chain_hash "([A-Za-z0-9_-]{{43}})"',
                    str(details.get("message", "")),
                )
                if match:
                    write_chain_state(match.group(1))
                    log_error("Chain hash resynced to server: " + match.group(1))
        raise RuntimeError("Audit API returned HTTP " + str(error.code)) from error


def main():
    try:
        submit_operation(read_event(), read_runtime_config())
    except Exception as error:
        log_error(error)
        if FAIL_CLOSED:
            raise


if __name__ == "__main__":
    try:
        main()
    except Exception as error:
        sys.stderr.write("[Elydora audit] " + repr(error) + "\\n")
        raise SystemExit(1)
    if SUCCESS_OUTPUT:
        sys.stdout.write(SUCCESS_OUTPUT)
'''