Coverage for src/lexigram/auth/di/sub_providers/authentication_provider.py: 90%
111 statements
« prev ^ index » next coverage.py v7.15.4, created at 2026-08-26 00:58 +0800
1"""Authentication provider - handles user authentication only."""
3from __future__ import annotations
5import contextlib
6import secrets
7from typing import TYPE_CHECKING, Annotated, Any, cast
9from pydantic import SecretStr
11from lexigram.auth.authn.password_hasher import (
12 Argon2idKeyDerivation,
13 Argon2idPasswordHasher,
14 ComposedPasswordHasher,
15)
16from lexigram.auth.authn.security import PasswordHasher, PasswordPolicy
17from lexigram.auth.authn.services import AuthenticationService
18from lexigram.auth.storage.token_store import InMemoryUserStore, UserStoreProtocol
19from lexigram.contracts import HealthCheckResult, HealthStatus, ProviderPriority
20from lexigram.contracts.auth import PasswordHasherProtocol, PasswordPolicyProtocol
21from lexigram.contracts.core import HookRegistryProtocol
22from lexigram.contracts.core.config import Environment
23from lexigram.contracts.exceptions import ConfigurationError
24from lexigram.di.markers import Inject
25from lexigram.di.provider import Provider
26from lexigram.logging import get_logger
28__all__ = ["AuthenticationProvider"]
31from lexigram.result import Err, Result
33if TYPE_CHECKING:
34 from lexigram.auth.config import AuthConfig
35 from lexigram.auth.mfa.manager import MFAManager
36 from lexigram.contracts.auth.exceptions import TokenError
37 from lexigram.contracts.auth.token import VerifiedToken
38 from lexigram.contracts.core.di import (
39 BootContainerProtocol,
40 ContainerRegistrarProtocol,
41 )
43logger = get_logger(__name__)
46def _build_token_manager(
47 secret_key: str | None,
48 jwt_algorithm: str,
49 cache_service: Any,
50 required_audience: str | None = None,
51) -> Any:
52 """Build a JWTTokenManager from the supplied credentials.
54 Generates an ephemeral RSA keypair when ``jwt_algorithm`` is RS-family
55 and no PEM key is provided via ``secret_key``.
57 Args:
58 secret_key: HMAC secret or PEM-encoded RSA private key.
59 jwt_algorithm: JWT signing algorithm (e.g. "HS256", "RS256").
60 cache_service: Optional cache backend for token blacklisting.
61 required_audience: Forwarded to :class:`JWTTokenManager`; when set,
62 every verified token must carry a matching ``aud`` claim.
64 Returns:
65 A configured :class:`~lexigram.auth.authn.jwt.JWTTokenManager`.
66 """
67 from lexigram.auth.authn.jwt import JWTTokenManager
69 if jwt_algorithm.startswith(("RS", "PS")):
70 if secret_key and "-----BEGIN" in secret_key:
71 keys: dict[str, Any] = {"default": {"private": secret_key}}
72 else:
73 try:
74 from cryptography.hazmat.primitives import serialization
75 from cryptography.hazmat.primitives.asymmetric import rsa
76 except (ImportError, ModuleNotFoundError) as exc:
77 raise RuntimeError(
78 "cryptography is required for RS/PS algorithms"
79 ) from exc
81 private_key = rsa.generate_private_key(public_exponent=65537, key_size=2048)
82 private_pem = private_key.private_bytes(
83 encoding=serialization.Encoding.PEM,
84 format=serialization.PrivateFormat.PKCS8,
85 encryption_algorithm=serialization.NoEncryption(),
86 ).decode("utf-8")
87 public_pem = (
88 private_key.public_key()
89 .public_bytes(
90 encoding=serialization.Encoding.PEM,
91 format=serialization.PublicFormat.SubjectPublicKeyInfo,
92 )
93 .decode("utf-8")
94 )
95 keys = {
96 "default": {
97 "private": SecretStr(private_pem),
98 "public": SecretStr(public_pem),
99 }
100 }
101 logger.warning(
102 "AuthenticationProvider: generated ephemeral RSA keypair; "
103 "provide persistent keys in production",
104 )
105 return JWTTokenManager(
106 current_key_id="default",
107 keys=keys,
108 algorithm=jwt_algorithm,
109 cache_service=cache_service,
110 required_audience=required_audience,
111 )
113 # Symmetric algorithm: use secret_key directly (single-secret mode)
114 env = Environment.from_env()
115 _STRICT_ENVS = {Environment.PRODUCTION, Environment.STAGING}
116 if not secret_key:
117 if env in _STRICT_ENVS:
118 raise ConfigurationError(
119 f"CRITICAL SECURITY: JWT secret_key is required in {env.value.upper()} "
120 "but none was provided. "
121 "Set LEX_AUTH__TOKEN__SECRET_KEY (or token.secret_key in config)."
122 )
123 effective_secret = secrets.token_urlsafe(32)
124 logger.warning(
125 "jwt_ephemeral_secret_generated",
126 environment=env.value,
127 reason="No JWT secret_key provided; using a generated ephemeral secret. "
128 "Tokens will be invalidated on restart. Provide a stable secret for production.",
129 )
130 else:
131 effective_secret = secret_key
132 return JWTTokenManager(
133 current_key_id="default",
134 keys={"default": SecretStr(effective_secret)},
135 algorithm=jwt_algorithm,
136 cache_service=cache_service,
137 required_audience=required_audience,
138 )
141class AuthenticationProvider(Provider):
142 """User authentication ONLY (login/logout/validation).
144 ``config`` is the primary parameter and drives all defaults. The optional
145 keyword arguments are *override points* for callers that need to supply
146 custom implementations (e.g., a production-grade :class:`UserStoreProtocol` backed
147 by a database, or a pre-built token manager).
149 Args:
150 config: The resolved :class:`~lexigram.auth.config.AuthConfig`.
151 JWT credentials (``config.token.secret_key``, ``config.token.algorithm``)
152 are used to build a :class:`~lexigram.auth.authn.jwt.JWTTokenManager`
153 automatically when no ``token_manager`` override is provided.
154 password_policy: Override the password policy derived from ``config.password``.
155 When *None* the policy is built from config; falls back to ``PasswordPolicy()``
156 when config is also absent.
157 user_store: Override the user store. Defaults to :class:`InMemoryUserStore`.
158 token_manager: Supply a fully constructed token manager. When *None* and
159 ``config.token`` is present the manager is built automatically from the
160 JWT configuration.
161 cache_service: Optional cache backend passed to the auto-built token manager
162 for token blacklisting. Ignored when ``token_manager`` is provided directly.
163 mfa_service: Optional MFA manager for multi-factor authentication flows.
164 """
166 def __init__(
167 self,
168 config: Annotated[AuthConfig, Inject] | None = None,
169 *,
170 password_policy: PasswordPolicy | None = None,
171 user_store: UserStoreProtocol | None = None,
172 token_manager: Any = None,
173 cache_service: Any = None,
174 mfa_service: MFAManager | None = None,
175 ) -> None:
176 super().__init__(name="authentication", priority=ProviderPriority.SECURITY)
177 self.config = config
178 if password_policy is not None:
179 self.password_policy = password_policy
180 elif config is not None and config.password is not None:
181 self.password_policy = PasswordPolicy.from_config(config.password)
182 else:
183 self.password_policy = PasswordPolicy()
184 self.user_store = user_store or InMemoryUserStore()
185 self.mfa_service = mfa_service
187 # Auto-create a JWTTokenManager from config when no override is supplied.
188 if (
189 token_manager is None
190 and config is not None
191 and getattr(config, "token", None) is not None
192 ):
193 token_manager = _build_token_manager(
194 secret_key=config.token.secret_key.get_secret_value(),
195 jwt_algorithm=config.token.algorithm,
196 cache_service=cache_service,
197 required_audience=config.token.required_audience,
198 )
200 self.token_manager = token_manager
201 self.delegation_manager: Any = None
202 self._service: AuthenticationService | None = None
203 self._hasher: PasswordHasherProtocol | None = None
205 @property
206 def password_hasher(self) -> PasswordHasherProtocol:
207 """Composed Argon2id-default hasher with a bcrypt legacy shim.
209 Builds the single composed instance (ODD-1 Option A): new hashes use
210 Argon2id with OWASP-2024 parameters; stored bcrypt hashes keep
211 verifying through the legacy shim and upgrade on the next login.
212 """
213 if self._hasher is None:
214 primary = Argon2idPasswordHasher(
215 kdf=Argon2idKeyDerivation(
216 config=self.config.password if self.config else None,
217 ),
218 )
219 self._hasher = ComposedPasswordHasher(
220 primary=primary,
221 legacy=PasswordHasher(),
222 )
223 return self._hasher
225 @property
226 def service(self) -> AuthenticationService:
227 """Get or create the authentication service."""
228 if self._service is None:
229 self._service = AuthenticationService(
230 password_policy=self.password_policy,
231 user_store=self.user_store,
232 token_manager=self.token_manager,
233 password_hasher=self.password_hasher,
234 )
235 return self._service
237 async def get_user(self, user_id: str) -> Any | None:
238 """Fetch a user by their ID.
240 Satisfies :class:`~lexigram.contracts.auth.AuthProviderProtocol`.
242 Args:
243 user_id: The unique identifier of the user to retrieve.
245 Returns:
246 The user object, or ``None`` if not found.
247 """
248 return await self.user_store.get_user_by_id(user_id)
250 async def register(self, container: ContainerRegistrarProtocol) -> None:
251 """Register authentication services with the container.
253 Registers the protocols for dependency injection across extensions
254 without direct imports. The single ``PasswordHasherProtocol``
255 binding is the composed hasher (Argon2id default, bcrypt legacy
256 shim) — one registration per protocol per AGENTS.md §2.6.
257 """
258 # Register concrete implementations
259 container.singleton(PasswordPolicy, lambda: self.password_policy)
260 container.singleton(UserStoreProtocol, lambda: self.user_store)
261 container.singleton(AuthenticationService, lambda: self.service)
263 # Register protocol mappings for cross-extension compatibility (CROSS-EXT-02)
264 # This allows other extensions to depend on protocols rather than concrete classes
265 container.singleton(PasswordHasherProtocol, self.password_hasher)
266 container.singleton(PasswordPolicyProtocol, lambda: self.password_policy)
267 from lexigram.contracts.auth import AuthProviderProtocol
269 # Only register self as AuthProviderProtocol when no other provider
270 # has already claimed this slot (e.g., an application-level AuthService).
271 if not container.has(AuthProviderProtocol):
272 container.singleton(AuthProviderProtocol, lambda: self)
274 async def boot(self, container: BootContainerProtocol) -> None:
275 """Initialize authentication provider and register with kernel health registry."""
276 logger.info("AuthenticationProvider started")
277 hooks = await container.resolve_optional(HookRegistryProtocol)
278 self.service.set_hook_registry(hooks)
279 if self.token_manager is not None and hasattr(
280 self.token_manager, "set_hook_registry"
281 ):
282 self.token_manager.set_hook_registry(hooks)
284 # Token revocation needs a cache backend; when none was supplied at
285 # construction, adopt the application default cache so
286 # ``logout_all_user_tokens`` works without extra wiring.
287 if self.token_manager is not None and hasattr(
288 self.token_manager, "set_blacklist_resolver"
289 ):
290 # Ordering-proof: auth boots before cache providers, so hand a
291 # deferred source over the root resolver instead of resolving now.
292 self.token_manager.set_blacklist_resolver(
293 lambda: _resolve_default_cache_sync(container)
294 )
296 if container is not None and hasattr(container, "resolve"):
297 with contextlib.suppress(
298 ImportError, AttributeError, RuntimeError, TypeError
299 ):
300 from lexigram.contracts.observability.metrics import (
301 HealthCheckRegistryProtocol,
302 )
304 registry = await container.resolve_optional(HealthCheckRegistryProtocol)
305 if registry is not None:
306 registry.add("authentication", self.health_check)
308 async def shutdown(self) -> None:
309 """Shutdown authentication provider."""
310 logger.info("AuthenticationProvider shutdown")
312 async def verify_token(self, token: str) -> Result[VerifiedToken, TokenError]:
313 """Verify a JWT token and return a ``Result`` with the decoded payload.
315 Satisfies :class:`~lexigram.contracts.auth.AuthProviderProtocol`.
317 Delegates to ``token_manager.verify_token()`` when a token manager
318 is configured. Returns an ``Err`` result if no token manager is set
319 or if the token is invalid for any expected domain reason.
321 Infrastructure failures (cache unavailable, network errors) are still
322 raised as exceptions and must be handled by the caller.
324 Args:
325 token: The JWT token string to verify.
327 Returns:
328 ``Ok(VerifiedToken)`` if the token is valid and not revoked,
329 or ``Err(TokenError)`` for expected domain failures.
330 """
331 from lexigram.auth.exceptions import TokenInvalidError
333 if self.token_manager is None:
334 return Err(TokenInvalidError("Authentication not configured")) # type: ignore[arg-type]
335 return cast(
336 "Result[VerifiedToken, TokenError]",
337 await self.token_manager.verify_token(token),
338 )
340 async def validate_session(self, token: str) -> Any:
341 """Validate a session token and return user information.
343 Delegates to the underlying authentication service's
344 ``get_user_from_token`` method. Returns a
345 ``Result[VerifiedToken, TokenError]`` or ``None`` when no token
346 manager is configured.
348 Args:
349 token: The session or JWT token to validate.
351 Returns:
352 ``Result[VerifiedToken, TokenError]`` on success/failure, or ``None``.
353 """
354 if self.token_manager is None:
355 return None
356 return await self.service.get_user_from_token(token)
358 async def health_check(self, timeout: float = 5.0) -> HealthCheckResult:
359 """Check authentication provider health."""
360 return HealthCheckResult(
361 component=self.name,
362 status=HealthStatus.HEALTHY,
363 details={
364 "service": "authentication",
365 "password_policy": str(self.password_policy),
366 "user_store_type": type(self.user_store).__name__,
367 },
368 )