Merge pull request #5330 from IBM/fix/keycloak-sso-role-claims

fix(sso): merge Keycloak realm/client roles from access_token, not just userinfo/id_token

Author: cafalchio

.secrets.baseline

@@ -3,7 +3,7 @@
     "files": "(?x)( package-lock\\.json$ |Cargo\\.lock$ |uv\\.lock$ |go\\.sum$ |mcpgateway/sri_hashes\\.json$ )|^.secrets.baseline$",
     "lines": null
   },
-  "generated_at": "2026-06-18T13:53:55Z",
+  "generated_at": "2026-06-20T09:31:24Z",
   "plugins_used": [
     {
       "name": "AWSKeyDetector"
@@ -2896,31 +2896,31 @@
         "hashed_secret": "c58994eefb19ae2d0bfc26a106de438359e53fb6",
         "is_secret": false,
         "is_verified": false,
-        "line_number": 314,
+        "line_number": 325,
         "type": "Secret Keyword",
         "verified_result": null
       },
       {
         "hashed_secret": "8c9fdbd88e905d33102d0be537adeab8595fe0da",
         "is_secret": false,
         "is_verified": false,
-        "line_number": 344,
+        "line_number": 355,
         "type": "Secret Keyword",
         "verified_result": null
       },
       {
         "hashed_secret": "6ea1c94703f93074853165df24e0ded6917472af",
         "is_secret": false,
         "is_verified": false,
-        "line_number": 370,
+        "line_number": 381,
         "type": "Secret Keyword",
         "verified_result": null
       },
       {
         "hashed_secret": "b05b46675e3b6cecdb5ed7d0c24a8a183abd4de8",
         "is_secret": false,
         "is_verified": false,
-        "line_number": 740,
+        "line_number": 751,
         "type": "Secret Keyword",
         "verified_result": null
       }
@@ -3896,7 +3896,7 @@
         "hashed_secret": "1e5c2f367f02e47a8c160cda1cd9d91decbac441",
         "is_secret": false,
         "is_verified": false,
-        "line_number": 172,
+        "line_number": 200,
         "type": "Secret Keyword",
         "verified_result": null
       }

docs/docs/manage/sso-keycloak-tutorial.md

@@ -240,15 +240,26 @@ To include roles in JWT tokens:
 **realm roles**:
 
 - Maps realm roles to `realm_access.roles` claim
-- Should be enabled by default
+- Exists by default
 
 **client roles**:
 
 - Maps client roles to `resource_access.{client_id}.roles` claim
-- Should be enabled by default
+- Exists by default
 
 If missing, create them manually using **Add mapper** → **By configuration** → **User Realm Role** or **User Client Role**.
 
+**Important — enable on ID token and userinfo, not just the access token:**
+
+Keycloak's built-in **realm roles** and **client roles** mappers ship with only **Add to access token** turned on. **Add to ID token** and **Add to userinfo** are off by default. ContextForge reads roles from the userinfo response (and falls back to the ID token in some configurations) — it does not parse the access token — so with the defaults left as-is, `realm_access`/`resource_access` never reach the gateway and `SSO_KEYCLOAK_ROLE_MAPPINGS`/`SSO_KEYCLOAK_MAP_REALM_ROLES` silently have nothing to map, even though the role is assigned correctly in Keycloak.
+
+For each of the **realm roles** and **client roles** mappers, open it and switch on:
+
+- **Add to ID token**
+- **Add to userinfo**
+
+(in addition to the default **Add to access token**). The local `infra/keycloak/realm-export.json` dev seed already configures these mappers correctly — this step is only needed when wiring up your own Keycloak realm by hand.
+
 ## Step 5: Configure User Attributes and Groups
 
 ### 5.1 Create Groups (Optional)
@@ -839,10 +850,11 @@ SSO_KEYCLOAK_PUBLIC_BASE_URL=http://localhost:8180   # Browser-facing (auth URL,
 **Problem**: User roles not included in JWT token
 **Solution**: Configure role mappers and enable role mapping
 
-1. Verify role mappers exist:
+1. Verify role mappers exist and reach the ID token / userinfo:
 
    - Go to **Client scopes** → **roles** → **Mappers**
    - Ensure **realm roles** and **client roles** mappers exist
