feat(oauth2): set the role of OAuth2 users from an identity provider claim

docs/en_US/oauth2.rst

@@ -95,6 +95,8 @@ and secure.
     "OAUTH2_ADDITIONAL_CLAIMS", "If a dictionary is provided, pgAdmin will check for a matching key and value on the **ID token first** (for OIDC providers), then fall back to the userinfo endpoint response. In case there is no match with the provided config, the user will receive an authorization error. Useful for checking AzureAD_ *wids* or *groups*, GitLab_ *owner*, *maintainer* and *reporter* claims."
     "OAUTH2_SERVER_GROUP_CLAIM", "The name of a claim listing the server groups whose shared servers the user may access. If set, shared servers owned by other users are only visible and accessible if the name of their server group is in the claim. See :ref:`oauth2_server_group_claims` below."
     "OAUTH2_SERVER_GROUP_CLAIM_MAPPING", "An optional dictionary mapping values of the OAUTH2_SERVER_GROUP_CLAIM claim to a server group name or a list of server group names. Values without a mapping are used as server group names unchanged."
+    "OAUTH2_ROLE_CLAIM", "The name of a claim used to set the pgAdmin role of the user each time they log in. Requires OAUTH2_ROLE_CLAIM_MAPPING. See :ref:`oauth2_role_claims` below."
+    "OAUTH2_ROLE_CLAIM_MAPPING", "A dictionary mapping values of the OAUTH2_ROLE_CLAIM claim to pgAdmin role names. The first entry whose value is in the claim is used; if none matches, the user is given the *User* role."
     "OAUTH2_SSL_CERT_VERIFICATION", "Set this variable to False to disable SSL certificate verification for OAuth2 provider.
     This may need to set False, in case of self-signed certificates."
     "OAUTH2_CHALLENGE_METHOD", "Enable PKCE workflow. PKCE method name, only *S256* is supported"
@@ -435,6 +437,54 @@ Example:
         # servers in the 'Staging' and 'Reporting' server groups.
     }]
 
+.. _oauth2_role_claims:
+
+Assigning Roles with a Claim
+----------------------------
+
+**OAUTH2_ROLE_CLAIM** and **OAUTH2_ROLE_CLAIM_MAPPING** set the pgAdmin role
+of an OAuth2 user from a claim of the identity provider, such as a list of
+groups. When they are set:
+
+* The role is set each time the user logs in, including when the account is
+  first created by OAUTH2_AUTO_CREATE_USER.
+* The first entry of OAUTH2_ROLE_CLAIM_MAPPING whose value is in the claim
+  gives the role, so more privileged roles should be listed first.
+* If no entry matches, or the claim is missing, the user is given the *User*
+  role.
+* A role changed in the pgAdmin user management dialog is overwritten at the
+  user's next login.
+* If a mapped role does not exist, a warning is logged and the role of the
+  user is not changed.
+* Users who log in by other methods are not affected.
+
+The role claim does not decide who may log in. It is applied only after the
+login has been authorised, so use **OAUTH2_ADDITIONAL_CLAIMS** to refuse
+users who are in none of the expected groups.
+
+The claim is read from the ID token first, then from the userinfo endpoint
+response. Its value may be a string or a list of strings; values of any other
+type are ignored. Values are matched exactly.
+
+Since the identity provider decides who is an administrator, make sure that
+at least one administrator can still log in in another way, for example as an
+internal user, in case the claim is not returned as expected.
+
+Example:
+
+.. code-block:: python
+
+    OAUTH2_CONFIG = [{
+        # ... other config ...
+        'OAUTH2_ROLE_CLAIM': 'groups',
+        'OAUTH2_ROLE_CLAIM_MAPPING': {
+            'pgadmin-admins': 'Administrator',
+            'pgadmin-users': 'User'
+        },
+        # A user in the 'pgadmin-admins' group is an Administrator; every
+        # other OAuth2 user is given the User role.
+    }]
+
 Legacy OAuth2 Configuration (Without OIDC)
 -------------------------------------------
 

web/config.py

@@ -960,6 +960,20 @@
         # },
         'OAUTH2_SERVER_GROUP_CLAIM': None,
         'OAUTH2_SERVER_GROUP_CLAIM_MAPPING': None,
+        # The claim used to set the pgAdmin role of the user at each login
+        # (ID token first, then userinfo). Requires
+        # OAUTH2_ROLE_CLAIM_MAPPING, which maps claim values to role names.
+        # The first entry whose value is in the claim wins, so list more
+        # privileged roles first; if none matches, the user gets the 'User'
+        # role. Roles assigned in the pgAdmin UI are overwritten at the next
+        # login.
+        # Example:
+        # 'OAUTH2_ROLE_CLAIM': 'groups',
+        # 'OAUTH2_ROLE_CLAIM_MAPPING': {
+        #     'pgadmin-admins': 'Administrator'
+        # },
+        'OAUTH2_ROLE_CLAIM': None,
+        'OAUTH2_ROLE_CLAIM_MAPPING': None,
         # Set this variable to False to disable SSL certificate verification
         # for OAuth2 provider.
         # This may need to set False, in case of self-signed certificates.

web/pgadmin/authenticate/oauth2.py

