Coverage for src/backend/InvenTree/users/models.py: 67%

1"""Database model definitions for the 'users' app."""

3import datetime

5from django.conf import settings

6from django.contrib import admin

7from django.contrib.auth import get_user_model

8from django.contrib.auth.models import Group, User

9from django.contrib.contenttypes.fields import GenericForeignKey

10from django.contrib.contenttypes.models import ContentType

11from django.core.validators import MinLengthValidator

12from django.db import models

13from django.db.models import Q, UniqueConstraint

14from django.db.models.signals import m2m_changed, post_delete, post_save

15from django.db.utils import IntegrityError

16from django.dispatch import receiver

17from django.urls import reverse

18from django.utils.translation import gettext_lazy as _

20import structlog

21from allauth.account.models import EmailAddress

22from rest_framework.authtoken.models import Token as AuthToken

24import InvenTree.helpers

25import InvenTree.models

26from common.settings import get_global_setting

27from InvenTree.ready import isImportingData, isReadOnlyCommand

29from .ruleset import RULESET_CHOICES, get_ruleset_models

31logger = structlog.get_logger('inventree')

34# OVERRIDE START

35# Overrides Django User model __str__ with a custom function to be able to change

36# string representation of a user

37def user_model_str(self):

38 """Function to override the default Django User __str__."""

39 if get_global_setting('DISPLAY_FULL_NAMES', cache=True): 39 ↛ 40line 39 didn't jump to line 40 because the condition on line 39 was never true

40 if self.first_name or self.last_name:

41 return f'{self.first_name} {self.last_name}'

42 return self.username

45User.add_to_class('__str__', user_model_str) # Overriding User.__str__

46# OVERRIDE END

49if settings.LDAP_AUTH: 49 ↛ 50line 49 didn't jump to line 50 because the condition on line 49 was never true

50 from django_auth_ldap.backend import populate_user # ty: ignore[unresolved-import]

52 @receiver(populate_user)

53 def create_email_address(user, **kwargs):

54 """If a django user is from LDAP and has an email attached to it, create an allauth email address for them automatically.

56 https://django-auth-ldap.readthedocs.io/en/latest/users.html#populating-users

57 https://django-auth-ldap.readthedocs.io/en/latest/reference.html#django_auth_ldap.backend.populate_user

58 """

59 # User must exist in the database before we can create their EmailAddress. By their recommendation,

60 # we can just call .save() now

61 user.save()

63 # if they got an email address from LDAP, create it now and make it the primary

64 if (

65 user.email

66 and not EmailAddress.objects.filter(user=user, email=user.email).exists()

67 ):

68 EmailAddress.objects.create(user=user, email=user.email, primary=True)

71def default_token():

72 """Generate a default value for the token."""

73 return ApiToken.generate_key()

76def default_token_expiry():

77 """Generate an expiry date for a newly created token."""

78 return InvenTree.helpers.current_date() + datetime.timedelta(days=365)

81class ApiToken(AuthToken, InvenTree.models.MetadataMixin):

82 """Extends the default token model provided by djangorestframework.authtoken.

84 Extensions:

85 - Adds an 'expiry' date - tokens can be set to expire after a certain date

86 - Adds a 'name' field - tokens can be given a custom name (in addition to the user information)

87 """

89 class Meta:

90 """Metaclass defines model properties."""

92 verbose_name = _('API Token')

93 verbose_name_plural = _('API Tokens')

94 abstract = False

96 def __str__(self):

97 """String representation uses the redacted token."""

98 return self.token

100 @classmethod

101 def generate_key(cls, prefix='inv-'):

102 """Generate a new token key - with custom prefix."""

103 # Suffix is the date of creation

104 suffix = '-' + str(datetime.datetime.now().date().isoformat().replace('-', ''))

105

106 return prefix + str(AuthToken.generate_key()) + suffix

107

108 # Override the 'key' field - force it to be unique

109 key = models.CharField(

110 default=default_token,

111 verbose_name=_('Key'),

112 db_index=True,

113 unique=True,

114 max_length=100,

115 validators=[MinLengthValidator(50)],

116 )

117

118 # Override the 'user' field, to allow multiple tokens per user

119 user = models.ForeignKey(

120 settings.AUTH_USER_MODEL,

121 on_delete=models.CASCADE,

122 verbose_name=_('User'),

123 related_name='api_tokens',

124 )

