Add allowed_scopes to all authenticators to allow some users based on granted scopes

oauthenticator/oauth2.py

@@ -456,6 +456,24 @@ def _logout_redirect_url_default(self):
         """,
     )
 
+    required_scopes = List(
+        Unicode(),
+        config=True,
+        help="""
+        List of scopes that must be granted to allow login.
+
+        All the scopes listed in this config must be present in the OAuth2 grant
+        from the authorizing server to allow the user to login. We request all
+        the scopes listed in the 'scope' config, but only a subset of these may
+        be granted by the authorization server. This may happen if the user does not
+        have permissions to access a requested scope, or has chosen to not give consent
+        for a particular scope. If the scopes listed in this config are not granted,
+        the user will not be allowed to log in.
+
+        See the OAuth documentation of your OAuth provider for various options.
+        """,
+    )
+
     extra_authorize_params = Dict(
         config=True,
         help="""
@@ -1025,6 +1043,18 @@ async def check_allowed(self, username, auth_model):
         if username in self.allowed_users:
             return True
 
+        # If we specific scope grants are required, validate that they have been granted
+        if self.required_scopes:
+            granted_scopes = auth_model.get('auth_state', {}).get('scope', [])
+            missing_scopes = set(self.required_scopes) - set(granted_scopes)
+            if missing_scopes:
+                self.log.info(
+                    f"Denying access to user {username} - scopes {missing_scopes} were not granted"
+                )
+                return False
+            else:
+                return True
+
         # users should be explicitly allowed via config, otherwise they aren't
         return False
 

oauthenticator/tests/mocks.py

@@ -104,6 +104,7 @@ def setup_oauth_mock(
     user_path=None,
     token_type='Bearer',
     token_request_style='post',
+    scope="",
 ):
     """setup the mock client for OAuth
 
@@ -125,6 +126,7 @@ def setup_oauth_mock(
         access_token_path (str): The path for the access token request (e.g. /access_token)
         user_path (str): The path for requesting  (e.g. /user)
         token_type (str): the token_type field for the provider
+        scope (str): The scope field returned by the provider
     """
 
     if user_path is None and token_request_style != "jwt":
@@ -164,6 +166,8 @@ def access_token(request):
             'access_token': token,
             'token_type': token_type,
         }
+        if scope:
+            model['scope'] = scope
         if token_request_style == 'jwt':
             model['id_token'] = user['id_token']
         return model
@@ -175,6 +179,7 @@ def get_user(request):
             token = auth_header.split(None, 1)[1]
         else:
             query = parse_qs(urlparse(request.url).query)
+
             if 'access_token' in query:
                 token = query['access_token'][0]
             else:

oauthenticator/tests/test_generic.py

@@ -25,6 +25,7 @@ def generic_client(client):
         host='generic.horse',
         access_token_path='/oauth/access_token',
         user_path='/oauth/userinfo',
+        scope='basic',
     )
     return client
 
@@ -200,6 +201,23 @@ async def test_generic_data(get_authenticator, generic_client):
     assert auth_model
 
 
+@mark.parametrize(
+    ["requested_scopes", "allowed"], [(["advanced"], False), (["basic"], True)]
+)
+async def test_required_scopes(
+    get_authenticator, generic_client, requested_scopes, allowed
+):
+    c = Config()
+    c.GenericOAuthenticator.required_scopes = requested_scopes
+    c.GenericOAuthenticator.scope = list(requested_scopes)
+    authenticator = get_authenticator(config=c)
+
+    handled_user_model = user_model("user1")
+    handler = generic_client.handler_for_user(handled_user_model)
+    auth_model = await authenticator.authenticate(handler)
+    assert allowed == await authenticator.check_allowed(auth_model["name"], auth_model)
+
+
 async def test_generic_callable_username_key(get_authenticator, generic_client):
     c = Config()
     c.GenericOAuthenticator.allow_all = True