fix(ui): set MAX_MEMBERS_PER_TEAM in team forms (#3650)

* fix(ui): set MAX_MEMBERS_PER_TEAM in team forms

The "Maximum Members" input in the Create Team and Edit Team modals
had max="1000" hardcoded, causing two problems:
- non-admin users could set max_members above MAX_MEMBERS_PER_TEAM
- when MAX_MEMBERS_PER_TEAM > 1000, the browser rejected valid values

Fix by surfacing the setting to the UI with role-aware behaviour:
- admins: no max attribute (no browser-side cap, matching skip_limits in the backend)
- non-admins: max attribute set to MAX_MEMBERS_PER_TEAM (default 100)

The default value in the Create modal is min(50, MAX_MEMBERS_PER_TEAM)
so tighter operator-configured limits are respected out of the box.

Closes #3589

Signed-off-by: Gabriel Costa <gabrielcg@proton.me>

* fix(teams): use extracted is_admin var in update_team, add boundary test

Use the already-extracted `is_admin` variable for `skip_limits` in
update_team instead of re-evaluating `bool(current_user.get("is_admin"))`.

Add missing boundary test for non-admin update at exactly the configured
max_members_per_team limit, matching the create-path coverage.

Signed-off-by: Mihai Criveti <crivetimihai@gmail.com>

* fix(ui): handle over-limit max_members in non-admin team edit form

When a non-admin user edits a team whose max_members already exceeds the
configured MAX_MEMBERS_PER_TEAM (e.g. set by an admin), the HTML form
would render value > max, causing browser validation to block submission.

Fix by showing an empty field with a hint displaying the current value
and the allowed limit. Submitting empty preserves the existing value
server-side. Admins continue to see the actual value with no cap.

Add regression tests for both admin and non-admin over-limit edit cases.

Signed-off-by: Mihai Criveti <crivetimihai@gmail.com>

* fix(docs): add missing docstring sections in passthrough_headers

Add missing Args and Returns to _loopback_skip_set,
safe_extract_headers_for_loopback, and
safe_extract_and_filter_for_loopback to satisfy flake8 DAR checks.

Signed-off-by: Mihai Criveti <crivetimihai@gmail.com>

---------

Signed-off-by: Gabriel Costa <gabrielcg@proton.me>
Signed-off-by: Mihai Criveti <crivetimihai@gmail.com>
Co-authored-by: Mihai Criveti <crivetimihai@gmail.com>

Author: gcgoncalves

mcpgateway/admin.py

@@ -3982,6 +3982,7 @@ def _to_dict_and_filter(raw_list):
             # Token policy flags
             "require_token_expiration": getattr(settings, "require_token_expiration", True),
             "sri_hashes": load_sri_hashes(),
+            "max_members_per_team": getattr(settings, "max_members_per_team", 100),
         },
     )
 
