fix(auth): enforce catalog scope permissions

Restrict gateway ownership transfers to the caller's token teams and
require gateways.create for admin catalog registration routes.

Add deny-path regression coverage for scoped transfers and insufficient
gateway permissions.

Signed-off-by: Lang-Akshay <akshay.shinde26@ibm.com>

Author: Lang-Akshay

mcpgateway/admin.py

@@ -13138,13 +13138,15 @@ async def transfer_gateway_ownership(
         Updated GatewayRead with new ownership.
     """
     actor_email = get_user_email(_user)
+    token_teams = extract_token_team_ids(_user)
     try:
         result = await gateway_service.transfer_gateway_ownership(
             db=db,
             gateway_id=gateway_id,
             target_owner_email=transfer.target_owner_email,
             actor_email=actor_email,
             target_team_id=transfer.target_team_id,
+            token_teams=token_teams,
         )
         return result
     except GatewayNotFoundError as e:
@@ -17980,6 +17982,7 @@ async def list_catalog_servers(
 
 @admin_router.post("/mcp-registry/{server_id}/register", response_model=CatalogServerRegisterResponse)
 @require_permission("servers.create", allow_admin_bypass=False)
+@require_permission("gateways.create", allow_admin_bypass=False)
 async def register_catalog_server(
     server_id: str,
     http_request: Request,
@@ -18119,6 +18122,7 @@ async def check_catalog_server_status(
 
 @admin_router.post("/mcp-registry/bulk-register", response_model=CatalogBulkRegisterResponse)
 @require_permission("servers.create", allow_admin_bypass=False)
+@require_permission("gateways.create", allow_admin_bypass=False)
 async def bulk_register_catalog_servers(
     http_request: Request,
     request: CatalogBulkRegisterRequest,

mcpgateway/middleware/token_scoping.py

@@ -208,6 +208,7 @@ class ResourceOwnershipResult(Enum):
     ("POST", re.compile(r"^/admin/gateways/?$"), Permissions.GATEWAYS_CREATE),
     ("POST", re.compile(r"^/admin/gateways/[^/]+/delete(?:$|/)"), Permissions.GATEWAYS_DELETE),
     ("POST", re.compile(r"^/admin/gateways/[^/]+/(?:edit|state)(?:$|/)"), Permissions.GATEWAYS_UPDATE),
+    ("POST", re.compile(r"^/admin/gateways/[^/]+/transfer-ownership(?:$|/)"), Permissions.GATEWAYS_UPDATE),
     ("GET", re.compile(r"^/admin/gateways(?:$|/)"), Permissions.GATEWAYS_READ),
     # Server management
     ("POST", re.compile(r"^/admin/servers/?$"), Permissions.SERVERS_CREATE),

mcpgateway/services/gateway_service.py

@@ -5529,6 +5529,7 @@ async def transfer_gateway_ownership(
         target_owner_email: str,
         actor_email: str,
         target_team_id: Optional[str] = None,
+        token_teams: Optional[List[str]] = None,
     ) -> GatewayRead:
         """Transfer gateway ownership to another user.
 
@@ -5538,13 +5539,14 @@ async def transfer_gateway_ownership(
             target_owner_email: Email of the new owner
             actor_email: Email of the user performing the transfer
             target_team_id: Optional new team ID for the gateway
+            token_teams: Normalized caller team scope; None means unrestricted admin scope
 
         Returns:
             GatewayRead with updated ownership
 
         Raises:
             GatewayNotFoundError: Gateway does not exist
-            ValueError: Target user invalid or team membership check fails
+            ValueError: Target user invalid, team membership check fails, or gateway is outside scope
         """
         # Validate target user exists and is active
         target_user = db.execute(select(DbEmailUser).where(DbEmailUser.email == target_owner_email, DbEmailUser.is_active == True)).scalar_one_or_none()  # noqa: E712  # pylint: disable=singleton-comparison
@@ -5554,6 +5556,12 @@ async def transfer_gateway_ownership(
         gateway = get_for_update(db, DbGateway, gateway_id)
         if not gateway:
             raise GatewayNotFoundError(f"Gateway not found: {gateway_id}")
+        if token_teams is not None:
+            current_team_id = gateway.team_id
+            if not current_team_id or current_team_id not in token_teams:
+                raise ValueError("Gateway is outside the caller's token team scope")
+            if target_team_id and target_team_id not in token_teams:
+                raise ValueError("Target team is outside the caller's token team scope")
 
         previous_owner = gateway.owner_email
         previous_team = gateway.team_id

tests/unit/mcpgateway/middleware/test_token_scoping.py

@@ -795,6 +795,12 @@ def test_permission_restrictions_llm_proxy_default_prefix(self, middleware):
         assert middleware._check_permission_restrictions("/v1/chat/completions", "POST", [Permissions.LLM_INVOKE]) is True
         assert middleware._check_permission_restrictions("/v1/chat/completions", "POST", [Permissions.LLM_READ]) is False
 
+    def test_transfer_ownership_requires_gateway_update_permission(self, middleware):
+        """Ownership transfer must be covered by the gateway update token scope."""
+        path = "/admin/gateways/gw-1/transfer-ownership"
+        assert middleware._check_permission_restrictions(path, "POST", [Permissions.GATEWAYS_UPDATE]) is True
+        assert middleware._check_permission_restrictions(path, "POST", [Permissions.GATEWAYS_READ]) is False
+
     def test_permission_restrictions_llm_proxy_custom_prefix(self, middleware, monkeypatch):
         """LLM proxy path mapping should follow settings.llm_api_prefix."""
         monkeypatch.setattr("mcpgateway.middleware.token_scoping.settings.llm_api_prefix", "/gateway-llm")

tests/unit/mcpgateway/services/test_gateway_service_transfer.py

@@ -249,3 +249,26 @@ async def test_transfer_private_gateway_to_team_coerces_visibility(service, mock
     assert tool.visibility == "team"
     assert resource.visibility == "team"
     assert prompt.visibility == "team"
+
+
+@pytest.mark.asyncio
+async def test_transfer_scoped_token_cannot_mutate_gateway_outside_scope(service, mock_db):
+    """Scoped administrators cannot transfer gateways outside their token teams."""
+    gateway = MagicMock()
+    gateway.visibility = "public"
+    gateway.team_id = "foreign-team"
+    gateway.owner_email = "old@example.com"
+
+    mock_db.execute.return_value = _scalar_result(MagicMock())  # target user found
+
+    with patch("mcpgateway.services.gateway_service.get_for_update", return_value=gateway):
+        with pytest.raises(ValueError, match="outside.*scope"):
+            await service.transfer_gateway_ownership(
+                mock_db,
+                "gw-1",
+                "target@example.com",
+                "actor@example.com",
+                token_teams=["allowed-team"],
+            )
+
+    mock_db.commit.assert_not_called()

tests/unit/mcpgateway/test_admin.py

@@ -26702,6 +26702,32 @@ async def test_transfer_gateway_ownership_success(self, monkeypatch, allow_permi
         )
         assert result is expected
 
+    @pytest.mark.asyncio
+    async def test_transfer_gateway_ownership_passes_token_team_scope(self, monkeypatch, allow_permission, mock_db):
+        """Transfer route forwards the normalized token-team scope to the service."""
+        expected = MagicMock(spec=GatewayRead)
+        transfer_service = AsyncMock(return_value=expected)
+        monkeypatch.setattr("mcpgateway.admin.gateway_service.transfer_gateway_ownership", transfer_service)
+        monkeypatch.setattr("mcpgateway.admin.get_user_email", MagicMock(return_value="admin@x.com"))
+
+        result = await transfer_gateway_ownership(
+            gateway_id="gw-1",
+            transfer=GatewayOwnershipTransferRequest(target_owner_email="new@x.com", target_team_id="team-1"),
+            db=mock_db,
+            _user={"email": "admin@x.com", "is_admin": True, "token_teams": ["team-1"]},
+        )
+
+        assert result is expected
+        transfer_service.assert_awaited_once_with(
+            db=mock_db,
+            gateway_id="gw-1",
+            target_owner_email="new@x.com",
+            actor_email="admin@x.com",
+            target_team_id="team-1",
+            token_teams=["team-1"],
+        )
+
+
     @pytest.mark.asyncio
     async def test_transfer_gateway_ownership_not_found(self, monkeypatch, allow_permission, mock_db):
         monkeypatch.setattr(
@@ -26791,6 +26817,43 @@ async def test_admin_register_catalog_permission_error(self, monkeypatch, allow_
             await register_catalog_server("srv-1", request, db=mock_db, _user={"email": "admin@test.com"})
         assert exc_info.value.status_code == 403
 
+
+    @pytest.mark.asyncio
+    async def test_admin_register_catalog_requires_gateways_create(self, monkeypatch, allow_permission, mock_db):
+        """Catalog registration must require gateway creation permission too."""
+        monkeypatch.setattr("mcpgateway.admin.settings.mcpgateway_catalog_enabled", True, raising=False)
+        allow_permission.check_permission = AsyncMock(side_effect=[True, False])
+        monkeypatch.setattr(
+            "mcpgateway.admin.catalog_service.register_catalog_server",
+            AsyncMock(return_value=MagicMock()),
+        )
+
+        request = MagicMock(spec=Request)
+        request.headers = {}
+        with pytest.raises(HTTPException) as exc_info:
+            await register_catalog_server("srv-1", request, db=mock_db, _user={"email": "admin@test.com"})
+
+        assert exc_info.value.status_code == 403
+
+    @pytest.mark.asyncio
+    async def test_admin_bulk_register_catalog_requires_gateways_create(self, monkeypatch, allow_permission, mock_db):
+        """Bulk catalog registration must require gateway creation permission too."""
+        monkeypatch.setattr("mcpgateway.admin.settings.mcpgateway_catalog_enabled", True, raising=False)
+        allow_permission.check_permission = AsyncMock(side_effect=[True, False])
+
+        from mcpgateway.schemas import CatalogBulkRegisterRequest
+
+        request = MagicMock(spec=Request)
+        with pytest.raises(HTTPException) as exc_info:
+            await bulk_register_catalog_servers(
+                request,
+                CatalogBulkRegisterRequest(server_ids=["srv-1"]),
+                db=mock_db,
+                _user={"email": "admin@test.com"},
+            )
+
+        assert exc_info.value.status_code == 403
+
     @pytest.mark.asyncio
     async def test_admin_bulk_register_catalog_permission_error(self, monkeypatch, allow_permission, mock_db):
         monkeypatch.setattr("mcpgateway.admin.settings.mcpgateway_catalog_enabled", True, raising=False)