+   - Open each mapper and confirm **Add to ID token** and **Add to userinfo** are both ON — Keycloak ships these mappers with only **Add to access token** enabled by default, and ContextForge reads roles from userinfo/ID token, not the access token. This is the most common cause of "role assigned in Keycloak but gateway treats user as default role" — see [Step 4.4](#44-configure-role-mappers).
 
 2. Enable role mapping in gateway:
 ```bash

infra/keycloak/realm-export.json

@@ -94,6 +94,34 @@
             "userinfo.token.claim": "true",
             "claim.name": "groups"
           }
+        },
+        {
+          "name": "realm roles (id+userinfo)",
+          "protocol": "openid-connect",
+          "protocolMapper": "oidc-usermodel-realm-role-mapper",
+          "consentRequired": false,
+          "config": {
+            "multivalued": "true",
+            "id.token.claim": "true",
+            "access.token.claim": "true",
+            "userinfo.token.claim": "true",
+            "claim.name": "realm_access.roles",
+            "jsonType.label": "String"
+          }
+        },
+        {
+          "name": "client roles (id+userinfo)",
+          "protocol": "openid-connect",
+          "protocolMapper": "oidc-usermodel-client-role-mapper",
+          "consentRequired": false,
+          "config": {
+            "multivalued": "true",
+            "id.token.claim": "true",
+            "access.token.claim": "true",
+            "userinfo.token.claim": "true",
+            "claim.name": "resource_access.${client_id}.roles",
+            "jsonType.label": "String"
+          }
         }
       ]
     }

mcpgateway/services/sso_service.py

@@ -1450,10 +1450,23 @@ async def _enrich_user_data_from_claims(
             return
 
         # Keycloak: merge realm_access, resource_access, and groups from id_token
-        if provider.id == "keycloak" and verified_id_token_claims:
+        # and, failing that, the access_token. Keycloak's built-in "realm roles"
+        # and "client roles" client-scope mappers default access.token.claim=true
+        # but id.token.claim=userinfo.token.claim=false, so on a stock Keycloak
+        # setup these claims are normally present ONLY on the access_token —
+        # never on the userinfo response or id_token. The access_token was
+        # already received directly from the trusted token endpoint, so decoding
+        # it here (without re-verifying the signature) carries the same trust
+        # level as the verified_id_token_claims fallback above.
+        if provider.id == "keycloak":
+            access_token_claims = self._decode_jwt_claims(access_token) or {}
             for claim in ["realm_access", "resource_access", "groups"]:
-                if claim in verified_id_token_claims and claim not in user_data:
+                if claim in user_data:
+                    continue
+                if verified_id_token_claims and claim in verified_id_token_claims:
                     user_data[claim] = verified_id_token_claims[claim]
+                elif claim in access_token_claims:
+                    user_data[claim] = access_token_claims[claim]
             return
 
         # Generic OIDC (including Okta, IBM Verify, and any custom provider):
@@ -1577,7 +1590,15 @@ async def _get_user_info(self, provider: SSOProvider, access_token: str, token_d
                     public_base_url,
                     metadata.get("base_url"),
                 )
-                return self._normalize_user_info(provider, verified_id_token_claims)
+                fallback_claims = dict(verified_id_token_claims)
+                # id_token also lacks realm_access/resource_access/groups by default
+                # on a stock Keycloak setup (see _enrich_user_data_from_claims) -
+                # pull them from the access_token, which carries them by default.
+                access_token_claims = self._decode_jwt_claims(access_token) or {}
+                for claim in ["realm_access", "resource_access", "groups"]:
+                    if claim not in fallback_claims and claim in access_token_claims:
+                        fallback_claims[claim] = access_token_claims[claim]
+                return self._normalize_user_info(provider, fallback_claims)
 
         logger.error("User info request failed for %s: HTTP %s - %s", provider.name, response.status_code, response.text)
 
@@ -2283,6 +2304,12 @@ async def _map_groups_to_roles(self, user_email: str, user_groups: List[str], pr
             logger.debug("No role mappings configured for provider %s, skipping role sync", provider.id)
             return role_assignments
 
+        # Match role_mappings keys case-insensitively, consistent with _should_user_be_admin().
+        # Without this, an IdP group/role casing that differs from the configured mapping key
+        # (e.g. Keycloak role "gateway-admin" vs a mapping key "Gateway-Admin") would silently
+        # skip RBAC role assignment even though is_admin could already be granted.
+        lower_role_mappings = {str(k).lower(): v for k, v in role_mappings.items()}
+
         personal_team_id: Optional[str] = None
         personal_team_checked = False
 
@@ -2339,10 +2366,9 @@ async def _resolve_team_scope_id_if_needed(role_scope: str) -> Optional[str]:
         # Batch role lookups: collect all role names that need to be looked up
         role_names_to_lookup = set()
         for group in user_groups:
-            if group in role_mappings:
-                role_name = role_mappings[group]
-                if role_name not in ["admin", settings.default_admin_role]:
-                    role_names_to_lookup.add(role_name)
+            role_name = lower_role_mappings.get(group.lower())
+            if role_name and role_name not in ["admin", settings.default_admin_role]:
+                role_names_to_lookup.add(role_name)
 
         # Add default role to lookup if needed
         if has_provider_default_role and provider_default_role:
@@ -2361,8 +2387,8 @@ async def _resolve_team_scope_id_if_needed(role_scope: str) -> Optional[str]:
 
         # Process role mappings for ALL providers
         for group in user_groups:
-            if group in role_mappings:
-                role_name = role_mappings[group]
+            role_name = lower_role_mappings.get(group.lower())
+            if role_name:
                 # Special case for "admin" shorthand or configured admin role name
                 if role_name in ["admin", settings.default_admin_role]:
                     role_assignments.append({"role_name": settings.default_admin_role, "scope": "global", "scope_id": None})

tests/unit/mcpgateway/services/test_sso_service.py

@@ -754,6 +754,7 @@ async def test_exchange_code_for_tokens_basic_auth_header_format(self, sso_servi
         assert auth_header.startswith("Basic ")
         # Verify it's valid base64
         import base64
+
         payload = auth_header[len("Basic "):]
         decoded = base64.b64decode(payload).decode("utf-8")
         assert decoded.startswith("test-client-id:")
@@ -809,6 +810,7 @@ async def test_build_basic_auth_header_url_encodes_special_chars(self, sso_servi
         header = sso_service._build_basic_auth_header("client@id", "secret:+?")
         assert header.startswith("Basic ")
         import base64
+
         payload = base64.b64decode(header[len("Basic "):]).decode("utf-8")
         assert payload == "client%40id:secret%3A%2B%3F"
 
@@ -1287,6 +1289,94 @@ async def test_get_user_info_keycloak_falls_back_to_id_token_when_userinfo_fails
         assert "admin" in result["groups"]
         assert "/team" in result["groups"]
 
+    @pytest.mark.asyncio
+    async def test_get_user_info_keycloak_merges_realm_access_from_access_token_when_userinfo_succeeds(self, sso_service):
+        """Keycloak's default mappers only put realm_access/resource_access on the access_token
+        (userinfo.token.claim and id.token.claim default to false). Even when userinfo returns
+        200 and there's no id_token claim to fall back to, the gateway should still extract
+        roles by decoding the access_token it already received from the token endpoint."""
+        # Standard
+        import base64
+
+        # Third-Party
+        import orjson
+
+        user_response = MagicMock()
+        user_response.status_code = 200
+        user_response.json.return_value = {"email": "user@kc.com", "name": "KC User", "preferred_username": "kcuser", "sub": "kc-123"}
+
+        mock_client = AsyncMock()
+        mock_client.get = AsyncMock(return_value=user_response)
+
+        provider = _make_provider(
+            id="keycloak",
+            name="keycloak",
+            provider_type="oidc",
+            provider_metadata={"map_realm_roles": True, "map_client_roles": True},
+        )
+
+        access_token_payload = orjson.dumps({"realm_access": {"roles": ["gateway-admin"]}, "resource_access": {"app": {"roles": ["edit"]}}})
+        access_token_payload_b64 = base64.urlsafe_b64encode(access_token_payload).decode().rstrip("=")
+        fake_access_token = f"eyJhbGciOiJSUzI1NiJ9.{access_token_payload_b64}.sig"
+
+        # No id_token / verified id_token claims at all - the bug surfaces even without
+        # the split-host fallback machinery being involved.
+        token_data = {"access_token": fake_access_token}
+
+        with patch("mcpgateway.services.http_client_service.get_http_client", new_callable=AsyncMock) as mock_get_client, patch("mcpgateway.services.sso_service.settings") as mock_settings:
+            mock_get_client.return_value = mock_client
+            mock_settings.sso_github_admin_orgs = []
+            result = await sso_service._get_user_info(provider, fake_access_token, token_data)
+
+        assert result is not None
+        assert result["provider"] == "keycloak"
+        assert "gateway-admin" in result["groups"]
+        assert "app:edit" in result["groups"]
+
+    @pytest.mark.asyncio
+    async def test_get_user_info_keycloak_fallback_merges_realm_access_from_access_token_when_id_token_lacks_it(self, sso_service):
+        """On the 401 + split-host fallback path, the id_token may itself lack realm_access
+        (same default-mapper gap as userinfo). The access_token, already in hand from the
+        token endpoint, should be decoded to fill in the missing roles."""
+        # Standard
+        import base64
+
+        # Third-Party
+        import orjson
+
+        fail_response = MagicMock()
+        fail_response.status_code = 401
+        fail_response.text = ""
+
+        mock_client = AsyncMock()
+        mock_client.get = AsyncMock(return_value=fail_response)
+
+        provider = _make_provider(
+            id="keycloak",
+            name="keycloak",
+            provider_type="oidc",
+            provider_metadata={"map_realm_roles": True, "map_client_roles": False, "base_url": "http://keycloak:8080", "public_base_url": "http://localhost:8180"},
+        )
+
+        # id_token claims have no realm_access (matches Keycloak's default mapper config).
+        id_token_claims = {"sub": "kc-123", "email": "user@kc.com", "name": "KC User", "preferred_username": "kcuser"}
+
+        access_token_payload = orjson.dumps({"realm_access": {"roles": ["gateway-admin"]}})
+        access_token_payload_b64 = base64.urlsafe_b64encode(access_token_payload).decode().rstrip("=")
+        fake_access_token = f"eyJhbGciOiJSUzI1NiJ9.{access_token_payload_b64}.sig"
+
+        token_data = {"access_token": fake_access_token, "_verified_id_token_claims": id_token_claims}
+
+        with patch("mcpgateway.services.http_client_service.get_http_client", new_callable=AsyncMock) as mock_get_client, patch("mcpgateway.services.sso_service.settings") as mock_settings:
+            mock_get_client.return_value = mock_client
+            mock_settings.sso_github_admin_orgs = []
+            result = await sso_service._get_user_info(provider, fake_access_token, token_data)
+
+        assert result is not None
+        assert result["provider"] == "keycloak"
+        assert result["email"] == "user@kc.com"
+        assert "gateway-admin" in result["groups"]
+
     @pytest.mark.asyncio
     async def test_get_user_info_generic_oidc_merges_groups_from_id_token(self, sso_service):
         """Generic OIDC provider merges configured groups claim from id_token when userinfo omits it."""
@@ -2959,6 +3049,43 @@ async def test_role_mapping_with_custom_role(self, sso_service):
         assert len(result) == 1
         assert result[0]["role_name"] == "developer"
 
+    @pytest.mark.asyncio
+    async def test_role_mapping_matches_case_insensitively(self, sso_service):
+        """A group/role name whose casing differs from the configured role_mappings key
+        must still resolve. _should_user_be_admin() already matches case-insensitively;
+        _map_groups_to_roles() must do the same so is_admin and the RBAC role assignment
+        stay consistent (otherwise a user can get is_admin=True with no platform_admin
+        role row, or vice versa)."""
+        mock_role = SimpleNamespace(name="developer", scope="team", id="r1")
+        provider = _make_provider(provider_metadata={"role_mappings": {"Dev-Group": "developer"}})
+
+        with patch("mcpgateway.services.sso_service.settings") as mock_settings, patch("mcpgateway.services.role_service.RoleService") as MockRoleService:
+            mock_settings.sso_entra_admin_groups = []
+            mock_settings.sso_entra_default_role = None
+            mock_settings.sso_entra_role_mappings = {}
+            role_svc_instance = AsyncMock()
+            role_svc_instance.get_role_by_name = AsyncMock(return_value=mock_role)
+            MockRoleService.return_value = role_svc_instance
+            # IdP returned "dev-group" (lowercase), mapping key is "Dev-Group".
+            result = await sso_service._map_groups_to_roles("user@test.com", ["dev-group"], provider)
+
+        assert len(result) == 1
+        assert result[0]["role_name"] == "developer"
+
+    @pytest.mark.asyncio
+    async def test_admin_shorthand_matches_case_insensitively(self, sso_service):
+        """Same case-insensitivity guarantee for the 'admin'/default_admin_role shorthand path."""
+        provider = _make_provider(provider_metadata={"role_mappings": {"Gateway-Admin": "admin"}})
+        with patch("mcpgateway.services.sso_service.settings") as mock_settings, patch("mcpgateway.services.role_service.RoleService"):
+            mock_settings.sso_entra_admin_groups = []
+            mock_settings.sso_entra_default_role = None
+            mock_settings.sso_entra_role_mappings = {}
+            mock_settings.default_admin_role = "platform_admin"
+            # IdP returned the role with Keycloak-typical lowercase-hyphen casing.
+            result = await sso_service._map_groups_to_roles("user@test.com", ["gateway-admin"], provider)
+
+        assert any(r["role_name"] == "platform_admin" for r in result)
+
     @pytest.mark.asyncio
     async def test_entra_default_role_fallback(self, sso_service):
         mock_role = SimpleNamespace(name="viewer", scope="global", id="r2")
@@ -4707,6 +4834,7 @@ async def test_get_user_info_adfs_with_id_token(self, sso_service):
     @pytest.mark.asyncio
     async def test_get_user_info_adfs_missing_id_token(self, sso_service):
         """ADFS provider raises error when id_token is missing."""
+        # First-Party
         from mcpgateway.services.sso_service import SSOProviderConfigError
 
         provider = _make_provider(id="adfs", name="adfs", provider_type="oidc")