Coverage for src / lexigram / admin / auth / permissions.py: 35%

1"""Permission definitions and checks for lexigram-admin. 

2 

3Integrates with lexigram-auth authorization_service for RBAC. 

4Provides declarative permission decorators and guard middleware. 

5""" 

6 

7from __future__ import annotations 

8 

9from dataclasses import dataclass, field 

10from enum import StrEnum 

11from functools import wraps 

12from typing import ( 

13 TYPE_CHECKING, 

14 Any, 

15 ParamSpec, 

16 Protocol, 

17 TypeVar, 

18 runtime_checkable, 

19) 

20 

21from lexigram.contracts.auth.guard import AuthorizerProtocol as AuthorizationService 

22 

23if TYPE_CHECKING: 

24 from collections.abc import Awaitable, Callable, Sequence 

25 

26 

27class Action(StrEnum): 

28 """Standard CRUD actions for resources.""" 

29 

30 LIST = "list" 

31 VIEW = "view" 

32 CREATE = "create" 

33 UPDATE = "update" 

34 DELETE = "delete" 

35 EXPORT = "export" 

36 BULK_DELETE = "bulk_delete" 

37 BULK_UPDATE = "bulk_update" 

38 

39 @classmethod 

40 def all_actions(cls) -> list[Action]: 

41 """All standard actions.""" 

42 return list(cls) 

43 

44 @classmethod 

45 def read_only(cls) -> list[Action]: 

46 """Read-only actions.""" 

47 return [cls.LIST, cls.VIEW, cls.EXPORT] 

48 

49 @classmethod 

50 def write(cls) -> list[Action]: 

51 """Write actions.""" 

52 return [cls.CREATE, cls.UPDATE, cls.DELETE, cls.BULK_DELETE, cls.BULK_UPDATE] 

53 

54 

55@dataclass(frozen=True, slots=True) 

56class Permission: 

57 """A single permission definition. 

58 

59 Permissions follow the pattern: resource.action 

60 Examples: "users.list", "posts.delete", "admin.settings" 

61 """ 

62 

63 resource: str 

64 action: str 

65 

66 def __str__(self) -> str: 

67 return f"{self.resource}.{self.action}" 

68 

69 @classmethod 

70 def from_string(cls, s: str) -> Permission: 

71 """Parse from "resource.action" string.""" 

72 parts = s.split(".", 1) 

73 if len(parts) != 2: 

74 raise ValueError(f"Invalid permission format: {s}") 

75 return cls(resource=parts[0], action=parts[1]) 

76 

77 @classmethod 

78 def for_resource( 

79 cls, 

80 resource: str, 

81 actions: Sequence[Action] | None = None, 

82 ) -> list[Permission]: 

83 """Generate permissions for a resource.""" 

84 actions = actions or Action.all_actions() 

85 return [cls(resource=resource, action=a.value) for a in actions] 

86 

87 

88@dataclass(slots=True) 

89class PermissionSet: 

90 """Collection of permissions with set operations.""" 

91 

92 permissions: set[str] = field(default_factory=set) 

93 

94 def add(self, *perms: str | Permission) -> PermissionSet: 

95 """Add permissions.""" 

96 for p in perms: 

97 self.permissions.add(str(p)) 

98 return self 

99 

100 def remove(self, *perms: str | Permission) -> PermissionSet: 

101 """Remove permissions.""" 

102 for p in perms: 

103 self.permissions.discard(str(p)) 

104 return self 

105 

106 def has(self, perm: str | Permission) -> bool: 

107 """Check if permission exists.""" 

108 perm_str = str(perm) 

109 # Check exact match 

110 if perm_str in self.permissions: 

111 return True 

112 # Check wildcard patterns 

113 if "*" in self.permissions: 

114 return True 

115 # Check resource wildcard (e.g., "users.*") 

116 parts = perm_str.split(".", 1) 

117 return bool(len(parts) == 2 and f"{parts[0]}.*" in self.permissions) 

118 

119 def has_any(self, *perms: str | Permission) -> bool: 

120 """Check if any permission exists.""" 

121 return any(self.has(p) for p in perms) 