125

126 name = models.CharField(

127 max_length=100,

128 blank=True,

129 verbose_name=_('Token Name'),

130 help_text=_('Custom token name'),

131 )

132

133 expiry = models.DateField(

134 default=default_token_expiry,

135 verbose_name=_('Expiry Date'),

136 help_text=_('Token expiry date'),

137 auto_now=False,

138 auto_now_add=False,

139 )

140

141 last_seen = models.DateField(

142 blank=True,

143 null=True,

144 verbose_name=_('Last Seen'),

145 help_text=_('Last time the token was used'),

146 )

147

148 revoked = models.BooleanField(

149 default=False, verbose_name=_('Revoked'), help_text=_('Token has been revoked')

150 )

151

152 @staticmethod

153 def sanitize_name(name: str) -> str:

154 """Sanitize the provide name value."""

155 name = str(name).strip()

156

157 # Remove any non-printable chars

158 name = InvenTree.helpers.remove_non_printable_characters(

159 name, remove_newline=True

160 )

161 name = InvenTree.helpers.strip_html_tags(name)

162

163 name = name.replace(' ', '-')

164 # Limit to 100 characters

165 name = name[:100]

166

167 return name

168

169 @property

170 @admin.display(description=_('Token'))

171 def token(self) -> str:

172 """Provide a redacted version of the token.

173

174 The *raw* key value should never be displayed anywhere!

175 """

176 # If the token has not yet been saved, return the raw key

177 if self.pk is None: 177 ↛ 178line 177 didn't jump to line 178 because the condition on line 177 was never true

178 return self.key # pragma: no cover

179

180 return InvenTree.helpers.sanitize_token(self.key)

181

182 @property

183 @admin.display(boolean=True, description=_('Expired'))

184 def expired(self) -> bool:

185 """Test if this token has expired."""

186 return (

187 self.expiry is not None and self.expiry < InvenTree.helpers.current_date()

188 )

189

190 @property

191 @admin.display(boolean=True, description=_('Active'))

192 def active(self) -> bool:

193 """Test if this token is active."""

194 return not self.revoked and not self.expired

195

196

197class RuleSet(models.Model):

198 """A RuleSet is somewhat like a superset of the django permission class, in that in encapsulates a bunch of permissions.

199

200 There are *many* apps models used within InvenTree,

201 so it makes sense to group them into "roles".

202

203 These roles translate (roughly) to the menu options available.

204

205 Each role controls permissions for a number of database tables,

206 which are then handled using the normal django permissions approach.

207 """

208

209 RULE_OPTIONS = ['can_view', 'can_add', 'can_change', 'can_delete']

210

211 class Meta:

212 """Metaclass defines additional model properties."""

213

214 unique_together = (('name', 'group'),)

215

216 @property

217 def label(self) -> str:

218 """Return the translated label for this ruleset."""

219 return dict(RULESET_CHOICES).get(self.name, self.name)

220

221 name = models.CharField(

222 max_length=50,

223 choices=RULESET_CHOICES,

224 blank=False,

225 help_text=_('Permission set'),

226 )

227

228 group = models.ForeignKey(

229 Group,

230 related_name='rule_sets',

231 blank=False,

232 null=False,

233 on_delete=models.CASCADE,

234 help_text=_('Group'),

235 )

236

237 can_view = models.BooleanField(

238 verbose_name=_('View'), default=False, help_text=_('Permission to view items')

239 )

240

241 can_add = models.BooleanField(

242 verbose_name=_('Add'), default=False, help_text=_('Permission to add items')

243 )

244

245 can_change = models.BooleanField(

246 verbose_name=_('Change'),

247 default=False,

248 help_text=_('Permissions to edit items'),

249 )

250

251 can_delete = models.BooleanField(

252 verbose_name=_('Delete'),

253 default=False,

254 help_text=_('Permission to delete items'),

255 )

256

257 def __str__(self, debug=False): # pragma: no cover

258 """Ruleset string representation."""

259 if debug:

260 # Makes debugging easier

261 return (

262 f'{str(self.group).ljust(15)}: {self.name.title().ljust(15)} | '

263 f'v: {str(self.can_view).ljust(5)} | a: {str(self.can_add).ljust(5)} | '

264 f'c: {str(self.can_change).ljust(5)} | d: {str(self.can_delete).ljust(5)}'

265 )

266 return self.name

267

