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

313 lines · 10.5 KBRaw

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