Coverage for src/lexigram/auth/web/guards.py: 42%
183 statements
« prev ^ index » next coverage.py v7.15.4, created at 2026-08-26 00:58 +0800
1"""GuardProtocol services for authorization in Lexigram Framework"""
3from __future__ import annotations
5from abc import ABC, abstractmethod
6from functools import wraps
7from typing import TYPE_CHECKING, Any, cast
9from lexigram.auth.types import GuardContext
10from lexigram.contracts.web import ResponseFactoryProtocol
11from lexigram.contracts.web.guard import GuardProtocol
12from lexigram.logging import get_logger
13from lexigram.primitives.context import Context, get_request_context
15if TYPE_CHECKING:
16 from collections.abc import Callable
18 from lexigram.contracts.web import ResponseProtocol
20logger = get_logger(__name__)
23def _get_request_resolver(request: Any) -> Any | None:
24 from lexigram.di.resolution.context import get_resolver
26 resolver = get_resolver(request)
27 if resolver is not None:
28 return resolver
30 scope = getattr(request, "scope", None)
31 if isinstance(scope, dict):
32 return scope.get("lexigram_resolver")
34 return None
37async def _get_request_context_user_id(request: Any) -> str | None:
38 resolver = _get_request_resolver(request)
39 if resolver is None:
40 return None
42 resolve_optional = getattr(resolver, "resolve_optional", None)
43 if callable(resolve_optional):
44 context = await resolve_optional(Context)
45 else:
46 context = await resolver.resolve(Context)
47 if context is None:
48 return None
50 current = get_request_context(context.registry)
51 return current.user_id if current is not None else None
54class _GuardBase(ABC):
55 """Private abstract base for auth guards; provides default handle_rejection."""
57 @abstractmethod
58 async def can_activate(self, context: GuardContext) -> bool:
59 """Check if the guard allows the request to proceed"""
61 async def handle_rejection(self, context: GuardContext) -> ResponseProtocol:
62 """Handle guard rejection by returning appropriate response."""
63 try:
64 resolver = _get_request_resolver(context.request)
65 if resolver is None:
66 raise ValueError("No resolver found in context")
67 response_factory = await resolver.resolve(ResponseFactoryProtocol)
68 except ValueError as exc:
69 raise RuntimeError(
70 "ResponseFactoryProtocol not available — ensure the DI container is configured",
71 ) from exc
73 return cast(
74 "ResponseProtocol",
75 response_factory.json(
76 {"error": "forbidden", "message": "Access denied"},
77 status_code=403,
78 ),
79 )
82class AuthGuard(_GuardBase):
83 """GuardProtocol that requires authentication"""
85 async def can_activate(self, context: GuardContext) -> bool:
86 """Check if user is authenticated"""
87 return context.user is not None or context.request_context_user_id is not None
89 async def handle_rejection(self, context: GuardContext) -> ResponseProtocol:
90 """Return 401 for unauthenticated requests."""
91 try:
92 resolver = _get_request_resolver(context.request)
93 if resolver is None:
94 raise ValueError("No resolver found in context")
95 response_factory = await resolver.resolve(ResponseFactoryProtocol)
96 except ValueError as exc:
97 raise RuntimeError(
98 "ResponseFactoryProtocol not available — ensure the DI container is configured",
99 ) from exc
101 return cast(
102 "ResponseProtocol",
103 response_factory.json(
104 {"error": "unauthorized", "message": "Authentication required"},
105 status_code=401,
106 ),
107 )
110class RoleGuard(_GuardBase):
111 """GuardProtocol that requires specific roles"""
113 def __init__(self, *roles: str) -> None:
114 self.required_roles = list(roles)
116 async def can_activate(self, context: GuardContext) -> bool:
117 """Check if user has required roles"""
118 if not context.user:
119 return False
120 return any(context.user.has_role(role) for role in self.required_roles)
123class PermissionGuard(_GuardBase):
124 """GuardProtocol that requires specific permissions"""
126 def __init__(self, *permissions: str) -> None:
127 self.required_permissions = list(permissions)
129 async def can_activate(self, context: GuardContext) -> bool:
130 """Check if user has required permissions"""
131 if not context.user:
132 return False
134 # Import here to avoid circular imports
136 try:
137 resolver = _get_request_resolver(context.request)
139 from lexigram.contracts.auth import AuthProviderProtocol
141 if resolver is None:
142 return False
144 auth_provider: Any = cast(
145 "Any",
146 await resolver.resolve(AuthProviderProtocol),
147 )
149 return bool(
150 auth_provider.has_any_permission(
151 cast("Any", context.user),
152 self.required_permissions,
153 ),
154 )
155 except (RuntimeError, ValueError, TypeError):
156 logger.warning("Failed to check permissions via container")
157 return False
160class CompositeGuard(_GuardBase):
161 """GuardProtocol that combines multiple guards with AND logic"""
163 def __init__(self, *guards: GuardProtocol) -> None:
164 self.guards = guards
166 async def can_activate(self, context: GuardContext) -> bool:
167 """Check if all guards pass"""
168 for guard in self.guards:
169 if not await guard.can_activate(context): # type: ignore[arg-type]
170 return False
171 return True
174class AdminGuard(RoleGuard):
175 """GuardProtocol that requires admin role"""
177 def __init__(self) -> None:
178 super().__init__("admin")
181class UserGuard(AuthGuard):
182 """GuardProtocol that requires any authenticated user"""
185def use_guards(
186 *guards: GuardProtocol,
187) -> Callable[[Callable[..., Any]], Callable[..., Any]]:
188 """Apply guards to a route handler (auth-scoped internal implementation).
190 .. note::
191 This is the auth-package-scoped version of ``use_guards``, intended for
192 internal use within ``lexigram-auth``. For general-purpose use outside
193 the auth subsystem, prefer ``lexigram.security.guards.use_guards`` which
194 integrates with ``GuardChain``.
195 """
197 def decorator(func: Callable) -> Callable:
198 @wraps(func)
199 async def wrapper(*args: Any, **kwargs: Any) -> Any:
200 # Extract request from args (Starlette pattern)
201 request = None
202 for arg in args:
203 if hasattr(arg, "state") and hasattr(arg, "headers"):
204 request = arg
205 break
207 if not request and "request" in kwargs:
208 request = kwargs["request"]
210 if not request:
211 # If no request found, assume guard passes (for testing)
212 return await func(*args, **kwargs)
214 # Always skip guards for OPTIONS requests (CORS preflight)
215 if hasattr(request, "method") and request.method == "OPTIONS":
216 return await func(*args, **kwargs)
218 # Get user from request state
219 user = getattr(request.state, "user", None)
220 request_context_user_id = await _get_request_context_user_id(request)
222 # Create guard context
223 context = GuardContext(
224 user=user,
225 request=request,
226 request_context_user_id=request_context_user_id,
227 )
229 # Check all guards
230 for guard in guards:
231 if not await guard.can_activate(context): # type: ignore[arg-type]
232 return await guard.handle_rejection(context) # type: ignore[arg-type]
234 # All guards passed, proceed
235 return await func(*args, **kwargs)
237 # Store guards on function for introspection
238 wrapper.__guards__ = guards # type: ignore[attr-defined]
239 return wrapper
241 return decorator
244class GuardFactory:
245 """Factory for creating guards via dependency injection.
247 This class handles the async resolution of guards from the DI container
248 and provides synchronous access for use in decorators.
249 """
251 _instances: dict[str, GuardProtocol] = {}
253 @classmethod
254 async def get_guard(
255 cls,
256 guard_type: type[GuardProtocol],
257 resolver: Any | None = None,
258 ) -> GuardProtocol:
259 """Get a guard instance, resolving from DI container if needed.
261 Args:
262 guard_type: The type of guard to get.
263 resolver: Optional resolver to use.
265 Returns:
266 The guard instance.
267 """
268 from lexigram.di.resolution.context import get_resolver
270 res = get_resolver(resolver)
271 if res:
272 guard = await res.resolve_optional(guard_type)
273 if guard is not None:
274 return guard
275 logger.debug(
276 "guard_resolution_failed",
277 guard=guard_type.__name__,
278 error="not registered",
279 )
281 key = guard_type.__name__
282 if key not in cls._instances:
283 cls._instances[key] = guard_type()
284 return cls._instances[key]
287def require_auth() -> Callable[[Callable[..., Any]], Callable[..., Any]]:
288 """Decorator requiring authentication.
290 Uses GuardFactory to get the guard instance properly.
291 """
293 def decorator(func: Callable) -> Callable:
294 @wraps(func)
295 async def wrapper(*args: Any, **kwargs: Any) -> Any:
296 request = None
297 for arg in args:
298 if hasattr(arg, "state") and hasattr(arg, "headers"):
299 request = arg
300 break
302 if not request and "request" in kwargs:
303 request = kwargs["request"]
305 if not request:
306 return await func(*args, **kwargs)
308 if hasattr(request, "method") and request.method == "OPTIONS":
309 return await func(*args, **kwargs)
311 user = getattr(request.state, "user", None)
312 context = GuardContext(request, user)
314 guard = await GuardFactory.get_guard(AuthGuard, request) # type: ignore[arg-type]
315 if not await guard.can_activate(context): # type: ignore[arg-type]
316 return await guard.handle_rejection(context) # type: ignore[arg-type]
318 return await func(*args, **kwargs)
320 wrapper.__guard_type__ = AuthGuard # type: ignore[attr-defined]
321 return wrapper
323 return decorator
326def require_admin() -> Callable[[Callable[..., Any]], Callable[..., Any]]:
327 """Decorator requiring admin role.
329 Uses GuardFactory to get the guard instance properly.
330 """
332 def decorator(func: Callable) -> Callable:
333 @wraps(func)
334 async def wrapper(*args: Any, **kwargs: Any) -> Any:
335 request = None
336 for arg in args:
337 if hasattr(arg, "state") and hasattr(arg, "headers"):
338 request = arg
339 break
341 if not request and "request" in kwargs:
342 request = kwargs["request"]
344 if not request:
345 return await func(*args, **kwargs)
347 if hasattr(request, "method") and request.method == "OPTIONS":
348 return await func(*args, **kwargs)
350 user = getattr(request.state, "user", None)
351 context = GuardContext(request, user)
353 guard = await GuardFactory.get_guard(AdminGuard, request) # type: ignore[arg-type]
354 if not await guard.can_activate(context): # type: ignore[arg-type]
355 return await guard.handle_rejection(context) # type: ignore[arg-type]
357 return await func(*args, **kwargs)
359 wrapper.__guard_type__ = AdminGuard # type: ignore[attr-defined]
360 return wrapper
362 return decorator
365def require_role(*roles: str) -> Callable[[Callable[..., Any]], Callable[..., Any]]:
366 """Decorator requiring specific roles"""
367 return use_guards(RoleGuard(*roles)) # type: ignore[arg-type]
370def require_permission(
371 *permissions: str,
372) -> Callable[[Callable[..., Any]], Callable[..., Any]]:
373 """Decorator requiring specific permissions"""
374 return use_guards(PermissionGuard(*permissions)) # type: ignore[arg-type]
377__all__ = [
378 "AdminGuard",
379 "AuthGuard",
380 "CompositeGuard",
381 "GuardContext",
382 # GuardProtocol classes
383 "GuardProtocol",
384 "PermissionGuard",
385 "RoleGuard",
386 "UserGuard",
387 "require_admin",
388 "require_auth",
389 "require_permission",
390 "require_role",
391 # Decorators (snake_case only)
392 "use_guards",
393]