"""passbook audit models""" from json import dumps, loads from logging import getLogger from django.conf import settings from django.contrib.auth.models import AnonymousUser from django.core.exceptions import ValidationError from django.db import models from django.utils.translation import gettext as _ from ipware import get_client_ip from reversion import register from passbook.lib.models import CreatedUpdatedModel, UUIDModel LOGGER = getLogger(__name__) @register() class AuditEntry(UUIDModel): """An individual audit log entry""" ACTION_LOGIN = 'login' ACTION_LOGIN_FAILED = 'login_failed' ACTION_LOGOUT = 'logout' ACTION_AUTHORIZE_APPLICATION = 'authorize_application' ACTION_SUSPICIOUS_REQUEST = 'suspicious_request' ACTION_SIGN_UP = 'sign_up' ACTION_PASSWORD_RESET = 'password_reset' # noqa ACTION_INVITE_CREATED = 'invitation_created' ACTION_INVITE_USED = 'invitation_used' ACTIONS = ( (ACTION_LOGIN, ACTION_LOGIN), (ACTION_LOGIN_FAILED, ACTION_LOGIN_FAILED), (ACTION_LOGOUT, ACTION_LOGOUT), (ACTION_AUTHORIZE_APPLICATION, ACTION_AUTHORIZE_APPLICATION), (ACTION_SUSPICIOUS_REQUEST, ACTION_SUSPICIOUS_REQUEST), (ACTION_SIGN_UP, ACTION_SIGN_UP), (ACTION_PASSWORD_RESET, ACTION_PASSWORD_RESET), (ACTION_INVITE_CREATED, ACTION_INVITE_CREATED), (ACTION_INVITE_USED, ACTION_INVITE_USED), ) user = models.ForeignKey(settings.AUTH_USER_MODEL, null=True, on_delete=models.SET_NULL) action = models.TextField(choices=ACTIONS) date = models.DateTimeField(auto_now_add=True) app = models.TextField() _context = models.TextField() _context_cache = None request_ip = models.GenericIPAddressField() created = models.DateTimeField(auto_now_add=True) @property def context(self): """Load context data and load json""" if not self._context_cache: self._context_cache = loads(self._context) return self._context_cache @staticmethod def create(action, request, **kwargs): """Create AuditEntry from arguments""" client_ip, _ = get_client_ip(request) user = request.user if isinstance(user, AnonymousUser): user = kwargs.get('user', None) entry = AuditEntry.objects.create( action=action, user=user, # User 255.255.255.255 as fallback if IP cannot be determined request_ip=client_ip or '255.255.255.255', _context=dumps(kwargs)) LOGGER.debug("Logged %s from %s (%s)", action, request.user, client_ip) return entry def save(self, *args, **kwargs): if not self._state.adding: raise ValidationError("you may not edit an existing %s" % self._meta.model_name) super().save(*args, **kwargs) class Meta: verbose_name = _('Audit Entry') verbose_name_plural = _('Audit Entries') class LoginAttempt(CreatedUpdatedModel): """Track failed login-attempts""" target_uid = models.CharField(max_length=254) request_ip = models.GenericIPAddressField() attempts = models.IntegerField(default=1) @staticmethod def attempt(target_uid, request): """Helper function to create attempt or count up existing one""" client_ip, _ = get_client_ip(request) # Since we can only use 254 chars for target_uid, truncate target_uid. target_uid = target_uid[:254] existing_attempts = LoginAttempt.objects.filter( target_uid=target_uid, request_ip=client_ip).order_by('created') # TODO: Add logic to group attempts by timeframe, i.e. within 10 minutes if existing_attempts.exists(): attempt = existing_attempts.first() attempt.attempts += 1 attempt.save() LOGGER.debug("Increased attempts on %s", attempt) else: attempt = LoginAttempt.objects.create( target_uid=target_uid, request_ip=client_ip) LOGGER.debug("Created new attempt %s", attempt) def __str__(self): return "LoginAttempt to %s from %s (x%d)" % (self.target_uid, self.request_ip, self.attempts) class Meta: unique_together = (('target_uid', 'request_ip', 'created'),)