surshar / claude-code

Claude Code is an agentic coding tool that lives in your terminal, understands your codebase, and helps you code faster by executing routine tasks, explaining complex code, and handling git workflows - all through natural language commands.

Built by Anthropic · github.com/anthropics

claude-code / plugins / hookify / core / rule_engine.py

Claudefix(hookify): add size limits to transcript reading to prevent OOM 449aa9d · on Jan 5, 2026 History

359 lines · 12.0 KBRaw

1	#!/usr/bin/env python3
2	"""Rule evaluation engine for hookify plugin."""
3
4	import os
5	import re
6	import sys
7	from functools import lru_cache
8	from typing import List, Dict, Any, Optional
9
10	# Import from local module
11	from hookify.core.config_loader import Rule, Condition
12
13	# Maximum transcript size to load into memory (10MB default)
14	# For larger files, only the tail is read to prevent OOM
15	MAX_TRANSCRIPT_SIZE_BYTES = 10 * 1024 * 1024 # 10MB
16
17	# Size threshold for warning about large transcripts
18	TRANSCRIPT_WARNING_SIZE_BYTES = 5 * 1024 * 1024 # 5MB
19
20
21	# Cache compiled regexes (max 128 patterns)
22	@lru_cache(maxsize=128)
23	def compile_regex(pattern: str) -> re.Pattern:
24	"""Compile regex pattern with caching.
25
26	Args:
27	pattern: Regex pattern string
28
29	Returns:
30	Compiled regex pattern
31	"""
32	return re.compile(pattern, re.IGNORECASE)
33
34
35	def read_transcript_safely(transcript_path: str) -> str:
36	"""Read transcript file with size limits to prevent OOM.
37
38	For large transcripts (>10MB), only reads the tail of the file
39	to prevent memory exhaustion. This preserves the most recent
40	conversation context which is typically what rules care about.
41
42	Args:
43	transcript_path: Path to the transcript file
44
45	Returns:
46	Transcript content as string, possibly truncated for large files
47	"""
48	try:
49	file_size = os.path.getsize(transcript_path)
50
51	# Warn about large transcripts
52	if file_size > TRANSCRIPT_WARNING_SIZE_BYTES:
53	size_mb = file_size / (1024 * 1024)
54	print(f"Warning: Large transcript ({size_mb:.1f}MB): {transcript_path}", file=sys.stderr)
55
56	# For files within limit, read normally
57	if file_size <= MAX_TRANSCRIPT_SIZE_BYTES:
58	with open(transcript_path, 'r') as f:
59	return f.read()
60
61	# For large files, read only the tail to prevent OOM
62	size_mb = file_size / (1024 * 1024)
63	limit_mb = MAX_TRANSCRIPT_SIZE_BYTES / (1024 * 1024)
64	print(f"Warning: Transcript too large ({size_mb:.1f}MB), reading last {limit_mb:.0f}MB only", file=sys.stderr)
65
66	with open(transcript_path, 'r') as f:
67	# Seek to position near end, leaving room for MAX_TRANSCRIPT_SIZE_BYTES
68	f.seek(file_size - MAX_TRANSCRIPT_SIZE_BYTES)
69	# Skip partial line at seek position
70	f.readline()
71	return f.read()
72
73	except FileNotFoundError:
74	print(f"Warning: Transcript file not found: {transcript_path}", file=sys.stderr)
75	return ''
76	except PermissionError:
77	print(f"Warning: Permission denied reading transcript: {transcript_path}", file=sys.stderr)
78	return ''
79	except (IOError, OSError) as e:
80	print(f"Warning: Error reading transcript {transcript_path}: {e}", file=sys.stderr)
81	return ''
82	except UnicodeDecodeError as e:
83	print(f"Warning: Encoding error in transcript {transcript_path}: {e}", file=sys.stderr)
84	return ''
85
86
87	class RuleEngine:
88	"""Evaluates rules against hook input data."""
89
90	def __init__(self):
91	"""Initialize rule engine."""
92	# No need for instance cache anymore - using global lru_cache
93	pass
94
95	def evaluate_rules(self, rules: List[Rule], input_data: Dict[str, Any]) -> Dict[str, Any]:
96	"""Evaluate all rules and return combined results.
97
98	Checks all rules and accumulates matches. Blocking rules take priority
99	over warning rules. All matching rule messages are combined.
100
101	Args:
102	rules: List of Rule objects to evaluate
103	input_data: Hook input JSON (tool_name, tool_input, etc.)
104
105	Returns:
106	Response dict with systemMessage, hookSpecificOutput, etc.
107	Empty dict {} if no rules match.
108	"""
109	hook_event = input_data.get('hook_event_name', '')
110	blocking_rules = []
111	warning_rules = []
112
113	for rule in rules:
114	if self._rule_matches(rule, input_data):
115	if rule.action == 'block':
116	blocking_rules.append(rule)
117	else:
118	warning_rules.append(rule)
119
120	# If any blocking rules matched, block the operation
121	if blocking_rules:
122	messages = [f"[{r.name}]\n{r.message}" for r in blocking_rules]
123	combined_message = "\n\n".join(messages)
124
125	# Use appropriate blocking format based on event type
126	if hook_event == 'Stop':
127	return {
128	"decision": "block",
129	"reason": combined_message,
130	"systemMessage": combined_message
131	}
132	elif hook_event in ['PreToolUse', 'PostToolUse']:
133	return {
134	"hookSpecificOutput": {
135	"hookEventName": hook_event,
136	"permissionDecision": "deny"
137	},
138	"systemMessage": combined_message
139	}
140	else:
141	# For other events, just show message
142	return {
143	"systemMessage": combined_message
144	}
145
146	# If only warnings, show them but allow operation
147	if warning_rules:
148	messages = [f"[{r.name}]\n{r.message}" for r in warning_rules]
149	return {
150	"systemMessage": "\n\n".join(messages)
151	}
152
153	# No matches - allow operation
154	return {}
155
156	def _rule_matches(self, rule: Rule, input_data: Dict[str, Any]) -> bool:
157	"""Check if rule matches input data.
158
159	Args:
160	rule: Rule to evaluate
161	input_data: Hook input data
162
163	Returns:
164	True if rule matches, False otherwise
165	"""
166	# Extract tool information
167	tool_name = input_data.get('tool_name', '')
168	tool_input = input_data.get('tool_input', {})
169
170	# Check tool matcher if specified
171	if rule.tool_matcher:
172	if not self._matches_tool(rule.tool_matcher, tool_name):
173	return False
174
175	# If no conditions, don't match
176	# (Rules must have at least one condition to be valid)
177	if not rule.conditions:
178	return False
179
180	# All conditions must match
181	for condition in rule.conditions:
182	if not self._check_condition(condition, tool_name, tool_input, input_data):
183	return False
184
185	return True
186
187	def _matches_tool(self, matcher: str, tool_name: str) -> bool:
188	"""Check if tool_name matches the matcher pattern.
189
190	Args:
191	matcher: Pattern like "Bash", "Edit\|Write", "*"
192	tool_name: Actual tool name
193
194	Returns:
195	True if matches
196	"""
197	if matcher == '*':
198	return True
199
200	# Split on \| for OR matching
201	patterns = matcher.split('\|')
202	return tool_name in patterns
203
204	def _check_condition(self, condition: Condition, tool_name: str,
205	tool_input: Dict[str, Any], input_data: Dict[str, Any] = None) -> bool:
206	"""Check if a single condition matches.
207
208	Args:
209	condition: Condition to check
210	tool_name: Tool being used
211	tool_input: Tool input dict
212	input_data: Full hook input data (for Stop events, etc.)
213
214	Returns:
215	True if condition matches
216	"""
217	# Extract the field value to check
218	field_value = self._extract_field(condition.field, tool_name, tool_input, input_data)
219	if field_value is None:
220	return False
221
222	# Apply operator
223	operator = condition.operator
224	pattern = condition.pattern
225
226	if operator == 'regex_match':
227	return self._regex_match(pattern, field_value)
228	elif operator == 'contains':
229	return pattern in field_value
230	elif operator == 'equals':
231	return pattern == field_value
232	elif operator == 'not_contains':
233	return pattern not in field_value
234	elif operator == 'starts_with':
235	return field_value.startswith(pattern)
236	elif operator == 'ends_with':
237	return field_value.endswith(pattern)
238	else:
239	# Unknown operator
240	return False
241
242	def _extract_field(self, field: str, tool_name: str,
243	tool_input: Dict[str, Any], input_data: Dict[str, Any] = None) -> Optional[str]:
244	"""Extract field value from tool input or hook input data.
245
246	Args:
247	field: Field name like "command", "new_text", "file_path", "reason", "transcript"
248	tool_name: Tool being used (may be empty for Stop events)
249	tool_input: Tool input dict
250	input_data: Full hook input (for accessing transcript_path, reason, etc.)
251
252	Returns:
253	Field value as string, or None if not found
254	"""
255	# Direct tool_input fields
256	if field in tool_input:
257	value = tool_input[field]
258	if isinstance(value, str):
259	return value
260	return str(value)
261
262	# For Stop events and other non-tool events, check input_data
263	if input_data:
264	# Stop event specific fields
265	if field == 'reason':
266	return input_data.get('reason', '')
267	elif field == 'transcript':
268	# Read transcript file with size limits to prevent OOM
269	transcript_path = input_data.get('transcript_path')
270	if transcript_path:
271	return read_transcript_safely(transcript_path)
272	elif field == 'user_prompt':
273	# For UserPromptSubmit events
274	return input_data.get('user_prompt', '')
275
276	# Handle special cases by tool type
277	if tool_name == 'Bash':
278	if field == 'command':
279	return tool_input.get('command', '')
280
281	elif tool_name in ['Write', 'Edit']:
282	if field == 'content':
283	# Write uses 'content', Edit has 'new_string'
284	return tool_input.get('content') or tool_input.get('new_string', '')
285	elif field == 'new_text' or field == 'new_string':
286	return tool_input.get('new_string', '')
287	elif field == 'old_text' or field == 'old_string':
288	return tool_input.get('old_string', '')
289	elif field == 'file_path':
290	return tool_input.get('file_path', '')
291
292	elif tool_name == 'MultiEdit':
293	if field == 'file_path':
294	return tool_input.get('file_path', '')
295	elif field in ['new_text', 'content']:
296	# Concatenate all edits
297	edits = tool_input.get('edits', [])
298	return ' '.join(e.get('new_string', '') for e in edits)
299
300	return None
301
302	def _regex_match(self, pattern: str, text: str) -> bool:
303	"""Check if pattern matches text using regex.
304
305	Args:
306	pattern: Regex pattern
307	text: Text to match against
308
309	Returns:
310	True if pattern matches
311	"""
312	try:
313	# Use cached compiled regex (LRU cache with max 128 patterns)
314	regex = compile_regex(pattern)
315	return bool(regex.search(text))
316
317	except re.error as e:
318	print(f"Invalid regex pattern '{pattern}': {e}", file=sys.stderr)
319	return False
320
321
322	# For testing
323	if __name__ == '__main__':
324	from hookify.core.config_loader import Condition, Rule
325
326	# Test rule evaluation
327	rule = Rule(
328	name="test-rm",
329	enabled=True,
330	event="bash",
331	conditions=[
332	Condition(field="command", operator="regex_match", pattern=r"rm\s+-rf")
333	],
334	message="Dangerous rm command!"
335	)
336
337	engine = RuleEngine()
338
339	# Test matching input
340	test_input = {
341	"tool_name": "Bash",
342	"tool_input": {
343	"command": "rm -rf /tmp/test"
344	}
345	}
346
347	result = engine.evaluate_rules([rule], test_input)
348	print("Match result:", result)
349
350	# Test non-matching input
351	test_input2 = {
352	"tool_name": "Bash",
353	"tool_input": {
354	"command": "ls -la"
355	}
356	}
357
358	result2 = engine.evaluate_rules([rule], test_input2)
359	print("Non-match result:", result2)