Coverage for src/lexigram/auth/policies/evaluator.py: 81%

1"""Condition evaluators for ABAC Policy Engine.""" 

2 

3from __future__ import annotations 

4 

5import re 

6import threading 

7from typing import TYPE_CHECKING, Any, Protocol, cast 

8 

9if TYPE_CHECKING: 

10 from collections.abc import Callable 

11 

12 from lexigram.auth.policies.types import Condition 

13 

14 

15class OperatorHandlerProtocol(Protocol): 

16 """Protocol for operator handlers.""" 

17 

18 def compare(self, actual: Any, expected: Any) -> bool: 

19 """Compare actual value against expected.""" 

20 ... 

21 

22 

23class EqualsOperator: 

24 def compare(self, actual: Any, expected: Any) -> bool: 

25 return cast("bool", actual == expected) 

26 

27 

28class NotEqualsOperator: 

29 def compare(self, actual: Any, expected: Any) -> bool: 

30 return cast("bool", actual != expected) 

31 

32 

33class ContainsOperator: 

34 def compare(self, actual: Any, expected: Any) -> bool: 

35 return expected in actual 

36 

37 

38class NotContainsOperator: 

39 def compare(self, actual: Any, expected: Any) -> bool: 

40 return expected not in actual 

41 

42 

43class InOperator: 

44 def compare(self, actual: Any, expected: Any) -> bool: 

45 return actual in expected 

46 

47 

48class NotInOperator: 

49 def compare(self, actual: Any, expected: Any) -> bool: 

50 return actual not in expected 

51 

52 

53class MatchesOperator: 

54 def compare(self, actual: Any, expected: Any) -> bool: 

55 return bool(re.match(str(expected), str(actual))) 

56 

57 

58class GreaterThanOperator: 

59 def compare(self, actual: Any, expected: Any) -> bool: 

60 try: 

61 return cast("bool", actual > expected) 

62 except TypeError: 

63 return False 

64 

65 

66class LessThanOperator: 

67 def compare(self, actual: Any, expected: Any) -> bool: 

68 try: 

69 return cast("bool", actual < expected) 

70 except TypeError: 

71 return False 

72 

73 

74class GreaterThanOrEqualsOperator: 

75 def compare(self, actual: Any, expected: Any) -> bool: 

76 try: 

77 return cast("bool", actual >= expected) 

78 except TypeError: 

79 return False 

80 

81 

82class LessThanOrEqualsOperator: 

83 def compare(self, actual: Any, expected: Any) -> bool: 

84 try: 

85 return cast("bool", actual <= expected) 

86 except TypeError: 

87 return False 

88 

89 

90class OperatorRegistry: 

91 """Registry for condition operators.""" 

92 

93 def __init__(self) -> None: 

94 self._lock = threading.Lock() 

95 self._handlers: dict[str, OperatorHandlerProtocol] = {} 

96 

97 @classmethod 

98 def with_defaults(cls) -> OperatorRegistry: 

99 """Create a registry pre-loaded with the standard operator handlers.""" 

100 instance = cls() 

101 instance._register_default_handlers() 

102 return instance 

103 

104 def _register_default_handlers(self) -> None: 

105 self._handlers = { 

106 "equals": EqualsOperator(), 

107 "not_equals": NotEqualsOperator(), 

108 "contains": ContainsOperator(), 

109 "not_contains": NotContainsOperator(), 

110 "in": InOperator(), 

111 "not_in": NotInOperator(), 

112 "matches": MatchesOperator(), 

113 "greater_than": GreaterThanOperator(), 

114 "less_than": LessThanOperator(), 

115 "greater_than_or_equals": GreaterThanOrEqualsOperator(), 

116 "less_than_or_equals": LessThanOrEqualsOperator(), 

117 } 

118 

119 def register_handler(self, operator: str, handler: OperatorHandlerProtocol) -> None: 

120 """Register a custom operator handler.""" 

121 with self._lock: 

122 self._handlers[operator] = handler 

123 

124 def compare(self, actual: Any, operator: str, expected: Any) -> bool: 

125 """Compare using the registered handler for the operator.""" 

126 with self._lock: 

127 handler = self._handlers.get(operator) 

128 if handler: 

129 return handler.compare(actual, expected) 

130 return False 

131 

132 

133class ConditionEvaluator: 

134 """Evaluates individual policy conditions against a request context.""" 

135 

136 def __init__(self) -> None: 

137 self._operator_registry = OperatorRegistry.with_defaults() 

138 # Cache compiled accessor functions keyed by attribute path string. 

139 # Each accessor is a Callable[[Any], Any] that traverses the path 

140 # without re-splitting on every evaluation call. 

141 self._path_cache: dict[str, Callable[[Any], Any]] = {} 

142 

143 @staticmethod 

144 def _compile_path(path: str) -> Callable[[Any], Any]: 

145 """Compile an attribute path string into a reusable accessor function. 

146 

147 Args: 

148 path: Dot-separated attribute path, e.g. ``"user.department"``. 

149 

150 Returns: 

151 A callable that accepts a context dict or object and traverses 

152 the path, returning ``None`` if any segment is absent. 

153 """ 

154 parts = path.split(".") 

155 

156 def accessor(obj: Any) -> Any: 

157 current = obj 

158 for part in parts: 

159 if isinstance(current, dict): 

160 if part not in current: 

161 return None 

162 current = current[part] 

163 elif hasattr(current, part): 

164 current = getattr(current, part) 

165 else: 

166 return None 

167 return current 

168 

169 return accessor 

170 

171 def evaluate(self, condition: Condition, context: dict[str, Any]) -> bool: 

172 """Evaluate a single condition against the context.""" 

173 actual_val = self._resolve_attribute(condition.attribute, context) 

174 

175 # Variable substitution in expected value (e.g., "expected": "${user.id}") 

176 expected_val = condition.value 

177 if ( 

178 isinstance(expected_val, str) 

179 and expected_val.startswith("${") 

180 and expected_val.endswith("}") 

181 ): 

182 var_path = expected_val[2:-1] 

183 expected_val = self._resolve_attribute(var_path, context) 

184 

185 return self._operator_registry.compare( 

186 actual_val, 

187 condition.operator, 

188 expected_val, 

189 ) 

190 

191 def _resolve_attribute(self, path: str, context: dict[str, Any]) -> Any: 

192 """Resolve a nested attribute from the context using a cached accessor.""" 

193 accessor = self._path_cache.get(path) 

194 if accessor is None: 

195 accessor = self._compile_path(path) 

196 self._path_cache[path] = accessor 

197 return accessor(context) 

198 

199 

200__all__ = [ 

201 "ConditionEvaluator", 

202 "ContainsOperator", 

203 "EqualsOperator", 

204 "GreaterThanOperator", 

205 "GreaterThanOrEqualsOperator", 

206 "InOperator", 

207 "LessThanOperator", 

208 "LessThanOrEqualsOperator", 

209 "MatchesOperator", 

210 "NotContainsOperator", 

211 "NotEqualsOperator", 

212 "NotInOperator", 

213 "OperatorHandlerProtocol", 

214 "OperatorRegistry", 

215]