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(security): Resolve symlinks before checking deny rules (CVE-2025-59829) ff0fdc0 · on Jan 8, 2026 History

357 lines · 12.4 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 _resolve_symlink_path(self, file_path: str) -> str:
145	"""Resolve symlinks in file path to get canonical path.
146
147	Security fix for CVE-2025-59829: Deny rules could be bypassed by creating
148	a symlink to a restricted file. This method resolves the symlink to its
149	target path so that deny rules are checked against the actual file.
150
151	Args:
152	file_path: The file path that may contain symlinks
153
154	Returns:
155	The canonical path with symlinks resolved, or original path if
156	resolution fails (e.g., file doesn't exist yet)
157	"""
158	import os
159
160	if not file_path:
161	return file_path
162
163	try:
164	# Expand user home directory first
165	expanded_path = os.path.expanduser(file_path)
166
167	# Use realpath to resolve all symlinks and get canonical path
168	# This handles nested symlinks and relative path components
169	resolved = os.path.realpath(expanded_path)
170
171	return resolved
172	except (OSError, ValueError):
173	# If resolution fails (e.g., permission denied, invalid path),
174	# return the original path to avoid blocking legitimate operations
175	return file_path
176
177	def _check_condition(self, condition: Condition, tool_name: str,
178	tool_input: Dict[str, Any], input_data: Dict[str, Any] = None) -> bool:
179	"""Check if a single condition matches.
180
181	Args:
182	condition: Condition to check
183	tool_name: Tool being used
184	tool_input: Tool input dict
185	input_data: Full hook input data (for Stop events, etc.)
186
187	Returns:
188	True if condition matches
189	"""
190	# Extract the field value to check
191	field_value = self._extract_field(condition.field, tool_name, tool_input, input_data)
192	if field_value is None:
193	return False
194
195	# Apply operator
196	operator = condition.operator
197	pattern = condition.pattern
198
199	if operator == 'regex_match':
200	return self._regex_match(pattern, field_value)
201	elif operator == 'contains':
202	return pattern in field_value
203	elif operator == 'equals':
204	return pattern == field_value
205	elif operator == 'not_contains':
206	return pattern not in field_value
207	elif operator == 'starts_with':
208	return field_value.startswith(pattern)
209	elif operator == 'ends_with':
210	return field_value.endswith(pattern)
211	else:
212	# Unknown operator
213	return False
214
215	def _extract_field(self, field: str, tool_name: str,
216	tool_input: Dict[str, Any], input_data: Dict[str, Any] = None) -> Optional[str]:
217	"""Extract field value from tool input or hook input data.
218
219	Args:
220	field: Field name like "command", "new_text", "file_path", "reason", "transcript"
221	tool_name: Tool being used (may be empty for Stop events)
222	tool_input: Tool input dict
223	input_data: Full hook input (for accessing transcript_path, reason, etc.)
224
225	Returns:
226	Field value as string, or None if not found
227	"""
228	# Direct tool_input fields
229	if field in tool_input:
230	value = tool_input[field]
231	if isinstance(value, str):
232	# Security fix: resolve symlinks for file_path fields to prevent bypass
233	# CVE-2025-59829: Deny rules could be bypassed via symlinks
234	if field == 'file_path':
235	value = self._resolve_symlink_path(value)
236	return value
237	return str(value)
238
239	# For Stop events and other non-tool events, check input_data
240	if input_data:
241	# Stop event specific fields
242	if field == 'reason':
243	return input_data.get('reason', '')
244	elif field == 'transcript':
245	# Read transcript file if path provided
246	transcript_path = input_data.get('transcript_path')
247	if transcript_path:
248	try:
249	with open(transcript_path, 'r') as f:
250	return f.read()
251	except FileNotFoundError:
252	print(f"Warning: Transcript file not found: {transcript_path}", file=sys.stderr)
253	return ''
254	except PermissionError:
255	print(f"Warning: Permission denied reading transcript: {transcript_path}", file=sys.stderr)
256	return ''
257	except (IOError, OSError) as e:
258	print(f"Warning: Error reading transcript {transcript_path}: {e}", file=sys.stderr)
259	return ''
260	except UnicodeDecodeError as e:
261	print(f"Warning: Encoding error in transcript {transcript_path}: {e}", file=sys.stderr)
262	return ''
263	elif field == 'user_prompt':
264	# For UserPromptSubmit events
265	return input_data.get('user_prompt', '')
266
267	# Handle special cases by tool type
268	if tool_name == 'Bash':
269	if field == 'command':
270	return tool_input.get('command', '')
271
272	elif tool_name in ['Write', 'Edit']:
273	if field == 'content':
274	# Write uses 'content', Edit has 'new_string'
275	return tool_input.get('content') or tool_input.get('new_string', '')
276	elif field == 'new_text' or field == 'new_string':
277	return tool_input.get('new_string', '')
278	elif field == 'old_text' or field == 'old_string':
279	return tool_input.get('old_string', '')
280	elif field == 'file_path':
281	# Security fix: resolve symlinks to prevent deny rule bypass
282	return self._resolve_symlink_path(tool_input.get('file_path', ''))
283
284	elif tool_name == 'Read':
285	# Security fix for CVE-2025-59829: Read tool symlink bypass
286	if field == 'file_path':
287	return self._resolve_symlink_path(tool_input.get('file_path', ''))
288
289	elif tool_name == 'MultiEdit':
290	if field == 'file_path':
291	# Security fix: resolve symlinks to prevent deny rule bypass
292	return self._resolve_symlink_path(tool_input.get('file_path', ''))
293	elif field in ['new_text', 'content']:
294	# Concatenate all edits
295	edits = tool_input.get('edits', [])
296	return ' '.join(e.get('new_string', '') for e in edits)
297
298	return None
299
300	def _regex_match(self, pattern: str, text: str) -> bool:
301	"""Check if pattern matches text using regex.
302
303	Args:
304	pattern: Regex pattern
305	text: Text to match against
306
307	Returns:
308	True if pattern matches
309	"""
310	try:
311	# Use cached compiled regex (LRU cache with max 128 patterns)
312	regex = compile_regex(pattern)
313	return bool(regex.search(text))
314
315	except re.error as e:
316	print(f"Invalid regex pattern '{pattern}': {e}", file=sys.stderr)
317	return False
318
319
320	# For testing
321	if __name__ == '__main__':
322	from hookify.core.config_loader import Condition, Rule
323
324	# Test rule evaluation
325	rule = Rule(
326	name="test-rm",
327	enabled=True,
328	event="bash",
329	conditions=[
330	Condition(field="command", operator="regex_match", pattern=r"rm\s+-rf")
331	],
332	message="Dangerous rm command!"
333	)
334
335	engine = RuleEngine()
336
337	# Test matching input
338	test_input = {
339	"tool_name": "Bash",
340	"tool_input": {
341	"command": "rm -rf /tmp/test"
342	}
343	}
344
345	result = engine.evaluate_rules([rule], test_input)
346	print("Match result:", result)
347
348	# Test non-matching input
349	test_input2 = {
350	"tool_name": "Bash",
351	"tool_input": {
352	"command": "ls -la"
353	}
354	}
355
356	result2 = engine.evaluate_rules([rule], test_input2)
357	print("Non-match result:", result2)