feat(oauth2): Map OAuth2 claims to Visible Server Groups

web/config.py

@@ -894,6 +894,19 @@
         #     'wids': ["cf1c38e5-3621-4004-a7cb-879624dced7c"],
         # }
         'OAUTH2_ADDITIONAL_CLAIMS': None,
+        # Optional server-group visibility claim controls.
+        # If OAUTH2_SERVER_GROUP_CLAIM is set, pgAdmin stores the resolved
+        # list in session['oauth2_server_group_claims'] and uses it to show
+        # additional server groups in browser tree.
+        # Without mapping, claim values are treated as server group names.
+        # Example:
+        # 'OAUTH2_SERVER_GROUP_CLAIM': 'pgadmin_server_groups',
+        # Optional mapping from claim value -> server group name(s):
+        # 'OAUTH2_SERVER_GROUP_CLAIM_MAPPING': {
+        #     'readonly': ['RO Server 1', 'RO Server 2']
+        # },
+        'OAUTH2_SERVER_GROUP_CLAIM': None,
+        'OAUTH2_SERVER_GROUP_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

@@ -537,6 +537,48 @@ def _resolve_username(self, id_token_claims, profile_dict):
 
         return None, None
 
+    def _extract_server_group_claims(self, profile_dict, id_token_claims):
+        """
+        Resolve allowed server groups from provider claim configuration.
+
+        Config keys (per provider):
+        - OAUTH2_SERVER_GROUP_CLAIM: claim name to read
+        - OAUTH2_SERVER_GROUP_CLAIM_MAPPING: optional dict mapping
+          claim-value -> list of server group names
+
+        Returns:
+            list[str] | None
+            - list of allowed server group names if configured
+            - None when claim-based server group filtering is not configured
+        """
+        provider = self.oauth2_config.get(self.oauth2_current_client) or {}
+        claim_name = provider.get('OAUTH2_SERVER_GROUP_CLAIM')
+        if not claim_name:
+            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:
+            return []
+
+        if not isinstance(claim_values, list):
+            claim_values = [claim_values]
+
+        mapping = provider.get('OAUTH2_SERVER_GROUP_CLAIM_MAPPING') or {}
+        server_groups = []
+        for value in claim_values:
+            mapped_groups = mapping.get(value)
+            if mapped_groups is None:
+                server_groups.append(value)
+            elif isinstance(mapped_groups, str):
+                server_groups.append(mapped_groups)
+            elif isinstance(mapped_groups, list):
+                server_groups.extend(mapped_groups)
+
+        # Remove duplicates
+        return list(set(server_groups))
+
     def login(self, form):
         if not self.oauth2_current_client:
             error_msg = gettext('No OAuth2 provider available.')
@@ -600,6 +642,15 @@ def login(self, form):
             current_app.logger.error(error_msg)
             return False, error_msg
 
+        oauth2_server_group_claims = self._extract_server_group_claims(
+            profile_dict, id_token_claims
+        )
+        if oauth2_server_group_claims is None:
+            session.pop('oauth2_server_group_claims', None)
+        else:
+            session['oauth2_server_group_claims'] = \
+                oauth2_server_group_claims
+
         additional_claims = None
         if 'OAUTH2_ADDITIONAL_CLAIMS' in self.oauth2_config[
                 self.oauth2_current_client]:

web/pgadmin/browser/tests/test_oauth2_with_mocking.py

@@ -114,6 +114,20 @@ class Oauth2LoginMockTestCase(BaseTestGenerator):
             profile={'email': 'claims@example.com'},
             id_token_claims={'groups': ['group-b']},
         )),
+        ('OIDC Server Group Claim Direct Mapping', dict(
+            oauth2_provider='oidc-server-groups-direct',
+            kind='login_success',
+            profile={'email': 'claims@example.com'},
+            id_token_claims={'pgadmin_server_groups': ['RO Server 1']},
+            expected_server_groups=['RO Server 1'],
+        )),
+        ('OIDC Server Group Claim Value Mapping', dict(
+            oauth2_provider='oidc-server-groups-mapped',
+            kind='login_success',
+            profile={'email': 'claims@example.com'},
+            id_token_claims={'pgadmin_server_groups': ['readonly']},
+            expected_server_groups=['RO Server 1', 'RO Server 2'],
+        )),
         ('OIDC get_user_profile Skips Userinfo', dict(
             oauth2_provider='oidc-basic',
             kind='oidc_get_user_profile_skip',
@@ -279,6 +293,37 @@ def setUp(self):
                 'OAUTH2_ADDITIONAL_CLAIMS': {
                     'groups': ['group-a']
                 }
+            },
+            {
+                'OAUTH2_NAME': 'oidc-server-groups-direct',
+                'OAUTH2_DISPLAY_NAME': 'OIDC Server Groups Direct',
+                '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_SERVER_GROUP_CLAIM': 'pgadmin_server_groups',
+            },
+            {
+                'OAUTH2_NAME': 'oidc-server-groups-mapped',
+                'OAUTH2_DISPLAY_NAME': 'OIDC Server Groups Mapped',
+                '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_SERVER_GROUP_CLAIM': 'pgadmin_server_groups',
+                'OAUTH2_SERVER_GROUP_CLAIM_MAPPING': {
+                    'readonly': ['RO Server 1', 'RO Server 2']
+                }
             }
         ]
 
