Add web/core/models_login.py
This commit is contained in:
parent
9a7ed59de7
commit
1620587ebb
54
web/core/models_login.py
Normal file
54
web/core/models_login.py
Normal file
@ -0,0 +1,54 @@
|
|||||||
|
from django.db import models
|
||||||
|
from django.utils import timezone
|
||||||
|
from datetime import timedelta
|
||||||
|
from django.dispatch import receiver
|
||||||
|
from django.contrib.auth.signals import user_logged_in, user_login_failed
|
||||||
|
|
||||||
|
class LoginAttempt(models.Model):
|
||||||
|
username = models.CharField(max_length=255, blank=True, default="")
|
||||||
|
ip_address = models.GenericIPAddressField(null=True, blank=True)
|
||||||
|
success = models.BooleanField(default=False)
|
||||||
|
timestamp = models.DateTimeField(auto_now_add=True)
|
||||||
|
|
||||||
|
class Meta:
|
||||||
|
indexes = [
|
||||||
|
models.Index(fields=["timestamp"]),
|
||||||
|
models.Index(fields=["username"]),
|
||||||
|
models.Index(fields=["success"]),
|
||||||
|
]
|
||||||
|
ordering = ["-timestamp"]
|
||||||
|
|
||||||
|
def __str__(self):
|
||||||
|
ok = "OK" if self.success else "FAIL"
|
||||||
|
return f"{self.timestamp:%Y-%m-%d %H:%M:%S} {ok} {self.username} {self.ip_address}"
|
||||||
|
|
||||||
|
def _get_client_ip(request):
|
||||||
|
if not request:
|
||||||
|
return None
|
||||||
|
xff = request.META.get("HTTP_X_FORWARDED_FOR")
|
||||||
|
if xff:
|
||||||
|
# first hop is the original client
|
||||||
|
return xff.split(",")[0].strip()
|
||||||
|
return request.META.get("REMOTE_ADDR")
|
||||||
|
|
||||||
|
def _prune_old():
|
||||||
|
cutoff = timezone.now() - timedelta(days=7)
|
||||||
|
_ = LoginAttempt.objects.filter(timestamp__lt=cutoff).delete()
|
||||||
|
|
||||||
|
@receiver(user_logged_in)
|
||||||
|
def _log_login_success(sender, request, user, **kwargs):
|
||||||
|
LoginAttempt.objects.create(
|
||||||
|
username=getattr(user, "username", "") or "",
|
||||||
|
ip_address=_get_client_ip(request),
|
||||||
|
success=True,
|
||||||
|
)
|
||||||
|
_prune_old()
|
||||||
|
|
||||||
|
@receiver(user_login_failed)
|
||||||
|
def _log_login_failed(sender, credentials, request, **kwargs):
|
||||||
|
LoginAttempt.objects.create(
|
||||||
|
username=(credentials or {}).get("username", "") or "",
|
||||||
|
ip_address=_get_client_ip(request),
|
||||||
|
success=False,
|
||||||
|
)
|
||||||
|
_prune_old()
|
||||||
Loading…
Reference in New Issue
Block a user