Fixes #23552: Restrict database writes in maintenance mode

docs/configuration/miscellaneous.md

@@ -178,7 +178,17 @@ The number of days to retain job results (scripts and reports). Set this to `0`
 
 Default: `False`
 
-Setting this to `True` will display a "maintenance mode" banner at the top of every page. Additionally, NetBox will no longer update a user's "last active" time upon login. This is to allow new logins when the database is in a read-only state. Recording of login times will resume when maintenance mode is disabled.
+Setting this to `True` places NetBox in a read-only state, and displays a "maintenance mode" banner (see [`BANNER_MAINTENANCE`](#banner_maintenance)) at the top of every page.
+
+While maintenance mode is enabled, any request which attempts to write to the database through the web UI or the REST API fails with an error message. The following remain available:
+
+* Logging in and out, including via single sign-on and remote authentication
+* Modifying the configuration under Admin > Configuration History, so that maintenance mode can be disabled again from the UI (unless it has been set in `configuration.py`)
+
+Additionally, NetBox will not update a user's "last login" time, or a token's "last used" time, while maintenance mode is enabled. This allows users to log in when the database itself is read-only. Note that logging in also saves the user's session, which is stored in the database unless [`SESSION_FILE_PATH`](./security.md#session_file_path) is set.
+
+!!! note
+    Maintenance mode applies only to web UI and API requests. It does not prevent background jobs (such as scheduled scripts, data source synchronization, and housekeeping tasks) or management commands from writing to the database.
 
 ---
 

netbox/account/views.py

@@ -8,8 +8,6 @@ from django.contrib.auth import logout as auth_logout
 from django.contrib.auth import update_session_auth_hash
 from django.contrib.auth.forms import AuthenticationForm, PasswordChangeForm
 from django.contrib.auth.mixins import LoginRequiredMixin
-from django.contrib.auth.models import update_last_login
-from django.contrib.auth.signals import user_logged_in
 from django.http import HttpResponseRedirect, JsonResponse
 from django.shortcuts import get_object_or_404, redirect, render, resolve_url
 from django.urls import reverse, reverse_lazy
@@ -117,12 +115,6 @@ class LoginView(View):
         if form.is_valid():
             logger.debug("Login form validation was successful")
 
-            # If maintenance mode is enabled, assume the database is read-only, and disable updating the user's
-            # last_login time upon authentication.
-            if get_config().MAINTENANCE_MODE:
-                logger.warning("Maintenance mode enabled: disabling update of most recent login time")
-                user_logged_in.disconnect(update_last_login, dispatch_uid='update_last_login')
-
             # Authenticate user
             auth_login(request, form.get_user())
             logger.info(f"User {request.user} successfully authenticated")

netbox/netbox/middleware.py

@@ -6,7 +6,7 @@ from django.contrib import auth, messages
 from django.contrib.auth.middleware import RemoteUserMiddleware as RemoteUserMiddleware_
 from django.core.exceptions import ImproperlyConfigured
 from django.core.signals import got_request_exception
-from django.db import ProgrammingError, connection
+from django.db import DatabaseError, ProgrammingError, connection
 from django.db.utils import InternalError
 from django.http import Http404, HttpResponseRedirect
 from django.middleware.common import CommonMiddleware as DjangoCommonMiddleware
@@ -256,12 +256,20 @@ class MaintenanceModeMiddleware:
         self.get_response = get_response
 
     def __call__(self, request):
-        if get_config().MAINTENANCE_MODE:
-            self._set_session_type(
-                allow_write=request.path_info.startswith(settings.MAINTENANCE_EXEMPT_PATHS)
-            )
+        if not get_config().MAINTENANCE_MODE or request.path_info.startswith(settings.MAINTENANCE_EXEMPT_PATHS):
+            return self.get_response(request)
 
-        return self.get_response(request)
+        # The read-only setting persists on the connection, so restore read/write access once the request has been
+        # handled. This also lets outer middleware (e.g. saving the session) write to the database.
+        self._set_session_type(allow_write=False)
+        try:
+            return self.get_response(request)
+        finally:
+            try:
+                self._set_session_type(allow_write=True)
+            except DatabaseError:
+                # Never leave a read-only connection to be reused by a later request
+                connection.close()
 
     @staticmethod
     def _set_session_type(allow_write):

netbox/netbox/settings.py

@@ -671,10 +671,10 @@ EXEMPT_EXCLUDE_MODELS = (
 
 # All URLs starting with a string listed here are exempt from maintenance mode enforcement
 MAINTENANCE_EXEMPT_PATHS = (
-    f'/{BASE_PATH}extras/config-revisions/',  # Allow modifying the configuration
+    f'/{BASE_PATH}core/config-revisions/',  # Allow modifying the configuration
     LOGIN_URL,
-    LOGIN_REDIRECT_URL,
-    LOGOUT_REDIRECT_URL
+    f'/{BASE_PATH}logout/',
+    f'/{BASE_PATH}oauth/',  # Single sign-on
 )
 
 

netbox/netbox/tests/test_middleware.py

@@ -3,13 +3,21 @@ from contextlib import contextmanager
 from types import SimpleNamespace
 from unittest.mock import Mock, patch
 
+from django.contrib.messages import get_messages
+from django.core.cache import cache
 from django.core.signals import got_request_exception
+from django.db import connection
 from django.db.utils import InternalError
-from django.test import RequestFactory, override_settings
+from django.test import RequestFactory, TransactionTestCase, override_settings
 from django.urls import reverse
 from rest_framework import status
 
+from core.models import ConfigRevision
+from dcim.models import Site
+from netbox.config import clear_config
 from netbox.middleware import CoreMiddleware, MaintenanceModeMiddleware
+from users.constants import TOKEN_PREFIX
+from users.models import Token, User
 from utilities.testing import TestCase
 
 
@@ -156,3 +164,165 @@ class CoreMiddlewareTestCase(TestCase):
             response = self.process_internal_error(request, 'Simulated maintenance mode GraphQL error')
 
         self.assert_json_500_response(response)
+
+
+def is_read_only():
+    with connection.cursor() as cursor:
+        cursor.execute('SHOW transaction_read_only')
+        return cursor.fetchone()[0] == 'on'
+
+
+@override_settings(MAINTENANCE_MODE=True)
+class MaintenanceModeTestCase(TransactionTestCase):
+    """
+    Maintenance mode is enforced on the database session, which has no effect on writes within the transaction that
+    wraps each TestCase test, so these tests run against a real connection.
+    """
+
+    def setUp(self):
+        self.user = User.objects.create_user(username='testuser', password='testpassword', is_superuser=True)
+
+    def tearDown(self):
+        # Don't leave a read-only connection behind for other tests
+        with connection.cursor() as cursor:
+            cursor.execute('SET SESSION CHARACTERISTICS AS TRANSACTION READ WRITE')
+
+    def get_read_only_during_request(self, path):
+        """
+        Return whether the database connection was read-only while handling a request for the given path.
+        """
+        read_only = []
+
+        def get_response(request):
+            read_only.append(is_read_only())
+
+        MaintenanceModeMiddleware(get_response)(RequestFactory().get(path))
+        return read_only[0]
+
+    def test_exempt_paths(self):
+        exempt_paths = (
+            reverse('login'),
+            reverse('logout'),
+            reverse('social:begin', kwargs={'backend': 'github'}),
+            reverse('social:complete', kwargs={'backend': 'github'}),
+            reverse('core:configrevision_list'),
+            reverse('core:configrevision_add'),
+        )
+        for path in exempt_paths:
+            with self.subTest(path=path):
+                self.assertFalse(self.get_read_only_during_request(path))
+                self.assertFalse(is_read_only())
+
+    def test_restricted_paths(self):
+        restricted_paths = (
+            reverse('home'),
+            reverse('dcim:site_add'),
+            reverse('account:preferences'),
+            reverse('api-root'),
+            reverse('graphql'),
+        )
+        for path in restricted_paths:
+            with self.subTest(path=path):
+                self.assertTrue(self.get_read_only_during_request(path))
+                self.assertFalse(is_read_only(), "Connection was left read-only after the request")
+
+    def test_maintenance_mode_disabled(self):
+        with override_settings(MAINTENANCE_MODE=False):
+            self.assertFalse(self.get_read_only_during_request(reverse('dcim:site_add')))
+
+    def test_ui_write_refused(self):
+        self.client.force_login(self.user)
+        url = reverse('dcim:site_add')
+
+        response = self.client.post(url, {'name': 'Site 1', 'slug': 'site-1', 'status': 'active'})
+
+        self.assertRedirects(response, url, fetch_redirect_response=False)
+        self.assertIn('maintenance mode', str(list(get_messages(response.wsgi_request))[0]))
+        self.assertFalse(Site.objects.exists())
+
+    def test_api_write_refused(self):
+        token = Token.objects.create(user=self.user)
+        header = {'HTTP_AUTHORIZATION': f'Bearer {TOKEN_PREFIX}{token.key}.{token.token}'}
+
+        response = self.client.post(
+            reverse('dcim-api:site-list'),
+            {'name': 'Site 1', 'slug': 'site-1', 'status': 'active'},
+            content_type='application/json',
+            **header,
+        )
+
+        self.assertEqual(response.status_code, status.HTTP_500_INTERNAL_SERVER_ERROR)
+        self.assertIn('maintenance mode', response.json()['error'])
+        self.assertFalse(Site.objects.exists())
+
+    def test_writes_resume_when_maintenance_mode_disabled(self):
+        self.client.force_login(self.user)
+        data = {'name': 'Site 1', 'slug': 'site-1', 'status': 'active'}
+        self.client.post(reverse('dcim:site_add'), data)
+
+        with override_settings(MAINTENANCE_MODE=False):
+            response = self.client.post(reverse('dcim:site_add'), data)
+
+        self.assertEqual(response.status_code, 302)
+        self.assertTrue(Site.objects.filter(slug='site-1').exists())
+
+    @override_settings(SESSION_SAVE_EVERY_REQUEST=True)
+    def test_session_saved_on_every_request(self):
+        """
+        LOGIN_PERSISTENCE saves the session on every request.
+        """
+        self.client.force_login(self.user)
+
+        response = self.client.get(reverse('dcim:site_list'))
+
+        self.assertEqual(response.status_code, 200)
+
+    def test_login_and_logout(self):
+        response = self.client.post(reverse('login'), {'username': 'testuser', 'password': 'testpassword'})
+        self.assertEqual(response.status_code, 302)
+        self.assertEqual(int(self.client.session['_auth_user_id']), self.user.pk)
+
+        # The login time isn't recorded in maintenance mode
+        self.user.refresh_from_db()
+        self.assertIsNone(self.user.last_login)
+
+        response = self.client.get(reverse('logout'))
+        self.assertEqual(response.status_code, 302)
+        self.assertNotIn('_auth_user_id', self.client.session)
+
+        # The login time is recorded once maintenance mode is disabled
+        with override_settings(MAINTENANCE_MODE=False):
+            self.client.post(reverse('login'), {'username': 'testuser', 'password': 'testpassword'})
+        self.user.refresh_from_db()
+        self.assertIsNotNone(self.user.last_login)
+
+    @override_settings(REMOTE_AUTH_ENABLED=True, LOGIN_REQUIRED=True)
+    def test_remote_auth_login(self):
+        User.objects.create(username='remoteuser1')
+
+        # A refused write must not leave the connection read-only for the next request
+        self.client.force_login(self.user)
+        self.client.post(reverse('dcim:site_add'), {'name': 'Site 1', 'slug': 'site-1', 'status': 'active'})
+        self.client.logout()
+
+        response = self.client.get(reverse('home'), follow=True, HTTP_REMOTE_USER='remoteuser1')
+
+        self.assertEqual(response.status_code, 200)
+        self.assertEqual(self.client.session['_auth_user_id'], str(User.objects.get(username='remoteuser1').pk))
+
+    # Use a per-process cache, so that activating the new revision doesn't affect other tests
+    @override_settings(CACHES={
+        'default': {
+            'BACKEND': 'django.core.cache.backends.locmem.LocMemCache',
+            'LOCATION': 'netbox-maintenance-mode-tests',
+        },
+    })
+    def test_modify_configuration(self):
+        self.addCleanup(cache.clear)
+        self.addCleanup(clear_config)
+        self.client.force_login(self.user)
+
+        response = self.client.post(reverse('core:configrevision_add'), {'comment': 'Disable maintenance mode'})
+
+        self.assertEqual(response.status_code, 302)
+        self.assertTrue(ConfigRevision.objects.filter(comment='Disable maintenance mode').exists())

netbox/users/signals.py

@@ -1,6 +1,7 @@
 import copy
 import logging
 
+from django.contrib.auth.models import update_last_login as django_update_last_login
 from django.contrib.auth.signals import user_logged_in, user_login_failed
 from django.db.models.signals import post_save
 from django.dispatch import receiver
@@ -24,6 +25,23 @@ def log_user_login_failed(sender, credentials, request, **kwargs):
         logger.info(f"Failed login attempt for username: {username}")
 
 
+# Replace Django's receiver, which always records the login time
+user_logged_in.disconnect(dispatch_uid='update_last_login')
+
+
+@receiver(user_logged_in, dispatch_uid='update_last_login')
+def update_last_login(sender, user, **kwargs):
+    """
+    Record the user's login time, unless maintenance mode is enabled. This allows users to log in while the database
+    is read-only, whichever authentication method they use.
+    """
+    if get_config().MAINTENANCE_MODE:
+        logger = logging.getLogger('netbox.auth.login')
+        logger.debug("Maintenance mode enabled: disabling update of most recent login time")
+        return
+    django_update_last_login(sender, user, **kwargs)
+
+
 @receiver(user_logged_in)
 def set_language_on_login(sender, user, request, **kwargs):
     """