Coverage for users/models/tokens.py: 71%
133 statements
« prev ^ index » next coverage.py v7.15.2, created at 2026-10-10 18:35 +0000
1import hashlib
2import hmac
3import secrets
4import zoneinfo
6from django.conf import settings
7from django.contrib.postgres.fields import ArrayField
8from django.core.exceptions import ValidationError
9from django.core.validators import MinLengthValidator
10from django.db import models
11from django.db.models import Q
12from django.urls import reverse
13from django.utils import timezone
14from django.utils.translation import gettext_lazy as _
15from netaddr import IPNetwork
17from ipam.fields import IPNetworkField
18from users.choices import TokenVersionChoices
19from users.constants import TOKEN_CHARSET, TOKEN_DEFAULT_LENGTH, TOKEN_KEY_LENGTH, TOKEN_PREFIX
20from users.utils import get_current_pepper
21from utilities.querysets import RestrictedQuerySet
23__all__ = (
24 'Token',
25)
28class Token(models.Model):
29 """
30 An API token used for user authentication. This extends the stock model to allow each user to have multiple tokens.
31 It also supports setting an expiration time and toggling write ability.
32 """
33 _token = None
35 version = models.PositiveSmallIntegerField(
36 verbose_name=_('version'),
37 choices=TokenVersionChoices,
38 default=TokenVersionChoices.V2,
39 )
40 user = models.ForeignKey(
41 to='users.User',
42 on_delete=models.CASCADE,
43 related_name='tokens'
44 )
45 description = models.CharField(
46 verbose_name=_('description'),
47 max_length=200,
48 blank=True
49 )
50 created = models.DateTimeField(
51 verbose_name=_('created'),
52 auto_now_add=True
53 )
54 expires = models.DateTimeField(
55 verbose_name=_('expires'),
56 blank=True,
57 null=True
58 )
59 last_used = models.DateTimeField(
60 verbose_name=_('last used'),
61 blank=True,
62 null=True
63 )
64 enabled = models.BooleanField(
65 verbose_name=_('enabled'),
66 default=True,
67 help_text=_('Disable to temporarily revoke this token without deleting it.'),
68 )
69 write_enabled = models.BooleanField(
70 verbose_name=_('write enabled'),
71 default=True,
72 help_text=_('Permit create/update/delete operations using this token')
73 )
74 # For legacy v1 tokens, this field stores the plaintext 40-char token value. Not used for v2.
75 plaintext = models.CharField(
76 verbose_name=_('plaintext'),
77 max_length=40,
78 unique=True,
79 blank=True,
80 null=True,
81 validators=[MinLengthValidator(40)],
82 )
83 key = models.CharField(
84 verbose_name=_('key'),
85 max_length=TOKEN_KEY_LENGTH,
86 unique=True,
87 blank=True,
88 null=True,
89 validators=[MinLengthValidator(TOKEN_KEY_LENGTH)],
90 help_text=_('v2 token identification key'),
91 )
92 pepper_id = models.PositiveSmallIntegerField(
93 verbose_name=_('pepper ID'),
94 blank=True,
95 null=True,
96 help_text=_('ID of the cryptographic pepper used to hash the token (v2 only)'),
97 )
98 hmac_digest = models.CharField(
99 verbose_name=_('digest'),
100 max_length=64,
101 blank=True,
102 null=True,
103 help_text=_('SHA256 hash of the token and pepper (v2 only)'),
104 )
105 allowed_ips = ArrayField(
106 base_field=IPNetworkField(),
107 blank=True,
108 null=True,
109 verbose_name=_('allowed IPs'),
110 help_text=_(
111 'Allowed IPv4/IPv6 networks from where the token can be used. Leave blank for no restrictions. '
112 'Ex: "10.1.1.0/24, 192.168.10.16/32, 2001:DB8:1::/64"'
113 ),
114 )
116 objects = RestrictedQuerySet.as_manager()
118 class Meta:
119 ordering = ('-created',)
120 indexes = (
121 models.Index(fields=('-created',)), # Default ordering
122 )
123 verbose_name = _('token')
124 verbose_name_plural = _('tokens')
125 constraints = [
126 models.CheckConstraint(
127 name='enforce_version_dependent_fields',
128 condition=(
129 Q(
130 version=1,
131 key__isnull=True,
132 pepper_id__isnull=True,
133 hmac_digest__isnull=True,
134 plaintext__isnull=False
135 ) |
136 Q(
137 version=2,
138 key__isnull=False,
139 pepper_id__isnull=False,
140 hmac_digest__isnull=False,
141 plaintext__isnull=True
142 )
143 ),
144 ),
145 ]
147 def __init__(self, *args, token=None, **kwargs):
148 super().__init__(*args, **kwargs)
150 # This stores the initial plaintext value (if given) on the creation of a new Token. If not provided, a
151 # random token value will be generated and assigned immediately prior to saving the Token instance.
152 self.token = token
154 def __str__(self):
155 return self.key if self.v2 else self.partial
157 def get_absolute_url(self):
158 return reverse('users:token', args=[self.pk])
160 @property
161 def v1(self):
162 return self.version == 1
164 @property
165 def v2(self):
166 return self.version == 2
168 @property
169 def partial(self):
170 """
171 Return a sanitized representation of a v1 token.
172 """
173 return f'**********************************{self.plaintext[-6:]}' if self.plaintext else ''
175 @property
176 def token(self):
177 return self._token
179 @token.setter
180 def token(self, value):
181 if not self._state.adding: 181 ↛ 182line 181 didn't jump to line 182 because the condition on line 181 was never true
182 raise ValueError("Cannot assign a new plaintext value for an existing token.")
183 self._token = value
184 if value is not None:
185 if self.v1: 185 ↛ 186line 185 didn't jump to line 186 because the condition on line 185 was never true
186 self.plaintext = value
187 elif self.v2: 187 ↛ exitline 187 didn't return from function 'token' because the condition on line 187 was always true
188 self.key = self.key or self.generate_key()
189 self.update_digest()
191 @property
192 def is_expired(self):
193 """
194 Check whether the token has expired.
195 """
196 if self.expires is None or timezone.now() < self.expires: 196 ↛ 198line 196 didn't jump to line 198 because the condition on line 196 was always true
197 return False
198 return True
200 @property
201 def is_active(self):
202 """
203 Check whether the token is active (enabled and not expired).
204 """
205 return self.enabled and not self.is_expired
207 def get_auth_header_prefix(self):
208 """
209 Return the HTTP Authorization header prefix for this token.
210 """
211 if self.v1: 211 ↛ 212line 211 didn't jump to line 212 because the condition on line 211 was never true
212 return 'Token '
213 if self.v2: 213 ↛ 215line 213 didn't jump to line 215 because the condition on line 213 was always true
214 return f'Bearer {TOKEN_PREFIX}{self.key}.'
215 return None
217 def clean(self):
218 super().clean()
220 if self.version == TokenVersionChoices.V2 and not settings.API_TOKEN_PEPPERS: 220 ↛ 221line 220 didn't jump to line 221 because the condition on line 220 was never true
221 raise ValidationError(_("Unable to save v2 tokens: API_TOKEN_PEPPERS is not defined."))
223 if self._state.adding: 223 ↛ 228line 223 didn't jump to line 228 because the condition on line 223 was never true
224 # Ensure a randomly-generated plaintext is always assigned to new tokens. A client-supplied value is
225 # never accepted via the REST API (the serializer's `token` field is read-only); generating it here
226 # guarantees the version-dependent key/digest fields are populated before constraint validation
227 # (full_clean) runs.
228 if self.token is None:
229 self.token = self.generate()
231 if self.pepper_id is not None and self.pepper_id not in settings.API_TOKEN_PEPPERS:
232 raise ValidationError(_(
233 "Invalid pepper ID: {id}. Check configured API_TOKEN_PEPPERS."
234 ).format(id=self.pepper_id))
236 # Prevent creating a token with a past expiration date
237 # while allowing updates to existing tokens.
238 if self.pk is None and self.is_expired: 238 ↛ 239line 238 didn't jump to line 239 because the condition on line 238 was never true
239 current_tz = zoneinfo.ZoneInfo(settings.TIME_ZONE)
240 now = timezone.now().astimezone(current_tz)
241 current_time_str = f'{now.date().isoformat()} {now.time().isoformat(timespec="seconds")}'
243 # Translators: {current_time} is the current server date and time in ISO format,
244 # {timezone} is the configured server time zone (for example, "UTC" or "Europe/Berlin").
245 message = _(
246 'Expiration time must be in the future. Current server time is {current_time} ({timezone}).'
247 ).format(current_time=current_time_str, timezone=current_tz.key)
249 raise ValidationError({'expires': message})
251 def save(self, *args, **kwargs):
252 # If creating a new Token and no token value has been specified, generate one
253 if self._state.adding and self.token is None:
254 self.token = self.generate()
256 return super().save(*args, **kwargs)
258 @classmethod
259 def generate_key(cls):
260 """
261 Generate and return a random alphanumeric key for v2 tokens.
262 """
263 return cls.generate(length=TOKEN_KEY_LENGTH)
265 @staticmethod
266 def generate(length=TOKEN_DEFAULT_LENGTH):
267 """
268 Generate and return a random token value of the given length.
269 """
270 return ''.join(secrets.choice(TOKEN_CHARSET) for _ in range(length))
272 def update_digest(self):
273 """
274 Recalculate and save the HMAC digest using the currently defined pepper and token values.
275 """
276 self.pepper_id, pepper = get_current_pepper()
277 self.hmac_digest = hmac.new(
278 pepper.encode('utf-8'),
279 self.token.encode('utf-8'),
280 hashlib.sha256
281 ).hexdigest()
283 def validate(self, token):
284 """
285 Validate the given plaintext against the token.
287 For v1 tokens, check that the given value is equal to the stored plaintext. For v2 tokens, calculate an HMAC
288 from the Token's pepper ID and the given plaintext value, and check whether the result matches the recorded
289 digest.
290 """
291 if self.v1: 291 ↛ 292line 291 didn't jump to line 292 because the condition on line 291 was never true
292 return hmac.compare_digest(token, self.plaintext)
293 if self.v2: 293 ↛ 302line 293 didn't jump to line 302 because the condition on line 293 was always true
294 token = token.removeprefix(TOKEN_PREFIX)
295 try:
296 pepper = settings.API_TOKEN_PEPPERS[self.pepper_id]
297 except KeyError:
298 # Invalid pepper ID
299 return False
300 digest = hmac.new(pepper.encode('utf-8'), token.encode('utf-8'), hashlib.sha256).hexdigest()
301 return hmac.compare_digest(digest, self.hmac_digest)
302 return False
304 def validate_client_ip(self, client_ip):
305 """
306 Validate the API client IP address against the source IP restrictions (if any) set on the token.
307 """
308 if not self.allowed_ips:
309 return True
311 for ip_network in self.allowed_ips:
312 if client_ip in IPNetwork(ip_network):
313 return True
315 return False