Coverage for src/lexigram/web/security/guards.py: 32%

1""" 

2GuardProtocol system for request authorization. 

3 

4Guards are executed before route handlers to validate authorization. 

5""" 

6 

7from __future__ import annotations 

8 

9from abc import ABC 

10from dataclasses import dataclass 

11from functools import wraps 

12import inspect 

13from typing import Any 

14 

15from starlette.requests import Request 

16from starlette.responses import JSONResponse 

17 

18from lexigram.contracts import AuthorizerProtocol 

19from lexigram.contracts.web.guard import GuardProtocol 

20from lexigram.result import Err, Ok, Result 

21from lexigram.web.protocols import ( 

22 ExecutionContextProtocol as _BaseExecutionContext, 

23) 

24 

25 

26@dataclass(frozen=True) 

27class GuardRejection: 

28 """Represents a guard denial with error details.""" 

29 

30 error: str = "Unauthorized" 

31 message: str = "Access denied" 

32 status_code: int = 403 

33 

34 

35def _extract_request(context: Any) -> Any: 

36 """Return the raw request from either an ExecutionContextProtocol or a plain request.""" 

37 if isinstance(context, _BaseExecutionContext): 

38 return context.request 

39 return context 

40 

41 

42class AuthGuard(ABC): 

43 """ 

44 GuardProtocol that checks if user is authenticated. 

45 

46 Looks for user in request.state.user (set by auth middleware). 

47 """ 

48 

49 async def can_activate(self, context: Any) -> bool: 

50 """Check if user is authenticated.""" 

51 request = _extract_request(context) 

52 user = getattr(request.state, "user", None) 

53 return user is not None 

54 

55 

56class RoleGuard(ABC): 

57 """ 

58 GuardProtocol that checks if user has required role(s). 

59 

60 The authorizer must be injected at guard instantiation time (constructor injection). 

61 

62 Usage: 

63 authorizer = await container.resolve(AuthorizerProtocol) 

64 @use_guards(RoleGuard(*required_roles, authorizer=authorizer)) 

65 async def admin_only(self): 

66 ... 

67 """ 

68 

69 def __init__(self, *required_roles: str, authorizer: AuthorizerProtocol) -> None: 

70 self.required_roles = set(required_roles) 

71 self._authorizer = authorizer 

72 

73 async def can_activate(self, context: Any) -> bool: 

74 """Check if user has required role using AuthorizerProtocol.""" 

75 request = _extract_request(context) 

76 user = getattr(request.state, "user", None) 

77 if not user: 

78 return False 

79 

80 # Check access with authorizer (handles superuser bypass and inheritance) 

81 return await self._authorizer.check_access(user, self.required_roles) 

82 

83 

84class PermissionGuard(ABC): 

85 """ 

86 GuardProtocol that checks if user has required permission(s). 

87 

88 The authorizer must be injected at guard instantiation time (constructor injection). 

89 

90 Usage: 

91 authorizer = await container.resolve(AuthorizerProtocol) 

92 @use_guards(PermissionGuard("users:write", authorizer=authorizer)) 

93 async def create_user(self): 

94 ... 

95 """ 

96 

97 def __init__( 

98 self, *required_permissions: str, authorizer: AuthorizerProtocol 

99 ) -> None: 

100 self.required_permissions = set(required_permissions) 

101 self._authorizer = authorizer 

102 

103 async def can_activate(self, context: Any) -> bool: 

104 """Check if user has required permissions using AuthorizerProtocol.""" 

105 request = _extract_request(context) 

106 user = getattr(request.state, "user", None) 

107 if not user: 

108 return False 

109 

110 # Check each permission against the authorizer 

111 for perm in self.required_permissions: 

112 # We treat permission string as "resource.action" if it contains a dot, 

113 # otherwise just pass as action (or resource, depending on how you view it). 

114 parts = perm.split(".", 1) 

115 resource = parts[0] 

116 action = parts[1] if len(parts) > 1 else perm 

117 

118 if await self._authorizer.can(user, action, resource): 

