Feature/sso group mapping

CHANGELOG.md

@@ -5,6 +5,27 @@ All notable changes to this project will be documented in this file.
 The format is based on [Keep a Changelog](https://keepachangelog.com/en/1.0.0/),
 and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0.html).
 
+## [8.7.0] - 2026-01-26
+
+### Added
+
+- **SSO Group Mapping** - Map external SSO groups (from Entra/Keycloak) to internal CIDX groups:
+  - Configure mappings with external group ID, optional display name, and target CIDX group
+  - First-match strategy: user assigned to first matching group in mapping list
+  - Graceful fallback to "users" group when no mappings match or mapped group doesn't exist
+  - Backward compatible: automatic migration from old dict format to new list format
+  - Optional display names for better UI readability in configuration
+
+### Changed
+
+- **ID Token-Based User Info** - OIDC authentication now parses ID token directly instead of calling userinfo endpoint:
+  - More reliable: works universally with Entra, Keycloak, and other OIDC providers
+  - Eliminates potential userinfo endpoint configuration issues
+  - Groups extracted directly from ID token claims
+  - All OIDC tests updated for new implementation
+
+---
+
 ## [8.6.17] - 2026-01-26
 
 ### Fixed

README.md

@@ -2,7 +2,7 @@
 
 AI-powered semantic code search for your codebase. Find code by meaning, not just keywords.
 
-**Version 8.6.17** - [Changelog](CHANGELOG.md) | [Migration Guide](docs/migration-to-v8.md) | [Architecture](docs/architecture.md)
+**Version 8.7.0** - [Changelog](CHANGELOG.md) | [Migration Guide](docs/migration-to-v8.md) | [Architecture](docs/architecture.md)
 
 ## Quick Navigation
 

RELEASE_NOTES.md

@@ -1,6 +1,6 @@
 # Release Notes
 
-**Current Version: 8.6.17** | [Full Changelog](CHANGELOG.md)
+**Current Version: 8.7.0** | [Full Changelog](CHANGELOG.md)
 
 ---
 

docs/architecture.md

@@ -298,7 +298,7 @@ CIDX can index and semantically search entire git commit history:
 
 **Initialize Handshake** (CRITICAL for Claude Code connection):
 - Method: `initialize` - MUST be first client-server interaction
-- Server Response: `{ "protocolVersion": "2025-06-18", "capabilities": { "tools": {} }, "serverInfo": { "name": "Neo", "version": "8.6.17" } }`
+- Server Response: `{ "protocolVersion": "2025-06-18", "capabilities": { "tools": {} }, "serverInfo": { "name": "Neo", "version": "8.7.0" } }`
 - Required for OAuth flow completion - Claude Code calls `initialize` after authentication
 
 **Version Notes**:

docs/query-guide.md

@@ -736,7 +736,7 @@ cidx query "anything" --time-range-all --quiet
 
 **Status**: FACT-CHECKED (2025-01-20)
 
-**Verification Scope**: All technical claims, parameter specifications, performance metrics, and code examples validated against CIDX implementation v8.6.17.
+**Verification Scope**: All technical claims, parameter specifications, performance metrics, and code examples validated against CIDX implementation v8.7.0.
 
 ### Corrections Made
 
@@ -880,7 +880,7 @@ cidx query "anything" --time-range-all --quiet
 
 **Fact-checker**: Claude Opus 4.5 (fact-checking agent)
 **Verification Date**: 2025-01-20
-**Version Reference**: v8.6.17
+**Version Reference**: v8.7.0
 
 ---
 

src/code_indexer/__init__.py

@@ -6,5 +6,5 @@
 HNSW graph indexing (O(log N) complexity).
 """
 
-__version__ = "8.6.17"
+__version__ = "8.7.0"
 __author__ = "Seba Battig"

src/code_indexer/server/auth/oidc/oidc_manager.py

@@ -60,16 +60,22 @@ def is_enabled(self):
         """Check if OIDC is enabled in configuration."""
         return self.config.enabled
 
-    def _ensure_group_membership(self, username: str) -> None:
+    def _ensure_group_membership(self, username: str, external_groups=None) -> None:
         """Ensure user has group membership via SSO provisioning hook.
 
         Story #708: SSO Auto-Provisioning with Default Group Assignment
         - AC1: New SSO users assigned to "users" group
         - AC3: Existing users' membership is NOT changed
         - AC6: Errors are logged but do not block authentication
 
+        Group Mapping Support:
+        - External groups from SSO provider can be mapped to CIDX groups via configuration
+        - First matched group is used for assignment
+        - Falls back to "users" group if no mappings match
+
         Args:
             username: The user's username to provision
