81 lines
3.1 KiB
Python
81 lines
3.1 KiB
Python
from django.db import models
|
||
from django.contrib.auth import get_user_model
|
||
|
||
|
||
User = get_user_model()
|
||
|
||
|
||
class UserProfile(models.Model):
|
||
user = models.OneToOneField(User, on_delete=models.CASCADE, related_name='profile')
|
||
display_name = models.CharField(max_length=100, blank=True, default='')
|
||
contact_phone = models.CharField(max_length=30, blank=True, default='')
|
||
default_free_traffic_gb_per_domain_override = models.PositiveIntegerField(null=True, blank=True)
|
||
|
||
class Meta:
|
||
verbose_name = '用户资料'
|
||
verbose_name_plural = '用户资料'
|
||
|
||
def __str__(self):
|
||
return self.display_name or self.user.get_username()
|
||
|
||
# Create your models here.
|
||
|
||
|
||
class LoginRecord(models.Model):
|
||
user = models.ForeignKey(User, on_delete=models.CASCADE, related_name='login_records')
|
||
ip_address = models.GenericIPAddressField(null=True, blank=True)
|
||
user_agent = models.CharField(max_length=255, blank=True, default='')
|
||
created_at = models.DateTimeField(auto_now_add=True)
|
||
|
||
class Meta:
|
||
verbose_name = '登录历史'
|
||
verbose_name_plural = '登录历史'
|
||
ordering = ['-created_at']
|
||
|
||
def __str__(self):
|
||
return f"{self.user.get_username()} @ {self.created_at}"
|
||
|
||
|
||
class LoginThrottle(models.Model):
|
||
user = models.ForeignKey(User, null=True, blank=True, on_delete=models.SET_NULL, related_name='login_throttles')
|
||
ip_address = models.GenericIPAddressField(null=True, blank=True)
|
||
fail_count = models.PositiveIntegerField(default=0)
|
||
total_fail_count = models.PositiveIntegerField(default=0)
|
||
last_failed_at = models.DateTimeField(null=True, blank=True)
|
||
banned_until = models.DateTimeField(null=True, blank=True)
|
||
note = models.CharField(max_length=255, blank=True, default='')
|
||
|
||
class Meta:
|
||
verbose_name = '登录节流'
|
||
verbose_name_plural = '登录节流'
|
||
ordering = ['-last_failed_at']
|
||
indexes = [
|
||
models.Index(fields=['ip_address']),
|
||
models.Index(fields=['user']),
|
||
models.Index(fields=['banned_until']),
|
||
]
|
||
|
||
def is_banned(self):
|
||
from django.utils import timezone
|
||
return bool(self.banned_until and self.banned_until > timezone.now())
|
||
|
||
def register_failure(self, ban_threshold=5, ban_minutes=15):
|
||
from django.utils import timezone
|
||
from datetime import timedelta
|
||
now = timezone.now()
|
||
self.fail_count += 1
|
||
self.total_fail_count += 1
|
||
self.last_failed_at = now
|
||
if self.fail_count >= ban_threshold:
|
||
self.banned_until = now + timedelta(minutes=int(ban_minutes))
|
||
self.note = f'自动封禁 {int(ban_minutes)} 分钟（超过失败阈值 {int(ban_threshold)}）'
|
||
self.fail_count = 0
|
||
self.save(update_fields=['fail_count', 'total_fail_count', 'last_failed_at', 'banned_until', 'note'])
|
||
|
||
def register_success(self):
|
||
if self.fail_count or self.banned_until:
|
||
self.fail_count = 0
|
||
self.banned_until = None
|
||
self.note = ''
|
||
self.save(update_fields=['fail_count', 'banned_until', 'note'])
|