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

1"""Authentication middleware base. 

2 

3This module provides HTTP transport layer authentication middleware. 

4Authentication logic (JWT, OAuth2, etc.) is implemented in lexigram-auth package. 

5 

6Architecture: 

7- lexigram-web: HTTP transport (middleware, headers, cookies) 

8- lexigram-auth: Security logic (JWT decoding, password hashing, OAuth flows) 

9""" 

10 

11from __future__ import annotations 

12 

13from starlette.requests import Request 

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

15 

16from lexigram.contracts.auth import IdentityResolverProtocol 

17from lexigram.contracts.auth.guard import ( 

18 AuthenticatorProtocol, 

19 AuthorizerProtocol, 

20) 

21from lexigram.contracts.exceptions import ( 

22 AuthenticationError as LexigramAuthenticationError, 

23) 

24from lexigram.contracts.exceptions import ( 

25 AuthorizationError as LexigramAuthorizationError, 

26) 

27from lexigram.logging import get_logger 

28 

29logger = get_logger(__name__) 

30 

31 

32class AuthenticationMiddleware: 

33 """HTTP transport layer authentication middleware. 

34 

35 This middleware handles the HTTP transport aspects of authentication: 

36 - Extracting credentials from HTTP headers/cookies 

37 - Setting user in request state 

38 - Returning 401 Unauthorized responses 

39 - Optionally resolving OAuth external IDs to internal UUIDs 

40 

41 The actual authentication logic (JWT decoding, password validation, etc.) 

42 is delegated to AuthenticatorProtocol implementations. 

43 

44 Example: 

45 from lexigram.web.middleware.auth import AuthenticationMiddleware 

46 

47 # Authenticators are injected or created elsewhere 

48 jwt_auth = ... # Some AuthenticatorProtocol implementation 

49 middleware = AuthenticationMiddleware(authenticators=[jwt_auth]) 

50 """ 

51 

52 def __init__( 

53 self, 

54 app: ASGIApp, 

55 authenticators: list[AuthenticatorProtocol] | None = None, 

56 authorizer: AuthorizerProtocol | None = None, 

57 exclude_paths: list[str] | None = None, 

58 enable_identity_resolution: bool = False, 

59 identity_resolver: IdentityResolverProtocol | None = None, 

60 ): 

61 """Initialize authentication middleware. 

62 

63 Args: 

64 app: ASGI application 

65 authenticators: List of authenticators implementing AuthenticatorProtocol protocol 

66 authorizer: Optional authorizer implementing AuthorizerProtocol protocol 

67 exclude_paths: Paths to skip authentication (e.g., ['/health', '/docs']) 

68 enable_identity_resolution: Whether to resolve OAuth external IDs to internal UUIDs 

69 identity_resolver: Optional resolver for OAuth identity resolution 

70 """ 

71 self.app = app 

72 self.authenticators = authenticators or [] # Allow empty list 

73 self.authorizer = authorizer 

74 self.exclude_paths = exclude_paths or [] 

75 self.enable_identity_resolution = enable_identity_resolution 

76 self.identity_resolver = identity_resolver 

77 

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

79 """Authenticate the request using configured authenticators.""" 

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

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

82 return 

83 

84 # Get path from scope 

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

86 

87 # Determine if authentication is mandatory for this path 

88 is_excluded = any( 

89 path.startswith(excl_path) for excl_path in self.exclude_paths 

90 ) 

91 

92 # Create request from scope for authenticators 

93 request = Request(scope) 

94 

95 # Try each authenticator in order 

96 user = None 

97 if self.authenticators: 

98 for authenticator in self.authenticators: 

99 user = await authenticator.authenticate(request) 

100 if user is not None: 

101 break 

102 

103 # If no authenticators are configured, allow request through without authentication 

104 if not self.authenticators: 

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

106 return 

107 

108 if user is None and not is_excluded: 

109 # Emit structured audit log for auth failure 

110 client_host = scope.get("client", ("unknown", 0))[0] 

111 method = scope.get("method", b"") 

112 method_str = method.decode() if isinstance(method, bytes) else str(method) 

113 logger.warning( 

114 "security.auth_failure", 

115 path=path, 

116 method=method_str, 

117 client_ip=client_host, 

118 reason="no_authenticator_succeeded", 

119 ) 

120 # Send 401 response 

121 await send( 

122 { 

123 "type": "http.response.start", 

124 "status": 401, 

125 "headers": [(b"content-type", b"application/json")], 

126 }, 

127 ) 

128 await send( 

129 { 

130 "type": "http.response.body", 

131 "body": b'{"error": "Authentication required"}', 

132 }, 

133 ) 

134 return 

135 

136 # Optionally resolve OAuth external IDs to internal UUIDs 

137 if ( 

138 self.enable_identity_resolution 

139 and self.identity_resolver is not None 

140 and user is not None 

141 ): 

142 

143 def _get(obj: object, key: str, default: object = None) -> object: 

144 if isinstance(obj, dict): 

145 return obj.get(key, default) 

146 return getattr(obj, key, default) 

147 

148 user_id = _get(user, "id") or _get(user, "user_id") 

149 provider = _get(user, "provider") or "google" 

150 

151 resolved_id = await self.identity_resolver.resolve_user_id( 

152 user_id, # type: ignore[arg-type] 

153 provider, # type: ignore[arg-type] 

154 ) 

155 if resolved_id: 

156 if isinstance(user, dict): 

157 user_dict: dict = {**user} 

158 else: 

159 user_dict = dict(user) 

160 user_dict["id"] = resolved_id 

161 user_dict["user_id"] = resolved_id 

162 user_dict["resolved"] = True 

163 user = user_dict 

164 

165 # Normalise user to dict 

166 if user is not None and not isinstance(user, dict): 

167 if isinstance(user, bool): 

168 user = None 

169 elif hasattr(user, "model_dump") and callable(user.model_dump): 

170 user = user.model_dump() 

171 elif hasattr(user, "__dict__"): 

172 import dataclasses 

173 

174 if dataclasses.is_dataclass(user): 

175 user = dataclasses.asdict(user) # type: ignore[arg-type] 

176 else: 

177 user = dict(vars(user)) 

178 else: 

179 try: 

180 user = dict(user) 

181 except (TypeError, ValueError): 

182 user = None 

183 

184 # Store user in scope state 

185 scope.setdefault("state", {})["user"] = user 

186 

187 # Extract and stringify user_id 

188 uid = user.get("id") or user.get("user_id") if user else None 

189 if uid is not None: 

190 uid = str(uid) 

191 

192 scope.setdefault("state", {})["user_id"] = uid 

193 

194 # Also store in scope extensions 

195 scope.setdefault("extensions", {})["user"] = user 

196 scope.setdefault("extensions", {})["user_id"] = uid 

197 

198 # Continue with request 

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

200 

201 

202AuthenticationError = LexigramAuthenticationError 

203AuthorizationError = LexigramAuthorizationError