Merge pull request #4 from dheerajoruganty/pr-3

UI Specific scopes

Author: aarora79

auth_server/scopes.yml

@@ -1,14 +1,62 @@
 # This file defines the scopes for the MCP servers, detailing which servers and tools are accessible under different scopes.
-mcp-registry-admin:
-  - list_service
-  - register_service
-  - health_check_service
-  - toggle_service
-  - modify_service
 
-mcp-registry-user:
-  - list_service:
-    - mcpgw
+UI-Scopes:
+  mcp-registry-admin:
+    list_service:
+      - all
+    register_service:
+      - all
+    health_check_service:
+      - all
+    toggle_service:
+      - all
+    modify_service:
+      - all
+
+  mcp-registry-user:
+    list_service:
+      - mcpgw
+      - realserverfaketools
+    health_check_service:
+      - mcpgw
+
+  mcp-registry-developer:
+    list_service:
+      - all
+    register_service:
+      - all
+    health_check_service:
+      - all
+
+  mcp-registry-operator:
+    list_service:
+      - all
+    health_check_service:
+      - all
+    toggle_service:
+      - all
+
+# Group mappings - maps Cognito groups to both UI scopes AND server scopes
+group_mappings:
+  mcp-registry-admin:
+    # Admin gets both UI permissions and unrestricted server access
+    - mcp-registry-admin
+    - mcp-servers-unrestricted/read
+    - mcp-servers-unrestricted/execute
+  mcp-registry-user:
+    # Users get limited UI permissions and restricted server access
+    - mcp-registry-user
+    - mcp-servers-restricted/read
+  mcp-registry-developer:
+    # Developers can register services and have restricted access
+    - mcp-registry-developer
+    - mcp-servers-restricted/read
+    - mcp-servers-restricted/execute
+  mcp-registry-operator:
+    # Operators can control services but not register new ones
+    - mcp-registry-operator
+    - mcp-servers-restricted/read
+    - mcp-servers-restricted/execute
 
 
 mcp-servers-unrestricted/read:

auth_server/server.py

@@ -50,34 +50,24 @@ def load_scopes_config():
 
 def map_cognito_groups_to_scopes(groups: List[str]) -> List[str]:
     """
-    Map Cognito groups to MCP scopes using the same format as M2M resource server scopes.
+    Map Cognito groups to MCP scopes using the group_mappings from scopes.yml configuration.
     
     Args:
         groups: List of Cognito group names
         
     Returns:
-        List of MCP scopes in resource server format
+        List of MCP scopes
     """
     scopes = []
+    group_mappings = SCOPES_CONFIG.get('group_mappings', {})
 
     for group in groups:
-        if group == 'mcp-admin':
-            # Admin gets unrestricted read and execute access
-            scopes.append('mcp-servers-unrestricted/read')
-            scopes.append('mcp-servers-unrestricted/execute')
-        elif group == 'mcp-user':
-            # Regular users get restricted read access by default
-            scopes.append('mcp-servers-restricted/read')
-        elif group.startswith('mcp-server-'):
-            # Server-specific groups grant access based on server permissions
-            # For now, grant restricted execute access for specific servers
-            # This allows access to the servers defined in the restricted scope
-            scopes.append('mcp-servers-restricted/execute')
-            
-            # Note: The actual server access control is handled by the
-            # validate_server_tool_access function which checks the scopes.yml
-            # configuration. The group names are preserved in the 'groups' field
-            # for potential future fine-grained access control.
+        if group in group_mappings:
+            group_scopes = group_mappings[group]
+            scopes.extend(group_scopes)
+            logger.debug(f"Mapped group '{group}' to scopes: {group_scopes}")
+        else:
+            logger.debug(f"No scope mapping found for group: {group}")
 
     # Remove duplicates while preserving order
     seen = set()
@@ -87,6 +77,7 @@ def map_cognito_groups_to_scopes(groups: List[str]) -> List[str]:
             seen.add(scope)
             unique_scopes.append(scope)
 
+    logger.info(f"Final mapped scopes: {unique_scopes}")
     return unique_scopes
 
 def validate_session_cookie(cookie_value: str) -> Dict[str, any]:

registry/api/server_routes.py

