Guard admin messages from leaking to frontend

apps/core/middleware.py

@@ -2,6 +2,8 @@ from __future__ import annotations
 
 import secrets
 
+from django.contrib.messages import get_messages
+
 from .consent import ConsentService
 
 
@@ -40,3 +42,25 @@ class SecurityHeadersMiddleware:
         )
         response["Permissions-Policy"] = "camera=(), microphone=(), geolocation=()"
         return response
+
+
+class AdminMessageGuardMiddleware:
+    ADMIN_PREFIXES = ("/cms/", "/django-admin/")
+
+    def __init__(self, get_response):
+        self.get_response = get_response
+
+    def __call__(self, request):
+        # The public site has no legitimate use of Django's shared flash queue.
+        # Drain any stale admin messages before frontend rendering can see them.
+        if not request.path.startswith(self.ADMIN_PREFIXES):
+            storage = get_messages(request)
+            list(storage)
+            storage._queued_messages = []
+            storage._loaded_data = []
+            for sub_storage in getattr(storage, "storages", []):
+                sub_storage._queued_messages = []
+                sub_storage._loaded_data = []
+                sub_storage.used = True
+            storage.used = True
+        return self.get_response(request)

apps/core/tests/test_message_handling.py

@@ -0,0 +1,78 @@
+from __future__ import annotations
+
+import pytest
+from django.contrib import messages
+from django.contrib.auth.models import AnonymousUser
+from django.contrib.messages import get_messages
+from django.contrib.messages.storage.fallback import FallbackStorage
+from django.contrib.sessions.middleware import SessionMiddleware
+from django.http import HttpResponse
+from django.shortcuts import render
+from django.test import RequestFactory, override_settings
+from django.urls import include, path
+
+from apps.core.middleware import AdminMessageGuardMiddleware
+
+
+def admin_message_test_view(request):
+    messages.success(request, "Page 'Test page' has been updated.")
+    messages.success(request, "Page 'Test page' has been updated.")
+    messages.success(request, "Page 'Test page' has been published.")
+    return render(request, "wagtailadmin/base.html", {})
+
+
+urlpatterns = [
+    path("cms/__tests__/admin-messages/", admin_message_test_view),
+    path("", include("config.urls")),
+]
+
+
+def _build_request(rf: RequestFactory, path: str):
+    request = rf.get(path)
+    SessionMiddleware(lambda req: None).process_request(request)
+    request.session.save()
+    request.user = AnonymousUser()
+    setattr(request, "_messages", FallbackStorage(request))
+    return request
+
+
+@pytest.mark.django_db
+def test_admin_message_guard_clears_stale_messages_on_frontend(rf):
+    request = _build_request(rf, "/articles/test/")
+    messages.success(request, "Page 'Test page' has been updated.")
+
+    response = AdminMessageGuardMiddleware(lambda req: HttpResponse("ok"))(request)
+
+    assert response.status_code == 200
+    assert list(get_messages(request)) == []
+
+
+@pytest.mark.django_db
+def test_admin_message_guard_preserves_admin_messages(rf):
+    request = _build_request(rf, "/cms/pages/1/edit/")
+    messages.success(request, "Page 'Test page' has been updated.")
+
+    response = AdminMessageGuardMiddleware(lambda req: HttpResponse("ok"))(request)
+    remaining = list(get_messages(request))
+
+    assert response.status_code == 200
+    assert len(remaining) == 1
+    assert remaining[0].message == "Page 'Test page' has been updated."
+
+
+@pytest.mark.django_db
+@override_settings(ROOT_URLCONF="apps.core.tests.test_message_handling")
+def test_wagtail_admin_template_deduplicates_consecutive_messages(client, django_user_model):
+    admin = django_user_model.objects.create_superuser(
+        username="admin-messages",
+        email="admin-messages@example.com",
+        password="admin-pass",
+    )
+    client.force_login(admin)
+
+    response = client.get("/cms/__tests__/admin-messages/")
+    content = response.content.decode()
+
+    assert response.status_code == 200
+    assert content.count("has been updated.") == 1
+    assert content.count("has been published.") == 1

config/settings/base.py

@@ -67,6 +67,7 @@ MIDDLEWARE = [
     "django.middleware.csrf.CsrfViewMiddleware",
     "django.contrib.auth.middleware.AuthenticationMiddleware",
     "django.contrib.messages.middleware.MessageMiddleware",
+    "apps.core.middleware.AdminMessageGuardMiddleware",
     "django.middleware.clickjacking.XFrameOptionsMiddleware",
     "django_htmx.middleware.HtmxMiddleware",
     "wagtail.contrib.redirects.middleware.RedirectMiddleware",

templates/wagtailadmin/base.html

@@ -0,0 +1,47 @@
+{% extends "wagtailadmin/admin_base.html" %}
+{% load wagtailadmin_tags wagtailcore_tags i18n %}
+
+{% block furniture %}
+    <template data-wagtail-sidebar-branding-logo>{% block branding_logo %}{% endblock %}</template>
+    {% sidebar_props %}
+    <aside id="wagtail-sidebar" class="sidebar-loading" data-wagtail-sidebar aria-label="{% trans 'Sidebar' %}"></aside>
+    {% keyboard_shortcuts_dialog %}
+    <main class="content-wrapper w-overflow-x-hidden" id="main">
+        <div class="content">
+            <div class="messages" role="status" data-controller="w-messages" data-action="w-messages:add@document->w-messages#add" data-w-messages-added-class="new" data-w-messages-show-class="appear" data-w-messages-show-delay-value="100">
+                <ul data-w-messages-target="container">
+                    {% if messages %}
+                        {% for message in messages %}
+                            {% message_level_tag message as level_tag %}
+                            {% ifchanged level_tag message.extra_tags message %}
+                                <li class="{% message_tags message %}">
+                                    {% if level_tag == "error" %}
+                                        {% icon name="warning" classname="messages-icon" %}
+                                    {% elif message.extra_tags == "lock" %}
+                                        {% icon name="lock" classname="messages-icon" %}
+                                    {% elif message.extra_tags == "unlock" %}
+                                        {% icon name="lock-open" classname="messages-icon" %}
+                                    {% else %}
+                                        {% icon name=level_tag classname="messages-icon" %}
+                                    {% endif %}
+                                    {{ message }}
+                                </li>
+                            {% endifchanged %}
+                        {% endfor %}
+                    {% endif %}
+                </ul>
+                <template data-w-messages-target="template" data-type="success">
+                    <li class="success">{% icon name="success" classname="messages-icon" %}<span></span></li>
+                </template>
+                <template data-w-messages-target="template" data-type="error">
+                    <li class="error">{% icon name="warning" classname="messages-icon" %}<span></span></li>
+                </template>
+                <template data-w-messages-target="template" data-type="warning">
+                    <li class="warning">{% icon name="warning" classname="messages-icon" %}<span></span></li>
+                </template>
+            </div>
+
+            {% block content %}{% endblock %}
+        </div>
+    </main>
+{% endblock %}