122 

123 def has_all(self, *perms: str | Permission) -> bool: 

124 """Check if all permissions exist.""" 

125 return all(self.has(p) for p in perms) 

126 

127 def __contains__(self, perm: str | Permission) -> bool: 

128 return self.has(perm) 

129 

130 def __iter__(self) -> Any: 

131 return iter(self.permissions) 

132 

133 def __len__(self) -> int: 

134 return len(self.permissions) 

135 

136 @classmethod 

137 def from_roles( 

138 cls, 

139 roles: Sequence[str], 

140 authorization_service: AuthorizationService, 

141 ) -> PermissionSet: 

142 """Build permission set from role names using authorization_service.""" 

143 perms = set() 

144 for role in roles: 

145 role_perms = authorization_service.get_role_permissions(role) # type: ignore[attr-defined] 

146 perms.update(role_perms) 

147 return cls(permissions=perms) 

148 

149 

150@runtime_checkable 

151class HasPermissions(Protocol): 

152 """Protocol for objects that have permissions.""" 

153 

154 @property 

155 def permissions(self) -> Sequence[str]: 

156 """Get list of permission strings.""" 

157 ... 

158 

159 @property 

160 def roles(self) -> Sequence[str]: 

161 """Get list of role names.""" 

162 ... 

163 

164 

165def get_user_permissions( 

166 user: Any, 

167 authorization_service: AuthorizationService, 

168) -> PermissionSet: 

169 """Extract permissions from a user object. 

170 

171 Handles both direct permissions and role-based permissions. 

172 """ 

173 perms = PermissionSet() 

174 

175 # Direct permissions 

176 direct = getattr(user, "permissions", []) or [] 

177 perms.add(*direct) 

178 

179 # Role-based permissions 

180 roles = getattr(user, "roles", []) or [] 

181 for role in roles: 

182 role_perms = authorization_service.get_role_permissions(role) # type: ignore[attr-defined] 

183 perms.add(*role_perms) 

184 

185 return perms 

186 

187 

188P = ParamSpec("P") 

189R = TypeVar("R") 

190 

191 

192def _get_permission_set(request: Any) -> PermissionSet | None: 

193 """Extract PermissionSet from request state.""" 

194 state = getattr(request, "state", None) 

195 if state is None: 

196 return None 

197 ps = getattr(state, "permissions", None) 

198 if isinstance(ps, PermissionSet): 

199 return ps 

200 return None 

201 

202 

203def require_permission(*permissions: str | Permission) -> Any: 

204 """Decorator to require any of the listed permissions on a handler. 

205 

206 Checks ``request.state.permissions`` (a PermissionSet). If not present 

207 or the required permission is missing, raises PermissionDeniedError. 

208 

209 Usage: 

210 @require_permission("users.list") 

211 async def list_users(request): ... 

212 

213 @require_permission("posts.create", "posts.update") 

214 async def manage_posts(request): ... 

215 """ 

216 perm_strings = list(map(str, permissions)) 

217 

218 def decorator(func: Callable[P, Awaitable[R]]) -> Callable[P, Awaitable[R]]: 

219 @wraps(func) 

220 async def wrapper(*args: P.args, **kwargs: P.kwargs) -> R: 

221 request = args[0] if args else kwargs.get("request") 

222 if request is None: 

223 raise ValueError("No request object found for permission check") 

224 

225 ps = _get_permission_set(request) 

226 if ps is None: 

227 from lexigram.admin.exceptions import PermissionDeniedError 

228 

229 raise PermissionDeniedError(message="Authentication required") 

230 

231 if not ps.has_any(*perm_strings): 

232 from lexigram.admin.exceptions import PermissionDeniedError 

233 

234 raise PermissionDeniedError( 

235 message=f"Requires permission: {' or '.join(perm_strings)}", 

236 ) 

237 

238 return await func(*args, **kwargs) 

239 

240 wrapper.__required_permissions__ = list(perm_strings) # type: ignore[attr-defined] 

241 return wrapper 

242 

243 return decorator 

244 

245 

246def require_all_permissions(*permissions: str | Permission) -> Any: 

