Coverage for src/lexigram/auth/decorators.py: 0%

1from __future__ import annotations 

2 

3from collections.abc import Callable 

4import functools 

5from typing import Any, TypeVar 

6 

7from lexigram.auth.exceptions import AuthenticationError, AuthorizationError 

8 

9F = TypeVar("F", bound=Callable[..., Any]) 

10 

11 

12def _extract_request(*args: Any, **kwargs: Any) -> Any: 

13 """Extract the request-like object from positional or keyword args.""" 

14 for arg in args: 

15 if hasattr(arg, "identity") or hasattr(arg, "state"): 

16 return arg 

17 return kwargs.get("request") 

18 

19 

20def require_auth(fn: F) -> F: 

21 """Require a valid authenticated identity on the request context. 

22 

23 The decorator reads ``request.identity`` from the first positional argument 

24 (assumed to be the request/context object). Raises ``AuthenticationError`` 

25 when no identity is present. 

26 

27 Args: 

28 fn: The async handler function to decorate. 

29 

30 Returns: 

31 Decorated async handler that validates identity before invocation. 

32 

33 Raises: 

34 AuthenticationError: When no identity is present on the request. 

35 

36 Example:: 

37 

38 @require_auth 

39 async def get_profile(self, request: Request) -> Response: 

40 return Response(data=request.identity.claims) 

41 """ 

42 

43 @functools.wraps(fn) 

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

45 request = _extract_request(*args, **kwargs) 

46 if request is None or not getattr(request, "identity", None): 

47 raise AuthenticationError("Authentication required") 

48 return await fn(*args, **kwargs) 

49 

50 return wrapper # type: ignore[return-value] 

51 

52 

53def require_roles(*roles: str) -> Callable[[F], F]: 

54 """Require that the authenticated identity holds at least one of the given roles. 

55 

56 Args: 

57 *roles: One or more role names. Access is granted if the identity 

58 has **any** of the listed roles. 

59 

60 Returns: 

61 Decorator that enforces role-based access control. 

62 

63 Raises: 

64 AuthenticationError: When no identity is present on the request. 

65 AuthorizationError: When the identity lacks all required roles. 

66 

67 Example:: 

68 

69 @require_roles("admin", "moderator") 

70 async def delete_post(self, request: Request, post_id: str) -> Response: ... 

71 """ 

72 

73 def decorator(fn: F) -> F: 

74 @functools.wraps(fn) 

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

76 request = _extract_request(*args, **kwargs) 

77 identity = ( 

78 getattr(request, "identity", None) if request is not None else None 

79 ) 

80 if identity is None: 

81 raise AuthenticationError("Authentication required") 

82 identity_roles: set[str] = set(getattr(identity, "roles", [])) 

83 if not identity_roles.intersection(roles): 

84 raise AuthorizationError( 

85 f"Required roles: {roles!r}. Identity has: {sorted(identity_roles)!r}" 

86 ) 

87 return await fn(*args, **kwargs) 

88 

89 return wrapper # type: ignore[return-value] 

90 

91 return decorator 

92 

93 

94__all__ = ["require_auth", "require_roles"]