Coverage for src/lexigram/auth/web/middleware/jwt_authenticator.py: 79%

1"""JWT token authentication component.""" 

2 

3from __future__ import annotations 

4 

5from typing import Any 

6 

7from lexigram.contracts import AuthProviderProtocol 

8from lexigram.logging import LoggerProtocol as Logger 

9from lexigram.logging import get_logger 

10 

11logger = get_logger(__name__) 

12 

13 

14class JwtAuthenticator: 

15 """Handles JWT token validation and user resolution.""" 

16 

17 def __init__( 

18 self, 

19 auth_provider: AuthProviderProtocol, 

20 config: Any | None = None, 

21 logger: Logger | None = None, 

22 ) -> None: 

23 self.auth_provider = auth_provider 

24 from lexigram.auth.web.middleware.token_extractor import TokenExtractor 

25 

26 self.token_extractor = TokenExtractor( 

27 config 

28 or type( 

29 "Config", (), {"header_name": "Authorization", "scheme": "Bearer"} 

30 )() 

31 ) 

32 

33 if logger is None: 

34 from lexigram.logging import get_logger 

35 

36 logger = get_logger(__name__) 

37 self.logger = logger.bind(middleware="JwtAuthenticator") 

38 

39 async def authenticate(self, request: Any) -> Any | None: 

40 """Validate JWT token from request and return user if valid.""" 

41 token = self.token_extractor.extract_token(request) 

42 if not token: 

43 return None 

44 

45 try: 

46 token_result = await self.auth_provider.verify_token(token) 

47 except (ValueError, TypeError, RuntimeError) as e: 

48 self.logger.exception( 

49 "Unexpected error during JWT authentication", 

50 exc_info=e, 

51 ) 

52 return None 

53 

54 if token_result is None or token_result.is_err(): 

55 self.logger.info( 

56 "JWT verification failed: %s", 

57 token_result.unwrap_err() if token_result is not None else "no result", 

58 ) 

59 return None 

60 

61 verified = token_result.unwrap() 

62 user_id = verified.user_id 

63 if not user_id: 

64 return None 

65 

66 # Fetch fresh user data from store 

67 user = await self.auth_provider.get_user(user_id) 

68 if not user: 

69 return None 

70 

71 # Pre-fetch delegations for sync check_access 

72 if self.auth_provider.delegation_manager: 

73 request.state.delegations = ( 

74 await self.auth_provider.delegation_manager.get_active_delegations( 

75 user.user_id, 

76 ) 

77 ) 

78 

79 return user 

80 

81 

82__all__ = [ 

83 "JwtAuthenticator", 

84 "logger", 

85]