JupyterHub + Jupyter Server read-only test

jupyter_server_config.py

import os
from typing import Dict, List
from unittest import mock

from jupyter_server.auth import Authorizer, IdentityProvider, User
from jupyterhub.services.auth import HubOAuth, HubOAuthCallbackHandler, check_scopes
from tornado.httputil import url_concat
from tornado.log import app_log
from traitlets import Instance, default


class JupyterHubUser(User):
    """Subclass User to store JupyterHub data"""

    # not a dataclass field
    hub_scopes: List
    hub_user: Dict

    def __init__(self, hub_user):
        self.hub_user = hub_user
        # translate jupyterhub custom scopes into server permissions
        self.hub_scopes = hub_user["scopes"]
        super().__init__(username=self.hub_user["name"])


class JupyterHubOAuthCallbackHandler(HubOAuthCallbackHandler):
    def initialize(self, hub_auth):
        self.hub_auth = hub_auth


class JupyterHubIdentityProvider(IdentityProvider):
    """Identity Provider for JupyterHub OAuth

    Replacement for JupyterHub's HubAuthenticated mixin
    """

    hub_auth = Instance(HubOAuth)

    @default("hub_auth")
    def _default_hub_auth(self):
        return HubOAuth(parent=self)

    def _patch_get_login_url(self, handler):
        original_get_login_url = handler.get_login_url

        def get_login_url():
            """Return the Hub's login URL"""
            login_url = self.hub_auth.login_url
            # add state argument to OAuth url
            state = self.hub_auth.set_state_cookie(
                handler, next_url=handler.request.uri
            )
            login_url = url_concat(login_url, {'state': state})
            # temporary override at setting level,
            # to allow any subclass overrides of get_login_url to preserve their effect
            # for example, APIHandler raises 403 to prevent redirects
            with mock.patch.dict(
                handler.application.settings, {"login_url": login_url}
            ):
                app_log.debug("Redirecting to login url: %s", login_url)
                return original_get_login_url()

        handler.get_login_url = get_login_url

    def get_user(self, handler):
        if hasattr(handler, "_jupyterhub_user"):
            return handler._jupyterhub_user
        self._patch_get_login_url(handler)
        user = self.hub_auth.get_user(handler)
        if user is None:
            handler._jupyterhub_user = None
            return None
        # check scopes
        self.log.debug(
            f"Checking user {user['name']} with scopes {user['scopes']} against {self.hub_auth.oauth_scopes}"
        )
        scopes = self.hub_auth.check_scopes(self.hub_auth.oauth_scopes, user)
        if scopes:
            self.log.debug(f"Allowing user {user['name']} with scopes {scopes}")
        else:
            self.log.warning(f"Not allowing user {user['name']}")
            return None
        handler._jupyterhub_user = JupyterHubUser(user)
        return handler._jupyterhub_user

    def get_handlers(self):
        return [
            (
                "/oauth_callback",
                JupyterHubOAuthCallbackHandler,
                {"hub_auth": self.hub_auth},
            )
        ]


class JupyterHubAuthorizer(Authorizer):
    """Authorizer that looks for permissions in JupyterHub scopes"""

    def is_authorized(self, handler, user, action, resource):
        # print(user)
        scopes = user.hub_scopes
        # authorize if any of these permissions are present
        # filters check for access to this specific server
        filters = [
            f"!user={os.environ['JUPYTERHUB_USER']}",
            f"!server={os.environ['JUPYTERHUB_USER']}/{os.environ['JUPYTERHUB_SERVER_NAME']}",
        ]
        required_scopes = set()
        for f in filters:
            required_scopes.update(
                {
                    f"custom:jupyter_server:{action}:{resource}{f}",
                    f"custom:jupyter_server:{action}:*{f}",
                }
            )
        # self.log.info(f"Required scopes are: {required_scopes}")
        have_scopes = check_scopes(required_scopes, scopes)

        # self.log.debug(f"{user['name']} has permissions: { user['scopes']}")
        self.log.debug(
            f"{user.username} has permissions {have_scopes} required to {action} on {resource}"
        )
        return bool(have_scopes)


c = get_config()  # noqa

c.ServerApp.identity_provider_class = JupyterHubIdentityProvider
c.ServerApp.authorizer_class = JupyterHubAuthorizer

# c.LabApp.collaborative = True

# reimplement service prefix
from urllib.parse import urlparse

url = urlparse(os.environ['JUPYTERHUB_SERVICE_URL'])
c.ServerApp.port = url.port
c.ServerApp.ip = url.hostname
c.ServerApp.base_url = url.path
c.ServerApp.open_browser = False

jupyterhub_config.py

c = get_config()  # noqa

c.Authenticator.allowed_users = {"tst01", "tst02"}
c.JupyterHub.authenticator_class = "dummy"
c.JupyterHub.spawner_class = "simple"
c.JupyterHub.log_level = 10
c.JupyterHub.cleanup_servers = True


c.JupyterHub.custom_scopes = {
    "custom:jupyter_server:read:*": {
        "description": "read-only access to your server",
    },
    "custom:jupyter_server:write:*": {
        "description": "access to modify files on your server. Does not include execution.",
        "subscopes": ["custom:jupyter_server:read:*"],
    },
    "custom:jupyter_server:execute:*": {
        "description": "Execute permissions on servers.",
        "subscopes": [
            "custom:jupyter_server:write:*",
            "custom:jupyter_server:read:*",
        ],
    },
}

c.JupyterHub.load_roles = [
    {
        "name": "read-only",
        "scopes": [
            "access:servers!user=tst01",
            "custom:jupyter_server:read:*!user=tst01",
        ],
        "users": ["tst01", "tst02"],
    },
    {
        "name": "full-access",
        "scopes": [
            "custom:jupyter_server:execute:*!user=tst01",
        ],
        "users": ["tst01"],
    },
    # {
    #     "name": "shared",
    #     "scopes": [
    #         "access:services!service=shared",
    #         "custom:jupyter_server:execute:*",
    #     ],
    # },
]

c.Spawner.oauth_roles = ["read-only", "full-access"]

import os

here = os.path.dirname(__file__)
c.Spawner.cmd = [
    "jupyter-lab",
    "--debug",
    f"--config={here}/jupyter_server_config.py",
]
# c.JupyterHub.services = [
#     {
#         "name": "shared",
#         "command": ['jupyter-labhub', '--debug', '--LabApp.collaborative=True'],
#         "url": "http://127.0.0.1:9999",
#         "oauth_roles": ["shared"],
#     },
# ]

requirements.txt

jupyterhub>=3
jupyter-server>=2
jupyterlab