Add selectable CASAN IDE integrations
This commit is contained in:
thanhnv
@@ -0,0 +1,212 @@
|
||||
#!/usr/bin/env python3
|
||||
"""Project-local bootstrap for CASAN's globally installed agentic adapters.
|
||||
|
||||
This file is intentionally small and stdlib-only. It is the stable command
|
||||
target committed by `casan init`; the policy implementation remains in the
|
||||
versioned global CASAN installation. On every hook invocation it:
|
||||
|
||||
1. locates the project and loads `.casan/config.json`;
|
||||
2. applies the project's enforcement/integration settings to the process;
|
||||
3. resolves the pinned global harness and verifies its live integrity hash;
|
||||
4. dispatches stdin/stdout to the selected client adapter.
|
||||
|
||||
The bootstrap never calls a model.
|
||||
"""
|
||||
|
||||
from __future__ import print_function
|
||||
|
||||
import importlib.util
|
||||
import json
|
||||
import os
|
||||
import runpy
|
||||
import sys
|
||||
|
||||
|
||||
ADAPTERS = {
|
||||
"claude": ("claude-code", "claude_hook.py"),
|
||||
"codex": ("codex", "codex_hook.py"),
|
||||
"vscode-copilot": ("vscode", "vscode_hook.py"),
|
||||
}
|
||||
|
||||
|
||||
def load_json(path, default=None):
|
||||
try:
|
||||
with open(path, "r", encoding="utf-8") as handle:
|
||||
return json.load(handle)
|
||||
except (OSError, ValueError):
|
||||
return {} if default is None else default
|
||||
|
||||
|
||||
def find_project_root(start):
|
||||
current = os.path.abspath(start or os.getcwd())
|
||||
while current and current != os.path.dirname(current):
|
||||
if os.path.isfile(os.path.join(current, ".casan", "config.json")):
|
||||
return current
|
||||
current = os.path.dirname(current)
|
||||
return None
|
||||
|
||||
|
||||
def harness_candidates(lock):
|
||||
values = []
|
||||
explicit = os.environ.get("CASAN_HARNESS_ROOT")
|
||||
if explicit:
|
||||
values.append(explicit)
|
||||
install = os.environ.get("CASAN_INSTALL_ROOT")
|
||||
if install:
|
||||
values.append(os.path.join(install, "packages", "casan-harness"))
|
||||
home = os.environ.get("CASAN_HOME")
|
||||
if home:
|
||||
values.append(os.path.join(home, "current", "packages", "casan-harness"))
|
||||
values.append(os.path.join(os.path.expanduser("~"), ".casan", "current",
|
||||
"packages", "casan-harness"))
|
||||
local = os.environ.get("LOCALAPPDATA")
|
||||
if local:
|
||||
values.append(os.path.join(local, "casan", "current",
|
||||
"packages", "casan-harness"))
|
||||
recorded = lock.get("install_root")
|
||||
if recorded:
|
||||
values.append(os.path.join(recorded, "packages", "casan-harness"))
|
||||
return values
|
||||
|
||||
|
||||
def resolve_harness(lock):
|
||||
for candidate in harness_candidates(lock):
|
||||
root = os.path.abspath(os.path.expanduser(candidate))
|
||||
if os.path.isfile(os.path.join(root, "scripts", "python",
|
||||
"agentic_bridge.py")):
|
||||
return root
|
||||
return None
|
||||
|
||||
|
||||
def live_hash(harness):
|
||||
module_path = os.path.join(harness, "scripts", "python", "harness_hash.py")
|
||||
spec = importlib.util.spec_from_file_location("casan_harness_hash", module_path)
|
||||
if spec is None or spec.loader is None:
|
||||
raise RuntimeError("harness_hash module is unavailable")
|
||||
module = importlib.util.module_from_spec(spec)
|
||||
spec.loader.exec_module(module)
|
||||
return module.compute(harness)
|
||||
|
||||
|
||||
def emit_failure(client, event, reason, enforce):
|
||||
"""Render a fail-closed response in the native client contract."""
|
||||
if client == "claude":
|
||||
if event == "PreToolUse":
|
||||
payload = {"hookSpecificOutput": {
|
||||
"hookEventName": "PreToolUse",
|
||||
"permissionDecision": "deny",
|
||||
"permissionDecisionReason": "CASAN bootstrap: " + reason,
|
||||
}}
|
||||
elif event == "UserPromptSubmit" and enforce:
|
||||
payload = {"decision": "block", "reason": "CASAN bootstrap: " + reason}
|
||||
elif event == "UserPromptSubmit":
|
||||
payload = {"hookSpecificOutput": {
|
||||
"hookEventName": "UserPromptSubmit",
|
||||
"additionalContext": "[CASAN observed_only] " + reason,
|
||||
}}
|
||||
else:
|
||||
payload = {}
|
||||
print(json.dumps(payload, ensure_ascii=False))
|
||||
return 0
|
||||
|
||||
if client == "codex":
|
||||
deny = event == "PreToolUse" or (event == "UserPromptSubmit" and enforce)
|
||||
if event == "PreToolUse":
|
||||
payload = {"systemMessage": "CASAN bootstrap denied tool: " + reason}
|
||||
code = 2
|
||||
elif deny:
|
||||
payload = {
|
||||
"continue": False,
|
||||
"stopReason": "CASAN bootstrap: " + reason,
|
||||
"systemMessage": "CASAN admission unavailable",
|
||||
}
|
||||
code = 0
|
||||
else:
|
||||
payload = {
|
||||
"continue": True,
|
||||
"systemMessage": "[CASAN observed_only] " + reason,
|
||||
}
|
||||
code = 0
|
||||
print(json.dumps(payload, ensure_ascii=False))
|
||||
return code
|
||||
|
||||
print(json.dumps({
|
||||
"decision": "block",
|
||||
"reason": "CASAN bootstrap: " + reason,
|
||||
"certification_strength": "observed_only",
|
||||
}, ensure_ascii=False))
|
||||
return 2
|
||||
|
||||
|
||||
def main(argv=None):
|
||||
args = list(argv if argv is not None else sys.argv[1:])
|
||||
client = None
|
||||
event = None
|
||||
index = 0
|
||||
while index < len(args):
|
||||
if args[index] == "--client" and index + 1 < len(args):
|
||||
client = args[index + 1]
|
||||
index += 2
|
||||
elif args[index] == "--event" and index + 1 < len(args):
|
||||
event = args[index + 1]
|
||||
index += 2
|
||||
else:
|
||||
index += 1
|
||||
if client not in ADAPTERS or not event:
|
||||
print(json.dumps({"decision": "block",
|
||||
"reason": "invalid CASAN hook arguments"}))
|
||||
return 64
|
||||
|
||||
root = find_project_root(os.environ.get("CASAN_APP_ROOT") or os.getcwd())
|
||||
if not root:
|
||||
return emit_failure(client, event, "project is not initialized", True)
|
||||
config = load_json(os.path.join(root, ".casan", "config.json"))
|
||||
mode = str(config.get("enforcement_mode") or "observe")
|
||||
enforce = mode == "enforce"
|
||||
enabled = config.get("clients") or []
|
||||
if client not in enabled:
|
||||
return emit_failure(client, event, "client is not enabled for this project",
|
||||
enforce)
|
||||
|
||||
os.environ["CASAN_APP_ROOT"] = root
|
||||
os.environ["CASAN_AGENTIC_BRIDGE_ENABLED"] = "1"
|
||||
os.environ["CASAN_AGENTIC_ENFORCEMENT_MODE"] = mode
|
||||
os.environ["CASAN_AGENTIC_INTEGRATION_MODE"] = str(
|
||||
config.get("integration_mode") or "project_hook")
|
||||
os.environ["CASAN_AGENTIC_CLIENT_ALLOWLIST"] = ",".join(
|
||||
{"claude": "claude-code", "codex": "codex",
|
||||
"vscode-copilot": "vscode"}.get(item, item) for item in enabled)
|
||||
|
||||
lock = load_json(os.path.join(root, ".casan", "version.lock"))
|
||||
harness = resolve_harness(lock)
|
||||
if not harness:
|
||||
return emit_failure(client, event, "pinned global harness not found",
|
||||
enforce)
|
||||
expected = lock.get("harness_hash")
|
||||
try:
|
||||
actual = live_hash(harness)
|
||||
except Exception as exc: # noqa: BLE001 - boundary must render native failure
|
||||
return emit_failure(client, event, "integrity check failed: %s" % exc,
|
||||
enforce)
|
||||
if not expected or str(expected).startswith("unavailable") or actual != expected:
|
||||
return emit_failure(client, event, "HARNESS_INTEGRITY_DRIFT", enforce)
|
||||
|
||||
os.environ["CASAN_HARNESS_ROOT"] = harness
|
||||
adapter_dir, adapter_name = ADAPTERS[client]
|
||||
adapter = os.path.realpath(os.path.join(harness, "adapters", adapter_dir,
|
||||
adapter_name))
|
||||
allowed_root = os.path.realpath(os.path.join(harness, "adapters")) + os.sep
|
||||
if not adapter.startswith(allowed_root) or not os.path.isfile(adapter):
|
||||
return emit_failure(client, event, "adapter is unavailable", enforce)
|
||||
|
||||
old_argv = sys.argv
|
||||
try:
|
||||
sys.argv = [adapter, "--event", event]
|
||||
runpy.run_path(adapter, run_name="__main__")
|
||||
finally:
|
||||
sys.argv = old_argv
|
||||
return 0
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
raise SystemExit(main())
|
||||
Reference in New Issue
Block a user