| """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) |
| |
| |
| 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}) |
| |
| 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') |
| |
| 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' |
| |
| 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: |
| |
| 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() |
|
|
