Coverage for src/lexigram/auth/web/middleware/token_extractor.py: 52%

1"""Token extraction utilities for authentication middleware.""" 

2 

3from __future__ import annotations 

4 

5from typing import TYPE_CHECKING, Any 

6 

7from lexigram.auth import constants as const 

8from lexigram.logging import get_logger 

9 

10if TYPE_CHECKING: 

11 from lexigram.contracts.web import RequestProtocol as Request 

12 

13logger = get_logger(__name__) 

14 

15 

16class TokenExtractor: 

17 """Handles extraction of authentication tokens from HTTP requests.""" 

18 

19 def __init__(self, config: Any) -> None: 

20 """Initialize with middleware configuration. 

21 

22 Args: 

23 config: AuthMiddlewareConfig with header_name, scheme, etc. 

24 """ 

25 self.config = config 

26 

27 def extract_token(self, request: Request) -> str | None: 

28 """Extract authentication token from request. 

29 

30 Tries multiple sources in order: 

31 1. Authorization header 

32 2. Query parameter 

33 3. Cookie 

34 

35 Args: 

36 request: HTTP request object 

37 

38 Returns: 

39 Extracted token or None 

40 """ 

41 # Try header first 

42 auth_header = request.headers.get(self.config.header_name) 

43 if isinstance(auth_header, str): 

44 # Log header presence (do not log full token for security) 

45 try: 

46 short_hdr = ( 

47 auth_header.split()[1][:10] 

48 if len(auth_header.split()) > 1 

49 else None 

50 ) 

51 except (IndexError, TypeError): 

52 short_hdr = None 

53 logger.info( 

54 "TokenExtractor.extract_token: header present short=%s", 

55 short_hdr, 

56 ) 

57 if auth_header.startswith(f"{self.config.scheme} "): 

58 return auth_header[len(f"{self.config.scheme} ") :] 

59 if auth_header.startswith("ApiKey "): 

60 return auth_header[7:] 

61 if ( 

62 self.config.scheme.lower() == const.DEFAULT_TOKEN_TYPE.lower() 

63 and not auth_header.startswith( 

64 f"{const.DEFAULT_TOKEN_TYPE} ", 

65 ) 

66 ): 

67 # Allow bare tokens for Bearer scheme 

68 return auth_header 

69 

70 # Try query parameter 

71 token = request.query_params.get("token") 

72 if isinstance(token, str): 

73 logger.info( 

74 "TokenExtractor.extract_token: token found in query (len=%d)", 

75 len(token), 

76 ) 

77 return token 

78 

79 # Try cookie 

80 token = request.cookies.get("access_token") 

81 if isinstance(token, str): 

82 logger.info( 

83 "TokenExtractor.extract_token: token found in cookie (len=%d)", 

84 len(token), 

85 ) 

86 return token 

87 

88 logger.debug("TokenExtractor.extract_token: no token found") 

89 return None 

90 

91 

92__all__ = ["TokenExtractor"]