227 lines
8.3 KiB
Python
227 lines
8.3 KiB
Python
#!/usr/bin/env python3
|
|
"""Project-local bootstrap for CASAN's managed or vendored agentic adapters.
|
|
|
|
This file is intentionally small and stdlib-only. It is the stable command
|
|
target committed by `casan init`; the policy implementation is resolved from
|
|
the project lock. 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 managed/vendored Core 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, project_root):
|
|
values = []
|
|
runtime_path = lock.get("runtime_path")
|
|
if lock.get("runtime_mode") == "vendored":
|
|
if not isinstance(runtime_path, str) or os.path.isabs(runtime_path):
|
|
return values
|
|
normalized = os.path.normpath(runtime_path)
|
|
candidate = os.path.abspath(os.path.join(project_root, normalized))
|
|
try:
|
|
inside_project = os.path.commonpath(
|
|
(project_root, candidate)) == project_root
|
|
except ValueError:
|
|
inside_project = False
|
|
if not inside_project:
|
|
return values
|
|
return [os.path.join(candidate, "packages", "casan-harness")]
|
|
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, project_root):
|
|
for candidate in harness_candidates(lock, project_root):
|
|
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, root)
|
|
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())
|