@@ -20,7 +20,7 @@
 from flask_security.utils import logout_user
 
 from pgadmin.authenticate.internal import BaseAuthentication
-from pgadmin.model import User
+from pgadmin.model import Role, User
 from pgadmin.tools.user_management import create_user
 from pgadmin.utils.constants import OAUTH2, MessageType
 from pgadmin.utils import PgAdminModule, get_safe_post_login_redirect, \
@@ -656,6 +656,75 @@ def _extract_server_group_claims(self, profile_dict, id_token_claims):
         # Remove duplicates, preserving order
         return list(dict.fromkeys(server_groups))
 
+    def _resolve_role_from_claims(self, profile_dict, id_token_claims):
+        """
+        Resolve the pgAdmin role of the user from provider claim
+        configuration.
+
+        Config keys (per provider):
+        - OAUTH2_ROLE_CLAIM: claim name to read
+        - OAUTH2_ROLE_CLAIM_MAPPING: dict mapping claim-value -> pgAdmin
+          role name. The first entry whose value is in the claim wins, so
+          more privileged roles should be listed first.
+
+        Claim values are matched exactly. Values that aren't strings are
+        ignored.
+
+        Returns:
+            str | None
+            - the role name to assign; 'User' if no mapping entry matches
+            - None when claim-based roles are not configured
+        """
+        provider = self.oauth2_config.get(self.oauth2_current_client) or {}
+        claim_name = provider.get('OAUTH2_ROLE_CLAIM')
+        if not claim_name:
+            return None
+
+        mapping = provider.get('OAUTH2_ROLE_CLAIM_MAPPING')
+        if not isinstance(mapping, dict) or not mapping:
+            current_app.logger.warning(
+                'OAUTH2_ROLE_CLAIM is set but OAUTH2_ROLE_CLAIM_MAPPING is '
+                'not a non-empty dictionary; user roles are not changed.')
+            return None
+
+        claim_values = id_token_claims.get(claim_name)
+        if claim_values is None:
+            claim_values = profile_dict.get(claim_name)
+        if claim_values is None:
+            claim_values = []
+        if not isinstance(claim_values, list):
+            claim_values = [claim_values]
+        claim_values = {v for v in claim_values if isinstance(v, str)}
+
+        for value, role_name in mapping.items():
+            if value in claim_values:
+                if not isinstance(role_name, str):
+                    current_app.logger.warning(
+                        f'OAUTH2_ROLE_CLAIM_MAPPING has a role that is not '
+                        f'a string for "{value}"; user roles are not '
+                        f'changed.')
+                    return None
+                return role_name
+        return 'User'
+
+    def _sync_user_role(self, user, role_name):
+        """Give the user the given role, replacing any other role."""
+        role = Role.query.filter_by(name=role_name).first()
+        if role is None:
+            current_app.logger.warning(
+                f'Role "{role_name}" from OAUTH2_ROLE_CLAIM_MAPPING does not '
+                f'exist; the role of user {user.username} is not changed.')
+            return
+
+        if [r.name for r in user.roles] == [role.name]:
+            return
+
+        user.roles = [role]
+        db.session.commit()
+        current_app.logger.info(
+            f'OAUTH2 user {user.username} was given the role "{role.name}" '
+            f'based on OAUTH2_ROLE_CLAIM.')
+
     def login(self, form):
         if not self.oauth2_current_client:
             error_msg = gettext('No OAuth2 provider available.')
@@ -798,10 +867,16 @@ def login(self, form):
             session['oauth2_server_group_claims'] = \
                 oauth2_server_group_claims
 
+        role_name = self._resolve_role_from_claims(
+            profile_dict, id_token_claims
+        )
+
         user, msg = self.__auto_create_user(username, email)
         if user:
             user = db.session.query(User).filter_by(
                 username=username, auth_source=OAUTH2).first()
+            if role_name is not None:
+                self._sync_user_role(user, role_name)
             current_app.login_manager.logout_view = \
                 OAuth2Authentication.LOGOUT_VIEW
             current_app.logger.info(
@@ -853,10 +928,11 @@ def get_user_profile(self):
                 username_claim = provider.get('OAUTH2_USERNAME_CLAIM')
                 additional_claims = provider.get('OAUTH2_ADDITIONAL_CLAIMS')
                 server_group_claim = provider.get('OAUTH2_SERVER_GROUP_CLAIM')
+                role_claim = provider.get('OAUTH2_ROLE_CLAIM')
 
                 # If custom username claim, additional authorization
-                # claims, or server-group claims are configured, they may
-                # exist only in userinfo; don't skip userinfo unless ID
+                # claims, server-group or role claims are configured, they
+                # may exist only in userinfo; don't skip userinfo unless ID
                 # token has them.
                 needs_userinfo = False
                 if username_claim and username_claim not in id_token_claims:
@@ -871,6 +947,8 @@ def get_user_profile(self):
                 if (server_group_claim and
                         server_group_claim not in id_token_claims):
                     needs_userinfo = True
+                if role_claim and role_claim not in id_token_claims:
+                    needs_userinfo = True
 
             if has_sufficient_claims and not needs_userinfo:
                 current_app.logger.debug(