@@ -37,6 +37,14 @@ async def read_root(
     except HTTPException as e:
         logger.info(f"Authentication failed at root route: {e.detail}, redirecting to login")
         return RedirectResponse(url="/login", status_code=302)
+        
+    from ..auth.dependencies import user_has_ui_permission_for_service
+    
+    # Helper function for templates
+    def can_perform_action(permission: str, service_name: str) -> bool:
+        """Check if user has UI permission for a specific service"""
+        return user_has_ui_permission_for_service(permission, service_name, user_context.get('ui_permissions', {}))
+    
     service_data = []
     search_query = query.lower() if query else ""
 
@@ -55,10 +63,18 @@ async def read_root(
         key=lambda p: all_servers[p]["server_name"]
     )
 
+    # Filter services based on UI permissions
+    accessible_services = user_context.get('accessible_services', [])
+    
     for path in sorted_server_paths:
         server_info = all_servers[path]
         server_name = server_info["server_name"]
 
+        # Check if user can list this service
+        if 'all' not in accessible_services and server_name not in accessible_services:
+            logger.debug(f"Filtering out service '{server_name}' - user doesn't have list_service permission")
+            continue
+        
         # Include description and tags in search
         searchable_text = f"{server_name.lower()} {server_info.get('description', '').lower()} {' '.join(server_info.get('tags', []))}"
         if not search_query or search_query in searchable_text:
@@ -88,7 +104,8 @@ async def read_root(
             "request": request, 
             "services": service_data, 
             "username": user_context['username'],
-            "user_context": user_context  # Pass full user context to template
+            "user_context": user_context,  # Pass full user context to template
+            "can_perform_action": can_perform_action  # Helper function for permission checks
         },
     )
 
@@ -100,25 +117,28 @@ async def toggle_service_route(
     enabled: Annotated[str | None, Form()] = None,
     user_context: Annotated[dict, Depends(enhanced_auth)] = None,
 ):
-    """Toggle a service on/off (requires modification permissions)."""
+    """Toggle a service on/off (requires toggle_service UI permission)."""
     from ..search.service import faiss_service
     from ..health.service import health_service
     from ..core.nginx_service import nginx_service
-    
-    # Check if user can modify servers
-    if not user_context['can_modify_servers']:
-        logger.warning(f"User {user_context['username']} attempted to toggle service {service_path} without modify permissions")
-        raise HTTPException(
-            status_code=status.HTTP_403_FORBIDDEN, 
-            detail="You do not have permission to modify servers"
-        )
+    from ..auth.dependencies import user_has_ui_permission_for_service
 
     if not service_path.startswith("/"):
         service_path = "/" + service_path
 
     server_info = server_service.get_server_info(service_path)
     if not server_info:
         raise HTTPException(status_code=404, detail="Service path not registered")
+    
+    service_name = server_info["server_name"]
+    
+    # Check if user has toggle_service permission for this specific service
+    if not user_has_ui_permission_for_service('toggle_service', service_name, user_context.get('ui_permissions', {})):
+        logger.warning(f"User {user_context['username']} attempted to toggle service {service_name} without toggle_service permission")
+        raise HTTPException(
+            status_code=status.HTTP_403_FORBIDDEN, 
+            detail=f"You do not have permission to toggle {service_name}"
+        )
 
     # For non-admin users, check if they have access to this specific server
     if not user_context['is_admin']:
@@ -194,17 +214,21 @@ async def register_service(
     license_str: Annotated[str, Form(alias="license")] = "N/A",
     user_context: Annotated[dict, Depends(enhanced_auth)] = None,
 ):
-    """Register a new service (requires modification permissions)."""
+    """Register a new service (requires register_service UI permission)."""
     from ..search.service import faiss_service
     from ..health.service import health_service
     from ..core.nginx_service import nginx_service
+    from ..auth.dependencies import user_has_ui_permission_for_service
+    
+    # Check if user has register_service permission for any service
+    ui_permissions = user_context.get('ui_permissions', {})
+    register_permissions = ui_permissions.get('register_service', [])
 
-    # Check if user can modify servers
-    if not user_context['can_modify_servers']:
-        logger.warning(f"User {user_context['username']} attempted to register service without modify permissions")
+    if not register_permissions:
+        logger.warning(f"User {user_context['username']} attempted to register service without register_service permission")
         raise HTTPException(
             status_code=status.HTTP_403_FORBIDDEN, 
-            detail="You do not have permission to register new servers"
+            detail="You do not have permission to register new services"
         )
 
     logger.info(f"Service registration request from user '{user_context['username']}'")
@@ -270,14 +294,8 @@ async def edit_server_form(
     service_path: str, 
     user_context: Annotated[dict, Depends(enhanced_auth)]
 ):
