Coverage for src/lexigram/auth/authn/_jwt_creation.py: 88%
51 statements
« prev ^ index » next coverage.py v7.15.4, created at 2026-08-26 00:58 +0800
1"""Token creation mixin for :class:`~lexigram.auth.authn.jwt.JWTTokenManager`.
3This module is an internal implementation detail; import
4:class:`~lexigram.auth.authn.jwt.JWTTokenManager` directly.
5"""
7from __future__ import annotations
9from datetime import timedelta
10import secrets
11from typing import Any
13import jwt
15from lexigram.auth.authn._binding import TokenBindingConfig, compute_binding_hash
16from lexigram.auth.models import AuthToken
17from lexigram.auth.models.user import User
18from lexigram.contracts.core.identity import IdGeneratorProtocol
19from lexigram.primitives import clock as ambient_clock
22class _JWTCreationMixin:
23 """Mixin providing JWT token-creation methods for :class:`JWTTokenManager`.
25 All public attributes referenced here are initialised by
26 ``JWTTokenManager.__init__``; they are declared below as class-level
27 annotations solely to satisfy static type checkers.
28 """
30 # ── Attributes set by JWTTokenManager.__init__ ───────────────────────────
31 algorithm: str
32 access_expiration_hours: int
33 refresh_expiration_days: int
34 _required_audience: str | None
35 _binding_config: TokenBindingConfig | None
37 def __init__(
38 self,
39 ids: IdGeneratorProtocol | None = None,
40 ) -> None:
41 """Initialize the JWT creation mixin.
43 Args:
44 ids: Optional ID generator for JTI claims.
45 """
46 self._ids = ids
48 @property
49 def current_key_id(self) -> str: # pragma: no cover
50 """Active signing key ID — provided by JWTTokenManager."""
51 raise NotImplementedError
53 def _get_signing_key(self) -> str: # pragma: no cover
54 """Return raw signing key — provided by JWTTokenManager."""
55 raise NotImplementedError
57 # ─────────────────────────────────────────────────────────────────────────
59 def create_access_token(
60 self,
61 user: User,
62 additional_claims: dict[str, Any] | None = None,
63 binding_context: dict[str, str] | None = None,
64 ) -> str:
65 """Create a JWT access token for a user with current key."""
66 now = ambient_clock.now()
67 expires_at = now + timedelta(hours=self.access_expiration_hours)
69 # Previously we included the user's username in the token. The
70 # field was deprecated in favour of ``name`` so callers can refer to a
71 # human-readable label without using the deprecated word "username".
72 # Authentication logic should still use ``sub`` (user_id) and email.
73 payload = {
74 "sub": user.user_id,
75 "email": user.email,
76 # include name for compatibility with tests and helpers
77 "name": user.name,
78 "roles": user.roles,
79 "permissions": user.permissions,
80 "type": "access",
81 # Audience minted only when required_audience is configured,
82 # keeping default deployments byte-compatible.
83 **({"aud": self._required_audience} if self._required_audience else {}),
84 "jti": self._ids.generate_for("Token")
85 if self._ids
86 else secrets.token_urlsafe(16),
87 "iat": int(now.timestamp()),
88 "exp": int(expires_at.timestamp()),
89 }
91 if additional_claims:
92 payload.update(additional_claims)
94 if self._binding_config and binding_context:
95 bind_hash = compute_binding_hash(self._binding_config, binding_context)
96 if bind_hash:
97 payload["bind"] = bind_hash
99 # Include key ID in header for verification
100 headers = {"kid": self.current_key_id}
102 signing_key = self._get_signing_key()
103 return jwt.encode(
104 payload,
105 signing_key,
106 algorithm=self.algorithm,
107 headers=headers,
108 )
110 def create_refresh_token(
111 self,
112 user: User,
113 binding_context: dict[str, str] | None = None,
114 ) -> str:
115 """Create a JWT refresh token for a user."""
116 now = ambient_clock.now()
117 expires_at = now + timedelta(days=self.refresh_expiration_days)
119 payload = {
120 "sub": user.user_id,
121 "type": "refresh",
122 # Audience minted only when required_audience is configured,
123 # keeping default deployments byte-compatible.
124 **({"aud": self._required_audience} if self._required_audience else {}),
125 "jti": self._ids.generate_for("Token")
126 if self._ids
127 else secrets.token_urlsafe(16),
128 "iat": int(now.timestamp()),
129 "exp": int(expires_at.timestamp()),
130 }
132 if self._binding_config and binding_context:
133 bind_hash = compute_binding_hash(self._binding_config, binding_context)
134 if bind_hash:
135 payload["bind"] = bind_hash
137 # Include key ID in header for verification
138 headers = {"kid": self.current_key_id}
140 signing_key = self._get_signing_key()
141 return jwt.encode(
142 payload,
143 signing_key,
144 algorithm=self.algorithm,
145 headers=headers,
146 )
148 def create_token_pair(
149 self,
150 user: User,
151 additional_claims: dict[str, Any] | None = None,
152 binding_context: dict[str, str] | None = None,
153 ) -> AuthToken:
154 """Create both access and refresh tokens."""
155 access_token = self.create_access_token(
156 user, additional_claims, binding_context
157 )
158 refresh_token = self.create_refresh_token(user, binding_context)
160 now = ambient_clock.now()
161 access_expires = now + timedelta(hours=self.access_expiration_hours)
162 refresh_expires = now + timedelta(days=self.refresh_expiration_days)
164 return AuthToken(
165 token=access_token,
166 expires_at=access_expires,
167 refresh_token=refresh_token,
168 refresh_expires_at=refresh_expires,
169 )
171 def create_token(
172 self,
173 user: Any,
174 additional_claims: dict[str, Any] | None = None,
175 binding_context: dict[str, str] | None = None,
176 ) -> AuthToken:
177 """Create auth tokens for a user, satisfying the TokenManagerProtocol protocol.
179 Delegates to :meth:`create_token_pair`. This alias exists so that
180 ``isinstance(manager, TokenManagerProtocol)`` passes the runtime-checkable
181 protocol check from ``lexigram.contracts.auth.token``.
183 Args:
184 user: Authenticated user object.
185 additional_claims: Optional extra JWT payload fields.
186 binding_context: Optional request context dict for token binding.
187 Recognised keys: ``ip``, ``fingerprint``.
189 Returns:
190 AuthToken containing the access and refresh tokens.
191 """
192 return self.create_token_pair(user, additional_claims, binding_context)