Coverage for src / lexigram / admin / rbac / middleware.py: 0%

1"""Starlette middleware for RBAC.""" 

2 

3from __future__ import annotations 

4 

5from collections.abc import Callable 

6from typing import Any 

7 

8from starlette.middleware.base import BaseHTTPMiddleware, RequestResponseEndpoint 

9from starlette.requests import Request 

10from starlette.responses import Response 

11 

12from lexigram.admin.rbac.service import PermissionService 

13from lexigram.di.decorators import inject 

14from lexigram.logging import get_logger 

15 

16logger = get_logger(__name__) 

17 

18 

19@inject 

20class RBACMiddleware(BaseHTTPMiddleware): 

21 """Middleware that attaches user permissions to the request context.""" 

22 

23 def __init__( 

24 self, 

25 app: Any, 

26 user_provider: Callable[[Request], Any] | None = None, 

27 role_extractor: Callable[[Any], set[str]] | None = None, 

28 permission_service: PermissionService | None = None, 

29 ): 

30 super().__init__(app) 

31 self.user_provider = user_provider or (lambda r: getattr(r.state, "user", None)) 

32 self.role_extractor = role_extractor or (lambda u: set(getattr(u, "roles", []))) 

33 self._permission_service = permission_service 

34 

35 async def dispatch( 

36 self, 

37 request: Request, 

38 call_next: RequestResponseEndpoint, 

39 ) -> Response: 

40 # 1. Resolve User 

41 user = self.user_provider(request) 

42 

43 if user: 

44 # 2. Attach injected permissions service to request state 

45 request.state.permissions = self._permission_service 

46 

47 # 3. Optional: Pre-fetch or cache user roles in request state 

48 roles = self.role_extractor(user) 

49 request.state.user_roles = roles 

50 

51 logger.debug( 

52 "RBAC: User %s has roles %s", 

53 getattr(user, "id", "unknown"), 

54 roles, 

55 ) 

56 

57 return await call_next(request)