[All] Authorize allowed_users, admin_users, _or_ other allowed/admin groups

docs/source/how-to/example-oauthenticator.py

@@ -88,9 +88,9 @@ def build_auth_state_dict(self, token_info, user_info):
     # Updates `auth_model` dict if any fields have changed or additional information is available
     # or returns the unchanged `auth_model`.
     # Returns the model unchanged by default.
-    # Should be overridden to take into account  additional checks such as against group/admin/team membership.
+    # Should be overridden to take into account additional checks such as against group/admin/team membership.
     # if the OAuth provider has such a concept
-    async def update_auth_model(self, auth_model, **kwargs):
+    async def update_auth_model(self, username, auth_model):
         pass
 
 

docs/source/reference/changelog.md

@@ -6,6 +6,24 @@ command line for details.
 
 ## [Unreleased]
 
+### Breaking changes
+
+- [All] Users are now authorized based on _either_ being part of
+  `Authenticator.admin_users`, `Authenticator.allowed_users`, an Authenticator
+  specific allowed team/group/organization, or declared in
+  `JupyterHub.load_roles` or `JupyterHub.load_groups`.
+- [Google] If `GoogleOAuthenticator.admin_google_groups` is configured, users
+  logging in not explicitly there or in `Authenticator.admin_users` will get
+  their admin status revoked.
+- [Generic, Google] `GenericOAuthenticator.allowed_groups`,
+  `GenericOAuthenticator.allowed_groups`
+  `GoogleOAuthenticator.allowed_google_groups`, and
+  `GoogleOAuthenticator.admin_google_groups` are now Set based configuration
+  instead of List based configuration. It is still possible to set these with
+  lists as as they are converted to sets automatically, but anyone reading and
+  adding entries must now use set logic and not list logic.
+- [Google] Authentication state's `google_groups` is now a set, not a list.
+
 (changelog:version-15)=
 
 ## 15.0

oauthenticator/auth0.py

@@ -99,5 +99,4 @@ def _userdata_url_default(self):
 
 
 class LocalAuth0OAuthenticator(LocalAuthenticator, Auth0OAuthenticator):
-
     """A version that mixes in local system user creation"""

oauthenticator/bitbucket.py

@@ -40,38 +40,64 @@ def _userdata_url_default(self):
         config=True, help="Automatically allow members of selected teams"
     )
 
-    async def user_is_authorized(self, auth_model):
-        access_token = auth_model["auth_state"]["token_response"]["access_token"]
-        token_type = auth_model["auth_state"]["token_response"]["token_type"]
-        username = auth_model["name"]
-
-        # Check if user is a member of any allowed teams.
-        # This check is performed here, as the check requires `access_token`.
-        if self.allowed_teams:
-            user_in_team = await self._check_membership_allowed_teams(
-                username, access_token, token_type
-            )
-            if not user_in_team:
-                self.log.warning(f"{username} not in team allowed list of users")
-                return False
-
-        return True
-
-    async def _check_membership_allowed_teams(self, username, access_token, token_type):
+    async def _fetch_user_teams(self, access_token, token_type):
+        """
+        Get user's team memberships via bitbucket's API.
+        """
         headers = self.build_userdata_request_headers(access_token, token_type)
-        # We verify the team membership by calling teams endpoint.
         next_page = url_concat(
             "https://api.bitbucket.org/2.0/workspaces", {'role': 'member'}
         )
+
+        user_teams = set()
         while next_page:
             resp_json = await self.httpfetch(next_page, method="GET", headers=headers)
             next_page = resp_json.get('next', None)