268 def save(self, *args, **kwargs):

269 """Intercept the 'save' functionality to make additional permission changes.

270

271 It does not make sense to be able to change / create something,

272 but not be able to view it!

273 """

274 if self.can_add or self.can_change or self.can_delete: 274 ↛ 275line 274 didn't jump to line 275 because the condition on line 274 was never true

275 self.can_view = True

276

277 if self.can_add or self.can_delete: 277 ↛ 278line 277 didn't jump to line 278 because the condition on line 277 was never true

278 self.can_change = True

279

280 super().save(*args, **kwargs)

281

282 if self.group: 282 ↛ exitline 282 didn't return from function 'save' because the condition on line 282 was always true

283 # Update the group too!

284 # Note: This will trigger the 'update_group_roles' signal

285 self.group.save()

286

287 def get_models(self):

288 """Return the database tables / models that this ruleset covers."""

289 return get_ruleset_models().get(self.name, [])

290

291

292class Owner(models.Model):

293 """The Owner class is a proxy for a Group or User instance.

294

295 Owner can be associated to any InvenTree model (part, stock, build, etc.)

296

297 owner_type: Model type (Group or User)

298 owner_id: Group or User instance primary key

299 owner: Returns the Group or User instance combining the owner_type and owner_id fields

300 """

301

302 class Meta:

303 """Metaclass defines extra model properties."""

304

305 # Ensure all owners are unique

306 constraints = [

307 UniqueConstraint(fields=['owner_type', 'owner_id'], name='unique_owner')

308 ]

309

310 @classmethod

311 def get_owners_matching_user(cls, user):

312 """Return all "owner" objects matching the provided user.

313

314 Includes:

315 - An exact match for the user

316 - Any groups that the user is a part of

317 """

318 user_type = ContentType.objects.get(app_label='auth', model='user')

319 group_type = ContentType.objects.get(app_label='auth', model='group')

320

321 owners = []

322

323 try:

324 owners.append(cls.objects.get(owner_id=user.pk, owner_type=user_type))

325 except Exception: # pragma: no cover

326 pass

327

328 for group in user.groups.all(): 328 ↛ 329line 328 didn't jump to line 329 because the loop on line 328 never started

329 try:

330 owner = cls.objects.get(owner_id=group.pk, owner_type=group_type)

331 owners.append(owner)

332 except Exception: # pragma: no cover

333 pass

334

335 return owners

336

337 @staticmethod

338 def get_api_url(): # pragma: no cover

339 """Returns the API endpoint URL associated with the Owner model."""

340 return reverse('api-owner-list')

341

342 owner_type = models.ForeignKey(

343 ContentType, on_delete=models.CASCADE, null=True, blank=True

344 )

345

346 owner_id = models.PositiveIntegerField(null=True, blank=True)

347

348 owner = GenericForeignKey('owner_type', 'owner_id')

349

350 def __str__(self):

351 """Defines the owner string representation."""

352 if self.owner_type.name == 'user' and get_global_setting(

353 'DISPLAY_FULL_NAMES', cache=True

354 ):

355 display_name = self.owner.get_full_name()

356 else:

357 display_name = str(self.owner)

358 return f'{display_name} ({self.owner_type.name})'

359

360 def name(self):

361 """Return the 'name' of this owner."""

362 if self.owner_type.name == 'user' and get_global_setting( 362 ↛ 365line 362 didn't jump to line 365 because the condition on line 362 was never true

363 'DISPLAY_FULL_NAMES', cache=True

364 ):

365 if self.owner and hasattr(self.owner, 'get_full_name'):

366 # Use the get_full_name method if available

367 return self.owner.get_full_name() or str(self.owner)

368 else:

369 return str(self.owner)

370 return str(self.owner)

371

372 def label(self):

373 """Return the 'type' label of this owner i.e. 'user' or 'group'."""

374 return str(self.owner_type.name)

375

376 @classmethod

377 def create(cls, obj):

378 """Check if owner exist then create new owner entry."""

379 # Check for existing owner

380 existing_owner = cls.get_owner(obj)

381

382 if not existing_owner:

383 # Create new owner

384 try:

385 return cls.objects.create(owner=obj)

386 except IntegrityError: # pragma: no cover

387 return None

388

389 return existing_owner

390

391 @classmethod

392 def get_owner(cls, user_or_group):

393 """Get owner instance for a group or user."""

