Add web/core/models_login.py
This commit is contained in:
parent
8d2bf2d8e8
commit
4ab80ef1d1
57
web/core/models_login.py
Normal file
57
web/core/models_login.py
Normal file
@ -0,0 +1,57 @@
|
|||||||
|
1 from django.db import models
|
||||||
|
2 from django.utils import timezone
|
||||||
|
3 from datetime import timedelta
|
||||||
|
4 from django.dispatch import receiver
|
||||||
|
5 from django.contrib.auth.signals import user_logged_in, user_login_failed
|
||||||
|
6
|
||||||
|
7 class LoginAttempt(models.Model):
|
||||||
|
8 username = models.CharField(max_length=255, blank=True, default="")
|
||||||
|
9 ip_address = models.GenericIPAddressField(null=True, blank=True)
|
||||||
|
10 success = models.BooleanField(default=False)
|
||||||
|
11 timestamp = models.DateTimeField(auto_now_add=True)
|
||||||
|
12
|
||||||
|
13 class Meta:
|
||||||
|
14 indexes = [
|
||||||
|
15 models.Index(fields=["timestamp"]),
|
||||||
|
16 models.Index(fields=["username"]),
|
||||||
|
17 models.Index(fields=["success"]),
|
||||||
|
18 ]
|
||||||
|
19 ordering = ["-timestamp"]
|
||||||
|
20
|
||||||
|
21 def __str__(self):
|
||||||
|
22 ok = "OK" if self.success else "FAIL"
|
||||||
|
23 return f"{self.timestamp:%Y-%m-%d %H:%M:%S} {ok} {self.username} {self.ip_address}"
|
||||||
|
24
|
||||||
|
25
|
||||||
|
26 def _get_client_ip(request):
|
||||||
|
27 if not request:
|
||||||
|
28 return None
|
||||||
|
29 xff = request.META.get("HTTP_X_FORWARDED_FOR")
|
||||||
|
30 if xff:
|
||||||
|
31 # first hop is the original client
|
||||||
|
32 return xff.split(",")[0].strip()
|
||||||
|
33 return request.META.get("REMOTE_ADDR")
|
||||||
|
34
|
||||||
|
35 def _prune_old():
|
||||||
|
36 cutoff = timezone.now() - timedelta(days=7)
|
||||||
|
37 _ = LoginAttempt.objects.filter(timestamp__lt=cutoff).delete()
|
||||||
|
38
|
||||||
|
39 @receiver(user_logged_in)
|
||||||
|
40 def _log_login_success(sender, request, user, **kwargs):
|
||||||
|
41 from .models_login import LoginAttempt # local import to avoid circulars
|
||||||
|
42 LoginAttempt.objects.create(
|
||||||
|
43 username=getattr(user, "username", "") or "",
|
||||||
|
44 ip_address=_get_client_ip(request),
|
||||||
|
45 success=True,
|
||||||
|
46 )
|
||||||
|
47 _prune_old()
|
||||||
|
48
|
||||||
|
49 @receiver(user_login_failed)
|
||||||
|
50 def _log_login_failed(sender, credentials, request, **kwargs):
|
||||||
|
51 from .models_login import LoginAttempt # local import to avoid circulars
|
||||||
|
52 LoginAttempt.objects.create(
|
||||||
|
53 username=(credentials or {}).get("username", "") or "",
|
||||||
|
54 ip_address=_get_client_ip(request),
|
||||||
|
55 success=False,
|
||||||
|
56 )
|
||||||
|
57 _prune_old()
|
||||||
Loading…
Reference in New Issue
Block a user