Merge branch 'fix-license-check-export-workspace-audit-log' into 'develop'

Fix license check export audit log workspace level

See merge request baserow/baserow!2775

changelog/entries/unreleased/bug/fix_license_check_export_workspace_audit_log.json

@@ -0,0 +1,7 @@
+{
+  "type": "bug",
+  "message": "Fix license check when exporting audit log from workspace as non-staff.",
+  "issue_number": null,
+  "bullet_points": [],
+  "created_at": "2024-10-08"
+}

enterprise/backend/src/baserow_enterprise/api/audit_log/views.py

@@ -1,6 +1,3 @@
-from typing import Optional
-
-from django.contrib.auth.models import AbstractBaseUser
 from django.db import transaction
 from django.utils import translation
 
@@ -8,7 +5,6 @@ from baserow_premium.api.admin.views import APIListingView
 from baserow_premium.license.handler import LicenseHandler
 from drf_spectacular.types import OpenApiTypes
 from drf_spectacular.utils import OpenApiParameter, extend_schema
-from rest_framework.exceptions import PermissionDenied
 from rest_framework.permissions import IsAdminUser, IsAuthenticated
 from rest_framework.response import Response
 from rest_framework.status import HTTP_202_ACCEPTED
@@ -25,15 +21,14 @@ from baserow.api.jobs.serializers import JobSerializer
 from baserow.api.schemas import CLIENT_SESSION_ID_SCHEMA_PARAMETER, get_error_schema
 from baserow.core.actions import DeleteWorkspaceActionType, OrderWorkspacesActionType
 from baserow.core.exceptions import WorkspaceDoesNotExist
-from baserow.core.handler import CoreHandler
 from baserow.core.jobs.exceptions import MaxJobCountExceeded
 from baserow.core.jobs.handler import JobHandler
 from baserow.core.jobs.registries import job_type_registry
 from baserow.core.models import User, Workspace
 from baserow_enterprise.audit_log.job_types import AuditLogExportJobType
 from baserow_enterprise.audit_log.models import AuditLogEntry
