Coverage for src/lexigram/web/integrations/auth.py: 23%

1"""Authentication integration for WebProvider.""" 

2 

3from __future__ import annotations 

4 

5import inspect 

6from typing import Any, cast 

7 

8from starlette.applications import Starlette 

9 

10from lexigram.di.container import Container 

11from lexigram.logging import get_logger 

12 

13logger = get_logger(__name__) 

14 

15 

16class AuthIntegration: 

17 """Handles authentication middleware configuration.""" 

18 

19 @staticmethod 

20 def _is_invocable_authenticator(candidate: Any) -> bool: 

21 """Check the candidate can be called as ``authenticate(request)``. 

22 

23 Args: 

24 candidate: Resolved service candidate. 

25 

26 Returns: 

27 True if the candidate is safe to invoke with a single request argument. 

28 """ 

29 authenticate = getattr(candidate, "authenticate", None) 

30 if not callable(authenticate): 

31 return False 

32 try: 

33 signature = inspect.signature(authenticate) 

34 except (TypeError, ValueError): 

35 return True 

36 positional: list[inspect.Parameter] = [ 

37 p 

38 for p in signature.parameters.values() 

39 if p.kind in (p.POSITIONAL_ONLY, p.POSITIONAL_OR_KEYWORD) 

40 ] 

41 if positional and positional[0].name in ("self", "cls"): 

42 positional = positional[1:] 

43 required = [p for p in positional if p.default is p.empty] 

44 return len(required) <= 1 

45 

46 @staticmethod 

47 async def configure(app: Starlette, container: Container, web_config: Any) -> None: 

48 """Configure authentication middleware.""" 

49 from lexigram.contracts.auth import IdentityResolverProtocol 

50 from lexigram.contracts.auth.guard import AuthenticatorProtocol 

51 from lexigram.contracts.exceptions import ContainerError 

52 from lexigram.web.middleware.auth import AuthenticationMiddleware 

53 

54 identity_resolver = await container.resolve_optional( 

55 cast("Any", IdentityResolverProtocol) 

56 ) 

57 

58 authenticators = [] 

59 try: 

60 candidates = await container.resolve_all(cast("Any", AuthenticatorProtocol)) 

61 for candidate in candidates: 

62 if AuthIntegration._is_invocable_authenticator(candidate): 

63 authenticators.append(candidate) 

64 else: 

65 logger.warning( 

66 "skipped_incompatible_authenticator", 

67 service=type(candidate).__name__, 

68 ) 

69 logger.info("resolved_authenticators", count=len(authenticators)) 

70 except (LookupError, RuntimeError, ContainerError) as e: 

71 logger.warning("failed_to_resolve_authenticators", error=str(e)) 

72 

73 app.add_middleware( 

74 AuthenticationMiddleware, 

75 authenticators=authenticators, 

76 exclude_paths=web_config.auth_exclude_paths, 

77 enable_identity_resolution=web_config.enable_identity_resolution, 

78 identity_resolver=identity_resolver, 

79 ) 

80 logger.info("Authentication middleware configured") 

81 

82 await AuthIntegration._add_role_guard(app, container, web_config) 

83 

84 @staticmethod 

85 async def _add_role_guard( 

86 app: Starlette, container: Container, web_config: Any 

87 ) -> None: 

88 """Register the role guard middleware after authentication. 

89 

90 Roles are resolved per request from the container-bound 

91 :class:`~lexigram.web.middleware.role_guard.RoleResolverProtocol`, 

92 never from JWT claims. When rules are declared but no resolver is 

93 bound, startup fails fast — enforcement must never be silently off. 

94 

95 Args: 

96 app: The Starlette application to configure. 

97 container: The DI container. 

98 web_config: The WebConfig driving the rules. 

99 """ 

100 from starlette.middleware import Middleware 

101 

102 from lexigram.contracts.exceptions import ContainerError 

103 from lexigram.contracts.exceptions.config import ConfigurationError 

104 from lexigram.web.middleware.role_guard import ( 

105 RoleGuardMiddleware, 

106 RoleGuardRule, 

107 RoleResolverProtocol, 

108 ) 

109 

110 role_guard = getattr(web_config, "role_guard", None) 

111 rules = role_guard.rules if role_guard is not None else [] 

112 if not rules: 

113 return 

114 

115 resolver = None 

116 try: 

117 resolver = await container.resolve(cast("Any", RoleResolverProtocol)) 

118 except (LookupError, RuntimeError, ContainerError) as e: 

119 logger.warning("failed_to_resolve_role_resolver", error=str(e)) 

120 

121 if resolver is None: 

122 raise ConfigurationError( 

123 "web.role_guard.rules declared but no RoleResolverProtocol is " 

124 "bound in the container. Bind one to enable role enforcement." 

125 ) 

126 

127 app.user_middleware.append( 

128 Middleware( 

129 RoleGuardMiddleware, 

130 rules=[ 

131 RoleGuardRule(path=rule.path, roles=list(rule.roles)) 

132 for rule in rules 

133 ], 

134 resolver=resolver, 

135 ) 

136 ) 

137 logger.info("Role guard middleware configured", rules=len(rules))