-    """Show edit form for a service (requires modification permissions)."""
-    # Check if user can modify servers
-    if not user_context['can_modify_servers']:
-        logger.warning(f"User {user_context['username']} attempted to access edit form for {service_path} without modify permissions")
-        raise HTTPException(
-            status_code=status.HTTP_403_FORBIDDEN, 
-            detail="You do not have permission to edit servers"
-        )
+    """Show edit form for a service (requires modify_service UI permission)."""
+    from ..auth.dependencies import user_has_ui_permission_for_service
 
     if not service_path.startswith('/'):
         service_path = '/' + service_path
@@ -286,6 +304,16 @@ async def edit_server_form(
     if not server_info:
         raise HTTPException(status_code=404, detail="Service path not found")
 
+    service_name = server_info["server_name"]
+    
+    # Check if user has modify_service permission for this specific service
+    if not user_has_ui_permission_for_service('modify_service', service_name, user_context.get('ui_permissions', {})):
+        logger.warning(f"User {user_context['username']} attempted to access edit form for {service_name} without modify_service permission")
+        raise HTTPException(
+            status_code=status.HTTP_403_FORBIDDEN, 
+            detail=f"You do not have permission to modify {service_name}"
+        )
+    
     # For non-admin users, check if they have access to this specific server
     if not user_context['is_admin']:
         if not server_service.user_can_access_server_path(service_path, user_context['accessible_servers']):
@@ -319,24 +347,29 @@ async def edit_server_submit(
     is_python: Annotated[bool | None, Form()] = False,  
     license_str: Annotated[str, Form(alias="license")] = "N/A", 
 ):
-    """Handle server edit form submission (requires modification permissions)."""
+    """Handle server edit form submission (requires modify_service UI permission)."""
     from ..search.service import faiss_service
     from ..core.nginx_service import nginx_service
-    
-    # Check if user can modify servers
-    if not user_context['can_modify_servers']:
-        logger.warning(f"User {user_context['username']} attempted to edit service {service_path} without modify permissions")
-        raise HTTPException(
-            status_code=status.HTTP_403_FORBIDDEN, 
-            detail="You do not have permission to edit servers"
-        )
+    from ..auth.dependencies import user_has_ui_permission_for_service
 
     if not service_path.startswith('/'):
         service_path = '/' + service_path
 
-    # Check if the server exists
-    if not server_service.get_server_info(service_path):
+    # Check if the server exists and get service name
+    server_info = server_service.get_server_info(service_path)
+    if not server_info:
         raise HTTPException(status_code=404, detail="Service path not found")
+    
+    service_name = server_info["server_name"]
+    
+    # Check if user has modify_service permission for this specific service
+    if not user_has_ui_permission_for_service('modify_service', service_name, user_context.get('ui_permissions', {})):
+        logger.warning(f"User {user_context['username']} attempted to edit service {service_name} without modify_service permission")
+        raise HTTPException(
+            status_code=status.HTTP_403_FORBIDDEN, 
+            detail=f"You do not have permission to modify {service_name}"
+        )
+
 
     # For non-admin users, check if they have access to this specific server
     if not user_context['is_admin']:
@@ -537,18 +570,29 @@ async def refresh_service(
     service_path: str, 
     user_context: Annotated[dict, Depends(enhanced_auth)]
 ):
-    """Refresh service health and tool information (requires read access)."""
+    """Refresh service health and tool information (requires health_check_service permission)."""
     from ..search.service import faiss_service
     from ..health.service import health_service
     from ..core.mcp_client import mcp_client_service
     from ..core.nginx_service import nginx_service
+    from ..auth.dependencies import user_has_ui_permission_for_service
 
     if not service_path.startswith('/'):
         service_path = '/' + service_path
 
     server_info = server_service.get_server_info(service_path)
     if not server_info:
         raise HTTPException(status_code=404, detail="Service path not registered")
+    
+    service_name = server_info["server_name"]
+    
+    # Check if user has health_check_service permission for this specific service
+    if not user_has_ui_permission_for_service('health_check_service', service_name, user_context.get('ui_permissions', {})):
+        logger.warning(f"User {user_context['username']} attempted to refresh service {service_name} without health_check_service permission")
+        raise HTTPException(
+            status_code=status.HTTP_403_FORBIDDEN, 
+            detail=f"You do not have permission to refresh {service_name}"
+        )
 
     # For non-admin users, check if they have access to this specific server
     if not user_context['is_admin']: