Coverage for src/lexigram/auth/authn/api_key.py: 100%

1"""API Key AuthenticatorProtocol for Lexigram Auth. 

2 

3Authenticates incoming requests by extracting an API key from the 

4``X-API-Key`` request header or the ``api_key`` query parameter, then 

5delegating validation to a configurable lookup function or static dict. 

6""" 

7 

8from __future__ import annotations 

9 

10from collections.abc import Awaitable, Callable 

11from dataclasses import dataclass, field 

12from typing import TYPE_CHECKING, Any 

13 

14from lexigram.auth.exceptions import AuthenticationError 

15from lexigram.auth.models.user import User 

16from lexigram.logging import get_logger 

17 

18if TYPE_CHECKING: 

19 from lexigram.result import Result 

20 

21logger = get_logger(__name__) 

22 

23# A lookup can be a plain dict mapping key → User, or an async/sync callable 

24# that accepts the raw key string and returns a User or None. 

25LookupCallable = Callable[[str], Awaitable[User | None] | (User | None)] 

26 

27 

28@dataclass 

29class APIKeyConfig: 

30 """Configuration for :class:`APIKeyAuthenticator`. 

31 

32 Attributes: 

33 header_name: HTTP header to inspect for the API key. 

34 Defaults to ``"X-API-Key"``. 

35 query_param: URL query parameter name to fall back to when the header 

36 is absent. Defaults to ``"api_key"``. 

37 lookup: Either a ``dict[str, User]`` mapping raw key strings to their 

38 owners, or an async/sync callable that accepts the raw key and 

39 returns ``User | None``. This field is **required**. 

40 """ 

41 

42 lookup: LookupCallable | dict[str, Any] 

43 header_name: str = "X-API-Key" 

44 query_param: str = "api_key" 

45 # Injected via field so subclasses can override without touching __init__ 

46 _lookup_is_dict: bool = field(init=False, repr=False) 

47 

48 def __post_init__(self) -> None: 

49 self._lookup_is_dict = isinstance(self.lookup, dict) 

50 

51 

52class APIKeyAuthenticator: 

53 """Authenticates requests by validating an API key. 

54 

55 Key extraction order: 

56 1. ``request_context["headers"][config.header_name]`` 

57 2. ``request_context["query_params"][config.query_param]`` 

58 

59 If the key is absent **or** the lookup returns ``None`` an 

60 ``Err(AuthenticationError)`` is returned. On success ``Ok(User)`` 

61 is returned. 

62 

63 Example:: 

64 

65 config = APIKeyConfig( 

66 lookup={"sk_live_abc123": admin_user}, 

67 ) 

68 authenticator = APIKeyAuthenticator(config) 

69 

70 result = await authenticator.authenticate({ 

71 "headers": {"X-API-Key": "sk_live_abc123"}, 

72 "query_params": {}, 

73 }) 

74 

75 if result.is_ok(): 

76 user = result.unwrap() 

77 """ 

78 

79 def __init__(self, config: APIKeyConfig) -> None: 

80 """Initialise the authenticator. 

81 

82 Args: 

83 config: Key-extraction and lookup configuration. 

84 """ 

85 self._config = config 

86 

87 def _extract_key(self, request_context: dict[str, Any]) -> str | None: 

88 """Extract the raw API key from the request context. 

89 

90 Checks the configured header first, then falls back to the query 

91 parameter. Header matching is case-insensitive. 

92 

93 Args: 

94 request_context: Mapping with optional ``"headers"`` and 

95 ``"query_params"`` sub-dicts. 

96 

97 Returns: 

98 Raw key string, or ``None`` if not present. 

99 """ 

100 headers: dict[str, str] = request_context.get("headers") or {} 

101 query_params: dict[str, str] = request_context.get("query_params") or {} 

102 

103 # Case-insensitive header lookup 

104 header_name_lower = self._config.header_name.lower() 

105 for name, value in headers.items(): 

106 if name.lower() == header_name_lower: 

107 return value or None 

108 

109 return query_params.get(self._config.query_param) or None 

110 

111 async def _resolve_user(self, raw_key: str) -> User | None: 

112 """Invoke the configured lookup to resolve the user. 

113 

114 Args: 

115 raw_key: Plain-text API key submitted by the caller. 

116 

117 Returns: 

118 :class:`~lexigram.auth.models.user.User` if found, ``None`` 

119 otherwise. 

120 """ 

121 import inspect 

122 

123 lookup = self._config.lookup 

124 

125 if self._config._lookup_is_dict: 

126 return lookup.get(raw_key) # type: ignore[union-attr] 

127 

128 result = lookup(raw_key) # type: ignore[operator] 

129 if inspect.isawaitable(result): 

130 return await result 

131 return result 

132 

133 async def authenticate( 

134 self, 

135 request_context: dict[str, Any], 

136 ) -> Result[User, AuthenticationError]: 

137 """Authenticate a request using its API key. 

138 

139 Args: 

140 request_context: A mapping that **must** contain at least one of: 

141 

142 * ``"headers"``: ``dict[str, str]`` of HTTP request headers. 

143 * ``"query_params"``: ``dict[str, str]`` of URL query params. 

144 

145 Returns: 

146 ``Ok(User)`` when a valid API key is found. 

147 ``Err(AuthenticationError)`` when the key is absent, invalid, or 

148 the lookup returns ``None``. 

149 """ 

150 from lexigram.result import Err, Ok 

151 

152 raw_key = self._extract_key(request_context) 

153 if not raw_key: 

154 logger.debug("api_key_missing", context_keys=list(request_context.keys())) 

155 return Err(AuthenticationError("API key is missing from the request")) 

156 

157 user = await self._resolve_user(raw_key) 

158 if user is None: 

159 logger.warning("api_key_invalid", key_prefix=raw_key[:8]) 

160 return Err(AuthenticationError("Invalid or unrecognised API key")) 

161 

162 logger.info( 

163 "api_key_authenticated", 

164 user_id=getattr(user, "user_id", None), 

165 ) 

166 return Ok(user) 

167 

168 

169__all__ = ["APIKeyAuthenticator", "APIKeyConfig"]