aminghadersohi commented on code in PR #44694:
URL: https://github.com/apache/superset/pull/44694#discussion_r4114036362
##########
docs/docs/using-superset/using-ai-with-superset.mdx:
##########
@@ -257,6 +257,7 @@ Ask your admin for the MCP server URL and any
authentication tokens you need.
| `health_check` | Verify the MCP server is running and connected
|
| `get_instance_info` | Get instance statistics (dataset, chart, dashboard
counts) |
| `get_schema` | Discover available charts, datasets, and dashboards
with schema info |
+| `get_catalog` | Compact, permission-filtered page of databases,
datasets, charts, or dashboards (id, uuid, name, description, link) |
Review Comment:
Valid; fixed in 6194b05aebc4ac502879db20b67caeb54b3006bb. The reference now
names `url`, matching CatalogItem. Added a documentation/schema regression test
that failed before the correction and passes after.
##########
superset/mcp_service/catalog/tool/get_catalog.py:
##########
@@ -0,0 +1,411 @@
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements. See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership. The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing,
+# software distributed under the License is distributed on an
+# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+# KIND, either express or implied. See the License for the
+# specific language governing permissions and limitations
+# under the License.
+
+"""Permission-filtered, bounded asset catalog tool.
+
+``get_catalog`` returns one small page of the databases, datasets, charts or
+dashboards the calling user can see, reduced to a fixed field allowlist.
+
+Security properties:
+
+- Every call is a live read made as the calling user. Visibility comes from
+ the same DAO ``base_filter`` used by list_databases, list_datasets,
+ list_charts and list_dashboards, so a revoked grant disappears on the next
+ call. Nothing is cached or persisted (the tool is also force-excluded from
+ response caching, see ``superset.mcp_service.caching``).
+- Class-level RBAC is checked per asset type, mirroring the matching list
+ tool; a denial raises the generic permission error.
+- Users without data-model metadata access (see ``privacy.py``) get an
+ explicit ``restricted`` page for databases and datasets instead of an error.
+- Only id, uuid, name, description, changed_on and url are read from the
+ database and returned; SQL, ``extra``, params, metrics and connection
+ details are never loaded.
+- Output is bounded: at most 100 items and ``CATALOG_MAX_RESPONSE_BYTES`` of
+ JSON per call, with an opaque keyset cursor for the next page.
+"""
+
+import base64
+import binascii
+import hashlib
+import logging
+from dataclasses import dataclass
+from datetime import datetime
+from typing import Any, Callable
+
+from fastmcp import Context
+from superset_core.mcp.decorators import tool, ToolAnnotations
+
+from superset.extensions import event_logger
+from superset.mcp_service.auth import _token_scope_allows,
MCPPermissionDeniedError
+from superset.mcp_service.catalog.schemas import (
+ CATALOG_MAX_DESCRIPTION_LENGTH,
+ CATALOG_MAX_NAME_LENGTH,
+ CATALOG_MAX_RESPONSE_BYTES,
+ CatalogAssetType,
+ CatalogItem,
+ CatalogResponse,
+ GetCatalogRequest,
+)
+from superset.mcp_service.privacy import (
+ DATA_MODEL_METADATA_ERROR_MESSAGE,
+ user_can_view_data_model_metadata,
+)
+from superset.mcp_service.utils.url_utils import get_superset_base_url
+from superset.utils import json
+
+logger = logging.getLogger(__name__)
+
+_CURSOR_VERSION = 1
+
+
+def _dataset_url(row: Any) -> str | None:
+ return (
+ f"{get_superset_base_url()}/explore/"
+ f"?datasource_type=table&datasource_id={row.id}"
+ )
+
+
+def _chart_url(row: Any) -> str | None:
+ return f"{get_superset_base_url()}/explore/?slice_id={row.id}"
+
+
+def _dashboard_url(row: Any) -> str | None:
+ return f"{get_superset_base_url()}/dashboard/{row.slug or row.id}/"
+
+
+def _no_url(row: Any) -> str | None:
+ return None
+
+
+def _database_dao() -> Any:
+ from superset.daos.database import DatabaseDAO
+
+ return DatabaseDAO
+
+
+def _dataset_dao() -> Any:
+ from superset.daos.dataset import DatasetDAO
+
+ return DatasetDAO
+
+
+def _chart_dao() -> Any:
+ from superset.daos.chart import ChartDAO
+
+ return ChartDAO
+
+
+def _dashboard_dao() -> Any:
+ from superset.daos.dashboard import DashboardDAO
+
+ return DashboardDAO
+
+
+@dataclass(frozen=True)
+class _AssetSpec:
+ """How to read one asset type through its DAO."""
+
+ dao: Callable[[], Any]
+ class_permission: str
+ name_column: str
+ description_column: str | None
+ requires_data_model_access: bool
+ url: Callable[[Any], str | None]
+ extra_columns: tuple[str, ...] = ()
+
+ @property
+ def columns(self) -> list[str]:
+ """The only model columns ever loaded (all plain columns)."""
+ columns = ["id", "uuid", self.name_column, "changed_on"]
+ if self.description_column:
+ columns.append(self.description_column)
+ columns.extend(self.extra_columns)
+ return columns
+
+
+_ASSET_SPECS: dict[str, _AssetSpec] = {
+ "databases": _AssetSpec(
+ dao=_database_dao,
+ class_permission="Database",
+ name_column="database_name",
+ description_column=None,
+ requires_data_model_access=True,
+ url=_no_url,
+ ),
+ "datasets": _AssetSpec(
+ dao=_dataset_dao,
+ class_permission="Dataset",
+ name_column="table_name",
+ description_column="description",
+ requires_data_model_access=True,
+ url=_dataset_url,
+ ),
+ "charts": _AssetSpec(
+ dao=_chart_dao,
+ class_permission="Chart",
+ name_column="slice_name",
+ description_column="description",
+ requires_data_model_access=False,
+ url=_chart_url,
+ ),
+ "dashboards": _AssetSpec(
+ dao=_dashboard_dao,
+ class_permission="Dashboard",
+ name_column="dashboard_title",
+ description_column="description",
+ requires_data_model_access=False,
+ url=_dashboard_url,
+ extra_columns=("slug",),
+ ),
+}
+
+
+def _search_digest(search: str | None) -> str:
+ if not search:
+ return ""
+ return hashlib.sha256(search.encode("utf-8")).hexdigest()[:16]
+
+
+def encode_cursor(asset_type: str, search: str | None, after_id: int) -> str:
+ """Encode an opaque keyset cursor bound to the asset type and search.
+
+ The cursor only positions the next page; it carries no authorization.
+ Every page is re-filtered for the calling user.
+ """
+ payload = json.dumps(
+ {
+ "v": _CURSOR_VERSION,
+ "t": asset_type,
+ "s": _search_digest(search),
+ "a": after_id,
+ },
+ separators=(",", ":"),
+ )
+ return
base64.urlsafe_b64encode(payload.encode("utf-8")).decode("ascii").rstrip("=")
+
+
+def decode_cursor(cursor: str, asset_type: str, search: str | None) -> int:
+ """Return the last-seen ID encoded in ``cursor``.
+
+ Raises ValueError for malformed cursors or cursors issued for a different
+ asset type or search.
+ """
+ try:
+ padded = cursor + "=" * (-len(cursor) % 4)
+ payload = json.loads(base64.urlsafe_b64decode(padded.encode("ascii")))
+ except (binascii.Error, UnicodeError, ValueError, TypeError) as ex:
+ raise ValueError("Invalid cursor") from ex
+ if not isinstance(payload, dict):
+ raise ValueError("Invalid cursor")
+ after_id = payload.get("a")
+ if (
+ payload.get("v") != _CURSOR_VERSION
+ or payload.get("t") != asset_type
+ or payload.get("s") != _search_digest(search)
+ or not isinstance(after_id, int)
+ or isinstance(after_id, bool)
+ or after_id < 0
+ ):
+ raise ValueError(
+ "Invalid cursor: it does not match this asset_type and search. "
+ "Restart without a cursor."
+ )
+ return after_id
+
+
+def _check_class_permission(spec: _AssetSpec, asset_type: str) -> None:
+ """Apply the same class-level RBAC as the matching list tool."""
+ from flask import current_app, g
+
+ from superset import security_manager
+
+ rbac_allows = not current_app.config.get(
+ "MCP_RBAC_ENABLED", True
+ ) or security_manager.can_access("can_read", spec.class_permission)
+ if rbac_allows and _token_scope_allows("read", spec.class_permission):
+ return
+ user_str = getattr(getattr(g, "user", None), "username", None)
+ logger.warning(
+ "get_catalog RBAC denied: user=%s type=%s view=%s",
+ user_str,
+ asset_type,
+ spec.class_permission,
+ )
+ raise MCPPermissionDeniedError(
+ permission_name="can_read",
+ view_name=spec.class_permission,
+ user=user_str,
+ tool_name="get_catalog",
+ )
+
+
+def _shorten(value: Any, limit: int) -> tuple[str | None, bool]:
+ if value is None:
+ return None, False
+ text = str(value)
+ if len(text) <= limit:
+ return text, False
+ return text[: limit - 1] + "…", True
+
+
+def _to_item(spec: _AssetSpec, row: Any) -> tuple[CatalogItem, bool]:
+ name, name_cut = _shorten(getattr(row, spec.name_column),
CATALOG_MAX_NAME_LENGTH)
+ description, description_cut = (
+ _shorten(getattr(row, spec.description_column),
CATALOG_MAX_DESCRIPTION_LENGTH)
+ if spec.description_column
+ else (None, False)
+ )
+ changed_on = row.changed_on
+ uuid = row.uuid
+ item = CatalogItem(
+ id=row.id,
+ uuid=str(uuid) if uuid is not None else None,
+ name=name or "",
+ description=description or None,
+ changed_on=changed_on.isoformat() if isinstance(changed_on, datetime)
else None,
+ url=spec.url(row),
+ )
+ return item, name_cut or description_cut
+
+
+def _response_size(response: CatalogResponse) -> int:
+ return len(
+ json.dumps(response.model_dump(mode="json"), separators=(",",
":")).encode(
+ "utf-8"
+ )
+ )
+
+
+def _fetch_rows(
+ spec: _AssetSpec, after_id: int, search: str | None, limit: int
+) -> tuple[list[Any], bool]:
+ """Read one keyset page through the DAO, as the current user.
+
+ ``DAO.list`` always applies the model's ``base_filter`` (the same
+ per-user visibility filter the list tools use) and soft-deleted rows
+ are excluded by the global soft-delete listener.
+ """
+ from superset.daos.base import ColumnOperator
+
+ dao = spec.dao()
+ column_operators = (
+ [ColumnOperator(col="id", opr="gt", value=after_id)] if after_id else
None
+ )
+ rows, total_count = dao.list(
+ column_operators=column_operators,
+ order_column="id",
+ order_direction="asc",
+ page=0,
+ page_size=limit,
+ search=search,
+ search_columns=[spec.name_column],
+ columns=spec.columns,
+ )
Review Comment:
Valid; fixed in 6194b05aebc4ac502879db20b67caeb54b3006bb. _fetch_rows
requests one lookahead row and derives continuation from that page rather than
the separately queried count. The regression inserts an asset after Query.count
and verifies that the cursor reaches it; it failed before and passes after.
##########
superset/mcp_service/catalog/tool/get_catalog.py:
##########
@@ -0,0 +1,411 @@
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements. See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership. The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing,
+# software distributed under the License is distributed on an
+# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+# KIND, either express or implied. See the License for the
+# specific language governing permissions and limitations
+# under the License.
+
+"""Permission-filtered, bounded asset catalog tool.
+
+``get_catalog`` returns one small page of the databases, datasets, charts or
+dashboards the calling user can see, reduced to a fixed field allowlist.
+
+Security properties:
+
+- Every call is a live read made as the calling user. Visibility comes from
+ the same DAO ``base_filter`` used by list_databases, list_datasets,
+ list_charts and list_dashboards, so a revoked grant disappears on the next
+ call. Nothing is cached or persisted (the tool is also force-excluded from
+ response caching, see ``superset.mcp_service.caching``).
+- Class-level RBAC is checked per asset type, mirroring the matching list
+ tool; a denial raises the generic permission error.
+- Users without data-model metadata access (see ``privacy.py``) get an
+ explicit ``restricted`` page for databases and datasets instead of an error.
+- Only id, uuid, name, description, changed_on and url are read from the
+ database and returned; SQL, ``extra``, params, metrics and connection
+ details are never loaded.
+- Output is bounded: at most 100 items and ``CATALOG_MAX_RESPONSE_BYTES`` of
+ JSON per call, with an opaque keyset cursor for the next page.
+"""
+
+import base64
+import binascii
+import hashlib
+import logging
+from dataclasses import dataclass
+from datetime import datetime
+from typing import Any, Callable
+
+from fastmcp import Context
+from superset_core.mcp.decorators import tool, ToolAnnotations
+
+from superset.extensions import event_logger
+from superset.mcp_service.auth import _token_scope_allows,
MCPPermissionDeniedError
+from superset.mcp_service.catalog.schemas import (
+ CATALOG_MAX_DESCRIPTION_LENGTH,
+ CATALOG_MAX_NAME_LENGTH,
+ CATALOG_MAX_RESPONSE_BYTES,
+ CatalogAssetType,
+ CatalogItem,
+ CatalogResponse,
+ GetCatalogRequest,
+)
+from superset.mcp_service.privacy import (
+ DATA_MODEL_METADATA_ERROR_MESSAGE,
+ user_can_view_data_model_metadata,
+)
+from superset.mcp_service.utils.url_utils import get_superset_base_url
+from superset.utils import json
+
+logger = logging.getLogger(__name__)
+
+_CURSOR_VERSION = 1
+
+
+def _dataset_url(row: Any) -> str | None:
+ return (
+ f"{get_superset_base_url()}/explore/"
+ f"?datasource_type=table&datasource_id={row.id}"
+ )
+
+
+def _chart_url(row: Any) -> str | None:
+ return f"{get_superset_base_url()}/explore/?slice_id={row.id}"
+
+
+def _dashboard_url(row: Any) -> str | None:
+ return f"{get_superset_base_url()}/dashboard/{row.slug or row.id}/"
+
+
+def _no_url(row: Any) -> str | None:
+ return None
+
+
+def _database_dao() -> Any:
+ from superset.daos.database import DatabaseDAO
+
+ return DatabaseDAO
+
+
+def _dataset_dao() -> Any:
+ from superset.daos.dataset import DatasetDAO
+
+ return DatasetDAO
+
+
+def _chart_dao() -> Any:
+ from superset.daos.chart import ChartDAO
+
+ return ChartDAO
+
+
+def _dashboard_dao() -> Any:
+ from superset.daos.dashboard import DashboardDAO
+
+ return DashboardDAO
+
+
+@dataclass(frozen=True)
+class _AssetSpec:
+ """How to read one asset type through its DAO."""
+
+ dao: Callable[[], Any]
+ class_permission: str
+ name_column: str
+ description_column: str | None
+ requires_data_model_access: bool
+ url: Callable[[Any], str | None]
+ extra_columns: tuple[str, ...] = ()
+
+ @property
+ def columns(self) -> list[str]:
+ """The only model columns ever loaded (all plain columns)."""
+ columns = ["id", "uuid", self.name_column, "changed_on"]
+ if self.description_column:
+ columns.append(self.description_column)
+ columns.extend(self.extra_columns)
+ return columns
+
+
+_ASSET_SPECS: dict[str, _AssetSpec] = {
+ "databases": _AssetSpec(
+ dao=_database_dao,
+ class_permission="Database",
+ name_column="database_name",
+ description_column=None,
+ requires_data_model_access=True,
+ url=_no_url,
+ ),
+ "datasets": _AssetSpec(
+ dao=_dataset_dao,
+ class_permission="Dataset",
+ name_column="table_name",
+ description_column="description",
+ requires_data_model_access=True,
+ url=_dataset_url,
+ ),
+ "charts": _AssetSpec(
+ dao=_chart_dao,
+ class_permission="Chart",
+ name_column="slice_name",
+ description_column="description",
+ requires_data_model_access=False,
+ url=_chart_url,
+ ),
+ "dashboards": _AssetSpec(
+ dao=_dashboard_dao,
+ class_permission="Dashboard",
+ name_column="dashboard_title",
+ description_column="description",
+ requires_data_model_access=False,
+ url=_dashboard_url,
+ extra_columns=("slug",),
+ ),
+}
+
+
+def _search_digest(search: str | None) -> str:
+ if not search:
+ return ""
+ return hashlib.sha256(search.encode("utf-8")).hexdigest()[:16]
+
+
+def encode_cursor(asset_type: str, search: str | None, after_id: int) -> str:
+ """Encode an opaque keyset cursor bound to the asset type and search.
+
+ The cursor only positions the next page; it carries no authorization.
+ Every page is re-filtered for the calling user.
+ """
+ payload = json.dumps(
+ {
+ "v": _CURSOR_VERSION,
+ "t": asset_type,
+ "s": _search_digest(search),
+ "a": after_id,
+ },
+ separators=(",", ":"),
+ )
+ return
base64.urlsafe_b64encode(payload.encode("utf-8")).decode("ascii").rstrip("=")
+
+
+def decode_cursor(cursor: str, asset_type: str, search: str | None) -> int:
+ """Return the last-seen ID encoded in ``cursor``.
+
+ Raises ValueError for malformed cursors or cursors issued for a different
+ asset type or search.
+ """
+ try:
+ padded = cursor + "=" * (-len(cursor) % 4)
+ payload = json.loads(base64.urlsafe_b64decode(padded.encode("ascii")))
+ except (binascii.Error, UnicodeError, ValueError, TypeError) as ex:
+ raise ValueError("Invalid cursor") from ex
+ if not isinstance(payload, dict):
+ raise ValueError("Invalid cursor")
+ after_id = payload.get("a")
+ if (
+ payload.get("v") != _CURSOR_VERSION
+ or payload.get("t") != asset_type
+ or payload.get("s") != _search_digest(search)
+ or not isinstance(after_id, int)
+ or isinstance(after_id, bool)
+ or after_id < 0
+ ):
+ raise ValueError(
+ "Invalid cursor: it does not match this asset_type and search. "
+ "Restart without a cursor."
+ )
+ return after_id
+
+
+def _check_class_permission(spec: _AssetSpec, asset_type: str) -> None:
+ """Apply the same class-level RBAC as the matching list tool."""
+ from flask import current_app, g
+
+ from superset import security_manager
+
+ rbac_allows = not current_app.config.get(
+ "MCP_RBAC_ENABLED", True
+ ) or security_manager.can_access("can_read", spec.class_permission)
+ if rbac_allows and _token_scope_allows("read", spec.class_permission):
+ return
+ user_str = getattr(getattr(g, "user", None), "username", None)
+ logger.warning(
+ "get_catalog RBAC denied: user=%s type=%s view=%s",
+ user_str,
+ asset_type,
+ spec.class_permission,
+ )
+ raise MCPPermissionDeniedError(
+ permission_name="can_read",
+ view_name=spec.class_permission,
+ user=user_str,
+ tool_name="get_catalog",
+ )
+
+
+def _shorten(value: Any, limit: int) -> tuple[str | None, bool]:
+ if value is None:
+ return None, False
+ text = str(value)
+ if len(text) <= limit:
+ return text, False
+ return text[: limit - 1] + "…", True
+
+
+def _to_item(spec: _AssetSpec, row: Any) -> tuple[CatalogItem, bool]:
+ name, name_cut = _shorten(getattr(row, spec.name_column),
CATALOG_MAX_NAME_LENGTH)
+ description, description_cut = (
+ _shorten(getattr(row, spec.description_column),
CATALOG_MAX_DESCRIPTION_LENGTH)
+ if spec.description_column
+ else (None, False)
+ )
+ changed_on = row.changed_on
+ uuid = row.uuid
+ item = CatalogItem(
+ id=row.id,
+ uuid=str(uuid) if uuid is not None else None,
+ name=name or "",
+ description=description or None,
+ changed_on=changed_on.isoformat() if isinstance(changed_on, datetime)
else None,
+ url=spec.url(row),
+ )
+ return item, name_cut or description_cut
+
+
+def _response_size(response: CatalogResponse) -> int:
+ return len(
+ json.dumps(response.model_dump(mode="json"), separators=(",",
":")).encode(
+ "utf-8"
+ )
+ )
+
+
+def _fetch_rows(
+ spec: _AssetSpec, after_id: int, search: str | None, limit: int
+) -> tuple[list[Any], bool]:
+ """Read one keyset page through the DAO, as the current user.
+
+ ``DAO.list`` always applies the model's ``base_filter`` (the same
+ per-user visibility filter the list tools use) and soft-deleted rows
+ are excluded by the global soft-delete listener.
+ """
+ from superset.daos.base import ColumnOperator
+
+ dao = spec.dao()
+ column_operators = (
+ [ColumnOperator(col="id", opr="gt", value=after_id)] if after_id else
None
+ )
+ rows, total_count = dao.list(
+ column_operators=column_operators,
+ order_column="id",
+ order_direction="asc",
+ page=0,
+ page_size=limit,
+ search=search,
+ search_columns=[spec.name_column],
+ columns=spec.columns,
+ )
+ return list(rows), total_count > len(rows)
+
+
+def build_catalog_page(request: GetCatalogRequest) -> CatalogResponse:
+ """Build one catalog page for the current user (``g.user``)."""
+ asset_type: CatalogAssetType = request.asset_type
+ spec = _ASSET_SPECS[asset_type]
+
+ _check_class_permission(spec, asset_type)
+
+ if spec.requires_data_model_access and not
user_can_view_data_model_metadata():
+ return CatalogResponse(
+ asset_type=asset_type,
+ restricted=True,
+ message=DATA_MODEL_METADATA_ERROR_MESSAGE,
+ )
+
+ after_id = (
+ decode_cursor(request.cursor, asset_type, request.search)
+ if request.cursor
+ else 0
+ )
+
+ rows, has_more = _fetch_rows(spec, after_id, request.search,
request.page_size)
+
+ items: list[CatalogItem] = []
+ truncated = False
+ for row in rows:
+ item, shortened = _to_item(spec, row)
+ items.append(item)
+ truncated = truncated or shortened
+
+ def _page(page_items: list[CatalogItem], more: bool) -> CatalogResponse:
+ return CatalogResponse(
+ asset_type=asset_type,
+ items=page_items,
+ next_cursor=encode_cursor(asset_type, request.search,
page_items[-1].id)
+ if more and page_items
+ else None,
+ truncated=truncated,
+ )
+
+ response = _page(items, has_more)
+ # Enforce the byte bound by deferring trailing items to the next page.
+ # Per-item text caps guarantee a single item always fits.
+ while _response_size(response) > CATALOG_MAX_RESPONSE_BYTES and len(items)
> 1:
+ items = items[:-1]
+ truncated = True
+ response = _page(items, True)
+ return response
Review Comment:
Valid; fixed in 6194b05aebc4ac502879db20b67caeb54b3006bb. URLs exceeding
2,048 characters are omitted (null), not cut into broken links, and mark the
page truncated. This keeps a single item within the byte budget. ASCII and
Unicode oversized-URL regressions failed before and pass after, including
continuation to the next asset.
##########
superset/mcp_service/catalog/tool/get_catalog.py:
##########
@@ -0,0 +1,411 @@
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements. See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership. The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing,
+# software distributed under the License is distributed on an
+# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+# KIND, either express or implied. See the License for the
+# specific language governing permissions and limitations
+# under the License.
+
+"""Permission-filtered, bounded asset catalog tool.
+
+``get_catalog`` returns one small page of the databases, datasets, charts or
+dashboards the calling user can see, reduced to a fixed field allowlist.
+
+Security properties:
+
+- Every call is a live read made as the calling user. Visibility comes from
+ the same DAO ``base_filter`` used by list_databases, list_datasets,
+ list_charts and list_dashboards, so a revoked grant disappears on the next
+ call. Nothing is cached or persisted (the tool is also force-excluded from
+ response caching, see ``superset.mcp_service.caching``).
+- Class-level RBAC is checked per asset type, mirroring the matching list
+ tool; a denial raises the generic permission error.
+- Users without data-model metadata access (see ``privacy.py``) get an
+ explicit ``restricted`` page for databases and datasets instead of an error.
+- Only id, uuid, name, description, changed_on and url are read from the
+ database and returned; SQL, ``extra``, params, metrics and connection
+ details are never loaded.
+- Output is bounded: at most 100 items and ``CATALOG_MAX_RESPONSE_BYTES`` of
+ JSON per call, with an opaque keyset cursor for the next page.
+"""
+
+import base64
+import binascii
+import hashlib
+import logging
+from dataclasses import dataclass
+from datetime import datetime
+from typing import Any, Callable
+
+from fastmcp import Context
+from superset_core.mcp.decorators import tool, ToolAnnotations
+
+from superset.extensions import event_logger
+from superset.mcp_service.auth import _token_scope_allows,
MCPPermissionDeniedError
+from superset.mcp_service.catalog.schemas import (
+ CATALOG_MAX_DESCRIPTION_LENGTH,
+ CATALOG_MAX_NAME_LENGTH,
+ CATALOG_MAX_RESPONSE_BYTES,
+ CatalogAssetType,
+ CatalogItem,
+ CatalogResponse,
+ GetCatalogRequest,
+)
+from superset.mcp_service.privacy import (
+ DATA_MODEL_METADATA_ERROR_MESSAGE,
+ user_can_view_data_model_metadata,
+)
+from superset.mcp_service.utils.url_utils import get_superset_base_url
+from superset.utils import json
+
+logger = logging.getLogger(__name__)
+
+_CURSOR_VERSION = 1
+
+
+def _dataset_url(row: Any) -> str | None:
+ return (
+ f"{get_superset_base_url()}/explore/"
+ f"?datasource_type=table&datasource_id={row.id}"
+ )
+
+
+def _chart_url(row: Any) -> str | None:
+ return f"{get_superset_base_url()}/explore/?slice_id={row.id}"
+
+
+def _dashboard_url(row: Any) -> str | None:
+ return f"{get_superset_base_url()}/dashboard/{row.slug or row.id}/"
+
+
+def _no_url(row: Any) -> str | None:
+ return None
+
+
+def _database_dao() -> Any:
+ from superset.daos.database import DatabaseDAO
+
+ return DatabaseDAO
+
+
+def _dataset_dao() -> Any:
+ from superset.daos.dataset import DatasetDAO
+
+ return DatasetDAO
+
+
+def _chart_dao() -> Any:
+ from superset.daos.chart import ChartDAO
+
+ return ChartDAO
+
+
+def _dashboard_dao() -> Any:
+ from superset.daos.dashboard import DashboardDAO
+
+ return DashboardDAO
+
+
+@dataclass(frozen=True)
+class _AssetSpec:
+ """How to read one asset type through its DAO."""
+
+ dao: Callable[[], Any]
+ class_permission: str
+ name_column: str
+ description_column: str | None
+ requires_data_model_access: bool
+ url: Callable[[Any], str | None]
+ extra_columns: tuple[str, ...] = ()
+
+ @property
+ def columns(self) -> list[str]:
+ """The only model columns ever loaded (all plain columns)."""
+ columns = ["id", "uuid", self.name_column, "changed_on"]
+ if self.description_column:
+ columns.append(self.description_column)
+ columns.extend(self.extra_columns)
+ return columns
+
+
+_ASSET_SPECS: dict[str, _AssetSpec] = {
+ "databases": _AssetSpec(
+ dao=_database_dao,
+ class_permission="Database",
+ name_column="database_name",
+ description_column=None,
+ requires_data_model_access=True,
+ url=_no_url,
+ ),
+ "datasets": _AssetSpec(
+ dao=_dataset_dao,
+ class_permission="Dataset",
+ name_column="table_name",
+ description_column="description",
+ requires_data_model_access=True,
+ url=_dataset_url,
+ ),
+ "charts": _AssetSpec(
+ dao=_chart_dao,
+ class_permission="Chart",
+ name_column="slice_name",
+ description_column="description",
+ requires_data_model_access=False,
+ url=_chart_url,
+ ),
+ "dashboards": _AssetSpec(
+ dao=_dashboard_dao,
+ class_permission="Dashboard",
+ name_column="dashboard_title",
+ description_column="description",
+ requires_data_model_access=False,
+ url=_dashboard_url,
+ extra_columns=("slug",),
+ ),
+}
+
+
+def _search_digest(search: str | None) -> str:
+ if not search:
+ return ""
+ return hashlib.sha256(search.encode("utf-8")).hexdigest()[:16]
+
+
+def encode_cursor(asset_type: str, search: str | None, after_id: int) -> str:
+ """Encode an opaque keyset cursor bound to the asset type and search.
+
+ The cursor only positions the next page; it carries no authorization.
+ Every page is re-filtered for the calling user.
+ """
+ payload = json.dumps(
+ {
+ "v": _CURSOR_VERSION,
+ "t": asset_type,
+ "s": _search_digest(search),
+ "a": after_id,
+ },
+ separators=(",", ":"),
+ )
+ return
base64.urlsafe_b64encode(payload.encode("utf-8")).decode("ascii").rstrip("=")
+
+
+def decode_cursor(cursor: str, asset_type: str, search: str | None) -> int:
+ """Return the last-seen ID encoded in ``cursor``.
+
+ Raises ValueError for malformed cursors or cursors issued for a different
+ asset type or search.
+ """
+ try:
+ padded = cursor + "=" * (-len(cursor) % 4)
+ payload = json.loads(base64.urlsafe_b64decode(padded.encode("ascii")))
+ except (binascii.Error, UnicodeError, ValueError, TypeError) as ex:
+ raise ValueError("Invalid cursor") from ex
+ if not isinstance(payload, dict):
+ raise ValueError("Invalid cursor")
+ after_id = payload.get("a")
+ if (
+ payload.get("v") != _CURSOR_VERSION
+ or payload.get("t") != asset_type
+ or payload.get("s") != _search_digest(search)
+ or not isinstance(after_id, int)
+ or isinstance(after_id, bool)
+ or after_id < 0
+ ):
+ raise ValueError(
+ "Invalid cursor: it does not match this asset_type and search. "
+ "Restart without a cursor."
+ )
+ return after_id
+
+
+def _check_class_permission(spec: _AssetSpec, asset_type: str) -> None:
+ """Apply the same class-level RBAC as the matching list tool."""
+ from flask import current_app, g
+
+ from superset import security_manager
+
+ rbac_allows = not current_app.config.get(
+ "MCP_RBAC_ENABLED", True
+ ) or security_manager.can_access("can_read", spec.class_permission)
+ if rbac_allows and _token_scope_allows("read", spec.class_permission):
+ return
+ user_str = getattr(getattr(g, "user", None), "username", None)
+ logger.warning(
+ "get_catalog RBAC denied: user=%s type=%s view=%s",
+ user_str,
+ asset_type,
+ spec.class_permission,
+ )
+ raise MCPPermissionDeniedError(
+ permission_name="can_read",
+ view_name=spec.class_permission,
+ user=user_str,
+ tool_name="get_catalog",
+ )
+
+
+def _shorten(value: Any, limit: int) -> tuple[str | None, bool]:
+ if value is None:
+ return None, False
+ text = str(value)
+ if len(text) <= limit:
+ return text, False
+ return text[: limit - 1] + "…", True
+
+
+def _to_item(spec: _AssetSpec, row: Any) -> tuple[CatalogItem, bool]:
+ name, name_cut = _shorten(getattr(row, spec.name_column),
CATALOG_MAX_NAME_LENGTH)
+ description, description_cut = (
+ _shorten(getattr(row, spec.description_column),
CATALOG_MAX_DESCRIPTION_LENGTH)
+ if spec.description_column
+ else (None, False)
+ )
+ changed_on = row.changed_on
+ uuid = row.uuid
+ item = CatalogItem(
+ id=row.id,
+ uuid=str(uuid) if uuid is not None else None,
+ name=name or "",
+ description=description or None,
+ changed_on=changed_on.isoformat() if isinstance(changed_on, datetime)
else None,
+ url=spec.url(row),
+ )
+ return item, name_cut or description_cut
+
+
+def _response_size(response: CatalogResponse) -> int:
+ return len(
+ json.dumps(response.model_dump(mode="json"), separators=(",",
":")).encode(
+ "utf-8"
+ )
+ )
+
+
+def _fetch_rows(
+ spec: _AssetSpec, after_id: int, search: str | None, limit: int
+) -> tuple[list[Any], bool]:
+ """Read one keyset page through the DAO, as the current user.
+
+ ``DAO.list`` always applies the model's ``base_filter`` (the same
+ per-user visibility filter the list tools use) and soft-deleted rows
+ are excluded by the global soft-delete listener.
+ """
+ from superset.daos.base import ColumnOperator
+
+ dao = spec.dao()
+ column_operators = (
+ [ColumnOperator(col="id", opr="gt", value=after_id)] if after_id else
None
+ )
+ rows, total_count = dao.list(
+ column_operators=column_operators,
+ order_column="id",
+ order_direction="asc",
+ page=0,
+ page_size=limit,
+ search=search,
+ search_columns=[spec.name_column],
+ columns=spec.columns,
+ )
+ return list(rows), total_count > len(rows)
+
+
+def build_catalog_page(request: GetCatalogRequest) -> CatalogResponse:
+ """Build one catalog page for the current user (``g.user``)."""
+ asset_type: CatalogAssetType = request.asset_type
+ spec = _ASSET_SPECS[asset_type]
+
+ _check_class_permission(spec, asset_type)
+
+ if spec.requires_data_model_access and not
user_can_view_data_model_metadata():
+ return CatalogResponse(
+ asset_type=asset_type,
+ restricted=True,
+ message=DATA_MODEL_METADATA_ERROR_MESSAGE,
+ )
+
+ after_id = (
+ decode_cursor(request.cursor, asset_type, request.search)
+ if request.cursor
+ else 0
+ )
+
+ rows, has_more = _fetch_rows(spec, after_id, request.search,
request.page_size)
+
+ items: list[CatalogItem] = []
+ truncated = False
+ for row in rows:
+ item, shortened = _to_item(spec, row)
+ items.append(item)
+ truncated = truncated or shortened
+
+ def _page(page_items: list[CatalogItem], more: bool) -> CatalogResponse:
+ return CatalogResponse(
+ asset_type=asset_type,
+ items=page_items,
+ next_cursor=encode_cursor(asset_type, request.search,
page_items[-1].id)
+ if more and page_items
+ else None,
+ truncated=truncated,
+ )
+
+ response = _page(items, has_more)
+ # Enforce the byte bound by deferring trailing items to the next page.
+ # Per-item text caps guarantee a single item always fits.
+ while _response_size(response) > CATALOG_MAX_RESPONSE_BYTES and len(items)
> 1:
+ items = items[:-1]
+ truncated = True
+ response = _page(items, True)
+ return response
+
+
+@tool(
+ tags=["discovery"],
+ annotations=ToolAnnotations(
+ title="Get asset catalog",
+ readOnlyHint=True,
+ destructiveHint=False,
+ openWorldHint=False,
+ ),
Review Comment:
Valid; fixed in 6194b05aebc4ac502879db20b67caeb54b3006bb. I checked
get_schema's manual class checks and the shared
mcp_auth_hook/check_tool_permission path: the outer scope check does run first.
The catalog now explicitly declares its manual resource-class alternatives for
that outer gate, while _check_class_permission still enforces RBAC and scope
for the requested asset type. MCP round-trip regressions for all four
resource-only read scopes failed before and pass after, with RBAC
enabled/disabled; mismatched resource scopes and missing class grants remain
denied. No other tool is opted into this behavior.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