-
-            user_teams = {entry["name"] for entry in resp_json["values"]}
-            # check if any of the organizations seen thus far are in the allowed list
-            if len(self.allowed_teams & user_teams) > 0:
+            user_teams |= {entry["name"] for entry in resp_json["values"]}
+        return user_teams
+
+    async def update_auth_model(self, auth_model):
+        """
+        Fetch and store `user_teams` in auth state if `allowed_teams` is
+        configured.
+        """
+        if self.allowed_teams:
+            access_token = auth_model["auth_state"]["token_response"]["access_token"]
+            token_type = auth_model["auth_state"]["token_response"]["token_type"]
+            user_teams = await self._fetch_user_teams(access_token, token_type)
+            auth_model["auth_state"]["user_teams"] = user_teams
+
+        return auth_model
+
+    async def check_allowed(self, username, auth_model):
+        """
+        Returns True for users allowed to be authorized.
+
+        Overrides the OAuthenticator.check_allowed implementation to allow users
+        either part of `allowed_users` or `allowed_teams`, and not just those
+        part of `allowed_users`.
+        """
+        # A workaround for JupyterHub<=4.0.1, described in
+        # https://github.com/jupyterhub/oauthenticator/issues/621
+        if auth_model is None:
+            return True
+
+        # allow admin users recognized via admin_users or update_auth_model
+        if auth_model["admin"]:
+            return True
+
+        # if allowed_users or allowed_teams is configured, we deny users not
+        # part of either
+        if self.allowed_users or self.allowed_teams:
+            user_teams = auth_model["auth_state"]["user_teams"]
+            if username in self.allowed_users:
+                return True
+            if any(user_teams & self.allowed_teams):
                 return True
-        return False
+            return False
+
+        # otherwise, authorize all users
+        return True
 
 
 class LocalBitbucketOAuthenticator(LocalAuthenticator, BitbucketOAuthenticator):

oauthenticator/cilogon.py