119 return True 

120 

121 return bool(not self.required_permissions) 

122 

123 

124def use_guards(*guards: type[Any] | Any) -> Any: 

125 """ 

126 Decorator to attach guards to a route handler or controller. 

127 

128 Guards are executed in order before the route handler. 

129 If any guard returns False, the request is rejected. 

130 

131 **Guards requiring dependencies (like RoleGuard, PermissionGuard) must 

132 be instantiated with those dependencies passed to the constructor.** 

133 

134 Usage: 

135 @use_guards(AuthGuard) 

136 async def protected_route(self): 

137 ... 

138 

139 # With injected dependencies 

140 authorizer = await container.resolve(AuthorizerProtocol) 

141 @use_guards(AuthGuard, RoleGuard("admin", authorizer=authorizer)) 

142 async def admin_route(self): 

143 ... 

144 """ 

145 

146 def decorator(target: Any) -> Any: 

147 # Store guards metadata 

148 guard_instances = [] 

149 for guard in guards: 

150 if isinstance(guard, type): 

151 guard_instances.append(guard()) 

152 else: 

153 guard_instances.append(guard) 

154 

155 target.__guards__ = guard_instances 

156 

157 # Wrap the function to execute guards 

158 if inspect.iscoroutinefunction(target): 

159 

160 @wraps(target) 

161 async def wrapper(*args: Any, **kwargs: Any) -> Any: 

162 # Find request in kwargs first, then scan positional args 

163 request = kwargs.get("request") 

164 if request is None: 

165 for arg in args: 

166 if isinstance(arg, Request): 

167 request = arg 

168 break 

169 if request is None: 

170 raise RuntimeError( 

171 "GuardProtocol requires request context. Ensure the handler " 

172 "declares 'request: Request' as a parameter.", 

173 ) 

174 

175 result = await execute_guards(guard_instances, request) 

176 if result.is_err(): 

177 rejection = result.unwrap_err() 

178 return JSONResponse( 

179 {"error": rejection.error, "message": rejection.message}, 

180 status_code=rejection.status_code, 

181 ) 

182 

183 return await target(*args, **kwargs) 

184 

185 return wrapper 

186 return target 

187 

188 return decorator 

189 

190 

191async def execute_guards( 

192 guards: list[Any], 

193 request: Request, 

194) -> Result[None, GuardRejection]: 

195 """ 

196 Execute a list of guards. 

197 

198 Args: 

199 guards: List of guard instances 

200 request: The incoming request 

201 

202 Returns: 

203 Ok(None) if all guards pass, Err(GuardRejection) if any guard fails 

204 """ 

205 for guard in guards: 

206 can_proceed = await guard.can_activate(request) 

207 if not can_proceed: 

208 return Err(GuardRejection()) 

209 

210 return Ok(None) 

211 

212 

213def require_admin(target: Any = None, *, authorizer: Any = None) -> Any: 

214 """ 

215 Convenience decorator for admin-only routes. 

216 

217 The ``authorizer`` dependency must be injected. 

218 

219 Can be used either as: 

220 @require_admin(authorizer=auth) 

221 async def handler(...): 

222 ... 

223 

224 or as: 

225 @require_admin 

226 async def handler(...): 

227 ... 

228 

229 Args: 

230 target: When used without parentheses, the decorated function. 

231 authorizer: **Required** AuthorizerProtocol instance. 

232 """ 

233 if authorizer is None: 

234 raise ValueError( 

235 "authorizer parameter is required. Inject it from the container at startup." 

236 ) 

237 

238 decorator = use_guards(RoleGuard("admin", authorizer=authorizer)) 

239 # If used without parentheses 

240 if target is not None: 

241 return decorator(target) 

242 return decorator 

243 

244 

245__all__ = [ 

246 "AuthGuard", 

247 "GuardProtocol", 

248 "GuardRejection", 

249 "PermissionGuard", 

250 "RoleGuard", 

251 "execute_guards", 

252 "require_admin", 

253 "use_guards", 

254]