247 """Decorator to require ALL listed permissions on a handler. 

248 

249 Checks ``request.state.permissions`` (a PermissionSet). Raises 

250 PermissionDeniedError if any permission is missing. 

251 

252 Usage: 

253 @require_all_permissions("users.list", "users.view") 

254 async def view_users(request): ... 

255 """ 

256 perm_strings = list(map(str, permissions)) 

257 

258 def decorator(func: Callable[P, Awaitable[R]]) -> Callable[P, Awaitable[R]]: 

259 @wraps(func) 

260 async def wrapper(*args: P.args, **kwargs: P.kwargs) -> R: 

261 request = args[0] if args else kwargs.get("request") 

262 if request is None: 

263 raise ValueError("No request object found for permission check") 

264 

265 ps = _get_permission_set(request) 

266 if ps is None: 

267 from lexigram.admin.exceptions import PermissionDeniedError 

268 

269 raise PermissionDeniedError(message="Authentication required") 

270 

271 if not ps.has_all(*perm_strings): 

272 from lexigram.admin.exceptions import PermissionDeniedError 

273 

274 raise PermissionDeniedError( 

275 message=f"Requires all permissions: {', '.join(perm_strings)}", 

276 ) 

277 

278 return await func(*args, **kwargs) 

279 

280 wrapper.__required_permissions__ = list(perm_strings) # type: ignore[attr-defined] 

281 wrapper.__require_all__ = True # type: ignore[attr-defined] 

282 return wrapper 

283 

284 return decorator 

285 

286 

287def require_role(*roles: str) -> Any: 

288 """Decorator to require the user to have any of the listed roles. 

289 

290 Checks ``request.user.roles`` (a list of role name strings). Raises 

291 PermissionDeniedError if the user has none of the required roles. 

292 

293 Usage: 

294 @require_role("admin") 

295 async def admin_panel(request): ... 

296 

297 @require_role("admin", "editor") 

298 async def editor_panel(request): ... 

299 """ 

300 

301 def decorator(func: Callable[P, Awaitable[R]]) -> Callable[P, Awaitable[R]]: 

302 @wraps(func) 

303 async def wrapper(*args: P.args, **kwargs: P.kwargs) -> R: 

304 request = args[0] if args else kwargs.get("request") 

305 if request is None: 

306 raise ValueError("No request object found for role check") 

307 

308 user = getattr(request, "user", None) 

309 if user is None: 

310 from lexigram.admin.exceptions import PermissionDeniedError 

311 

312 raise PermissionDeniedError(message="Authentication required") 

313 

314 user_roles = set(getattr(user, "roles", []) or []) 

315 if not user_roles.intersection(roles): 

316 from lexigram.admin.exceptions import PermissionDeniedError 

317 

318 raise PermissionDeniedError( 

319 message=f"Requires role: {' or '.join(roles)}", 

320 ) 

321 

322 return await func(*args, **kwargs) 

323 

324 wrapper.__required_roles__ = list(roles) # type: ignore[attr-defined] 

325 return wrapper 

326 

327 return decorator 

328 

329 

330def create_permission_context( 

331 user: Any, 

332 authorization_service: AuthorizationService, 

333) -> dict[str, Any]: 

334 """Create template context with permission helpers. 

335 

336 Returns a dict containing the user object and callables that delegate 

337 to ``get_user_permissions`` so templates can check permissions without 

338 depending on a ``PermissionChecker`` class. 

339 

340 Returns: 

341 Dict with ``user``, ``can``, ``can_any``, ``can_all``, ``has_role``, 

342 ``has_any_role``, and ``permissions`` keys. 

343 """ 

344 perms = get_user_permissions(user, authorization_service) 

345 user_roles: set[str] = set(getattr(user, "roles", []) or []) 

346 return { 

347 "user": user, 

348 "can": perms.has, 

349 "can_any": perms.has_any, 

350 "can_all": perms.has_all, 

351 "has_role": lambda role: role in user_roles, 

352 "has_any_role": lambda *roles: bool(user_roles.intersection(roles)), 

353 "permissions": perms, 

354 }