Read the script
#!/usr/bin/env python3
"""Small optim-plans hook guard for scoped read-only refinement sessions."""
from __future__ import annotations
import json
import os
import shlex
import sys
from pathlib import PurePosixPath
from typing import Any
RESERVED_GIT = {"commit", "reset", "clean", "worktree", "update-ref"}
def owned() -> bool:
return bool(
os.environ.get("OPTIM_PLANS_RUN_ID")
or os.environ.get("OPTIM_PLANS_WORKER_NONCE")
or os.environ.get("OPTIM_PLANS_STATE_PATH")
)
def validate_owned_state() -> str | None:
if not owned():
return None
state_path = os.environ.get("OPTIM_PLANS_STATE_PATH")
if not state_path:
return "owned optim-plans session is missing OPTIM_PLANS_STATE_PATH"
try:
with open(state_path, encoding="utf-8") as handle:
state = json.load(handle)
except (OSError, json.JSONDecodeError):
return "owned optim-plans session has invalid state"
if state.get("run_id") != os.environ.get("OPTIM_PLANS_RUN_ID"):
return "run ID does not match optim-plans state"
if state.get("worker_nonce") != os.environ.get("OPTIM_PLANS_WORKER_NONCE"):
return "session nonce does not match optim-plans state"
return None
def output(payload: dict[str, Any]) -> None:
print(json.dumps(payload, ensure_ascii=True, sort_keys=True))
def codex() -> bool:
return bool(os.environ.get("PLUGIN_DATA") or os.environ.get("OPTIM_PLANS_PLUGIN_ROOT"))
def pre_tool_output(decision: str, reason: str | None = None) -> dict[str, Any]:
if decision == "allow":
return {}
return {
"hookSpecificOutput": {
"hookEventName": "PreToolUse",
"permissionDecision": decision,
"permissionDecisionReason": reason or "",
}
}
def codex_session_context(context: str) -> dict[str, Any]:
return {"hookSpecificOutput": {"hookEventName": "SessionStart", "additionalContext": context}}
def normalize_hook_event(raw: dict[str, Any]) -> dict[str, Any]:
event = dict(raw)
if not isinstance(event.get("event"), str) and isinstance(raw.get("hook_event_name"), str):
event["event"] = raw["hook_event_name"]
if not isinstance(event.get("tool"), str) and isinstance(raw.get("tool_name"), str):
event["tool"] = raw["tool_name"]
tool_input = raw.get("tool_input")
if isinstance(tool_input, dict):
for key in ("command", "file_path", "path"):
if key not in event and isinstance(tool_input.get(key), str):
event[key] = tool_input[key]
return event
def in_scope(path: str) -> bool:
scopes = [item for item in os.environ.get("OPTIM_PLANS_SCOPES", "").split(os.pathsep) if item]
if not scopes:
return False
target = PurePosixPath(path.replace("\\", "/"))
if target.is_absolute() or ".." in target.parts:
return False
for scope in scopes:
scope_path = PurePosixPath(scope.replace("\\", "/"))
if target == scope_path or scope_path in target.parents:
return True
return False
def handle_session_start() -> dict[str, Any]:
ids = os.environ.get("OPTIM_PLANS_IDS", "")
scopes = os.environ.get("OPTIM_PLANS_SCOPES", "")
context = f"optim-plans scope: ids={ids or 'none'} scopes={scopes or 'repository'}"
if codex():
return codex_session_context(context)
return {"action": "inject", "context": context}
def is_reserved_git(command: str) -> bool:
try:
parts = shlex.split(command)
except ValueError:
return False
if len(parts) < 2 or parts[0] != "git":
return False
index = 1
options_with_values = {"-C", "-c", "--git-dir", "--work-tree"}
while index < len(parts) and parts[index].startswith("-"):
index += 2 if parts[index] in options_with_values else 1
return index < len(parts) and parts[index] in RESERVED_GIT
def handle_pre_tool(event: dict[str, Any]) -> dict[str, Any]:
if not owned():
return pre_tool_output("allow")
invalid = validate_owned_state()
if invalid:
return pre_tool_output("deny", invalid)
tool = event.get("tool")
command = event.get("command")
if tool in {"Shell", "Bash"} and isinstance(command, str) and is_reserved_git(command):
return pre_tool_output("deny", "optim-plans reserves destructive Git state changes")
if tool in {"Write", "Edit", "MultiEdit"}:
path = event.get("path") or event.get("file_path")
if not isinstance(path, str):
return pre_tool_output("deny", "write path is missing")
if not in_scope(path):
return pre_tool_output("deny", f"write path {path!r} is outside OPTIM_PLANS_SCOPES")
return pre_tool_output("allow")
def handle(raw: dict[str, Any]) -> dict[str, Any]:
event = normalize_hook_event(raw)
event_name = event.get("event")
if event_name == "SessionStart":
if not owned():
return {"action": "noop"}
invalid = validate_owned_state()
if invalid:
return {"action": "deny", "reason": invalid}
return handle_session_start()
if event_name == "PreToolUse":
return handle_pre_tool(event)
return {"action": "noop"}
def main() -> int:
try:
raw = json.load(sys.stdin)
except json.JSONDecodeError:
output({"action": "deny", "reason": "invalid hook JSON"})
return 0
output(handle(raw if isinstance(raw, dict) else {}))
return 0
if __name__ == "__main__":
raise SystemExit(main())