feat: allow override of export limits, and bypass for superuser (#3883)

* feat: allow environment variable override of export limits

* feat: ignore export limits when superuser

backend/mlarchive/archive/view_funcs.py

@@ -156,13 +156,17 @@ def get_export(search, export_type, request):
     # don't allow export of huge querysets and skip empty querysets
     count = search.count()
     redirect_url = '%s?%s' % (reverse('archive_search'), request.META['QUERY_STRING'])
-    if (count > settings.EXPORT_LIMIT) or (count > settings.ANONYMOUS_EXPORT_LIMIT and not request.user.is_authenticated):  # noqa
-        messages.error(request, 'Too many messages to export.')
-        return redirect(redirect_url)
-    elif count == 0:
+    if count == 0:
         messages.error(request, 'No messages to export.')
         return redirect(redirect_url)
-
+    elif request.user.is_superuser:
+        pass
+    elif not request.user.is_authenticated and count > settings.ANONYMOUS_EXPORT_LIMIT:
+        messages.error(request, f'Export exceeds message limit of {settings.ANONYMOUS_EXPORT_LIMIT}')
+        return redirect(redirect_url)
+    elif count > settings.EXPORT_LIMIT:  # noqa
+        messages.error(request, f'Export exceeds message limit of {settings.EXPORT_LIMIT}')
+        return redirect(redirect_url)
     search = search.params(preserve_order=True)
     results = list(search.scan())
     apply_objects(results)

backend/mlarchive/settings/base.py

@@ -29,6 +29,7 @@
     # set casting, default value
     ADMINS=(list, []),
     ALLOWED_HOSTS=(list, ['*']),
+    ANONYMOUS_EXPORT_LIMIT=(int, 100),
     CELERY_BROKER_URL=(str, 'amqp://'),
     CLOUDFLARE_AUTH_EMAIL=(str, ''),
     CLOUDFLARE_AUTH_KEY=(str, ''),
@@ -47,6 +48,7 @@
     ELASTICSEARCH_HOST=(str, '127.0.0.1'),
     ELASTICSEARCH_PASSWORD=(str, 'changeme'),
     ELASTICSEARCH_SIGNAL_PROCESSOR=(str, 'mlarchive.archive.signals.CelerySignalProcessor'),
+    EXPORT_LIMIT=(int, 5000),
     HTAUTH_PASSWD_FILENAME=(str, ''),
     IMPORT_MESSAGE_APIKEY=(str, ''),
     INTERNAL_IPS=(list, []),
@@ -269,9 +271,12 @@
 ARCHIVE_MBOX_DIR = os.path.join(DATA_ROOT, 'archive_mbox')
 CONSOLE_STATS_FILE = os.path.join(DATA_ROOT, 'log', 'console.json')
 
-EXPORT_LIMIT = 5000             # maximum number of messages we will export
-ANONYMOUS_EXPORT_LIMIT = 100    # maximum number of messages a non-logged in user can export
-FILTER_CUTOFF = 5000            # maximum results for which we'll provide filter options
+# maximum number of messages a non-superuser can export
+EXPORT_LIMIT = env('EXPORT_LIMIT')
+# maximum number of messages a non-authenticated user can export
+ANONYMOUS_EXPORT_LIMIT = env('ANONYMOUS_EXPORT_LIMIT')
+# maximum results for which we'll provide filter options
+FILTER_CUTOFF = 5000
 
 LOG_DIR = env('LOG_DIR')
 LOG_FILE = os.path.join(LOG_DIR, 'mlarchive.log')

backend/mlarchive/tests/archive/view_funcs.py

@@ -114,6 +114,16 @@ def test_get_export_anonymous_limit(client, admin_client, thread_messages, setti
     assert response.status_code == 200
 
 
+@pytest.mark.django_db(transaction=True)
+def test_get_export_superuser_limit(client, admin_client, thread_messages, settings):
+    settings.EXPORT_LIMIT = 1
+    url = '%s?%s' % (reverse('archive_export', kwargs={'type': 'mbox'}), 'q=anvil')
+    response = client.get(url)
+    assert response.status_code == 302
+    response = admin_client.get(url)
+    assert response.status_code == 200
+
+
 @pytest.mark.django_db(transaction=True)
 def test_get_export_mbox(client, thread_messages, tmpdir):
     url = '%s?%s' % (reverse('archive_export', kwargs={'type': 'mbox'}), 'q=anvil')