This repository has been archived on 2024-05-31. You can view files and clone it, but cannot push or open issues or pull requests.
authentik/passbook/audit/models.py

197 lines
6.5 KiB
Python

"""passbook audit models"""
from inspect import getmodule, stack
from typing import Any, Dict, Optional, Union
from uuid import UUID, uuid4
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.db.models.base import Model
from django.http import HttpRequest
from django.utils.translation import gettext as _
from django.views.debug import SafeExceptionReporterFilter
from guardian.utils import get_anonymous_user
from structlog import get_logger
from passbook.core.middleware import (
SESSION_IMPERSONATE_ORIGINAL_USER,
SESSION_IMPERSONATE_USER,
)
from passbook.core.models import User
from passbook.lib.utils.http import get_client_ip
LOGGER = get_logger("passbook.audit")
def cleanse_dict(source: Dict[Any, Any]) -> Dict[Any, Any]:
"""Cleanse a dictionary, recursively"""
final_dict = {}
for key, value in source.items():
try:
if SafeExceptionReporterFilter.hidden_settings.search(key):
final_dict[key] = SafeExceptionReporterFilter.cleansed_substitute
else:
final_dict[key] = value
except TypeError:
final_dict[key] = value
if isinstance(value, dict):
final_dict[key] = cleanse_dict(value)
return final_dict
def model_to_dict(model: Model) -> Dict[str, Any]:
"""Convert model to dict"""
name = str(model)
if hasattr(model, "name"):
name = model.name
return {
"app": model._meta.app_label,
"model_name": model._meta.model_name,
"pk": model.pk,
"name": name,
}
def get_user(user: User, original_user: Optional[User] = None) -> Dict[str, Any]:
"""Convert user object to dictionary, optionally including the original user"""
if isinstance(user, AnonymousUser):
user = get_anonymous_user()
user_data = {
"username": user.username,
"pk": user.pk,
"email": user.email,
}
if original_user:
original_data = get_user(original_user)
original_data["on_behalf_of"] = user_data
return original_data
return user_data
def sanitize_dict(source: Dict[Any, Any]) -> Dict[Any, Any]:
"""clean source of all Models that would interfere with the JSONField.
Models are replaced with a dictionary of {
app: str,
name: str,
pk: Any
}"""
final_dict = {}
for key, value in source.items():
if isinstance(value, dict):
final_dict[key] = sanitize_dict(value)
elif isinstance(value, models.Model):
final_dict[key] = sanitize_dict(model_to_dict(value))
elif isinstance(value, UUID):
final_dict[key] = value.hex
else:
final_dict[key] = value
return final_dict
class EventAction(models.TextChoices):
"""All possible actions to save into the audit log"""
LOGIN = "login"
LOGIN_FAILED = "login_failed"
LOGOUT = "logout"
SIGN_UP = "sign_up"
AUTHORIZE_APPLICATION = "authorize_application"
SUSPICIOUS_REQUEST = "suspicious_request"
PASSWORD_SET = "password_set" # noqa # nosec
INVITE_CREATED = "invitation_created"
INVITE_USED = "invitation_used"
IMPERSONATION_STARTED = "impersonation_started"
IMPERSONATION_ENDED = "impersonation_ended"
MODEL_CREATED = "model_created"
MODEL_UPDATED = "model_updated"
MODEL_DELETED = "model_deleted"
CUSTOM_PREFIX = "custom_"
class Event(models.Model):
"""An individual audit log event"""
event_uuid = models.UUIDField(primary_key=True, editable=False, default=uuid4)
user = models.JSONField(default=dict)
action = models.TextField(choices=EventAction.choices)
date = models.DateTimeField(auto_now_add=True)
app = models.TextField()
context = models.JSONField(default=dict, blank=True)
client_ip = models.GenericIPAddressField(null=True)
created = models.DateTimeField(auto_now_add=True)
@staticmethod
def _get_app_from_request(request: HttpRequest) -> str:
if not isinstance(request, HttpRequest):
return ""
return request.resolver_match.app_name
@staticmethod
def new(
action: Union[str, EventAction],
app: Optional[str] = None,
_inspect_offset: int = 1,
**kwargs,
) -> "Event":
"""Create new Event instance from arguments. Instance is NOT saved."""
if not isinstance(action, EventAction):
action = EventAction.CUSTOM_PREFIX + action
if not app:
app = getmodule(stack()[_inspect_offset][0]).__name__
cleaned_kwargs = cleanse_dict(sanitize_dict(kwargs))
event = Event(action=action, app=app, context=cleaned_kwargs)
return event
def from_http(
self, request: HttpRequest, user: Optional[settings.AUTH_USER_MODEL] = None
) -> "Event":
"""Add data from a Django-HttpRequest, allowing the creation of
Events independently from requests.
`user` arguments optionally overrides user from requests."""
if hasattr(request, "user"):
self.user = get_user(
request.user,
request.session.get(SESSION_IMPERSONATE_ORIGINAL_USER, None),
)
if user:
self.user = get_user(user)
# Check if we're currently impersonating, and add that user
if hasattr(request, "session"):
if SESSION_IMPERSONATE_ORIGINAL_USER in request.session:
self.user = get_user(request.session[SESSION_IMPERSONATE_ORIGINAL_USER])
self.user["on_behalf_of"] = get_user(
request.session[SESSION_IMPERSONATE_USER]
)
# User 255.255.255.255 as fallback if IP cannot be determined
self.client_ip = get_client_ip(request) or "255.255.255.255"
# If there's no app set, we get it from the requests too
if not self.app:
self.app = Event._get_app_from_request(request)
self.save()
return self
def save(self, *args, **kwargs):
if not self._state.adding:
raise ValidationError(
"you may not edit an existing %s" % self._meta.model_name
)
LOGGER.debug(
"Created Audit event",
action=self.action,
context=self.context,
client_ip=self.client_ip,
user=self.user,
)
return super().save(*args, **kwargs)
class Meta:
verbose_name = _("Audit Event")
verbose_name_plural = _("Audit Events")