@@ -309,7 +354,8 @@ def runTest(self):
             self._test_workload_identity_missing_token_file_fails_fast()
         elif self.kind == 'login_success':
             self._test_oauth2_login_success(
-                self.oauth2_provider, self.profile, self.id_token_claims
+                self.oauth2_provider, self.profile, self.id_token_claims,
+                getattr(self, 'expected_server_groups', None)
             )
         elif self.kind == 'login_failure':
             self._test_oauth2_login_failure(
@@ -386,7 +432,8 @@ def _fake_authenticate(self, _form):
                 )
 
     def _test_oauth2_login_success(
-            self, provider, profile, id_token_claims=None
+            self, provider, profile, id_token_claims=None,
+            expected_server_groups=None
     ):
         from pgadmin.authenticate.oauth2 import OAuth2Authentication
 
@@ -421,6 +468,14 @@ def _fake_get_user_profile(self):
             )
         self.assertEqual(res.status_code, 200)
         self._assert_oauth2_session_logged_in()
+        with self.tester.session_transaction() as sess:
+            if expected_server_groups is None:
+                self.assertNotIn('oauth2_server_group_claims', sess)
+            else:
+                self.assertEqual(
+                    sess.get('oauth2_server_group_claims'),
+                    expected_server_groups
+                )
 
     def _test_oauth2_login_failure(
             self, provider, profile, id_token_claims=None

web/pgadmin/utils/server_access.py

@@ -15,9 +15,11 @@
 """
 
 from sqlalchemy import or_, case, func, literal
+from flask import session
 from flask_security import current_user
 
 from pgadmin.model import db, Server, ServerGroup
+from pgadmin.utils.constants import OAUTH2
 import config
 
 
@@ -109,8 +111,10 @@ def get_server_groups_for_user(hide_shared=False, servergroup_id=None):
 def get_server_groups_for_user_query(hide_shared=False, servergroup_id=None):
     """Return a query for server groups visible to the current user.
 
-    Includes groups owned by the user plus groups containing shared
-    servers (Server.shared=True, visible to all authenticated users).
+    Includes groups owned by the user, groups containing shared
+    servers (Server.shared=True, visible to all authenticated users),
+    and groups resolved from OAuth2 claim configuration when
+    OAUTH2_SERVER_GROUP_CLAIM is configured for the provider.
 
     is_shared_group is an additional column indicating if the group is a group
     not owned by the user and contains shared servers.
@@ -154,7 +158,7 @@ def get_server_groups_for_user_query(hide_shared=False, servergroup_id=None):
     )
 
     if hide_shared:
-        query = query.filter(ServerGroup.user_id == current_user.id)
+        conditions = [ServerGroup.user_id == current_user.id]
     else:
         has_shared_servers = (
             db.session.query(Server.id)
@@ -165,12 +169,21 @@ def get_server_groups_for_user_query(hide_shared=False, servergroup_id=None):
             .exists()
         )
 
-        query = query.filter(
-            or_(
-                ServerGroup.user_id == current_user.id,
-                has_shared_servers
-            )
+        conditions = [
+            ServerGroup.user_id == current_user.id,
+            has_shared_servers
+        ]
+
+    if getattr(current_user, 'auth_source', None) == OAUTH2:
+        oauth2_allowed_server_groups = session.get(
+            'oauth2_server_group_claims'
         )
+        if isinstance(oauth2_allowed_server_groups, list):
+            conditions.append(
+                ServerGroup.name.in_(oauth2_allowed_server_groups)
+            )
+
+    query = query.filter(or_(*conditions))
 
     if servergroup_id is not None:
         query = query.filter(ServerGroup.id == servergroup_id)