-from baserow_enterprise.audit_log.operations import (
-    ListWorkspaceAuditLogEntriesOperationType,
+from baserow_enterprise.audit_log.utils import (
+    check_for_license_and_permissions_or_raise,
 )
 from baserow_enterprise.features import AUDIT_LOG
 
@@ -50,30 +45,6 @@ from .serializers import (
 )
 
 
-def check_for_license_and_permissions_or_raise(
-    user: AbstractBaseUser, workspace_id: Optional[int] = None
-):
-    """
-    Check if the user has the feature enabled and has the correct permissions to list
-    audit log entries. If not, an exception is raised.
-    """
-
-    if workspace_id is not None:
-        workspace = CoreHandler().get_workspace(workspace_id)
-        LicenseHandler.raise_if_user_doesnt_have_feature(AUDIT_LOG, user, workspace)
-        if not user.is_staff:
-            CoreHandler().check_permissions(
-                user,
-                ListWorkspaceAuditLogEntriesOperationType.type,
-                workspace=workspace,
-                context=workspace,
-            )
-    else:
-        LicenseHandler.raise_if_user_doesnt_have_feature_instance_wide(AUDIT_LOG, user)
-        if not user.is_staff:
-            raise PermissionDenied()
-
-
 class AuditLogView(APIListingView):
     permission_classes = (IsAuthenticated,)
     serializer_class = AuditLogSerializer

enterprise/backend/src/baserow_enterprise/audit_log/job_types.py

@@ -8,7 +8,6 @@ from django.utils.translation import gettext as _
 from django.utils.translation import override as translation_override
 
 import unicodecsv as csv
-from baserow_premium.license.handler import LicenseHandler
 from loguru import logger
 from rest_framework import serializers
 
@@ -26,9 +25,9 @@ from baserow.core.action.registries import action_type_registry
 from baserow.core.jobs.registries import JobType
 from baserow.core.storage import get_default_storage
 from baserow.core.utils import ChildProgressBuilder
-from baserow_enterprise.features import AUDIT_LOG
 
 from .models import AuditLogEntry, AuditLogExportJob
+from .utils import check_for_license_and_permissions_or_raise
 
 AUDIT_LOG_CSV_COLUMN_NAMES = OrderedDict(
     {
@@ -266,9 +265,7 @@ class AuditLogExportJobType(JobType):
         :progress: The progress object that can be used to update the progress bar.
         """
 
-        LicenseHandler.raise_if_user_doesnt_have_feature_instance_wide(
-            AUDIT_LOG, job.user
-        )
+        check_for_license_and_permissions_or_raise(job.user, job.filter_workspace_id)
 
         queryset = self.get_filtered_queryset(job)
 

enterprise/backend/src/baserow_enterprise/audit_log/utils.py

@@ -0,0 +1,36 @@
+from typing import Optional
+
+from django.contrib.auth.models import AbstractBaseUser
+
+from baserow_premium.license.handler import LicenseHandler
+from rest_framework.exceptions import PermissionDenied
+
+from baserow.core.handler import CoreHandler
+from baserow_enterprise.audit_log.operations import (
+    ListWorkspaceAuditLogEntriesOperationType,
+)
+from baserow_enterprise.features import AUDIT_LOG
+
+
+def check_for_license_and_permissions_or_raise(
+    user: AbstractBaseUser, workspace_id: Optional[int] = None
+):
+    """
+    Check if the user has the feature enabled and has the correct permissions to list
+    audit log entries. If not, an exception is raised.
+    """
+
+    if workspace_id is not None:
+        workspace = CoreHandler().get_workspace(workspace_id)
+        LicenseHandler.raise_if_user_doesnt_have_feature(AUDIT_LOG, user, workspace)
+        if not user.is_staff:
+            CoreHandler().check_permissions(
+                user,
+                ListWorkspaceAuditLogEntriesOperationType.type,
+                workspace=workspace,
+                context=workspace,
+            )
+    else:
+        LicenseHandler.raise_if_user_doesnt_have_feature_instance_wide(AUDIT_LOG, user)
+        if not user.is_staff:
+            raise PermissionDenied()

enterprise/backend/tests/baserow_enterprise_tests/audit_log/test_audit_log_export_job.py

@@ -5,6 +5,7 @@ from unittest.mock import MagicMock, patch
 from django.test.utils import override_settings
 
 import pytest
+from baserow_premium.license.exceptions import FeaturesNotAvailableError
 from freezegun import freeze_time
 
 from baserow.contrib.database.export.handler import ExportHandler
@@ -288,3 +289,89 @@ def test_audit_log_export_workspace_csv_correctly(
     )
 
     close()
+
+
+@pytest.mark.django_db
+@override_settings(DEBUG=True)
+@patch("baserow.contrib.database.export.handler.get_default_storage")
+def test_audit_log_export_workspace_csv_correctly_if_feature_is_enable_for_the_user(
+    get_storage_mock, enterprise_data_fixture, synced_roles
+):
+    storage_mock = MagicMock()
+    get_storage_mock.return_value = storage_mock
+    user = enterprise_data_fixture.create_user()
+    workspace = enterprise_data_fixture.create_workspace(user=user)
+
+    with freeze_time("2023-01-01 12:00:00"):
+        app_1 = CreateApplicationActionType.do(
+            user, workspace, "database", name="App 1"
+        )
+
+    with freeze_time("2023-01-01 12:00:10"):
+        app_2 = CreateApplicationActionType.do(
+            user, workspace, "database", name="App 2"
+        )
+
+    csv_settings = {
+        "csv_column_separator": ",",
+        "csv_first_row_header": True,
+        "export_charset": "utf-8",
+        "exclude_columns": "workspace_id,workspace_name",
+    }
+
+    stub_file = BytesIO()
+    storage_mock.open.return_value = stub_file
+    close = stub_file.close
+    stub_file.close = lambda: None
+
+    # The user cannot use the feature without a license, both instance-wide and for a
+    # specific workspace.
+    with pytest.raises(FeaturesNotAvailableError):
+        csv_export_job = JobHandler().create_and_start_job(
+            user, AuditLogExportJobType.type, sync=True, **csv_settings
+        )
+
+    with pytest.raises(FeaturesNotAvailableError):
+        csv_export_job = JobHandler().create_and_start_job(
+            user,
+            AuditLogExportJobType.type,
+            sync=True,
+            filter_workspace_id=workspace.id,
+            **csv_settings,
+        )
+
+    # Patch only the workspace feature, the instance-wide feature should still raise an
+    # exception.
+    with patch(
+        "baserow_enterprise.audit_log.utils.LicenseHandler.raise_if_user_doesnt_have_feature"
+    ):
+        with pytest.raises(FeaturesNotAvailableError):
+            csv_export_job = JobHandler().create_and_start_job(
+                user, AuditLogExportJobType.type, sync=True, **csv_settings
+            )
+
+        # Providing the workspace id should work as it checks if the feature is enabled
+        # for the workspace.
+        csv_export_job = JobHandler().create_and_start_job(
+            user,
+            AuditLogExportJobType.type,
+            sync=True,
+            filter_workspace_id=workspace.id,
+            **csv_settings,
+        )
+        csv_export_job.refresh_from_db()
+        assert csv_export_job.state == JOB_FINISHED
+
+        data = stub_file.getvalue().decode(csv_settings["export_charset"])
+        bom = "\ufeff"
+
+        assert data == (
+            bom
+            + "User Email,User ID,Action Type,Description,Timestamp,IP Address\r\n"
+            + f'{user.email},{user.id},Create application,"""{app_2.name}"" ({app_2.id}) database created '
+            + f'in group ""{workspace.name}"" ({workspace.id}).",2023-01-01 12:00:10+00:00,\r\n'
+            + f'{user.email},{user.id},Create application,"""{app_1.name}"" ({app_1.id}) database created '
+            + f'in group ""{workspace.name}"" ({workspace.id}).",2023-01-01 12:00:00+00:00,\r\n'
+        )
+
+        close()