@@ -246,90 +246,138 @@ def _validate_allowed_idps(self, proposal):
         This is useful for linked identities where not all of them return
         the primary username_claim.
         """,
+        default_value=["email"],
     )
 
-    def user_info_to_username(self, user_info):
-        claimlist = [self.username_claim]
+    def _get_final_username_claim_list(self, user_info):
+        """
+        The username claims that will be used to determine the hub username can be set through:
+         - `CILogonOAutnenticator.username_claim`, that can be extended through `CILogonOAutnenticator.additional_username_claims`
+         or
+         - `CILogonOAuthenticator.allowed_idps.<idp>.username_claim`, that
+            will overwrite any value set through CILogonOAuthenticator.username_claim
+            for this identity provider.
+
+        This function returns the username claim list that will be used for the current user trying to login
+        based on the idp that they have selected. If no `CILogonOAutnenticator.allowed_idps` is set, then
+        `CILogonOAutnenticator.username_claim` will be used.
+        """
+        username_claims = [self.username_claim]
         if self.additional_username_claims:
-            claimlist.extend(self.additional_username_claims)
-
+            username_claims.extend(self.additional_username_claims)
         if self.allowed_idps:
-            selected_idp = user_info.get("idp")
-            if selected_idp:
+            selected_idp = user_info["idp"]
+            if selected_idp in self.allowed_idps.keys():
                 # The username_claim which should be used for this idp
-                claimlist = [
+                return [
                     self.allowed_idps[selected_idp]["username_derivation"][
                         "username_claim"
                     ]
                 ]
+            else:
+                return username_claims
+        return username_claims
 
-        for claim in claimlist:
+    def _get_username_from_claim_list(self, user_info, username_claims):
+        username = None
+        for claim in username_claims:
             username = user_info.get(claim)
             if username:
-                return username
+                break
+
+        return username
+
+    def user_info_to_username(self, user_info):
+        username_claims = self._get_final_username_claim_list(user_info)
+        username = self._get_username_from_claim_list(user_info, username_claims)
 
         if not username:
             user_info_keys = sorted(user_info.keys())
             self.log.error(
-                f"No username claim in the list at {claimlist} was found in the response {user_info_keys}"
+                f"No username claim in the list at {username_claims} was found in the response {user_info_keys}"
             )
             raise web.HTTPError(500, "Failed to get username from CILogon")
 
-    async def user_is_authorized(self, auth_model):
-        username = auth_model["name"]
-        # Check if selected idp was marked as allowed
+        # Optionally strip idp domain or prefix the username
         if self.allowed_idps:
-            selected_idp = auth_model["auth_state"][self.user_auth_state_key].get("idp")
-            # Fail hard if idp wasn't allowed
+            selected_idp = user_info["idp"]
+            if selected_idp in self.allowed_idps.keys():
+                username_derivation = self.allowed_idps[selected_idp][
+                    "username_derivation"
+                ]
+                action = username_derivation.get("action")
+
+                if action == "strip_idp_domain":
+                    username = username.split("@", 1)[0]
+                elif action == "prefix":
+                    prefix = username_derivation["prefix"]
+                    username = f"{prefix}:{username}"
+
+        return username
+
+    async def check_allowed(self, username, auth_model):
+        """
+        Returns True for authorized users, raises errors for users
+        denied authorization.
+
+        Overrides the `OAuthenticator.check_allowed` implementation to only allow users
+        logging in using a provider that is  part of `allowed_idps`.
+        Following this, the user must either be part of `allowed_users` or `allowed_domains`
+        to be authorized if either is configured, otherwise all users are
+        authorized.
+        """
+        # A workaround for JupyterHub<=4.0.1, described in
+        # https://github.com/jupyterhub/oauthenticator/issues/621
+        if auth_model is None:
+            return True
+
+        # allow admin users recognized via admin_users or update_auth_model
+        if auth_model["admin"]:
+            return True
+
+        if self.allowed_idps:
+            user_info = auth_model["auth_state"][self.user_auth_state_key]
+            selected_idp = user_info["idp"]
             if selected_idp not in self.allowed_idps.keys():
                 self.log.error(
                     f"Trying to login from an identity provider that was not allowed {selected_idp}",
                 )
                 raise web.HTTPError(
-                    500,
+                    403,
                     "Trying to login using an identity provider that was not allowed",
                 )
 
-            allowed_domains = self.allowed_idps[selected_idp].get(
-                "allowed_domains", None
-            )
+            allowed_domains = self.allowed_idps[selected_idp].get("allowed_domains")
+            if self.allowed_users or allowed_domains:
+                if username in self.allowed_users:
+                    return True
 
-            if allowed_domains:
-                gotten_name, gotten_domain = username.split('@')
-                if gotten_domain not in allowed_domains:
-                    raise web.HTTPError(
-                        500,
-                        "Trying to login using a domain that was not allowed",
+                if allowed_domains:
+                    username_claims = self._get_final_username_claim_list(user_info)
+                    username_with_domain = self._get_username_from_claim_list(
+                        user_info, username_claims
                     )
-
+                    user_domain = username_with_domain.split("@", 1)[1]
+                    if user_domain in allowed_domains:
+                        return True
+                    else:
+                        raise web.HTTPError(
+                            403,
+                            "Trying to login using a domain that was not allowed",
+                        )
+
+                return False
+        # Although not recommended, it might be that `allowed_idps` is not specified
+        # In this case we need to make sure we still check `allowed_users` and don't assume
+        # everyone should be authorized
+        elif self.allowed_users:
+            if username in self.allowed_users:
+                return True
+            return False
+
+        # otherwise, authorize all users
         return True
 
-    async def update_auth_model(self, auth_model):
-        selected_idp = auth_model["auth_state"][self.user_auth_state_key].get("idp")
-
-        # Check if the requested username_claim exists in the response from the provider
-        username = auth_model["name"]
-
-        # Check if we need to strip/prefix username
-        if self.allowed_idps:
-            username_derivation_config = self.allowed_idps[selected_idp][
-                "username_derivation"
-            ]
-            action = username_derivation_config.get("action", None)
-            allowed_domains = self.allowed_idps[selected_idp].get(
-                "allowed_domains", None
-            )
-
-            if action == "strip_idp_domain":
-                gotten_name, gotten_domain = username.split('@')
-                username = gotten_name
-            elif action == "prefix":
-                prefix = username_derivation_config["prefix"]
-                username = f"{prefix}:{username}"
-
-        auth_model["name"] = username
-        return auth_model
-
 
 class LocalCILogonOAuthenticator(LocalAuthenticator, CILogonOAuthenticator):