394 user_model = get_user_model()

395 owner = None

396 content_type_id = 0

397 content_type_id_list = [

398 ContentType.objects.get_for_model(Group).id,

399 ContentType.objects.get_for_model(user_model).id,

400 ]

401

402 # If instance type is obvious: set content type

403 if isinstance(user_or_group, Group):

404 content_type_id = content_type_id_list[0]

405 elif isinstance(user_or_group, get_user_model()): 405 ↛ 408line 405 didn't jump to line 408 because the condition on line 405 was always true

406 content_type_id = content_type_id_list[1]

407

408 if content_type_id: 408 ↛ 416line 408 didn't jump to line 416 because the condition on line 408 was always true

409 try:

410 owner = Owner.objects.get(

411 owner_id=user_or_group.id, owner_type=content_type_id

412 )

413 except Owner.DoesNotExist:

414 pass

415

416 return owner

417

418 def get_related_owners(self, include_group=False):

419 """Get all owners "related" to an owner.

420

421 This method is useful to retrieve all "user-type" owners linked to a "group-type" owner

422 """

423 user_model = get_user_model()

424 related_owners = None

425

426 if type(self.owner) is Group:

427 users = user_model.objects.filter(groups__name=self.owner.name)

428

429 if include_group:

430 # Include "group-type" owner in the query

431 query = Q(

432 owner_id__in=users,

433 owner_type=ContentType.objects.get_for_model(user_model).id,

434 ) | Q(

435 owner_id=self.owner.id,

436 owner_type=ContentType.objects.get_for_model(Group).id,

437 )

438 else:

439 query = Q(

440 owner_id__in=users,

441 owner_type=ContentType.objects.get_for_model(user_model).id,

442 )

443

444 related_owners = Owner.objects.filter(query)

445

446 elif type(self.owner) is user_model:

447 related_owners = [self]

448

449 return related_owners

450

451 def is_user_allowed(self, user, include_group: bool = False):

452 """Check if user is allowed to access something owned by this owner."""

453 user_owner = Owner.get_owner(user)

454 return user_owner in self.get_related_owners(include_group=include_group)

455

456

457@receiver(post_save, sender=Group, dispatch_uid='create_owner')

458@receiver(post_save, sender=get_user_model(), dispatch_uid='create_owner')

459def create_owner(sender, instance, **kwargs):

460 """Callback function to create a new owner instance after either a new group or user instance is saved."""

461 # Ignore during data import process to avoid data duplication

462 if not isReadOnlyCommand() and not isImportingData(): 462 ↛ exitline 462 didn't return from function 'create_owner' because the condition on line 462 was always true

463 Owner.create(obj=instance)

464

465

466@receiver(post_delete, sender=Group, dispatch_uid='delete_owner')

467@receiver(post_delete, sender=get_user_model(), dispatch_uid='delete_owner')

468def delete_owner(sender, instance, **kwargs):

469 """Callback function to delete an owner instance after either a new group or user instance is deleted."""

470 owner = Owner.get_owner(instance)

471 owner.delete()

472

473

474@receiver(post_save, sender=Group, dispatch_uid='create_missing_rule_sets')

475def create_missing_rule_sets(sender, instance, **kwargs):

476 """Called *after* a Group object is saved.

477

478 As the linked RuleSet instances are saved *before* the Group, then we can now use these RuleSet values to update the group permissions.

479 """

480 from users.tasks import update_group_roles

481

482 update_group_roles(instance)

483

484

485class UserProfile(InvenTree.models.MetadataMixin):

486 """Model to store additional user profile information."""

487

488 class UserType(models.TextChoices):

489 """Enumeration for user types."""

490

491 BOT = 'bot', _('Bot')

492 INTERNAL = 'internal', _('Internal')

493 EXTERNAL = 'external', _('External')

494 GUEST = 'guest', _('Guest')

495

496 user = models.OneToOneField(

497 User, on_delete=models.CASCADE, related_name='profile', verbose_name=_('User')

498 )

499 language = models.CharField(

500 max_length=10,

501 blank=True,

502 null=True,

503 verbose_name=_('Language'),

504 help_text=_('Preferred language for the user'),

505 )

506 theme = models.JSONField(

507 blank=True,

508 null=True,

509 verbose_name=_('Theme'),

510 help_text=_('Settings for the web UI as JSON - do not edit manually!'),

511 )

