Filtering request using OpenAPI spec

jupyter_server/extension/application.py

@@ -1,6 +1,8 @@
 import logging
 import re
 import sys
+import typing
+from typing import Iterable, Optional
 
 from jinja2 import Environment
 from jinja2 import FileSystemLoader
@@ -96,7 +98,9 @@ def _prepare_templates(self):
         self.initialize_templates()
         # Add templates to web app settings if extension has templates.
         if len(self.template_paths) > 0:
-            self.settings.update({"{}_template_paths".format(self.name): self.template_paths})
+            self.settings.update(
+                {"{}_template_paths".format(self.name): self.template_paths}
+            )
 
         # Create a jinja environment for logging html templates.
         self.jinja2_env = Environment(
@@ -136,6 +140,12 @@ class method. This method can be set as a entry_point in
         the extensions setup.py
     """
 
+    # Filtering rules to apply on handlers registration
+    # Subclasses can override this list to filter handlers
+    # They will be applied on the ServerApp
+    __allowed_spec: Optional[dict] = None
+    __blocked_spec: Optional[dict] = None
+
     # Subclasses should override this trait. Tells the server if
     # this extension allows other other extensions to be loaded
     # side-by-side when launched directly.
@@ -180,6 +190,10 @@ def get_extension_package(cls):
     def get_extension_point(cls):
         return cls.__module__
 
+    @classmethod
+    def get_firewall_rules(cls):
+        return {"allowed": cls.__allowed_spec, "blocked": cls.__blocked_spec}
+
     # Extension URL sets the default landing page for this extension.
     extension_url = "/"
 
@@ -240,7 +254,9 @@ def _default_static_url_prefix(self):
         ),
     ).tag(config=True)
 
-    settings = Dict(help=_i18n("""Settings that will passed to the server.""")).tag(config=True)
+    settings = Dict(help=_i18n("""Settings that will passed to the server.""")).tag(
+        config=True
+    )
 
     handlers = List(help=_i18n("""Handlers appended to the server.""")).tag(config=True)
 
@@ -333,7 +349,9 @@ def _prepare_handlers(self):
     def _prepare_templates(self):
         # Add templates to web app settings if extension has templates.
         if len(self.template_paths) > 0:
-            self.settings.update({"{}_template_paths".format(self.name): self.template_paths})
+            self.settings.update(
+                {"{}_template_paths".format(self.name): self.template_paths}
+            )
         self.initialize_templates()
 
     def _jupyter_server_config(self):
@@ -452,7 +470,11 @@ def load_classic_server_extension(cls, serverapp):
                 (
                     r"/static/favicons/favicon.ico",
                     RedirectHandler,
-                    {"url": url_path_join(serverapp.base_url, "static/base/images/favicon.ico")},
+                    {
+                        "url": url_path_join(
+                            serverapp.base_url, "static/base/images/favicon.ico"
+                        )
+                    },
                 ),
                 (
                     r"/static/favicons/favicon-busy-1.ico",
@@ -495,7 +517,8 @@ def load_classic_server_extension(cls, serverapp):
                     RedirectHandler,
                     {
                         "url": url_path_join(
-                            serverapp.base_url, "static/base/images/favicon-notebook.ico"
+                            serverapp.base_url,
+                            "static/base/images/favicon-notebook.ico",
                         )
                     },
                 ),
@@ -504,14 +527,19 @@ def load_classic_server_extension(cls, serverapp):
                     RedirectHandler,
                     {
                         "url": url_path_join(
-                            serverapp.base_url, "static/base/images/favicon-terminal.ico"
+                            serverapp.base_url,
+                            "static/base/images/favicon-terminal.ico",
                         )
                     },
                 ),
                 (
                     r"/static/logo/logo.png",
                     RedirectHandler,
-                    {"url": url_path_join(serverapp.base_url, "static/base/images/logo.png")},
+                    {
+                        "url": url_path_join(
+                            serverapp.base_url, "static/base/images/logo.png"
+                        )
+                    },
                 ),
             ]
         )
@@ -532,7 +560,9 @@ def initialize_server(cls, argv=[], load_other_extensions=True, **kwargs):
             jpserver_extensions.update(cls.serverapp_config["jpserver_extensions"])
             cls.serverapp_config["jpserver_extensions"] = jpserver_extensions
             find_extensions = False
-        serverapp = ServerApp.instance(jpserver_extensions=jpserver_extensions, **kwargs)
+        serverapp = ServerApp.instance(
+            jpserver_extensions=jpserver_extensions, **kwargs
+        )
         serverapp.aliases.update(cls.aliases)
         serverapp.initialize(
             argv=argv,

jupyter_server/firewall.py

@@ -0,0 +1,75 @@
+from typing import Optional, Union
+
+from openapi_core import create_spec
+from tornado import httpclient, httputil
+from tornado.log import access_log
+from tornado_openapi3 import RequestValidator
+
+
+class FireWall:
+    """Validate server request against a list of allowed and blocked OpenAPI v3 specifications.
+
+    If allowed and blocked specifications are defined, the request must be allowed and not blocked;
+    i.e. blocked specification takes precedence.
+
+    Args:
+        allowed_spec: [optional] Allowed endpoints
+        blocked_spec: [optional] Blocked endpoints
+    """
+
+    def __init__(self, base_url: str, allowed_spec: Optional[dict], blocked_spec: Optional[dict]):
+        self.__allowed_validator: Optional[RequestValidator] = None
+        self.__blocked_validator: Optional[RequestValidator] = None
+
+        def add_base_url_server(spec: dict):
+            servers = spec.get("servers", [])
+            if not any(map(lambda s: s.get("url") == base_url, servers)):
+                servers.append({
+                    "url": base_url
+                })
+            spec["servers"] = servers
+
+        if allowed_spec is not None:
+            add_base_url_server(allowed_spec)
+            self.__allowed_validator = RequestValidator(create_spec(allowed_spec))
+        if blocked_spec is not None:
+            add_base_url_server(blocked_spec)
+            self.__blocked_validator = RequestValidator(create_spec(blocked_spec))
+
+    def validate(
+        self, request: Union[httpclient.HTTPRequest, httputil.HTTPServerRequest]
+    ) -> bool:
+        """Validate a request against allowed and blocked specifications.
+
+        Args:
+            request: Request to validate
+        Returns:
+            Whether the request is valid or not.
+        """
+        allowed_result = (
+            None
+            if self.__allowed_validator is None
+            else self.__allowed_validator.validate(request)
+        )
+
+        blocked_result = (
+            None
+            if self.__blocked_validator is None
+            else self.__blocked_validator.validate(request)
+        )
+
+        allowed = (allowed_result is None or len(allowed_result.errors) == 0)
+        not_blocked = (
+            blocked_result is None or len(blocked_result.errors) > 0
+        )
+
+        # The error raised if this is not valid will be logged
+        # So we only give the reason in debug level
+        if (not (allowed and not_blocked)):
+            if(not allowed):
+                # Provides only the first error
+                access_log.debug(f"Request not allowed: {allowed_result.errors[0]!s}")
+            elif (not not_blocked):
+                access_log.debug(f"Request blocked.")
+
+        return allowed and not_blocked

jupyter_server/serverapp.py

@@ -25,9 +25,15 @@
 import sys
 import threading
 import time
+import typing
 import urllib
 import webbrowser
 from base64 import encodebytes
+from typing import Iterable, Optional
+
+from tornado import httputil
+
+from jupyter_server.firewall import FireWall
 
 try:
     import resource
@@ -222,7 +228,16 @@ def __init__(
         default_url,
         settings_overrides,
         jinja_env_options,
+        endpoints_filters = None
     ):
+        if endpoints_filters is None:
+            self.__firewall = FireWall(base_url, None, None)
+        else:
+            self.__firewall = FireWall(
+                base_url,
+                endpoints_filters.get('allowed'),
+                endpoints_filters.get('blocked')
+            )
 
         settings = self.init_settings(
             jupyter_app,
@@ -433,6 +448,14 @@ def init_handlers(self, default_services, settings):
         new_handlers.append((r"(.*)", Template404))
         return new_handlers
 
+    def find_handler(
+        self, request: httputil.HTTPServerRequest, **kwargs: Any
+    ) -> "web._HandlerDelegate":
+        if self.__firewall.validate(request):
+            return super().find_handler(request, **kwargs)
+        else:
+            return self.get_handler_delegate(request, web.ErrorHandler, {"status_code": 403})
+
     def last_activity(self):
         """Get a UTC timestamp for when the server last did something.
 
@@ -756,6 +779,11 @@ class ServerApp(JupyterApp):
         "view",
     )
 
+    # Filtering rules to apply on handlers registration
+    # Subclasses can override this list to filter handlers
+    __allowed_spec: Optional[dict] = None
+    __blocked_spec: Optional[dict] = None
+
     _log_formatter_cls = LogFormatter
 
     @default("log_level")
@@ -1843,6 +1871,10 @@ def init_webapp(self):
             self.default_url,
             self.tornado_settings,
             self.jinja_environment_options,
+            endpoints_filters={
+                "allowed": self.__allowed_spec,
+                "blocked": self.__blocked_spec
+            }
         )
         if self.certfile:
             self.ssl_options["certfile"] = self.certfile
@@ -2316,6 +2348,10 @@ def initialize(
             # Set starter_app property.
             if point.app:
                 self._starter_app = point.app
+                # Apply endpoint filters from the extension app
+                firewall_rules = point.app.get_firewall_rules()
+                self.__allowed_spec = firewall_rules["allowed"]
+                self.__blocked_spec = firewall_rules["blocked"]
             # Load any configuration that comes from the Extension point.
             self.update_config(Config(point.config))