Add request filter if server is configured as readonly

The `readonly` setting is only leverage in `kinto.core` Resources.
With this change, services that are defined ad-hoc do not have to explicitly
check for this setting.

kinto/core/__init__.py

@@ -54,6 +54,7 @@
     "initialization_sequence": (
         "kinto.core.initialization.setup_request_bound_data",
         "kinto.core.initialization.setup_json_serializer",
+        "kinto.core.initialization.restrict_http_methods_if_readonly",
         "kinto.core.initialization.setup_csp_headers",
         "kinto.core.initialization.setup_logging",
         "kinto.core.initialization.setup_storage",

kinto/core/initialization.py

@@ -8,7 +8,12 @@
 from dateutil import parser as dateparser
 from pyramid.events import ApplicationCreated, NewRequest, NewResponse
 from pyramid.exceptions import ConfigurationError
-from pyramid.httpexceptions import HTTPBadRequest, HTTPGone, HTTPTemporaryRedirect
+from pyramid.httpexceptions import (
+    HTTPBadRequest,
+    HTTPGone,
+    HTTPMethodNotAllowed,
+    HTTPTemporaryRedirect,
+)
 from pyramid.interfaces import IAuthenticationPolicy
 from pyramid.renderers import JSON as JSONRenderer
 from pyramid.response import Response
@@ -78,6 +83,24 @@ def on_new_response(event):
     config.add_subscriber(on_new_response, NewResponse)
 
 
+def restrict_http_methods_if_readonly(config):
+    """Prevent write operations if server is configured as read-only.
+
+    This is an additional layer of security on top of the verifications done
+    in :method:`kinto.core.Resource.register_service`, in case an installed
+    plugin does not take this setting into account in the definition of its
+    ad-hoc Pyramid views.
+    """
+    settings = config.get_settings()
+
+    def on_new_request(event):
+        if event.request.method.lower() not in ("get", "head", "options"):
+            raise HTTPMethodNotAllowed()
+
+    if asbool(settings["readonly"]):
+        config.add_subscriber(on_new_request, NewRequest)
+
+
 def setup_version_redirection(config):
     """Add a view which redirects to the current version of the API."""
     settings = config.get_settings()

tests/core/test_initialization.py

@@ -463,14 +463,15 @@ def test_kinto_core_includes_are_included_manually(self):
                 config.include.assert_any_call("elastic")
                 config.include.assert_any_call("history")
 
-    def make_app(self):
+    def make_app(self, settings=None):
         config = Configurator(settings={**kinto.core.DEFAULT_SETTINGS})
         config.add_settings(
             {
                 "permission_backend": "kinto.core.permission.memory",
                 "includes": "tests.core.testplugin",
                 "multiauth.policies": "basicauth",
-            }
+            },
+            **settings,
         )
         kinto.core.initialize(config, "0.0.1", "name")
         return webtest.TestApp(config.make_wsgi_app())
@@ -486,3 +487,25 @@ def test_plugin_benefits_from_cors_setup(self):
         headers = {"Origin": "lolnet.org", "Access-Control-Request-Method": "POST"}
         resp = app.options("/v0/attachment", headers=headers, status=200)
         self.assertIn("Access-Control-Allow-Origin", resp.headers)
+
+    def test_write_http_methods_are_rejected_if_readonly(self):
+        app = self.make_app({"readonly": True})
+        # `pytest.mark.parametrize` is not available on `TestCase` methods.
+        for method, status in [
+            ("GET", 200),
+            ("HEAD", 200),
+            ("OPTIONS", 200),
+            ("POST", 405),
+            ("PUT", 405),
+            ("PATCH", 405),
+            ("DELETE", 405),
+        ]:
+            app.request(
+                "/v0/log",
+                method=method,
+                status=status,
+                headers={
+                    "Access-Control-Request-Method": method,
+                    "Origin": "http://server.com",
+                },
+            )

tests/core/testplugin/__init__.py

@@ -15,5 +15,38 @@ def attachment_post(request):
     return {"ok": True}
 
 
+log = Service(
+    name="log",
+    description="Test endpoint without permissions",
+    path="/log",
+)
+
+
+@log.get()
+def log_get(request):
+    return {}
+
+
+@log.post()
+def log_post(request):
+    return {}
+
+
+@log.delete()
+def log_delete(request):
+    return {}
+
+
+@log.put()
+def log_put(request):
+    return {}
+
+
+@log.patch()
+def log_patch(request):
+    return {}
+
+
 def includeme(config):
     config.add_cornice_service(attachment)
+    config.add_cornice_service(log)