Coverage for src/lexigram/web/middleware/role_guard.py: 40%

1"""Role guard middleware — declarative path-to-role enforcement. 

2 

3Guards HTTP paths against configured role requirements using the 

4authenticated identity set by the authentication middleware 

5(``scope["state"]["user_id"]``). Roles are resolved **per request** from the 

6identity store (never from JWT claims), so demotions take effect immediately. 

7 

8Rules are plain values: an exact path, or a path ending in ``/**`` which 

9matches every path under that prefix. The first matching rule wins. 

10 

11Registration order matters: this middleware must sit **inside** the 

12authentication middleware so an authenticated identity exists when it runs. 

13In :class:`~lexigram.web.di.provider.WebProvider` the auth middleware is 

14registered by :class:`~lexigram.web.integrations.auth.AuthIntegration`; the 

15role guard is registered right after it in the same step. 

16""" 

17 

18from __future__ import annotations 

19 

20from dataclasses import dataclass 

21from typing import Protocol, runtime_checkable 

22 

23from starlette.responses import JSONResponse 

24from starlette.types import ASGIApp, Receive, Scope, Send 

25 

26from lexigram.logging import get_logger 

27 

28logger = get_logger(__name__) 

29 

30 

31@dataclass(frozen=True) 

32class RoleGuardRule: 

33 """One role guard rule. 

34 

35 Attributes: 

36 path: Exact path to guard. A trailing ``/**`` matches every path 

37 under that prefix (e.g. ``/api/admin/**`` also matches 

38 ``/api/admin`` itself). 

39 roles: Role identifiers allowed to pass. 

40 """ 

41 

42 path: str 

43 roles: list[str] 

44 

45 

46@runtime_checkable 

47class RoleResolverProtocol(Protocol): 

48 """Resolve the role set of the authenticated user per request. 

49 

50 Implementations must query the identity store — never read JWT claims — 

51 so demotions take effect immediately. 

52 """ 

53 

54 async def resolve(self, user_id: str) -> list[str] | None: 

55 """Return the roles granted to *user_id*. 

56 

57 Args: 

58 user_id: Verified identity of the authenticated caller. 

59 

60 Returns: 

61 The roles granted to the user, or ``None`` when the user is 

62 unknown or deactivated. 

63 """ 

64 ... 

65 

66 

67class RoleGuardMiddleware: 

68 """ASGI middleware enforcing role rules against authenticated identities. 

69 

70 Args: 

71 app: Inner ASGI application. 

72 rules: Rules applied in declaration order; first match wins. 

73 resolver: Per-request role resolver. 

74 """ 

75 

76 def __init__( 

77 self, 

78 app: ASGIApp, 

79 rules: list[RoleGuardRule], 

80 resolver: RoleResolverProtocol, 

81 ) -> None: 

82 """Initialise the middleware with its rule set and resolver. 

83 

84 Args: 

85 app: Inner ASGI application. 

86 rules: Rules applied in declaration order; first match wins. 

87 resolver: Per-request role resolver. 

88 """ 

89 self.app = app 

90 self._rules = rules 

91 self._resolver = resolver 

92 

93 def _match(self, path: str) -> RoleGuardRule | None: 

94 """Return the first rule matching *path*. 

95 

96 Args: 

97 path: Request path to evaluate. 

98 

99 Returns: 

100 The first matching rule, or ``None`` when no rule applies. 

101 """ 

102 for rule in self._rules: 

103 if rule.path.endswith("/**"): 

104 prefix = rule.path[:-3].rstrip("/") 

105 if path == prefix or path.startswith(f"{prefix}/"): 

106 return rule 

107 elif path == rule.path: 

108 return rule 

109 return None 

110 

111 async def __call__(self, scope: Scope, receive: Receive, send: Send) -> None: 

112 """Enforce role rules for the request. 

113 

114 Args: 

115 scope: The ASGI connection scope. 

116 receive: The ASGI receive callable. 

117 send: The ASGI send callable. 

118 """ 

119 if scope["type"] != "http": 

120 await self.app(scope, receive, send) 

121 return 

122 

123 rule = self._match(scope.get("path", "")) 

124 if rule is None: 

125 await self.app(scope, receive, send) 

126 return 

127 

128 user_id = scope.setdefault("state", {}).get("user_id") 

129 if user_id is None: 

130 logger.warning("role_guard_unauthorized", path=scope.get("path")) 

131 await self._reject(scope, receive, send, 401, "Unauthorized") 

132 return 

133 

134 roles = await self._resolver.resolve(str(user_id)) 

135 if roles is None or not set(roles).intersection(rule.roles): 

136 logger.warning( 

137 "role_guard_forbidden", 

138 path=scope.get("path"), 

139 user_id=str(user_id), 

140 rule_roles=list(rule.roles), 

141 ) 

142 await self._reject(scope, receive, send, 403, "Forbidden") 

143 return 

144 

145 await self.app(scope, receive, send) 

146 

147 @staticmethod 

148 async def _reject( 

149 scope: Scope, 

150 receive: Receive, 

151 send: Send, 

152 status_code: int, 

153 error: str, 

154 ) -> None: 

155 """Respond with a JSON error body. 

156 

157 Args: 

158 scope: The ASGI connection scope. 

159 receive: The ASGI receive callable. 

160 send: The ASGI send callable. 

161 status_code: HTTP status of the rejection. 

162 error: Error message for the ``{"error": ...}`` body. 

163 """ 

164 response = JSONResponse({"error": error}, status_code=status_code) 

165 await response(scope, receive, send) 

166 

167 

168__all__ = [ 

169 "RoleGuardMiddleware", 

170 "RoleGuardRule", 

171 "RoleResolverProtocol", 

172]