Add logic for audit logging

Add audit logging mechanism. `AuditLoggingModelViewSet` is included to be able
to log all the actions done through a viewset. Viewset calls a separate logging
function which writes audit logs in a specified format into the DB. Failing to
write audit log will fail the whole operation i.e. action fails if audit log
cannot be written. Later logs will be sent from the DB to Elasticsearch.
Management commands for sending logs to Elasticsearch and to clean sent logs
are included. Management command can be called e.g. periodically from a cron.

Implementation is influenced by earlier work done for appartment-application-service,
yjdh, open-city-profile and youth-membership projects.

Log format follows the audit log specification for the project:
- https://helsinkisolutionoffice.atlassian.net/wiki/spaces/KAN/pages/6400704679/Audit+log+Asiointitietovaranto
- https://helsinkisolutionoffice.atlassian.net/wiki/spaces/KAN/pages/416972828/Helsinki+profile+audit+logging

- https://github.com/City-of-Helsinki/apartment-application-service/tree/develop/audit_log
- https://github.com/City-of-Helsinki/yjdh/tree/develop/backend/shared/shared/audit_log
- https://github.com/City-of-Helsinki/open-city-profile/blob/develop/profiles/audit_log.py
- https://github.com/City-of-Helsinki/youth-membership/blob/master/common_utils/audit_logging.py

Refs ATV-51

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()