From ec37dddc5cd34d54a3f31195900089875f28191b Mon Sep 17 00:00:00 2001 From: artikq Date: Tue, 10 Nov 2026 07:33:07 +0300 Subject: [PATCH] Allow the role of OAuth2 users to be set from an identity provider claim. #8115 --- docs/en_US/oauth2.rst | 50 ++++ web/config.py | 14 ++ web/pgadmin/authenticate/oauth2.py | 84 ++++++- .../browser/tests/test_oauth2_with_mocking.py | 225 +++++++++++++++++- 4 files changed, 368 insertions(+), 5 deletions(-) diff --git a/docs/en_US/oauth2.rst b/docs/en_US/oauth2.rst index 149972a1c93..6c40afd1ed1 100644 --- a/docs/en_US/oauth2.rst +++ b/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) ------------------------------------------- diff --git a/web/config.py b/web/config.py index 95516b3ab4b..5b485f0348c 100644 --- a/web/config.py +++ b/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. diff --git a/web/pgadmin/authenticate/oauth2.py b/web/pgadmin/authenticate/oauth2.py index 5828c36c52e..50f1404e3be 100644 --- a/web/pgadmin/authenticate/oauth2.py +++ b/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( diff --git a/web/pgadmin/browser/tests/test_oauth2_with_mocking.py b/web/pgadmin/browser/tests/test_oauth2_with_mocking.py index 7eb23ea2517..e3e72558738 100644 --- a/web/pgadmin/browser/tests/test_oauth2_with_mocking.py +++ b/web/pgadmin/browser/tests/test_oauth2_with_mocking.py @@ -11,6 +11,7 @@ from pgAdmin4 import app from pgadmin.utils.route import BaseTestGenerator from regression.python_test_utils import test_utils as utils +from regression.python_test_utils.user_utils import add_user from pgadmin.authenticate.registry import AuthSourceRegistry from unittest.mock import patch, MagicMock, mock_open from pgadmin.utils.constants import OAUTH2, INTERNAL @@ -174,6 +175,76 @@ class Oauth2LoginMockTestCase(BaseTestGenerator): profile={}, id_token_claims=None, )), + ('OIDC Role Claim Gives Mapped Role', dict( + oauth2_provider='oidc-role-claim', + kind='login_success', + profile={'email': 'role-admin@example.com'}, + id_token_claims={'groups': ['other', 'pgadmin-admins']}, + expected_role='Administrator', + )), + ('OIDC Role Claim Single String', dict( + oauth2_provider='oidc-role-claim', + kind='login_success', + profile={'email': 'role-string@example.com'}, + id_token_claims={'groups': 'pgadmin-admins'}, + expected_role='Administrator', + )), + ('OIDC Role Claim Without Match Gives User Role', dict( + oauth2_provider='oidc-role-claim', + kind='login_success', + profile={'email': 'role-user@example.com'}, + id_token_claims={'groups': ['other']}, + expected_role='User', + )), + ('OIDC Role Claim First Mapping Entry Wins', dict( + oauth2_provider='oidc-role-claim', + kind='login_success', + profile={'email': 'role-order@example.com'}, + id_token_claims={'groups': ['pgadmin-users', 'pgadmin-admins']}, + expected_role='Administrator', + )), + ('OIDC Role Claim From Userinfo', dict( + oauth2_provider='oidc-role-claim', + kind='login_success', + profile={'email': 'role-userinfo@example.com', + 'groups': ['pgadmin-admins']}, + id_token_claims={'sub': 'abc'}, + expected_role='Administrator', + )), + ('OIDC Role Claim Unknown Role Keeps Role', dict( + oauth2_provider='oidc-role-claim-unknown-role', + kind='login_success', + profile={'email': 'role-unknown@example.com'}, + id_token_claims={'groups': ['pgadmin-admins']}, + preexisting_admin=True, + expected_role='Administrator', + )), + ('OIDC Role Claim Non-String Role Keeps Role', dict( + oauth2_provider='oidc-role-claim-invalid-role', + kind='login_success', + profile={'email': 'role-invalid@example.com'}, + id_token_claims={'groups': ['pgadmin-admins']}, + preexisting_admin=True, + expected_role='Administrator', + )), + ('OIDC Role Claim Does Not Let Unauthorised Users In', dict( + oauth2_provider='oidc-role-claim-additional-claims', + kind='role_claim_rejected_login', + profile={'email': 'role-rejected@example.com'}, + id_token_claims={'groups': ['other']}, + )), + ('OIDC Role Claim Removed At Next Login', dict( + oauth2_provider='oidc-role-claim', + kind='role_claim_removed', + profile={'email': 'role-removed@example.com'}, + id_token_claims=None, + )), + ('OIDC get_user_profile Fetches Userinfo For Role Claim', dict( + oauth2_provider='oidc-role-claim', + kind='oidc_get_user_profile_role_claim', + profile={}, + id_token_claims=None, + )), ('OIDC get_user_profile Skips Userinfo', dict( oauth2_provider='oidc-basic', kind='oidc_get_user_profile_skip', @@ -399,6 +470,78 @@ def setUp(self): 'groups': ['group-a'] }, 'OAUTH2_SERVER_GROUP_CLAIM': 'pgadmin_server_groups', + }, + { + 'OAUTH2_NAME': 'oidc-role-claim', + 'OAUTH2_DISPLAY_NAME': 'OIDC Role Claim', + 'OAUTH2_CLIENT_ID': 'testclientid', + 'OAUTH2_CLIENT_SECRET': 'testclientsec', + 'OAUTH2_TOKEN_URL': 'https://oidc.example/token', + 'OAUTH2_AUTHORIZATION_URL': 'https://oidc.example/auth', + 'OAUTH2_API_BASE_URL': 'https://oidc.example/', + 'OAUTH2_USERINFO_ENDPOINT': 'userinfo', + 'OAUTH2_SCOPE': 'openid email profile', + 'OAUTH2_SERVER_METADATA_URL': + 'https://oidc.example/.well-known/openid-configuration', + 'OAUTH2_ROLE_CLAIM': 'groups', + 'OAUTH2_ROLE_CLAIM_MAPPING': { + 'pgadmin-admins': 'Administrator', + 'pgadmin-users': 'User' + }, + }, + { + 'OAUTH2_NAME': 'oidc-role-claim-unknown-role', + 'OAUTH2_DISPLAY_NAME': 'OIDC Role Claim Unknown Role', + 'OAUTH2_CLIENT_ID': 'testclientid', + 'OAUTH2_CLIENT_SECRET': 'testclientsec', + 'OAUTH2_TOKEN_URL': 'https://oidc.example/token', + 'OAUTH2_AUTHORIZATION_URL': 'https://oidc.example/auth', + 'OAUTH2_API_BASE_URL': 'https://oidc.example/', + 'OAUTH2_USERINFO_ENDPOINT': 'userinfo', + 'OAUTH2_SCOPE': 'openid email profile', + 'OAUTH2_SERVER_METADATA_URL': + 'https://oidc.example/.well-known/openid-configuration', + 'OAUTH2_ROLE_CLAIM': 'groups', + 'OAUTH2_ROLE_CLAIM_MAPPING': { + 'pgadmin-admins': 'No Such Role' + }, + }, + { + 'OAUTH2_NAME': 'oidc-role-claim-invalid-role', + 'OAUTH2_DISPLAY_NAME': 'OIDC Role Claim Invalid Role', + 'OAUTH2_CLIENT_ID': 'testclientid', + 'OAUTH2_CLIENT_SECRET': 'testclientsec', + 'OAUTH2_TOKEN_URL': 'https://oidc.example/token', + 'OAUTH2_AUTHORIZATION_URL': 'https://oidc.example/auth', + 'OAUTH2_API_BASE_URL': 'https://oidc.example/', + 'OAUTH2_USERINFO_ENDPOINT': 'userinfo', + 'OAUTH2_SCOPE': 'openid email profile', + 'OAUTH2_SERVER_METADATA_URL': + 'https://oidc.example/.well-known/openid-configuration', + 'OAUTH2_ROLE_CLAIM': 'groups', + 'OAUTH2_ROLE_CLAIM_MAPPING': { + 'pgadmin-admins': ['Administrator'] + }, + }, + { + 'OAUTH2_NAME': 'oidc-role-claim-additional-claims', + 'OAUTH2_DISPLAY_NAME': 'OIDC Role Claim With Claims', + 'OAUTH2_CLIENT_ID': 'testclientid', + 'OAUTH2_CLIENT_SECRET': 'testclientsec', + 'OAUTH2_TOKEN_URL': 'https://oidc.example/token', + 'OAUTH2_AUTHORIZATION_URL': 'https://oidc.example/auth', + 'OAUTH2_API_BASE_URL': 'https://oidc.example/', + 'OAUTH2_USERINFO_ENDPOINT': 'userinfo', + 'OAUTH2_SCOPE': 'openid email profile', + 'OAUTH2_SERVER_METADATA_URL': + 'https://oidc.example/.well-known/openid-configuration', + 'OAUTH2_ADDITIONAL_CLAIMS': { + 'groups': ['pgadmin-admins', 'pgadmin-users'] + }, + 'OAUTH2_ROLE_CLAIM': 'groups', + 'OAUTH2_ROLE_CLAIM_MAPPING': { + 'pgadmin-admins': 'Administrator' + }, } ] @@ -428,9 +571,25 @@ def runTest(self): elif self.kind == 'workload_identity_missing_token_file': self._test_workload_identity_missing_token_file_fails_fast() elif self.kind == 'login_success': + if getattr(self, 'preexisting_admin', False): + add_user(self, self.profile['email'], OAUTH2, + role_name='Administrator') self._test_oauth2_login_success( self.oauth2_provider, self.profile, self.id_token_claims, - getattr(self, 'expected_server_groups', None) + getattr(self, 'expected_server_groups', None), + getattr(self, 'expected_role', None) + ) + elif self.kind == 'role_claim_rejected_login': + self._test_oauth2_role_claim_rejected_login( + self.oauth2_provider, self.profile, self.id_token_claims + ) + elif self.kind == 'role_claim_removed': + self._test_oauth2_role_claim_removed( + self.oauth2_provider, self.profile + ) + elif self.kind == 'oidc_get_user_profile_role_claim': + self._test_oidc_get_user_profile_role_claim_userinfo( + self.oauth2_provider ) elif self.kind == 'login_failure': self._test_oauth2_login_failure( @@ -515,9 +674,39 @@ def _fake_authenticate(self, _form): str(e) ) + def _assert_oauth2_user_role(self, username, expected_role): + """Check that the OAuth2 user has exactly the expected role.""" + from pgadmin.model import User + with self.app.app_context(): + user = User.query.filter_by( + username=username, auth_source=OAUTH2).first() + self.assertIsNotNone(user) + self.assertEqual([r.name for r in user.roles], [expected_role]) + + def _test_oauth2_role_claim_rejected_login( + self, provider, profile, id_token_claims): + """The role claim does not decide who may log in: a user rejected + by OAUTH2_ADDITIONAL_CLAIMS is not created with the default role.""" + from pgadmin.model import User + self._test_oauth2_login_failure(provider, profile, id_token_claims) + with self.app.app_context(): + self.assertIsNone(User.query.filter_by( + username=profile['email'], auth_source=OAUTH2).first()) + + def _test_oauth2_role_claim_removed(self, provider, profile): + """A user who loses the claim value loses the role at next login.""" + self._test_oauth2_login_success( + provider, profile, {'groups': ['pgadmin-admins']}, + expected_role='Administrator' + ) + self.tester.logout() + self._test_oauth2_login_success( + provider, profile, {'groups': []}, expected_role='User' + ) + def _test_oauth2_login_success( self, provider, profile, id_token_claims=None, - expected_server_groups=None + expected_server_groups=None, expected_role=None ): from pgadmin.authenticate.oauth2 import OAuth2Authentication @@ -560,6 +749,8 @@ def _fake_get_user_profile(self): sess.get('oauth2_server_group_claims'), expected_server_groups ) + if expected_role is not None: + self._assert_oauth2_user_role(profile['email'], expected_role) def _test_oauth2_login_failure( self, provider, profile, id_token_claims=None @@ -760,6 +951,36 @@ def _test_oidc_get_user_profile_server_group_userinfo(self, provider): ['Reporting']) client.get.assert_called_once() + def _test_oidc_get_user_profile_role_claim_userinfo(self, provider): + """The ID token has the standard claims but not the role claim, so + userinfo must still be fetched.""" + from pgadmin.authenticate.oauth2 import OAuth2Authentication + + with self.app.test_request_context('/'): + oauth = OAuth2Authentication() + oauth.oauth2_current_client = provider + + client = MagicMock() + client.authorize_access_token = MagicMock(return_value={ + 'access_token': 't', + 'id_token': 'mock.jwt.token', + 'token_type': 'Bearer', + 'userinfo': {'email': 'oidc@example.com', 'sub': 'abc'} + }) + + resp = MagicMock() + resp.raise_for_status = MagicMock() + resp.json = MagicMock(return_value={ + 'email': 'oidc@example.com', + 'groups': ['pgadmin-admins'] + }) + client.get = MagicMock(return_value=resp) + + OAuth2Authentication.oauth2_clients[provider] = client + profile = oauth.get_user_profile() + self.assertEqual(profile.get('groups'), ['pgadmin-admins']) + client.get.assert_called_once() + def _test_oidc_get_user_profile_calls_userinfo(self, provider): from pgadmin.authenticate.oauth2 import OAuth2Authentication