+            external_groups: Optional list of external group names from OIDC provider
         """
         import logging
 
@@ -88,7 +94,12 @@ def _ensure_group_membership(self, username: str) -> None:
                 ensure_user_group_membership,
             )
 
-            result = ensure_user_group_membership(username, self.group_manager)
+            # Get group_mappings from config
+            group_mappings = self.config.group_mappings or {}
+
+            result = ensure_user_group_membership(
+                username, self.group_manager, external_groups, group_mappings
+            )
             if result:
                 logger.debug(
                     f"SSO provisioning completed for user {username}",
@@ -191,7 +202,9 @@ async def match_or_create_user(self, user_info):
                         extra={"correlation_id": get_correlation_id()},
                     )
                     # Story #708: Ensure group membership on every SSO login
-                    self._ensure_group_membership(existing_user.username)
+                    self._ensure_group_membership(
+                        existing_user.username, user_info.groups
+                    )
                     return existing_user
                 else:
                     # Stale OIDC link (defensive check - should be cleaned up on user deletion)
@@ -219,7 +232,9 @@ async def match_or_create_user(self, user_info):
                         email=user_info.email,
                     )
                     # Story #708: Ensure group membership on every SSO login
-                    self._ensure_group_membership(existing_user.username)
+                    self._ensure_group_membership(
+                        existing_user.username, user_info.groups
+                    )
                     return existing_user
 
         # Create new user via JIT provisioning if enabled
@@ -281,6 +296,6 @@ async def match_or_create_user(self, user_info):
             )
 
             # Story #708: Ensure group membership for new JIT-provisioned user
-            self._ensure_group_membership(new_user.username)
+            self._ensure_group_membership(new_user.username, user_info.groups)
 
             return new_user

src/code_indexer/server/auth/oidc/oidc_provider.py

@@ -2,7 +2,7 @@
 
 from code_indexer.server.middleware.correlation import get_correlation_id
 from dataclasses import dataclass
-from typing import Optional
+from typing import Optional, List
 
 
 @dataclass
@@ -20,6 +20,7 @@ class OIDCUserInfo:
     email: Optional[str] = None
     email_verified: bool = False
     username: Optional[str] = None
+    groups: Optional[List[str]] = None
 
 
 class OIDCProvider:
@@ -110,39 +111,53 @@ async def exchange_code_for_token(self, code, code_verifier, redirect_uri):
 
         return tokens
 
-    async def get_user_info(self, access_token):
-        import httpx
+    async def get_user_info(self, access_token, id_token):
+        """Parse ID token to extract user information and claims.
+
+        ID token contains all necessary user claims including groups.
+        This approach works universally with Entra, Keycloak, and other OIDC providers.
+
+        Args:
+            access_token: OAuth access token (kept for backward compatibility)
+            id_token: OIDC ID token (JWT) containing user claims
+
+        Returns:
+            OIDCUserInfo object with user claims including groups
+        """
+        import base64
+        import json
         import logging
 
         logger = logging.getLogger(__name__)
 
-        # Construct userinfo endpoint (typically from discovery, but fallback to standard path)
-        # Use userinfo endpoint from discovery metadata (preferred) or fallback
-        if self._metadata and self._metadata.userinfo_endpoint:
-            userinfo_endpoint = self._metadata.userinfo_endpoint
-        else:
-            userinfo_endpoint = (
-                f"{self.config.issuer_url}/protocol/openid-connect/userinfo"
-            )
+        # Parse ID token JWT (format: header.payload.signature)
+        if not id_token:
+            raise Exception("ID token is required but was not provided")
 
-        # Fetch user info from userinfo endpoint
-        headers = {"Authorization": f"Bearer {access_token}"}
         try:
-            async with httpx.AsyncClient() as client:
-                response = await client.get(userinfo_endpoint, headers=headers)
-                response.raise_for_status()  # Raise HTTPStatusError for 4xx/5xx
-                data = response.json()  # Not async in httpx
-        except httpx.HTTPStatusError as e:
-            raise Exception(
-                f"Failed to get user info: HTTP {e.response.status_code} - {e.response.text}"
-            ) from e
-        except httpx.RequestError as e:
-            raise Exception(f"Failed to connect to userinfo endpoint: {str(e)}") from e
+            parts = id_token.split('.')
+            if len(parts) != 3:
+                raise Exception(f"Invalid ID token format: expected 3 parts, got {len(parts)}")
+
+            # Decode payload (base64url decode with padding)
+            payload = parts[1]
+            # Add padding if needed (base64 requires length to be multiple of 4)
+            padding = 4 - (len(payload) % 4)
+            if padding != 4:
+                payload += '=' * padding
+
+            data = json.loads(base64.urlsafe_b64decode(payload))
+            logger.info(
+                f"Parsed ID token with claims: {list(data.keys())}",
+                extra={"correlation_id": get_correlation_id()},
+            )
+        except Exception as e:
+            raise Exception(f"Failed to parse ID token: {e}") from e
 
-        # Validate userinfo response has required fields
+        # Validate ID token has required fields
         if "sub" not in data or not data["sub"]:
             raise Exception(
-                "Invalid userinfo response: missing or empty sub (subject) claim"
+                "Invalid ID token: missing or empty sub (subject) claim"
             )
 
         # Log claim extraction for debugging
@@ -151,7 +166,7 @@ async def get_user_info(self, access_token):
             extra={"correlation_id": get_correlation_id()},
         )
         logger.info(
-            f"Available claims in userinfo: {list(data.keys())}",
+            f"Available claims in ID token: {list(data.keys())}",
             extra={"correlation_id": get_correlation_id()},
         )
 
@@ -167,12 +182,32 @@ async def get_user_info(self, access_token):
             extra={"correlation_id": get_correlation_id()},
         )
 
+        # Extract groups from configured groups_claim
+        groups_value = data.get(self.config.groups_claim)
+        logger.info(
+            f"Groups claim '{self.config.groups_claim}' raw value: {groups_value} (type: {type(groups_value).__name__})",
+            extra={"correlation_id": get_correlation_id()},
+        )
+        if groups_value and isinstance(groups_value, list):
+            groups_list = [str(g) for g in groups_value]
+            logger.info(
+                f"Extracted {len(groups_list)} groups from '{self.config.groups_claim}' claim: {groups_list}",
+                extra={"correlation_id": get_correlation_id()},
+            )
+        else:
+            groups_list = []
+            logger.info(
+                f"No groups found in '{self.config.groups_claim}' claim or claim value is not a list",
+                extra={"correlation_id": get_correlation_id()},
+            )
+
         # Create OIDCUserInfo from response
         user_info = OIDCUserInfo(
             subject=data.get("sub", ""),
             email=email_value,
             email_verified=data.get("email_verified", False),
             username=username_value,
+            groups=groups_list if groups_list else None,
         )
 
         return user_info

src/code_indexer/server/auth/oidc/routes.py

@@ -56,8 +56,14 @@ async def sso_callback(code: str, state: str, request: Request):
         code, code_verifier, callback_url
     )
 
-    # Get user info from provider
-    user_info = await oidc_manager.provider.get_user_info(tokens["access_token"])
+    # Parse ID token to get user info (includes groups for Entra/Keycloak)
+    if "id_token" not in tokens:
+        raise HTTPException(status_code=500, detail="ID token not returned by provider")
+
+    user_info = await oidc_manager.provider.get_user_info(
+        tokens["access_token"],
+        tokens["id_token"]
+    )
 
     # Match or create user (email-based)
     user = await oidc_manager.match_or_create_user(user_info)

src/code_indexer/server/services/config_service.py

@@ -191,6 +191,8 @@ def get_all_settings(self) -> Dict[str, Any]:
                 "require_email_verification": config.oidc_provider_config.require_email_verification,
                 "enable_jit_provisioning": config.oidc_provider_config.enable_jit_provisioning,
                 "default_role": config.oidc_provider_config.default_role,
+                "groups_claim": config.oidc_provider_config.groups_claim,
+                "group_mappings": config.oidc_provider_config.group_mappings,
             },
             # SCIP workspace cleanup (Story #647, Story #15 AC2: moved to scip_config)
             "scip_cleanup": {
@@ -580,6 +582,30 @@ def _update_oidc_setting(self, config: ServerConfig, key: str, value: Any) -> No
             oidc.enable_jit_provisioning = value in ["true", True]
         elif key == "default_role":
             oidc.default_role = str(value)
+        elif key == "groups_claim":
+            oidc.groups_claim = str(value)
+        elif key == "group_mappings":
+            # Parse JSON string, dict (old format), or list (new format)
+            import json
+
+            if isinstance(value, (dict, list)):
+                oidc.group_mappings = value
+            elif isinstance(value, str):
+                try:
+                    parsed = json.loads(value)
+                    if not isinstance(parsed, (dict, list)):
+                        raise ValueError(
+                            f"Invalid JSON for group_mappings: must be dict or list, got {type(parsed)}"
+                        )
+                    oidc.group_mappings = parsed
+                except json.JSONDecodeError:
+                    raise ValueError(
+                        f"Invalid JSON for group_mappings: {value}. Expected format: [{{'external_group_id': 'guid', 'cidx_group': 'admins'}}]"
+                    )
+            else:
+                raise ValueError(
+                    f"Invalid type for group_mappings: {type(value)}. Expected dict, list, or JSON string"
+                )
         else:
             raise ValueError(f"Unknown OIDC setting: {key}")