File size: 7,123 Bytes
aa73ff9 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 | """Claude Code command hook. Classifies Bash input; never executes that input."""
import argparse
import fcntl
import hashlib
import json
import math
import os
import signal
import sys
import tempfile
import time
import urllib.request
from pathlib import Path
MAX_BYTES=65536
def output(decision,reason):
return {'hookSpecificOutput':{'hookEventName':'PreToolUse','permissionDecision':decision.lower(),'permissionDecisionReason':'Laya: '+reason}}
def atomic(path,value):
fd,name=tempfile.mkstemp(dir=path.parent)
try:
with os.fdopen(fd,'w') as f:json.dump(value,f)
os.replace(name,path)
finally:
if os.path.exists(name):os.unlink(name)
def session_file(profile,event):
sid=event.get('session_id')
if not isinstance(sid,str) or not sid or len(sid)>256:raise ValueError('Missing session ID')
return Path(profile['state_dir'])/(hashlib.sha256(sid.encode()).hexdigest()+'.json')
def prompt_event(profile,event):
path=session_file(profile,event)
# Only the exact launch task is bound initially. Later prompts, including
# autonomous messages, invalidate auto-approval rather than gain authority.
with (path.parent/'context.lock').open('a') as lock:
fcntl.flock(lock,fcntl.LOCK_EX)
previous=json.loads(path.read_text()) if path.exists() else None
valid=previous is None and event.get('prompt')==profile['intent'] and Path(event.get('cwd','')).resolve()==Path(profile['project'])
atomic(path,{'valid':valid,'created':time.time(),'project':profile['project']})
return {}
def query(profile,state,environ=None):
environ=os.environ if environ is None else environ
key=(environ.get(profile['api_key_env'],'') if profile.get('api_key_env') else Path(profile['api_key_file']).read_text()).strip()
if not key:raise ValueError('Missing API credential')
body={'model':profile['model'],'state':state,'questions':profile['questions']}
request=urllib.request.Request(profile['endpoint'],data=json.dumps(body).encode(),headers={'Content-Type':'application/json','Authorization':'Bearer '+key})
# Use the selected endpoint directly; do not leak prompts into ambient proxies.
opener=urllib.request.build_opener(urllib.request.ProxyHandler({}),NoRedirect())
with opener.open(request,timeout=profile['timeout_seconds']) as r:
raw=r.read(MAX_BYTES+1)
if len(raw)>MAX_BYTES:raise ValueError('Oversized response')
result=json.loads(raw)
if result.get('error'):raise ValueError('Classifier unavailable')
answers=result.get('answers')
if not isinstance(answers,dict) or set(answers)!={'permission'}:raise ValueError('Malformed answer')
answer=answers['permission'];decision=answer.get('choice');p=answer.get('probabilities')
if answer.get('type')!='choice' or decision not in ('ALLOW','ASK','DENY'):raise ValueError('Invalid decision')
if not isinstance(p,dict) or set(p)!={'ALLOW','ASK','DENY'}:raise ValueError('Invalid probabilities')
if any(isinstance(v,bool) or not isinstance(v,(int,float)) or not math.isfinite(v) or not 0<=v<=1 for v in p.values()) or abs(sum(p.values())-1)>.002:raise ValueError('Invalid probability distribution')
# Keep the checkpoint's abstention. Independently enforce a minimum floor.
if decision=='ALLOW' and p['ALLOW']<profile['minimum_allow_probability']:decision='ASK'
return decision
class NoRedirect(urllib.request.HTTPRedirectHandler):
def redirect_request(self,*args,**kwargs):raise ValueError('Redirect refused')
def pretool(profile,event,client=query):
if event.get('tool_name')!='Bash':return {}
if Path(event.get('cwd','')).resolve()!=Path(profile['project']):return output('ASK','project context changed; manual review required')
path=session_file(profile,event)
context=json.loads(path.read_text()) if path.exists() else {}
if not context.get('valid'):return output('ASK','missing or changed trusted task; start a new adapter session for automatic decisions')
if event.get('agent_id'):return output('ASK','subagent context requires manual review')
if event.get('permission_mode','default')!='default':return output('ASK','use default permission mode')
tool=event.get('tool_input')
if not isinstance(tool,dict) or not isinstance(tool.get('command'),str) or not tool['command'].strip():raise ValueError('Missing Bash command')
if tool.get('dangerouslyDisableSandbox'):return output('ASK','request to leave the sandbox requires manual review')
state={'tool':'Bash','command':tool['command'],'trusted_user_request':profile['intent'],'environment':profile['environment'],
'trusted_context':'Trusted launch configuration. Project: '+profile['project']+'. '+profile['trusted_context'],
'untrusted_tool_output':json.dumps({'proposed_tool_metadata':{k:v for k,v in tool.items() if k!='command'}},ensure_ascii=False)}
decision=client(profile,state)
if profile['environment']=='production' and decision=='ALLOW':decision='ASK'
# A concurrent new prompt can invalidate authority while inference is running.
fresh=json.loads(path.read_text())
if not fresh.get('valid'):return output('ASK','trusted task changed during inference')
if profile.get('observe_only'):return {'systemMessage':'Laya observation: '+decision+'; normal Claude Code permission flow applies.'}
reasons={'ALLOW':'stage/local operation approved under the fixed permission policy',
'ASK':'manual confirmation required by permission policy',
'DENY':'operation blocked by permission policy; revise the command or clarify the task'}
return output(decision,reasons[decision])
def main():
p=argparse.ArgumentParser();p.add_argument('--profile',type=Path,required=True);a=p.parse_args()
event={};profile={}
def expired(signum,frame):raise TimeoutError('Hook deadline')
signal.signal(signal.SIGALRM,expired);signal.alarm(10)
try:
raw=sys.stdin.buffer.read(MAX_BYTES+1)
if len(raw)>MAX_BYTES:raise ValueError('Oversized hook input')
event=json.loads(raw);profile=json.loads(a.profile.read_text())
if event.get('hook_event_name')=='UserPromptSubmit':result=prompt_event(profile,event)
elif event.get('hook_event_name')=='PreToolUse':result=pretool(profile,event)
else:raise ValueError('Unsupported hook event')
except Exception:
# Do not echo exceptions: URLs, commands, credentials or bodies may be in them.
if event.get('hook_event_name')=='UserPromptSubmit':
print(json.dumps({'decision':'block','reason':'Laya: could not bind task context. Restart the adapter session.'}));return
result=output('ASK','classifier/context unavailable or malformed; manual review required')
finally:signal.alarm(0)
if profile.get('observe_only') and 'hookSpecificOutput' in result:
result={'systemMessage':'Laya observation: '+result['hookSpecificOutput'].get('permissionDecisionReason','manual review')+'; normal Claude Code permission flow applies.'}
print(json.dumps(result),flush=True)
if __name__=='__main__':main()
|