Implemented mock scopes in tests and fixed scopes

jupyterhub/apihandlers/hub.py

@@ -66,7 +66,6 @@ class RootAPIHandler(APIHandler):
 
 
 class InfoAPIHandler(APIHandler):
-    @needs_scope('admin')  # Todo: Probably too strict
     def get(self):
         """GET /api/info returns detailed info about the Hub and its API.
 

jupyterhub/apihandlers/users.py

@@ -28,7 +28,7 @@ class SelfAPIHandler(APIHandler):
     Based on the authentication info. Acts as a 'whoami' for auth tokens.
     """
 
-    @needs_scope('read:users')
+    @needs_scope('all')
     async def get(self):
         user = self.current_user
         if user is None:
@@ -382,7 +382,7 @@ class UserTokenAPIHandler(APIHandler):
 class UserServerAPIHandler(APIHandler):
     """Start and stop single-user servers"""
 
-    @needs_scope('user:servers')
+    @needs_scope('users:servers')
     async def post(self, name, server_name='', subset=None):
         user = self.find_user(name)
         if server_name:
@@ -432,7 +432,7 @@ class UserServerAPIHandler(APIHandler):
         self.set_header('Content-Type', 'text/plain')
         self.set_status(status)
 
-    @needs_scope('user:servers')
+    @needs_scope('users:servers')
     async def delete(self, name, server_name=''):
         user = self.find_user(name)
         options = self.get_json_body()

jupyterhub/tests/mocking.py

@@ -53,7 +53,7 @@ from ..spawner import SimpleLocalProcessSpawner
 from ..utils import random_port
 from ..utils import url_path_join
 from .utils import async_requests
-from .utils import get_all_scopes
+from .utils import get_scopes
 from .utils import public_host
 from .utils import public_url
 from .utils import ssl_setup
@@ -300,7 +300,7 @@ class MockHub(JupyterHub):
         super().init_tornado_application()
         # reconnect tornado_settings so that mocks can update the real thing
         self.tornado_settings = self.users.settings = self.tornado_application.settings
-        self.tornado_settings['mock_scopes'] = get_all_scopes()
+        self.tornado_settings['mock_scopes'] = get_scopes()
 
     def init_services(self):
         # explicitly expire services before reinitializing

jupyterhub/tests/test_api.py

@@ -26,6 +26,7 @@ from .utils import api_request
 from .utils import async_requests
 from .utils import auth_header
 from .utils import find_user
+from .utils import get_scopes
 
 
 # --------------------
@@ -168,7 +169,7 @@ TIMESTAMP = normalize_timestamp(datetime.now().isoformat() + 'Z')
 @mark.user
 async def test_get_users(app):
     db = app.db
-    r = await api_request(app, 'users')
+    r = await api_request(app, 'users', headers=auth_header(db, 'admin'))
     assert r.status_code == 200
 
     users = sorted(r.json(), key=lambda d: d['name'])
@@ -178,7 +179,9 @@ async def test_get_users(app):
         fill_user({'name': 'user', 'admin': False, 'last_activity': None}),
     ]
 
-    r = await api_request(app, 'users', headers=auth_header(db, 'user'))
+    r = await api_request(
+        app, 'users', headers=auth_header(db, 'user'), scopes=get_scopes('user')
+    )
     assert r.status_code == 403
 
 
@@ -205,13 +208,24 @@ async def test_get_self(app):
     )
     db.add(oauth_token)
     db.commit()
-    r = await api_request(app, 'user', headers={'Authorization': 'token ' + token})
+    app.log.warn("Scopes:" + ", ".join(app.tornado_settings['mock_scopes']))
+    r = await api_request(
+        app,
+        'user',
+        headers={'Authorization': 'token ' + token},
+        scopes=get_scopes('user'),
+    )
     r.raise_for_status()
     model = r.json()
     assert model['name'] == u.name
 
     # invalid auth gets 403
-    r = await api_request(app, 'user', headers={'Authorization': 'token notvalid'})
+    r = await api_request(
+        app,
+        'user',
+        headers={'Authorization': 'token notvalid'},
+        scopes=get_scopes('user'),
+    )
     assert r.status_code == 403
 
 
@@ -420,6 +434,7 @@ async def test_user_set_auth_state(app, auth_state_enabled):
         method='patch',
         data=json.dumps({'auth_state': auth_state}),
         headers=auth_header(app.db, name),
+        scopes=get_scopes('user'),
     )
 
     assert r.status_code == 403

jupyterhub/tests/utils.py

@@ -118,7 +118,13 @@ def auth_header(db, name):
 
 @check_db_locks
 async def api_request(
-    app, *api_path, method='get', noauth=False, bypass_proxy=False, **kwargs
+    app,
+    *api_path,
+    method='get',
+    noauth=False,
+    bypass_proxy=False,
+    scopes=None,
+    **kwargs
 ):
     """Make an API request"""
     if bypass_proxy:
@@ -128,7 +134,12 @@ async def api_request(
     else:
         base_url = public_url(app, path='hub')
     headers = kwargs.setdefault('headers', {})
-
+    old_scopes = ['Nothing here']
+    if scopes is not None:
+        old_scopes = app.tornado_settings[
+            'mock_scopes'
+        ]  # Store old scopes so request has no side effects
+        app.tornado_settings['mock_scopes'] = scopes
     if 'Authorization' not in headers and not noauth and 'cookies' not in kwargs:
         # make a copy to avoid modifying arg in-place
         kwargs['headers'] = h = {}
@@ -147,6 +158,8 @@ async def api_request(
         kwargs['cert'] = (app.internal_ssl_cert, app.internal_ssl_key)
         kwargs["verify"] = app.internal_ssl_ca
     resp = await f(url, **kwargs)
+    if scopes is not None:
+        app.tornado_settings['mock_scopes'] = old_scopes
     assert "frame-ancestors 'self'" in resp.headers['Content-Security-Policy']
     assert (
         ujoin(app.hub.base_url, "security/csp-report")
@@ -196,23 +209,33 @@ def public_url(app, user_or_service=None, path=''):
         return host + prefix
 
 
-def get_all_scopes():
-    scopes = [
-        'all',
-        'all',
-        'users',
-        'users:name',
-        'users:groups',
-        'users:activity',
-        'users:servers',
-        'users:tokens',
-        'admin:users',
-        'admin:users:servers',
-        'groups',
-        'admin:groups',
-        'read:services',
-        'proxy',
-        'shutdown',
-    ]
-    read_only = ["read:%s" % el for el in scopes]
+def get_scopes(role='admin'):
+    """Get all scopes for a role. Default role is admin, alternatives are user and service"""
+    all_scopes = {
+        'admin': [
+            'all',
+            'users',
+            'users:name',
+            'users:groups',
+            'users:activity',
+            'users:servers',
+            'users:tokens',
+            'admin:users',
+            'admin:users:servers',
+            'groups',
+            'admin:groups',
+            'services',
+            'proxy',
+            'shutdown',
+        ],
+        'user': ['all'],
+        'server': ['users:activity'],
+    }
+    scopes = all_scopes[role]
+    read_only = ["read:" + el for el in scopes]
     return scopes + read_only
+
+
+def limit_scopes(scopes, key, name):
+    new_scopes = ["{}!{}={}".format(scope, key, name) for scope in scopes]
+    return new_scopes