@@ -6011,6 +6012,22 @@ async def admin_get_team_edit(
 
         safe_team_name = html.escape(team.name, quote=True)
         safe_description = html.escape(team.description or "")
+        is_admin_edit = isinstance(_user, dict) and _user.get("is_admin")
+        max_members_limit = getattr(settings, "max_members_per_team", 100)
+        current_exceeds_limit = bool(team.max_members and team.max_members > max_members_limit)
+        max_attr = "" if is_admin_edit else f'max="{max_members_limit}"'
+        # When the existing value exceeds the configured limit for a non-admin,
+        # show an empty field to avoid browser validation blocking form submission.
+        # Submitting empty preserves the current value server-side.
+        if is_admin_edit:
+            max_members_value = team.max_members if team.max_members else ""
+            max_members_hint = "Admins can set any limit. Leave empty to keep current value."
+        elif current_exceeds_limit:
+            max_members_value = ""
+            max_members_hint = f"Current: {team.max_members} (above max {max_members_limit}). Leave empty to keep, or set a new value \u2264 {max_members_limit}."
+        else:
+            max_members_value = team.max_members if team.max_members else ""
+            max_members_hint = f"Max {max_members_limit}. Leave empty to keep current value."
         edit_form = rf"""
         <div class="space-y-4">
             <h3 class="text-lg font-semibold text-gray-900 dark:text-white mb-4">Edit Team</h3>
@@ -6043,9 +6060,9 @@ async def admin_get_team_edit(
                 </div>
                 <div>
                     <label class="block text-sm font-medium text-gray-700 dark:text-gray-300">Maximum Members</label>
-                    <input type="number" name="max_members" min="1" max="1000" value="{team.max_members if team.max_members else ''}"
+                    <input type="number" name="max_members" min="1" {max_attr} value="{max_members_value}"
                            class="mt-1 block w-full px-3 py-2 border border-gray-300 dark:border-gray-600 rounded-md shadow-sm focus:outline-none focus:ring-indigo-500 focus:border-indigo-500 dark:bg-gray-700 text-gray-900 dark:text-white">
-                    <p class="text-xs text-gray-500 dark:text-gray-400 mt-1">Leave empty to keep current value</p>
+                    <p class="text-xs text-gray-500 dark:text-gray-400 mt-1">{max_members_hint}</p>
                 </div>
                 <div class="flex justify-end space-x-3">
                     <button type="button" onclick="hideTeamEditModal()"

mcpgateway/routers/teams.py

@@ -96,17 +96,28 @@ async def create_team(request: TeamCreateRequest, current_user_ctx: dict = Depen
         True
     """
     try:
-        if not settings.allow_team_creation and not current_user_ctx.get("is_admin"):
+        is_admin = bool(current_user_ctx.get("is_admin"))
+
+        if not settings.allow_team_creation and not is_admin:
             raise HTTPException(status_code=status.HTTP_403_FORBIDDEN, detail="Team creation is currently disabled")
 
+        # Enforce max_members_per_team limit for non-admin users
+        if not is_admin and request.max_members is not None:
+            limit = getattr(settings, "max_members_per_team", 100)
+            if request.max_members > limit:
+                raise HTTPException(
+                    status_code=status.HTTP_400_BAD_REQUEST,
+                    detail=f"max_members cannot exceed {limit} for non-admin users",
+                )
+
         service = TeamManagementService(db)
         team = await service.create_team(
             name=request.name,
             description=request.description,
             created_by=current_user_ctx["email"],
             visibility=request.visibility,
             max_members=request.max_members,
-            skip_limits=bool(current_user_ctx.get("is_admin")),
+            skip_limits=is_admin,
         )
 
         # Build response BEFORE closing session to avoid lazy-load issues with get_member_count()
@@ -359,16 +370,24 @@ async def update_team(team_id: str, request: TeamUpdateRequest, current_user: di
         HTTPException: If team not found, access denied, or update fails
     """
     try:
+        is_admin = bool(current_user.get("is_admin"))
         service = TeamManagementService(db)
 
         # Check if user is team owner
         role = await service.get_user_role_in_team(current_user["email"], team_id)
         if role != "owner":
             raise HTTPException(status_code=status.HTTP_403_FORBIDDEN, detail=_ACCESS_DENIED_MSG)
 
-        success = await service.update_team(
-            team_id=team_id, name=request.name, description=request.description, visibility=request.visibility, max_members=request.max_members, skip_limits=bool(current_user.get("is_admin"))
-        )
+        # Enforce max_members_per_team limit for non-admin users
+        if not is_admin and request.max_members is not None:
+            limit = getattr(settings, "max_members_per_team", 100)
+            if request.max_members > limit:
+                raise HTTPException(
+                    status_code=status.HTTP_400_BAD_REQUEST,
+                    detail=f"max_members cannot exceed {limit} for non-admin users",
+                )
+
+        success = await service.update_team(team_id=team_id, name=request.name, description=request.description, visibility=request.visibility, max_members=request.max_members, skip_limits=is_admin)
 
         if not success:
             raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="Team not found or update failed")

mcpgateway/templates/admin.html

@@ -14645,12 +14645,16 @@ <h3 class="text-lg font-medium text-gray-900 dark:text-white">
                     name="max_members"
                     id="team-max-members"
                     min="1"
-                    max="1000"
-                    value="50"
+                    {% if not is_admin %}max="{{ max_members_per_team }}"{% endif %}
+                    value="{{ [50, max_members_per_team] | min }}"
                     class="block w-full border-gray-300 rounded-md shadow-sm focus:ring-indigo-500 focus:border-indigo-500 dark:bg-gray-700 dark:border-gray-600 dark:text-white"
                   />
                   <p class="mt-1 text-xs text-gray-500 dark:text-gray-400">
-                    Optional, defaults to 50
+                    {% if is_admin %}
+                    Optional. Admins can set any limit.
+                    {% else %}
+                    Optional, defaults to {{ [50, max_members_per_team] | min }}.
+                    {% endif %}
                   </p>
                 </div>
               </div>

mcpgateway/utils/passthrough_headers.py

@@ -568,6 +568,9 @@ def _loopback_skip_set() -> frozenset[str]:
     passthrough headers can never overwrite the gateway-internal proxy
     user identity — even if that header name is added to the passthrough
     allowlist by mistake.
+
+    Returns:
+        frozenset[str]: Header names to skip during loopback forwarding.
     """
     proxy = settings.proxy_user_header.lower()
     if proxy in _LOOPBACK_SKIP_HEADERS:
@@ -787,6 +790,13 @@ def safe_extract_headers_for_loopback(request_headers: Dict[str, str], transport
     Wraps :func:`extract_headers_for_loopback` so that SSE / WebSocket setup
     is never blocked by passthrough configuration issues.  ``ImportError``
     propagates (broken deployment should fail loudly).
+
+    Args:
+        request_headers: Incoming HTTP headers to extract from.
+        transport_name: Label for warning logs on failure.
+
+    Returns:
+        Dict[str, str]: Extracted passthrough headers, or empty dict on error.
     """
     try:
         return extract_headers_for_loopback(request_headers)
@@ -801,6 +811,12 @@ def safe_extract_and_filter_for_loopback(request_headers: Dict[str, str]) -> Dic
     Combines :func:`extract_headers_for_loopback` and
     :func:`filter_loopback_skip_headers` with error handling so that
     Streamable HTTP affinity loopback calls degrade gracefully.
+
+    Args:
+        request_headers: Incoming HTTP headers to extract and filter.
+
+    Returns:
+        Dict[str, str]: Filtered passthrough headers, or empty dict on error.
     """
     try:
         return filter_loopback_skip_headers(extract_headers_for_loopback(request_headers))

tests/unit/mcpgateway/routers/test_teams.py

@@ -433,7 +433,7 @@ async def test_get_team_access_denied(self, mock_user_context, mock_db, mock_tea
     async def test_update_team_success(self, mock_user_context, mock_db, mock_team):
         """Test updating a team successfully."""
         team_id = mock_team.id
-        request = TeamUpdateRequest(name="Updated Team", description="Updated description", visibility="public", max_members=200)
+        request = TeamUpdateRequest(name="Updated Team", description="Updated description", visibility="public", max_members=50)
 
         with patch("mcpgateway.routers.teams.TeamManagementService") as MockService:
             mock_service = AsyncMock(spec=TeamManagementService)
@@ -453,7 +453,7 @@ async def test_update_team_success(self, mock_user_context, mock_db, mock_team):
     async def test_update_team_insufficient_permissions(self, mock_user_context, mock_db):
         """Test updating a team without owner permissions."""
         team_id = str(uuid4())
-        request = TeamUpdateRequest(name="Updated Team", description="Updated description", visibility="public", max_members=200)
+        request = TeamUpdateRequest(name="Updated Team", description="Updated description", visibility="public", max_members=50)
 
         with patch("mcpgateway.routers.teams.TeamManagementService") as MockService:
             mock_service = AsyncMock(spec=TeamManagementService)
@@ -472,7 +472,7 @@ async def test_update_team_insufficient_permissions(self, mock_user_context, moc
     async def test_update_team_not_found(self, mock_user_context, mock_db):
         """Test updating a non-existent team."""
         team_id = str(uuid4())
-        request = TeamUpdateRequest(name="Updated Team", description="Updated description", visibility="public", max_members=200)
+        request = TeamUpdateRequest(name="Updated Team", description="Updated description", visibility="public", max_members=50)
 
         with patch("mcpgateway.routers.teams.TeamManagementService") as MockService:
             mock_service = AsyncMock(spec=TeamManagementService)
@@ -1126,6 +1126,121 @@ async def test_invitation_with_value_error(self, mock_user_context, mock_db):
             assert exc_info.value.status_code == status.HTTP_400_BAD_REQUEST
             assert "Invalid email format" in str(exc_info.value.detail)
 
+    # =========================================================================
+    # max_members Enforcement Tests
+    # =========================================================================
+
+    @pytest.mark.asyncio
+    async def test_create_team_non_admin_max_members_too_high(self, mock_user_context, mock_db):
+        """Non-admin users cannot set max_members above the configured limit."""
+        request = TeamCreateRequest(name="Test Team", description="desc", visibility="private", max_members=9999)
+
+        with patch("mcpgateway.routers.teams.settings") as mock_settings:
+            mock_settings.allow_team_creation = True
+            mock_settings.max_members_per_team = 100
+
+            from mcpgateway.routers.teams import create_team
+
+            with pytest.raises(HTTPException) as exc_info:
+                await create_team(request, current_user_ctx=mock_user_context, db=mock_db)
+
+            assert exc_info.value.status_code == status.HTTP_400_BAD_REQUEST
+            assert "max_members cannot exceed 100" in str(exc_info.value.detail)
+
+    @pytest.mark.asyncio
+    async def test_create_team_non_admin_max_members_at_limit(self, mock_user_context, mock_team, mock_db):
+        """Non-admin users can set max_members exactly at the configured limit."""
+        request = TeamCreateRequest(name="Test Team", description="desc", visibility="private", max_members=100)
+
+        with patch("mcpgateway.routers.teams.settings") as mock_settings, patch("mcpgateway.routers.teams.TeamManagementService") as MockService:
+            mock_settings.allow_team_creation = True
+            mock_settings.max_members_per_team = 100
+            mock_service = AsyncMock(spec=TeamManagementService)
+            mock_service.create_team = AsyncMock(return_value=mock_team)
+            MockService.return_value = mock_service
+
+            from mcpgateway.routers.teams import create_team
+
+            result = await create_team(request, current_user_ctx=mock_user_context, db=mock_db)
+            assert result.id == mock_team.id
+
+    @pytest.mark.asyncio
+    async def test_create_team_admin_can_exceed_max_members(self, mock_admin_context, mock_team, mock_db):
+        """Admin users can set max_members above the configured limit."""
+        request = TeamCreateRequest(name="Test Team", description="desc", visibility="private", max_members=9999)
+
+        with patch("mcpgateway.routers.teams.settings") as mock_settings, patch("mcpgateway.routers.teams.TeamManagementService") as MockService:
+            mock_settings.allow_team_creation = True
+            mock_settings.max_members_per_team = 100
+            mock_service = AsyncMock(spec=TeamManagementService)
+            mock_service.create_team = AsyncMock(return_value=mock_team)
+            MockService.return_value = mock_service
+
+            from mcpgateway.routers.teams import create_team
+
+            result = await create_team(request, current_user_ctx=mock_admin_context, db=mock_db)
+            assert result.id == mock_team.id
+            mock_service.create_team.assert_called_once()
+
+    @pytest.mark.asyncio
+    async def test_update_team_non_admin_max_members_too_high(self, mock_user_context, mock_db):
+        """Non-admin users cannot set max_members above the configured limit on update."""
+        team_id = str(uuid4())
+        request = TeamUpdateRequest(max_members=9999)
+
+        with patch("mcpgateway.routers.teams.settings") as mock_settings, patch("mcpgateway.routers.teams.TeamManagementService") as MockService:
+            mock_settings.max_members_per_team = 100
+            mock_service = AsyncMock(spec=TeamManagementService)
+            mock_service.get_user_role_in_team = AsyncMock(return_value="owner")
+            MockService.return_value = mock_service
+
+            from mcpgateway.routers.teams import update_team
+
+            with pytest.raises(HTTPException) as exc_info:
+                await update_team(team_id, request, current_user=mock_user_context, db=mock_db)
+
+            assert exc_info.value.status_code == status.HTTP_400_BAD_REQUEST
+            assert "max_members cannot exceed 100" in str(exc_info.value.detail)
+
+    @pytest.mark.asyncio
+    async def test_update_team_non_admin_max_members_at_limit(self, mock_user_context, mock_team, mock_db):
+        """Non-admin users can set max_members exactly at the configured limit on update."""
+        team_id = str(uuid4())
+        request = TeamUpdateRequest(max_members=100)
+
+        with patch("mcpgateway.routers.teams.settings") as mock_settings, patch("mcpgateway.routers.teams.TeamManagementService") as MockService:
+            mock_settings.max_members_per_team = 100
+            mock_service = AsyncMock(spec=TeamManagementService)
+            mock_service.get_user_role_in_team = AsyncMock(return_value="owner")
+            mock_service.update_team = AsyncMock(return_value=True)
+            mock_service.get_team_by_id = AsyncMock(return_value=mock_team)
+            MockService.return_value = mock_service
+
+            from mcpgateway.routers.teams import update_team
+
+            result = await update_team(team_id, request, current_user=mock_user_context, db=mock_db)
+            assert result.id == mock_team.id
+
+    @pytest.mark.asyncio
+    async def test_update_team_admin_can_exceed_max_members(self, mock_admin_context, mock_team, mock_db):
+        """Admin users can set max_members above the configured limit on update."""
+        team_id = str(uuid4())
+        request = TeamUpdateRequest(max_members=9999)
+
+        with patch("mcpgateway.routers.teams.settings") as mock_settings, patch("mcpgateway.routers.teams.TeamManagementService") as MockService:
+            mock_settings.max_members_per_team = 100
+            mock_service = AsyncMock(spec=TeamManagementService)
+            mock_service.get_user_role_in_team = AsyncMock(return_value="owner")
+            mock_service.update_team = AsyncMock(return_value=True)
+            mock_service.get_team_by_id = AsyncMock(return_value=mock_team)
+            MockService.return_value = mock_service
+
+            from mcpgateway.routers.teams import update_team
+
+            result = await update_team(team_id, request, current_user=mock_admin_context, db=mock_db)
+            assert result.id == mock_team.id
+            mock_service.update_team.assert_called_once()
+
     @pytest.mark.skip(reason="RBAC mocking complex - functionality covered by test_teams_v2.py")
     @pytest.mark.asyncio
     async def test_member_operations_with_invalid_role(self, mock_user_context, mock_db):

tests/unit/mcpgateway/routers/test_teams_v2.py

@@ -194,7 +194,7 @@ async def test_get_team_not_found(self, mock_user_context, mock_db):
     async def test_update_team_success(self, mock_user_context, mock_db, mock_team):
         """Test updating a team successfully."""
         team_id = mock_team.id
-        request = TeamUpdateRequest(name="Updated Team", description="Updated description", visibility="public", max_members=200)
+        request = TeamUpdateRequest(name="Updated Team", description="Updated description", visibility="public", max_members=100)
 
         with patch("mcpgateway.routers.teams.TeamManagementService") as MockService:
             mock_service = AsyncMock(spec=TeamManagementService)