512 widgets = models.JSONField(

513 blank=True,

514 null=True,

515 verbose_name=_('Widgets'),

516 help_text=_(

517 'Settings for the dashboard widgets as JSON - do not edit manually!'

518 ),

519 )

520 displayname = models.CharField(

521 max_length=255,

522 blank=True,

523 null=True,

524 verbose_name=_('Display Name'),

525 help_text=_('Chosen display name for the user'),

526 )

527 position = models.CharField(

528 max_length=255,

529 blank=True,

530 null=True,

531 verbose_name=_('Position'),

532 help_text=_('Main job title or position'),

533 )

534 status = models.CharField(

535 max_length=2000,

536 blank=True,

537 null=True,

538 verbose_name=_('Status'),

539 help_text=_('User status message'),

540 )

541 location = models.CharField(

542 max_length=2000,

543 blank=True,

544 null=True,

545 verbose_name=_('Location'),

546 help_text=_('User location information'),

547 )

548 active = models.BooleanField(

549 default=True,

550 verbose_name=_('Active'),

551 help_text=_('User is actively using the system'),

552 )

553 contact = models.CharField(

554 max_length=255,

555 blank=True,

556 null=True,

557 verbose_name=_('Contact'),

558 help_text=_('Preferred contact information for the user'),

559 )

560 type = models.CharField(

561 max_length=10,

562 choices=UserType.choices,

563 default=UserType.INTERNAL,

564 verbose_name=_('User Type'),

565 help_text=_('Which type of user is this?'),

566 )

567 organisation = models.CharField(

568 max_length=255,

569 blank=True,

570 null=True,

571 verbose_name=_('Organisation'),

572 help_text=_('Users primary organisation/affiliation'),

573 )

574 primary_group = models.ForeignKey(

575 Group,

576 on_delete=models.SET_NULL,

577 null=True,

578 blank=True,

579 related_name='primary_users',

580 verbose_name=_('Primary Group'),

581 help_text=_('Primary group for the user'),

582 )

583

584 def __str__(self):

585 """Return string representation of the user profile."""

586 return f'{self.user.username} user profile'

587

588 def save(self, *args, **kwargs):

589 """Ensure primary_group is a group that the user is a member of."""

590 if self.primary_group and self.primary_group not in self.user.groups.all(): 590 ↛ 591line 590 didn't jump to line 591 because the condition on line 590 was never true

591 self.primary_group = None

592 super().save(*args, **kwargs)

593

594

595# Signal to create or update user profile when user is saved

596@receiver(post_save, sender=User)

597def create_or_update_user_profile(sender, instance, created, **kwargs):

598 """Create or update user profile when user is saved."""

599 # Disable profile creation if importing data from file or running a read-only command

600 if isReadOnlyCommand() or isImportingData(): 600 ↛ 601line 600 didn't jump to line 601 because the condition on line 600 was never true

601 return

602

603 if created:

604 UserProfile.objects.create(user=instance)

605 instance.profile.save()

606

607

608# Validate groups

609@receiver(post_save, sender=Group)

610def validate_primary_group_on_save(sender, instance, **kwargs):

611 """Validate primary_group on user profiles when a group is created or updated."""

612 for user in instance.user_set.all(): 612 ↛ 613line 612 didn't jump to line 613 because the loop on line 612 never started

613 profile = user.profile

614 if profile.primary_group and profile.primary_group not in user.groups.all():

615 profile.primary_group = None

616 profile.save()

617

618

619@receiver(post_delete, sender=Group)

620def validate_primary_group_on_delete(sender, instance, **kwargs):

621 """Validate primary_group on user profiles when a group is deleted."""

622 for user in instance.user_set.all(): 622 ↛ 623line 622 didn't jump to line 623 because the loop on line 622 never started

623 profile = user.profile

624 if profile.primary_group == instance:

625 profile.primary_group = None

626 profile.save()

627

628

629@receiver(m2m_changed, sender=User.groups.through)

630def validate_primary_group_on_group_change(sender, instance, action, **kwargs):

631 """Validate primary_group on user profiles when a group is added or removed."""

632 # Disable user profile validation if importing data from file

633 if isImportingData():

634 return

635

636 if action in ['post_add', 'post_remove']:

637 profile = instance.profile

638 if profile.primary_group and profile.primary_group not in instance.groups.all():

639 profile.primary_group = None

640 profile.save()