Merge pull request #18 from City-of-Helsinki/ATV-51-add-audit-logging

ATV-51 | Add audit logging

Author: charn

atv/settings.py

@@ -45,6 +45,13 @@
     FIELD_ENCRYPTION_KEYS=(list, []),
     ENABLE_AUTOMATIC_ATTACHMENT_FILE_DELETION=(bool, True),
     ENABLE_SWAGGER_UI=(bool, True),
+    ELASTICSEARCH_APP_AUDIT_LOG_INDEX=(str, "atv_audit_log"),
+    ELASTICSEARCH_CLOUD_ID=(str, ""),
+    ELASTICSEARCH_API_ID=(str, ""),
+    ELASTICSEARCH_API_KEY=(str, ""),
+    ENABLE_SEND_AUDIT_LOG=(bool, False),
+    CLEAR_AUDIT_LOG_ENTRIES=(bool, False),
+    USE_X_FORWARDED_FOR=(bool, False),
 )
 if os.path.exists(env_file):
     env.read_env(env_file)
@@ -124,6 +131,7 @@
     "users",
     "services",
     "documents",
+    "audit_log",
 ]
 
 MIDDLEWARE = [
@@ -206,6 +214,18 @@
     "VERSION": "0.0.1",
 }
 
+# Audit logging
+
+AUDIT_LOG_ORIGIN = "atv"
+ELASTICSEARCH_APP_AUDIT_LOG_INDEX = env("ELASTICSEARCH_APP_AUDIT_LOG_INDEX")
+ELASTICSEARCH_CLOUD_ID = env("ELASTICSEARCH_CLOUD_ID")
+ELASTICSEARCH_API_ID = env("ELASTICSEARCH_API_ID")
+ELASTICSEARCH_API_KEY = env("ELASTICSEARCH_API_KEY")
+ENABLE_SEND_AUDIT_LOG = env("ENABLE_SEND_AUDIT_LOG")
+CLEAR_AUDIT_LOG_ENTRIES = env("CLEAR_AUDIT_LOG_ENTRIES")
+
+USE_X_FORWARDED_FOR = env("USE_X_FORWARDED_FOR")
+
 LOGGING = {
     "version": 1,
     "disable_existing_loggers": False,

audit_log/__init__.py

@@ -0,0 +1,18 @@
+from django.contrib import admin
+
+from audit_log.models import AuditLogEntry
+
+
+@admin.register(AuditLogEntry)
+class AuditLogEntryAdmin(admin.ModelAdmin):
+    fields = ("id", "created_at", "message")
+    readonly_fields = ("id", "created_at", "message")
+
+    def has_delete_permission(self, request, obj=None):
+        return False
+
+    def has_add_permission(self, request):
+        return False
+
+    def has_change_permission(self, request, obj=None):
+        return False

audit_log/admin.py

@@ -0,0 +1,8 @@
+from django.apps import AppConfig
+from django.utils.translation import gettext_lazy as _
+
+
+class AuditLogConfig(AppConfig):
+    default_auto_field = "django.db.models.BigAutoField"
+    name = "audit_log"
+    verbose_name = _("audit log")

audit_log/apps.py

@@ -0,0 +1,92 @@
+from datetime import datetime, timezone
+from typing import Callable, Optional, Union
+
+from django.conf import settings
+from django.contrib.auth import get_user_model
+from django.contrib.auth.models import AnonymousUser
+from django.db.models import Model
+from django.db.models.base import ModelBase
+
+from audit_log.enums import Operation, Role, Status
+from audit_log.models import AuditLogEntry
+
+User = get_user_model()
+
+
+def _now() -> datetime:
+    """Returns the current time in UTC timezone."""
+    return datetime.now(tz=timezone.utc)
+
+
+def _iso8601_date(time: datetime) -> str:
+    """Formats the timestamp in ISO-8601 format, e.g. '2020-06-01T00:00:00.000Z'."""
+    return f"{time.replace(tzinfo=None).isoformat(timespec='milliseconds')}Z"
+
+
+def log(
+    actor: Optional[Union[User, AnonymousUser]],
+    actor_backend: str,
+    operation: Operation,
+    target: Union[Model, ModelBase],
+    status: Status = Status.SUCCESS,
+    get_time: Callable[[], datetime] = _now,
+    ip_address: str = "",
+    additional_information: str = "",
+):
+    """
+    Write an event to the audit log.
+
+    Each audit log event has an actor (or None for system events),
+    an operation(e.g. READ or UPDATE), the target of the operation
+    (a Django model instance), status (e.g. SUCCESS), and a timestamp.
+    """
+    current_time = get_time()
+    user_id_field_name = getattr(target, "audit_log_id_field", "pk")
+    user_id = str(getattr(actor, user_id_field_name, ""))
+
+    if actor is None:
+        role = Role.SYSTEM
+    elif isinstance(actor, AnonymousUser):
+        role = Role.ANONYMOUS
+    elif actor.id == target.pk:
+        role = Role.OWNER
+    else:
+        role = Role.USER
+
+    message = {
+        "audit_event": {
+            "origin": settings.AUDIT_LOG_ORIGIN,
+            "status": status.value,
+            "date_time_epoch": int(current_time.timestamp() * 1000),
+            "date_time": _iso8601_date(current_time),
+            "actor": {
+                "role": role.value,
+                "user_id": user_id,
+                "provider": actor_backend if actor_backend else "",
+                "ip_address": ip_address,
+            },
+            "operation": operation.value,
+            "additional_information": additional_information,
+            "target": {
+                "id": _get_target_id(target),
+                "type": _get_target_type(target),
+            },
+        },
+    }
+    AuditLogEntry.objects.create(message=message)
+
+
+def _get_target_id(target: Union[Model, ModelBase]) -> str:
+    if isinstance(target, ModelBase):
+        return ""
+    field_name = getattr(target, "audit_log_id_field", "pk")
+    audit_log_id = getattr(target, field_name, "")
+    return str(audit_log_id)
+
+
+def _get_target_type(target: Union[Model, ModelBase]) -> str:
+    return (
+        str(target.__class__.__name__)
+        if isinstance(target, Model)
+        else str(target.__name__)
+    )

audit_log/audit_logging.py

@@ -0,0 +1,21 @@
+from django.db.models import TextChoices
+from django.utils.translation import gettext_lazy as _
+
+
+class Operation(TextChoices):
+    CREATE = "CREATE", _("Create")
+    READ = "READ", _("Read")
+    UPDATE = "UPDATE", _("Update")
+    DELETE = "DELETE", _("Delete")
+
+
+class Role(TextChoices):
+    OWNER = "OWNER", _("Owner")
+    USER = "USER", _("User")
+    SYSTEM = "SYSTEM", _("System")
+    ANONYMOUS = "ANONYMOUS", _("Anonymous")
+
+
+class Status(TextChoices):
+    SUCCESS = "SUCCESS", _("Success")
+    FORBIDDEN = "FORBIDDEN", _("Forbidden")

audit_log/enums.py

@@ -0,0 +1,10 @@
+from django.core.management.base import BaseCommand
+
+from audit_log.tasks import clear_audit_log_entries
+
+
+class Command(BaseCommand):
+    help = "Clear old sent audit log entries from database"
+
+    def handle(self, *args, **kwargs):
+        clear_audit_log_entries()

audit_log/management/__init__.py

@@ -0,0 +1,10 @@
+from django.core.management.base import BaseCommand
+
+from audit_log.tasks import send_audit_log_entries
+
+
+class Command(BaseCommand):
+    help = "Send unsent audit log entries to Elasticsearch"
+
+    def handle(self, *args, **kwargs):
+        send_audit_log_entries()