Enable support for read-only mode on APIs

deploy/rbac-clowdapp.yml

@@ -515,6 +515,8 @@ objects:
             value: ${IT_TOKEN_JKWS_CACHE_LIFETIME}
           - name: V2_APIS_ENABLED
             value: ${V2_APIS_ENABLED}
+          - name: READ_ONLY_API_MODE
+            value: ${READ_ONLY_API_MODE}
 
     jobs:
       - name: tenant-org-id-populator
@@ -920,4 +922,6 @@ parameters:
   value: "localhost:9000"
 - name: V2_APIS_ENABLED
   description: Flag to explicitly enable v2 API endpoints
+- name: READ_ONLY_API_MODE
+  description: Enforce GET only on RBAC APIs
   value: 'False'

rbac/rbac/middleware.py

@@ -420,3 +420,16 @@ def process_request(self, request):  # pylint: disable=no-self-use
 
         """
         setattr(request, "_dont_enforce_csrf_checks", True)
+
+
+class ReadOnlyApiMiddleware(MiddlewareMixin):  # pylint: disable=too-few-public-methods
+    """Middleware to enable read-only on APIs when configured."""
+
+    def process_request(self, request):  # pylint: disable=no-self-use
+        """Process request ReadOnlyApiMiddleware."""
+        if settings.READ_ONLY_API_MODE and request.method in ["POST", "PUT", "PATCH", "DELETE"]:
+            return HttpResponse(
+                json.dumps({"error": "This API is currently in read-only mode. Please try again later."}),
+                content_type="application/json",
+                status=405,
+            )

rbac/rbac/settings.py

@@ -129,6 +129,7 @@
     "django.middleware.clickjacking.XFrameOptionsMiddleware",
     "whitenoise.middleware.WhiteNoiseMiddleware",
     "django_prometheus.middleware.PrometheusAfterMiddleware",
+    "rbac.middleware.ReadOnlyApiMiddleware",
 ]
 
 DEVELOPMENT = ENVIRONMENT.bool("DEVELOPMENT", default=False)
@@ -486,3 +487,4 @@
 
 # Versioned API settings
 V2_APIS_ENABLED = ENVIRONMENT.bool("V2_APIS_ENABLED", default=False)
+READ_ONLY_API_MODE = ENVIRONMENT.get_value("READ_ONLY_API_MODE", default=False)

tests/rbac/test_middleware.py

@@ -22,6 +22,7 @@
 from unittest.mock import Mock
 from django.conf import settings
 from django.http import QueryDict
+from django.test.utils import override_settings
 
 from django.test import TestCase
 from django.urls import reverse
@@ -32,7 +33,7 @@
 from api.models import Tenant, User
 from api.serializers import create_tenant_name
 from tests.identity_request import IdentityRequest
-from rbac.middleware import HttpResponseUnauthorizedRequest, IdentityHeaderMiddleware
+from rbac.middleware import HttpResponseUnauthorizedRequest, IdentityHeaderMiddleware, ReadOnlyApiMiddleware
 from management.models import Access, Group, Permission, Principal, Policy, ResourceDefinition, Role
 
 
@@ -574,3 +575,53 @@ def test_principal_with_access_with_wildcard_access(self):
             "permission": {"read": ["*"], "write": ["*"]},
         }
         self.assertEqual(expected, access)
+
+
+class RBACReadOnlyApiMiddleware(IdentityRequest):
+    """Tests against the read-only API middleware."""
+
+    def setUp(self):
+        """Set up middleware tests."""
+        super().setUp()
+        self.request = Mock()
+        self.request.path = "/api/rbac/v1/roles/"
+        self.write_methods = ["POST", "PUT", "PATCH", "DELETE"]
+
+    def assertReadOnlyFailure(self, resp):
+        resp_body_str = resp.content.decode("utf-8")
+        self.assertEqual(
+            json.loads(resp_body_str)["error"], "This API is currently in read-only mode. Please try again later."
+        )
+        self.assertEqual(resp.status_code, 405)
+
+    @override_settings(READ_ONLY_API_MODE=True)
+    def test_get_read_only_true(self):
+        """Test GET and READ_ONLY_API_MODE=True."""
+        self.request.method = "GET"
+        middleware = ReadOnlyApiMiddleware(get_response=Mock())
+        resp = middleware.process_request(self.request)
+        self.assertEqual(resp, None)
+
+    @override_settings(READ_ONLY_API_MODE=True)
+    def test_write_methods_read_only_true(self):
+        """Test write methods and READ_ONLY_API_MODE=True."""
+        for method in self.write_methods:
+            self.request.method = method
+            middleware = ReadOnlyApiMiddleware(get_response=Mock())
+            resp = middleware.process_request(self.request)
+            self.assertReadOnlyFailure(resp)
+
+    def test_get_read_only_false(self):
+        """Test GET and READ_ONLY_API_MODE=False."""
+        self.request.method = "GET"
+        middleware = ReadOnlyApiMiddleware(get_response=Mock())
+        resp = middleware.process_request(self.request)
+        self.assertEqual(resp, None)
+
+    def test_write_methods_read_only_false(self):
+        """Test write methods and READ_ONLY_API_MODE=False."""
+        for method in self.write_methods:
+            self.request.method = method
+            middleware = ReadOnlyApiMiddleware(get_response=Mock())
+            resp = middleware.process_request(self.request)
+            self.assertEqual(resp, None)