fix(search): token-scope team results for admins in unified search (#5668)

* fix(search): token-scope team results for admins in unified search

Complements the non-admin team-scope fix: the admin branch of
admin_search_teams returns all teams (the admin management view) and ignored
token scope, so an admin using a token explicitly narrowed to a team subset
still saw every team through /v1/search, /admin/search, and /admin/teams/search.

Honor explicit token_teams even for admins (Layer 1 constrains visibility
independently of admin status): None means full admin bypass (unchanged), while
an explicit list narrows the result. The caller's own personal team stays
visible. Admin UI session tokens and unscoped admin API tokens resolve to None
and are unaffected.

Add real-data tests through /v1/search: an unscoped admin sees all teams
(including one they do not belong to), while a token scoped to one team narrows
to it. Verified empirically by disabling the filter and observing the scoped
admin token still see every team.

Signed-off-by: Pratik Gandhi <gandhipratik203@gmail.com>

* fix(search): apply admin team scope in the query before pagination

The admin branch of admin_search_teams fetched list_teams(page=1,
per_page=limit) and then filtered by token scope, so a scoped-admin token whose
allowed team sorted past the first `limit` matches returned empty/incomplete
results even though that team was visible.

Add an optional team_ids filter to TeamManagementService.list_teams (applied to
the query before pagination; an empty list matches nothing) and pass the
normalized token scope from the admin branch. Scope is now enforced in the DB
query rather than on an already-limited page. As a consequence an explicit
scope (including [] = public-only) no longer surfaces the personal team, which
aligns the admin branch with the token model; the non-admin branch carve-out is
addressed separately.

Verified empirically: an admin scoped to a team that sorts last among 12 teams
with limit=8 now returns that team instead of an empty page.

Signed-off-by: Pratik Gandhi <gandhipratik203@gmail.com>

* fix(search): drop personal-team carve-out; cover public-only and past-page scope

Finding 1 (non-admin branch): remove the getattr(is_personal) carve-out so an
explicit token scope (including [] = public-only) no longer surfaces the
caller's personal team. An unscoped caller's memberships already include their
personal team via _get_user_team_ids, so the common case is unchanged; this
matches normalize_token_teams()/get_team_from_token(), which define [] as
public-only with no personal fallback. (The admin branch was already aligned by
pushing scope into the query.)

Add the two regression cases the review flagged:
- public-only admin token (token_teams=[]) sees no teams, not a bypass view.
- an admin scoped to a team that sorts past the per-page limit still gets it
  (proves scope is applied in the query before pagination).

Verified empirically: disabling the list_teams team_ids filter makes all three
scope-dependent admin tests fail (narrowing leaks, public-only non-empty,
past-page team dropped).

Signed-off-by: Pratik Gandhi <gandhipratik203@gmail.com>

* fix(search): rename admin-branch team-id var to avoid mypy type collision

The admin branch declared scoped_team_ids: Optional[list[str]] while the
non-admin branch reused the same name for a set[str], which strict mypy flagged
as an incompatible reassignment (harmless at runtime since the branches are
mutually exclusive). Rename the admin-branch variable to admin_scoped_team_ids.

Signed-off-by: Pratik Gandhi <gandhipratik203@gmail.com>

* test(search): add unit coverage for admin scope extraction and list_teams team_ids

CI diff-cover flagged two changed lines as uncovered because they are only
exercised by the integration suite (gated behind --with-integration):
- admin.py: the admin-branch token-scope extraction
- team_management_service.py: the list_teams team_ids where-clause

Add unit tests that run in the standard pytest job:
- admin_search_teams with a scoped admin token (mixed str/dict token_teams)
  forwards the normalized team_ids to list_teams.
- list_teams(team_ids=[...]) restricts results in the query; None = no filter.

Signed-off-by: Pratik Gandhi <gandhipratik203@gmail.com>

---------

Signed-off-by: Pratik Gandhi <gandhipratik203@gmail.com>

Author: gandhipratik203

mcpgateway/admin.py

@@ -5581,9 +5581,25 @@ async def admin_search_teams(
     # The CALLER (admin.py) distinguishes.
 
     if current_user.is_admin:
-        # Admin sees all non-personal teams plus their own personal team (single query)
+        # Honor explicit token narrowing even for admins (Layer 1 constrains
+        # visibility independently of RBAC/admin status). token_teams is None for
+        # full admin bypass (unrestricted); an explicit list (including []) scopes
+        # the result. The scope is pushed into the query so it applies before
+        # pagination (an allowed team must not be dropped for sorting past the
+        # first page) and so an explicit scope no longer surfaces the personal team.
+        raw_token_teams = user.get("token_teams")
+        admin_scoped_team_ids: Optional[list[str]] = None
+        if raw_token_teams is not None:
+            admin_scoped_team_ids = [team["id"] if isinstance(team, dict) else team for team in raw_token_teams]
         result = await team_service.list_teams(
-            page=1, per_page=limit, include_inactive=include_inactive, visibility_filter=visibility, include_personal=False, search_query=search_query, personal_owner_email=user_email
+            page=1,
+            per_page=limit,
+            include_inactive=include_inactive,
+            visibility_filter=visibility,
+            include_personal=False,
+            search_query=search_query,
+            personal_owner_email=user_email,
+            team_ids=admin_scoped_team_ids,
         )
         # Result is dict {data, pagination...} (since page provided)
         teams = result["data"]
@@ -5595,12 +5611,14 @@ async def admin_search_teams(
         # returns every membership and ignores token scope, so a token narrowed to
         # a team subset would otherwise leak sibling teams the caller belongs to but
         # is scoped out of. _get_user_team_ids honors token_teams/_cached_team_ids;
-        # the caller's own personal team stays visible (owner is always visible).
+        # an unscoped caller's own memberships (including their personal team) are in
+        # this set, while an explicit scope (including [] = public-only) does not add
+        # a personal-team fallback, matching normalize_token_teams()/get_team_from_token().
         scoped_team_ids = set(await _get_user_team_ids(user, db))
         # Filter in memory
         filtered = []
         for t in all_teams:
-            if not getattr(t, "is_personal", False) and t.id not in scoped_team_ids:
+            if t.id not in scoped_team_ids:
                 continue
             if not include_inactive and not t.is_active:
                 continue

mcpgateway/services/team_management_service.py

@@ -1856,6 +1856,7 @@ async def list_teams(
         include_personal: bool = False,
         search_query: Optional[str] = None,
         personal_owner_email: Optional[str] = None,
+        team_ids: Optional[List[str]] = None,
     ) -> Union[Tuple[List[EmailTeam], Optional[str]], Dict[str, Any]]:
         """List teams with pagination support (cursor or page based).
 
@@ -1871,6 +1872,7 @@ async def list_teams(
             include_personal: Whether to include personal teams
             search_query: Search term for name/slug/description
             personal_owner_email: When set (and include_personal=False), includes this user's personal team alongside non-personal teams
+            team_ids: When set, restrict results to these team IDs before pagination (e.g. token-scoped callers). An empty list matches no teams.
 
         Returns:
             Union[Tuple[List[EmailTeam], Optional[str]], Dict[str, Any]]:
@@ -1887,6 +1889,11 @@ async def list_teams(
             search_description=True,
         )
 
+        # Restrict to specific team IDs before pagination so token-scoped callers
+        # are filtered in the query rather than on an already-limited page.
+        if team_ids is not None:
+            query = query.where(EmailTeam.id.in_(team_ids))
+
         # Choose ordering based on pagination mode:
         # - Page-based (UI): alphabetical by name for user-friendly display
         # - Cursor-based (API): created_at DESC, id DESC to match unified_paginate expectations

tests/integration/test_search_endpoint.py

@@ -287,15 +287,17 @@ def _real_data_env():
     now = datetime.now(timezone.utc)
     team_a = uuid.uuid4().hex
     team_b = uuid.uuid4().hex
+    team_c = uuid.uuid4().hex  # exists but user_b is NOT a member; only visible via the admin all-teams view
     # Unique emails per invocation: get_user_teams and role/permission lookups are
     # cached by email at process scope, so a fixed email would leak one test's
     # temp-DB team ids into another test using the same user.
     suffix = uuid.uuid4().hex
     user_b = f"user-b-{suffix}@example.com"
+    admin_user = f"admin-{suffix}@example.com"
     owner = f"owner-{suffix}@example.com"  # owns the tools so user_b access is never via ownership
 
-    def _user(email):
-        return EmailUser(id=uuid.uuid4().hex, email=email, password_hash="x", full_name=email, is_admin=False, is_active=True, auth_provider="local", email_verified_at=now)  # pragma: allowlist secret
+    def _user(email, is_admin=False):
+        return EmailUser(id=uuid.uuid4().hex, email=email, password_hash="x", full_name=email, is_admin=is_admin, is_active=True, auth_provider="local", email_verified_at=now)  # pragma: allowlist secret
 
     def _tool(name, visibility, team_id):
         return Tool(
@@ -316,24 +318,27 @@ def _tool(name, visibility, team_id):
         )
 
     db = TestSessionLocal()
-    db.add_all([_user(user_b), _user(owner)])
+    db.add_all([_user(user_b), _user(owner), _user(admin_user, is_admin=True)])
     db.add_all(
         [
             EmailTeam(id=team_a, name=f"{SEARCH_TERM} Team A", slug="team-a", created_by=owner, is_personal=False, visibility="public"),
             EmailTeam(id=team_b, name=f"{SEARCH_TERM} Team B", slug="team-b", created_by=owner, is_personal=False, visibility="public"),
+            EmailTeam(id=team_c, name=f"{SEARCH_TERM} Team C", slug="team-c", created_by=owner, is_personal=False, visibility="public"),
         ]
     )
     db.commit()
 
-    # user_b is a real member of BOTH teams (so team-A is genuinely accessible),
-    # and holds a real GLOBAL tools.read role (Layer 2 pass-through).
+    # user_b is a real member of teams A and B (so team-A is genuinely accessible),
+    # and holds a real GLOBAL tools.read+teams.read role (Layer 2 pass-through).
+    # admin_user is a DB admin holding the same role (teams.read has no admin bypass).
     role_id = uuid.uuid4().hex
     db.add_all(
         [
             EmailTeamMember(id=uuid.uuid4().hex, team_id=team_a, user_email=user_b, role="member", is_active=True),
             EmailTeamMember(id=uuid.uuid4().hex, team_id=team_b, user_email=user_b, role="member", is_active=True),
             Role(id=role_id, name="test-reader", scope="global", permissions=["tools.read", "teams.read"], created_by=owner, is_active=True),
             UserRole(id=uuid.uuid4().hex, user_email=user_b, role_id=role_id, scope="global", scope_id=None, granted_by=owner, is_active=True),
+            UserRole(id=uuid.uuid4().hex, user_email=admin_user, role_id=role_id, scope="global", scope_id=None, granted_by=owner, is_active=True),
         ]
     )
 
@@ -353,7 +358,9 @@ def _tool(name, visibility, team_id):
         "server_public": s_public.id,
         "team_a": team_a,
         "team_b": team_b,
+        "team_c": team_c,
         "user_b": user_b,
+        "admin_user": admin_user,
     }
     db.close()
 
@@ -368,22 +375,24 @@ def _tool(name, visibility, team_id):
     os.unlink(path)
 
 
-def _inject_identity(app, TestSessionLocal, email, token_teams=None):
-    """Override the auth dependency to yield a non-admin context, optionally token-scoped.
+def _inject_identity(app, TestSessionLocal, email, token_teams=None, is_admin=False):
+    """Override the auth dependency to yield a caller context, optionally token-scoped.
 
     Args:
         app: The FastAPI app to override on.
         TestSessionLocal: Session factory bound to the temp DB.
         email (str): Caller email.
         token_teams: When provided, narrows the caller's visible team scope
             (list of team-id strings); when ``None``, the key is omitted so the
-            caller's full DB team membership applies.
+            caller's full DB team membership applies (non-admin) or full admin
+            bypass applies (admin).
+        is_admin (bool): Whether the context is flagged admin.
     """
 
     async def _ctx():
         session = TestSessionLocal()
         try:
-            context = {"email": email, "is_admin": False, "ip_address": "127.0.0.1", "user_agent": "test-client", "db": session}
+            context = {"email": email, "is_admin": is_admin, "ip_address": "127.0.0.1", "user_agent": "test-client", "db": session}
             if token_teams is not None:
                 context["token_teams"] = token_teams
             yield context
@@ -503,3 +512,66 @@ def test_public_only_token_sees_only_public_tools(self, _real_data_env):
         assert ids["public"] in returned  # public tool visible
         assert ids["teama"] not in returned  # team-private hidden under public-only scope
         assert ids["teamb"] not in returned  # team-private hidden under public-only scope
+
+    def test_admin_unscoped_token_sees_all_teams(self, _real_data_env):
+        """An admin with no token narrowing sees every team, including one they don't belong to."""
+        app, TestSessionLocal, ids = _real_data_env
+        _inject_identity(app, TestSessionLocal, ids["admin_user"], token_teams=None, is_admin=True)
+
+        client = TestClient(app, raise_server_exceptions=False)
+        resp = client.get(f"/v1/search?q={SEARCH_TERM}&entity_types=teams", headers={"Authorization": "Bearer x"})
+
+        assert resp.status_code == 200
+        returned = {team["id"] for team in resp.json()["results"]["teams"]}
+        # Admin all-teams view: team-C is returned even though admin is not a member.
+        assert {ids["team_a"], ids["team_b"], ids["team_c"]} <= returned
+
+    def test_admin_scoped_token_narrows_teams(self, _real_data_env):
+        """An admin with a token scoped to team-B sees only team-B, not the other teams.
+
+        Explicit token_teams constrains Layer-1 visibility regardless of admin status;
+        None (full bypass) is the only unrestricted case.
+        """
+        app, TestSessionLocal, ids = _real_data_env
+        _inject_identity(app, TestSessionLocal, ids["admin_user"], token_teams=[ids["team_b"]], is_admin=True)
+
+        client = TestClient(app, raise_server_exceptions=False)
+        resp = client.get(f"/v1/search?q={SEARCH_TERM}&entity_types=teams", headers={"Authorization": "Bearer x"})
+
+        assert resp.status_code == 200
+        returned = {team["id"] for team in resp.json()["results"]["teams"]}
+        assert ids["team_b"] in returned  # in-scope team visible (positive control)
+        assert ids["team_a"] not in returned  # narrowed out
+        assert ids["team_c"] not in returned  # narrowed out
+
+    def test_admin_public_only_token_sees_no_teams(self, _real_data_env):
+        """A public-only admin token (token_teams=[]) sees no teams, not a full-bypass all-teams view.
+
+        [] is public-only per the token model (normalize_token_teams), and there is no
+        personal-team fallback, so an admin scoped to [] must see zero teams.
+        """
+        app, TestSessionLocal, ids = _real_data_env
+        _inject_identity(app, TestSessionLocal, ids["admin_user"], token_teams=[], is_admin=True)
+
+        client = TestClient(app, raise_server_exceptions=False)
+        resp = client.get(f"/v1/search?q={SEARCH_TERM}&entity_types=teams", headers={"Authorization": "Bearer x"})
+
+        assert resp.status_code == 200
+        assert resp.json()["results"]["teams"] == []  # empty scope -> no teams (not bypass)
+
+    def test_admin_scoped_team_returned_even_when_it_sorts_past_the_limit(self, _real_data_env):
+        """Scope is applied in the query before pagination, so a scoped team that sorts past the page limit is still returned.
+
+        Teams are ordered by name (Team A < B < C). With limit_per_type=1 the admin
+        page holds only Team A; a filter-after-pagination approach would drop the
+        scoped Team C entirely.
+        """
+        app, TestSessionLocal, ids = _real_data_env
+        _inject_identity(app, TestSessionLocal, ids["admin_user"], token_teams=[ids["team_c"]], is_admin=True)
+
+        client = TestClient(app, raise_server_exceptions=False)
+        resp = client.get(f"/v1/search?q={SEARCH_TERM}&entity_types=teams&limit_per_type=1", headers={"Authorization": "Bearer x"})
+
+        assert resp.status_code == 200
+        returned = {team["id"] for team in resp.json()["results"]["teams"]}
+        assert ids["team_c"] in returned  # returned despite sorting past the 1-item page

tests/unit/mcpgateway/services/test_team_management_service.py

@@ -1063,6 +1063,34 @@ async def test_list_teams_personal_owner_email(self):
         assert "shared-team" in names
         assert "other-personal" not in names
 
+    @pytest.mark.asyncio
+    async def test_list_teams_team_ids_filters_before_pagination(self):
+        """team_ids restricts results in the query; None applies no filter."""
+        from sqlalchemy import create_engine
+        from sqlalchemy.orm import Session as OrmSession
+
+        from mcpgateway.db import Base
+
+        engine = create_engine("sqlite:///:memory:", connect_args={"check_same_thread": False})
+        Base.metadata.create_all(engine)
+        with OrmSession(engine) as db:
+            db.add_all(
+                [
+                    EmailTeam(id="id-a", name="Alpha", slug="alpha-x", created_by="o@example.com", is_personal=False),
+                    EmailTeam(id="id-b", name="Beta", slug="beta-x", created_by="o@example.com", is_personal=False),
+                ]
+            )
+            db.commit()
+
+            svc = TeamManagementService(db)
+            scoped, _ = await svc.list_teams(team_ids=["id-b"])
+            scoped_names = {t.name for t in scoped}
+            unfiltered, _ = await svc.list_teams(team_ids=None)
+            unfiltered_names = {t.name for t in unfiltered}
+
+        assert scoped_names == {"Beta"}  # restricted to the given id
+        assert {"Alpha", "Beta"} <= unfiltered_names  # None = no filter
+
     @pytest.mark.asyncio
     async def test_list_teams_with_search_query_page(self, service, mock_db):
         """Test list_teams applies search_query and page-based ordering."""

tests/unit/mcpgateway/test_admin.py

@@ -19846,6 +19846,31 @@ async def test_admin_search_teams_admin(self, monkeypatch, allow_permission, moc
         assert len(result) == 1
         assert result[0]["name"] == "Alpha"
 
+    @pytest.mark.asyncio
+    async def test_admin_search_teams_admin_scoped_token_forwards_team_ids(self, monkeypatch, allow_permission, mock_db):
+        """A scoped admin token forwards normalized team_ids to list_teams (dict/str extraction)."""
+        mock_auth = MagicMock()
+        admin_user = SimpleNamespace(is_admin=True)
+        mock_auth.get_user_by_email = AsyncMock(return_value=admin_user)
+        monkeypatch.setattr("mcpgateway.admin.EmailAuthService", lambda db: mock_auth)
+
+        team = SimpleNamespace(id="tb", name="Beta", slug="beta", description="", visibility="public", is_active=True)
+        ts = MagicMock()
+        ts.list_teams = AsyncMock(return_value={"data": [team]})
+        monkeypatch.setattr("mcpgateway.admin.TeamManagementService", lambda db: ts)
+
+        # token_teams mixes a raw str id and a dict id to exercise both extraction branches
+        result = await admin_search_teams(
+            q="Beta",
+            include_inactive=False,
+            limit=10,
+            visibility=None,
+            db=mock_db,
+            user={"email": "admin@test.com", "token_teams": ["tb", {"id": "tc"}]},
+        )
+        assert result[0]["id"] == "tb"
+        assert ts.list_teams.await_args.kwargs["team_ids"] == ["tb", "tc"]
+
     @pytest.mark.asyncio
     async def test_admin_search_teams_non_admin_filters(self, monkeypatch, allow_permission, mock_db):
         """Cover visibility and q filter continue branches in admin_search_teams (non-admin)."""