Filter environments by user access

src/_nebari/stages/kubernetes_services/template/modules/kubernetes/services/jupyterhub/files/jupyterhub/02-spawner.py

@@ -46,10 +46,133 @@ def generate_paged_urls(base_url: str, total_records: int, page_size: int) -> li
     return urls
 
 
+def get_scoped_token(
+    conda_store_url: str,
+    admin_token: str,
+    name: str,
+    groups: list[str] = None,
+    admin: bool = False,
+):
+    """Generate a conda store auth token for a given set of groups. The
+    token will only have `view` permissions for each group (regardless if
+    the user has higher permissions associated with the group generally).
+
+    If the user is an admin, then the token will have `view` permissions
+    for all namespaces.
+
+    By default, the user will have view permissions for the following
+    groups:
+    - `default/*`
+    - `filesystem/*`
+    - `nebari-git/*`
+    - `global/*`
+
+    Parameters
+    ----------
+    conda_store_url : str
+        The URL of the Conda Store instance.
+    admin_token : str
+        The admin token used to authenticate with the Conda Store API.
+    name : str
+        The name of the user for whom the token will be generated (the token
+        will have `view` permission to this namespace).
+    groups : list[str], optional
+        A list of group names that the user should have access to. If not provided,
+        default roles will be assigned.
+    admin : bool, optional
+        Whether the user is an admin. If true, the token will have `view` permissions
+        for all namespaces. Defaults to False.
+
+    Returns
+    -------
+    str
+        The generated token.
+    """
+    import urllib3
+
+    token_endpoint = f"http://{conda_store_url}/conda-store/api/v1/token/"
+    http = urllib3.PoolManager()
+
+    # add default role bindings
+    role_bindings = {
+        "role_bindings": {
+            f"{name}/*": ["viewer"],
+            "default/*": ["viewer"],
+            "filesystem/*": ["viewer"],
+            "nebari-git/*": ["viewer"],
+            "global/*": ["viewer"],
+        }
+    }
+
+    # add role bindings for all the groups the user is part of
+    if groups is not None:
+        for group in groups:
+            group = group.replace("/", "")
+            role_bindings["role_bindings"][f"{group}/*"] = ["viewer"]
+
+    # if the user is an admin, they can view all namespace + environments
+    if admin:
+        role_bindings["role_bindings"]["*/*"] = ["viewer"]
+
+    encoded_body = json.dumps(role_bindings)
+
+    # generate a token with with the generated role bindings
+    token_response = http.request(
+        "POST",
+        str(token_endpoint),
+        headers={
+            "Authorization": f"Bearer {admin_token}",
+            "Content-Type": "application/json",
+        },
+        body=encoded_body,
+    )
+    token_data = json.loads(token_response.data.decode("UTF-8"))
+    return token_data.get("data", {}).get("token")
+
+
 # TODO: this should get unit tests. Currently, since this is not a python module,
 # adding tests in a traditional sense is not possible. See https://github.com/soapy1/nebari/tree/try-unit-test-spawner
 # for a demo on one approach to adding test.
 def get_conda_store_environments(user_info: dict):
+    """Gets the conda-store environments for a given user using the v1 environment
+    API.
+
+    This scopes permissions for the given user using the `groups` field in the
+    user_info dict. The user_info dict comes from Jupyter's user dict. The groups
+    in this dict come from the keycloak groups, which include the list of conda-store
+    namespaces the user has access to. For the purpose of this function, we can assume
+    that if the user is part of the group, it at least has `view` permissions on the
+    conda-store namespaces.
+
+    Parameters
+    ----------
+    user_info : dict
+        A dictionary containing user information originating from the JupyterHub
+        user info. The scheme of the user info is:
+        ```
+        {
+            "name": "string",
+            "admin": true,
+            "roles": [
+                "string"
+            ],
+            "groups": [
+                "string"
+            ],
+            "server": "string",
+            "pending": "spawn",
+            "last_activity": "2019-08-24T14:15:22Z",
+            "servers": {
+            },
+            "auth_state": {}
+        }
+        ```
+
+    Returns
+    -------
+    list[str]
+        A list of all conda-store environments for the given user
+    """
     import os
 
     import urllib3
@@ -65,27 +188,35 @@ def get_conda_store_environments(user_info: dict):
     endpoint = "conda-store/api/v1/environment"
 
     base_url = f"http://{external_url}/{endpoint}/"
-    http = urllib3.PoolManager()
+
+    groups = user_info["groups"]
+    name = user_info["name"]
+    admin = user_info["admin"]
+    # get token with appropriate scope for the user making the request
+    scoped_token = get_scoped_token(external_url, token, name, groups, admin)
 
     # get total number of records from the endpoint
-    total_records = get_total_records(base_url, token)
+    total_records = get_total_records(base_url, scoped_token)
 
     # will contain all the environment info returned from the api
     env_data = []
 
     # generate a list of urls to hit to build the response
     urls = generate_paged_urls(base_url, total_records, page_size)
 
+    http = urllib3.PoolManager()
+
     # get content from urls
     for url in urls:
         response = http.request(
-            "GET", url, headers={"Authorization": f"Bearer {token}"}
+            "GET", url, headers={"Authorization": f"Bearer {scoped_token}"}
         )
         decoded_response = json.loads(response.data.decode("UTF-8"))
         env_data += decoded_response.get("data", [])
 
     # Filter and return conda environments for the user
-    return [f"{env['namespace']['name']}-{env['name']}" for env in env_data]
+    envs = [f"{env['namespace']['name']}-{env['name']}" for env in env_data]
+    return envs
 
 
 c.Spawner.pre_spawn_hook = get_username_hook