mirror of
https://github.com/jumpserver/jumpserver.git
synced 2025-09-10 20:00:10 +00:00
perf: split user model to many file
This commit is contained in:
ibuler
310
apps/users/models/user/__init__.py
Normal file
310
apps/users/models/user/__init__.py
Normal file
@@ -0,0 +1,310 @@
|
||||
#!/usr/bin/env python
|
||||
# -*- coding: utf-8 -*-
|
||||
#
|
||||
import uuid
|
||||
|
||||
from django.conf import settings
|
||||
from django.contrib.auth.models import AbstractUser
|
||||
from django.db import models
|
||||
from django.shortcuts import reverse
|
||||
from django.utils import timezone
|
||||
from django.utils.translation import gettext_lazy as _
|
||||
from rest_framework.exceptions import PermissionDenied
|
||||
|
||||
from common.db import fields, models as jms_models
|
||||
from common.utils import (
|
||||
date_expired_default,
|
||||
get_logger,
|
||||
)
|
||||
from labels.mixins import LabeledMixin
|
||||
from orgs.utils import current_org
|
||||
from ._auth import AuthMixin, MFAMixin
|
||||
from ._json import JSONFilterMixin
|
||||
from ._role import RoleMixin
|
||||
from ._source import SourceMixin, Source
|
||||
from ._token import TokenMixin
|
||||
|
||||
logger = get_logger(__file__)
|
||||
__all__ = [
|
||||
"User",
|
||||
"UserPasswordHistory",
|
||||
"MFAMixin"
|
||||
]
|
||||
|
||||
|
||||
class User(
|
||||
AuthMixin,
|
||||
SourceMixin,
|
||||
TokenMixin,
|
||||
RoleMixin,
|
||||
MFAMixin,
|
||||
LabeledMixin,
|
||||
JSONFilterMixin,
|
||||
AbstractUser,
|
||||
):
|
||||
id = models.UUIDField(default=uuid.uuid4, primary_key=True)
|
||||
username = models.CharField(max_length=128, unique=True, verbose_name=_("Username"))
|
||||
name = models.CharField(max_length=128, verbose_name=_("Name"))
|
||||
email = models.EmailField(max_length=128, unique=True, verbose_name=_("Email"))
|
||||
groups = models.ManyToManyField(
|
||||
"users.UserGroup",
|
||||
related_name="users",
|
||||
blank=True,
|
||||
verbose_name=_("User group"),
|
||||
)
|
||||
role = models.CharField(
|
||||
default="User", max_length=10, blank=True, verbose_name=_("Role")
|
||||
)
|
||||
is_service_account = models.BooleanField(
|
||||
default=False, verbose_name=_("Is service account")
|
||||
)
|
||||
avatar = models.ImageField(upload_to="avatar", null=True, verbose_name=_("Avatar"))
|
||||
wechat = fields.EncryptCharField(
|
||||
max_length=128, blank=True, verbose_name=_("Wechat")
|
||||
)
|
||||
phone = fields.EncryptCharField(
|
||||
max_length=128, blank=True, null=True, verbose_name=_("Phone")
|
||||
)
|
||||
mfa_level = models.SmallIntegerField(
|
||||
default=0, choices=MFAMixin.MFA_LEVEL_CHOICES, verbose_name=_("MFA")
|
||||
)
|
||||
otp_secret_key = fields.EncryptCharField(
|
||||
max_length=128, blank=True, null=True, verbose_name=_("OTP secret key")
|
||||
)
|
||||
# Todo: Auto generate key, let user download
|
||||
private_key = fields.EncryptTextField(
|
||||
blank=True, null=True, verbose_name=_("Private key")
|
||||
)
|
||||
public_key = fields.EncryptTextField(
|
||||
blank=True, null=True, verbose_name=_("Public key")
|
||||
)
|
||||
comment = models.TextField(blank=True, null=True, verbose_name=_("Comment"))
|
||||
is_first_login = models.BooleanField(default=True, verbose_name=_("Is first login"))
|
||||
date_expired = models.DateTimeField(
|
||||
default=date_expired_default,
|
||||
blank=True,
|
||||
null=True,
|
||||
db_index=True,
|
||||
verbose_name=_("Date expired"),
|
||||
)
|
||||
created_by = models.CharField(
|
||||
max_length=30, default="", blank=True, verbose_name=_("Created by")
|
||||
)
|
||||
updated_by = models.CharField(
|
||||
max_length=30, default="", blank=True, verbose_name=_("Updated by")
|
||||
)
|
||||
date_password_last_updated = models.DateTimeField(
|
||||
auto_now_add=True,
|
||||
blank=True,
|
||||
null=True,
|
||||
verbose_name=_("Date password last updated"),
|
||||
)
|
||||
need_update_password = models.BooleanField(
|
||||
default=False, verbose_name=_("Need update password")
|
||||
)
|
||||
source = models.CharField(
|
||||
max_length=30,
|
||||
default=Source.local,
|
||||
choices=Source.choices,
|
||||
verbose_name=_("Source"),
|
||||
)
|
||||
wecom_id = models.CharField(
|
||||
null=True, default=None, max_length=128, verbose_name=_("WeCom")
|
||||
)
|
||||
dingtalk_id = models.CharField(
|
||||
null=True, default=None, max_length=128, verbose_name=_("DingTalk")
|
||||
)
|
||||
feishu_id = models.CharField(
|
||||
null=True, default=None, max_length=128, verbose_name=_("FeiShu")
|
||||
)
|
||||
lark_id = models.CharField(
|
||||
null=True, default=None, max_length=128, verbose_name="Lark"
|
||||
)
|
||||
slack_id = models.CharField(
|
||||
null=True, default=None, max_length=128, verbose_name=_("Slack")
|
||||
)
|
||||
date_api_key_last_used = models.DateTimeField(
|
||||
null=True, blank=True, verbose_name=_("Date api key used")
|
||||
)
|
||||
date_updated = models.DateTimeField(auto_now=True, verbose_name=_("Date updated"))
|
||||
DATE_EXPIRED_WARNING_DAYS = 5
|
||||
|
||||
def __str__(self):
|
||||
return "{0.name}({0.username})".format(self)
|
||||
|
||||
@classmethod
|
||||
def get_queryset(cls):
|
||||
queryset = cls.objects.all()
|
||||
if not current_org.is_root():
|
||||
queryset = current_org.get_members()
|
||||
queryset = queryset.exclude(is_service_account=True)
|
||||
return queryset
|
||||
|
||||
@property
|
||||
def secret_key(self):
|
||||
instance = self.preferences.filter(name="secret_key").first()
|
||||
if not instance:
|
||||
return
|
||||
return instance.decrypt_value
|
||||
|
||||
@property
|
||||
def receive_backends(self):
|
||||
try:
|
||||
return self.user_msg_subscription.receive_backends
|
||||
except:
|
||||
return []
|
||||
|
||||
@property
|
||||
def is_otp_secret_key_bound(self):
|
||||
return bool(self.otp_secret_key)
|
||||
|
||||
def get_absolute_url(self):
|
||||
return reverse("users:user-detail", args=(self.id,))
|
||||
|
||||
@property
|
||||
def is_expired(self):
|
||||
if self.date_expired and self.date_expired < timezone.now():
|
||||
return True
|
||||
else:
|
||||
return False
|
||||
|
||||
def is_password_authenticate(self):
|
||||
cas = self.Source.cas
|
||||
saml2 = self.Source.saml2
|
||||
oauth2 = self.Source.oauth2
|
||||
return self.source not in [cas, saml2, oauth2]
|
||||
|
||||
@property
|
||||
def expired_remain_days(self):
|
||||
date_remain = self.date_expired - timezone.now()
|
||||
return date_remain.days
|
||||
|
||||
@property
|
||||
def will_expired(self):
|
||||
if 0 <= self.expired_remain_days <= self.DATE_EXPIRED_WARNING_DAYS:
|
||||
return True
|
||||
else:
|
||||
return False
|
||||
|
||||
@property
|
||||
def lang(self):
|
||||
return self.preference.get_value("lang")
|
||||
|
||||
@lang.setter
|
||||
def lang(self, value):
|
||||
self.preference.set_value('lang', value)
|
||||
|
||||
@property
|
||||
def preference(self):
|
||||
from users.models.preference import PreferenceManager
|
||||
return PreferenceManager(self)
|
||||
|
||||
@property
|
||||
def is_valid(self):
|
||||
if self.is_active and not self.is_expired:
|
||||
return True
|
||||
return False
|
||||
|
||||
def set_required_attr_if_need(self):
|
||||
if not self.name:
|
||||
self.name = self.username
|
||||
if not self.email or "@" not in self.email:
|
||||
email = "{}@{}".format(self.username, settings.EMAIL_SUFFIX)
|
||||
if "@" in self.username:
|
||||
email = self.username
|
||||
self.email = email
|
||||
|
||||
def save(self, *args, **kwargs):
|
||||
self.set_required_attr_if_need()
|
||||
if self.username == "admin":
|
||||
self.role = "Admin"
|
||||
self.is_active = True
|
||||
return super().save(*args, **kwargs)
|
||||
|
||||
def is_member_of(self, user_group):
|
||||
if user_group in self.groups.all():
|
||||
return True
|
||||
return False
|
||||
|
||||
def set_avatar(self, f):
|
||||
self.avatar.save(self.username, f)
|
||||
|
||||
@classmethod
|
||||
def get_avatar_url(cls, username):
|
||||
user_default = settings.STATIC_URL + "img/avatar/user.png"
|
||||
return user_default
|
||||
|
||||
def avatar_url(self):
|
||||
admin_default = settings.STATIC_URL + "img/avatar/admin.png"
|
||||
user_default = settings.STATIC_URL + "img/avatar/user.png"
|
||||
if self.avatar:
|
||||
return self.avatar.url
|
||||
if self.is_superuser:
|
||||
return admin_default
|
||||
else:
|
||||
return user_default
|
||||
|
||||
def unblock_login(self):
|
||||
from users.utils import LoginBlockUtil, MFABlockUtils
|
||||
|
||||
LoginBlockUtil.unblock_user(self.username)
|
||||
MFABlockUtils.unblock_user(self.username)
|
||||
|
||||
@property
|
||||
def login_blocked(self):
|
||||
from users.utils import LoginBlockUtil, MFABlockUtils
|
||||
|
||||
if LoginBlockUtil.is_user_block(self.username):
|
||||
return True
|
||||
if MFABlockUtils.is_user_block(self.username):
|
||||
return True
|
||||
return False
|
||||
|
||||
def delete(self, using=None, keep_parents=False):
|
||||
if self.pk == 1 or self.username == "admin":
|
||||
raise PermissionDenied(_("Can not delete admin user"))
|
||||
return super(User, self).delete(using=using, keep_parents=keep_parents)
|
||||
|
||||
class Meta:
|
||||
ordering = ["username"]
|
||||
verbose_name = _("User")
|
||||
unique_together = (
|
||||
("dingtalk_id",),
|
||||
("wecom_id",),
|
||||
("feishu_id",),
|
||||
("lark_id",),
|
||||
("slack_id",),
|
||||
)
|
||||
permissions = [
|
||||
("invite_user", _("Can invite user")),
|
||||
("remove_user", _("Can remove user")),
|
||||
("match_user", _("Can match user")),
|
||||
]
|
||||
|
||||
def can_send_created_mail(self):
|
||||
if self.email and self.source == self.Source.local.value:
|
||||
return True
|
||||
return False
|
||||
|
||||
|
||||
class UserPasswordHistory(models.Model):
|
||||
id = models.UUIDField(default=uuid.uuid4, primary_key=True)
|
||||
password = models.CharField(max_length=128)
|
||||
user = models.ForeignKey(
|
||||
"users.User",
|
||||
related_name="history_passwords",
|
||||
on_delete=jms_models.CASCADE_SIGNAL_SKIP,
|
||||
verbose_name=_("User"),
|
||||
)
|
||||
date_created = models.DateTimeField(
|
||||
auto_now_add=True, verbose_name=_("Date created")
|
||||
)
|
||||
|
||||
def __str__(self):
|
||||
return f"{self.user} set at {self.date_created}"
|
||||
|
||||
def __repr__(self):
|
||||
return self.__str__()
|
||||
|
||||
class Meta:
|
||||
verbose_name = _("User password history")
|
||||
Reference in New Issue
Block a user