rebenitez1802 commented on code in PR #45575:
URL: https://github.com/apache/superset/pull/45575#discussion_r4352609612
##########
tests/unit_tests/mcp_service/test_sanitization_warning_inputs.py:
##########
@@ -524,3 +579,683 @@ def validate(request: BaseModel) -> str:
)
assert f"request.{field}: Contains disallowed content" in message
assert "private-input" not in message
+
+
[email protected]
[email protected](
+ ("tool", "payload", "field", "limit"),
+ [
+ ("generate_dashboard", {"chart_ids": [1]}, "dashboard_title", 500),
+ ("update_dashboard", {"identifier": 1}, "dashboard_title", 500),
+ ("duplicate_dashboard", {"dashboard_id": 1}, "dashboard_title", 500),
+ (
+ "generate_chart",
+ {
+ "dataset_id": 1,
+ "config": {"chart_type": "table", "columns": [{"name": "a"}]},
+ },
+ "chart_name",
+ 250,
+ ),
+ ("update_chart", {"identifier": 1}, "chart_name", 250),
+ ],
+)
[email protected]("alias", [None, "title", "name"])
[email protected]("length", [None, 600])
+async def test_title_length_wire_reason(
+ tool: str,
+ payload: dict[str, Any],
+ field: str,
+ limit: int,
+ alias: str | None,
+ length: int | None,
+) -> None:
+ """Length errors retain their reason through warning detection and MCP."""
+ rejected = "x" * (length or limit + 1)
+ async with Client(make_title_server()) as client:
+ result = await client.call_tool_mcp(
+ tool, {"request": {**payload, alias or field: rejected}}
+ )
+ assert result.isError is True
+ assert result.content == [
+ TextContent(
+ type="text",
+ text=f"Error: Validation error in {tool}: request.{field}: "
+ "String exceeds the allowed maximum length",
+ )
+ ]
+
+
[email protected]
[email protected]("field", ["dashboard_title", "title", "name"])
[email protected]("rejected", ["", " \t\n"])
+async def test_duplicate_dashboard_empty_title_wire_reason(
+ field: str,
+ rejected: str,
+) -> None:
+ """Blank required titles have a static, actionable reason on the wire."""
+ async with Client(make_title_server()) as client:
+ result = await client.call_tool_mcp(
+ "duplicate_dashboard", {"request": {"dashboard_id": 1, field:
rejected}}
+ )
+ assert result.isError is True
+ assert result.content == [
+ TextContent(
+ type="text",
+ text="Error: Validation error in duplicate_dashboard: "
+ "request.dashboard_title: Text must not be empty",
+ )
+ ]
+
+
+_REQUIRED_NAME_CASES: list[tuple[str, dict[str, Any], str, list[str | None]]]
= [
+ (
+ "create_virtual_dataset",
+ {"database_id": 1, "sql": "select 1"},
+ "dataset_name",
+ ["", " \t\n"],
+ ),
+ ("create_theme", {"json_data": {}}, "theme_name", ["", " \t\n"]),
+ (
+ "create_dataset_metric",
+ {"dataset_id": 1, "expression": "count(*)"},
+ "metric_name",
+ ["", " \t\n"],
+ ),
+ (
+ "update_dataset_metric",
+ {"dataset_id": 1, "metric": 1},
+ "metric_name",
+ ["", " \t\n", None],
+ ),
+ ("update_dataset", {"dataset_id": 1}, "table_name", ["", " \t\n", None]),
+ (
+ "save_sql_query",
+ {"database_id": 1, "sql": "select 1"},
+ "label",
+ ["", " \t\n"],
+ ),
+]
+
+
[email protected]
[email protected](
+ ("tool", "payload", "field", "rejected"),
+ [
+ (tool, payload, field, rejected)
+ for tool, payload, field, blanks in _REQUIRED_NAME_CASES
+ for rejected in blanks
+ ],
+)
+async def test_required_name_empty_wire_reason(
+ tool: str, payload: dict[str, Any], field: str, rejected: str | None
+) -> None:
+ """Blank object names report a safe reason at the declared field."""
+ async with Client(make_name_server()) as client:
+ result = await client.call_tool_mcp(
+ tool, {"request": {**payload, field: rejected}}
+ )
+ assert result.isError is True
+ assert result.content == [
+ TextContent(
+ type="text",
+ text=f"Error: Validation error in {tool}: request.{field}: "
+ "Text must not be empty",
+ )
+ ]
+
+
[email protected](
+ "model,payload",
+ [
+ (
+ GenerateChartRequest,
+ {
+ "dataset_id": 1,
+ "config": {"chart_type": "table", "columns": [{"name": "a"}]},
+ },
+ ),
+ (UpdateChartRequest, {"identifier": 1}),
+ ],
+)
[email protected]("length", [250, 251, 252, 253, 254, 255])
+def test_chart_name_model_column_boundary(
+ model: type[BaseModel], payload: dict[str, Any], length: int
+) -> None:
+ """Chart names must fit the persisted Slice name column."""
+ from superset.models.slice import Slice
+
+ limit = Slice.__table__.c.slice_name.type.length
+ assert limit == 250
+ if length <= limit:
+ assert (
+ model.model_validate({**payload, "chart_name": "x" *
length}).chart_name
+ == "x" * length
+ )
+ else:
+ with pytest.raises(ValidationError) as exc:
+ model.model_validate({**payload, "chart_name": "x" * length})
+ assert exc.value.errors()[0]["type"] == "string_too_long"
+ assert exc.value.errors()[0]["loc"] == ("chart_name",)
+
+
[email protected](
+ "model,payload",
+ [
+ (
+ GenerateChartRequest,
+ {
+ "dataset_id": 1,
+ "config": {"chart_type": "table", "columns": [{"name": "a"}]},
+ },
+ ),
+ (UpdateChartRequest, {"identifier": 1}),
+ ],
+)
[email protected]("alias", ["chart_name", "name", "title",
"slice_name"])
[email protected]("length", [250, 251])
+def test_padded_chart_name_boundary(
+ model: type[BaseModel], payload: dict[str, Any], alias: str, length: int
+) -> None:
+ """Validate the trimmed chart name against the model column boundary."""
+ value = " " + "x" * length + " "
+ if length == 250:
+ assert (
+ model.model_validate({**payload, alias: value}).chart_name == "x"
* length
+ )
+ else:
+ with pytest.raises(ValidationError) as exc:
+ model.model_validate({**payload, alias: value})
+ assert exc.value.errors()[0]["type"] == "string_too_long"
+
+
[email protected]("length", [250, 251, 310])
+def test_theme_name_model_column_boundary(length: int) -> None:
+ """Theme creation must respect its persisted name column."""
+ from superset.models.core import Theme
+
+ assert Theme.__table__.c.theme_name.type.length == 250
+ if length == 250:
+ assert (
+ CreateThemeRequest(theme_name="x" * length,
json_data={}).theme_name
+ == "x" * length
+ )
+ else:
+ with pytest.raises(ValidationError) as exc:
+ CreateThemeRequest(theme_name="x" * length, json_data={})
+ assert exc.value.errors()[0]["type"] == "string_too_long"
+
+
[email protected]
[email protected]("alias", ["dashboard_title", "title", "name"])
[email protected]("value", ["", " \t\n"])
+async def test_update_dashboard_blank_title_wire_reason(alias: str, value:
str) -> None:
+ """Reject blank renames rather than reporting a title that was not
saved."""
+ async with Client(make_title_server()) as client:
+ result = await client.call_tool_mcp(
+ "update_dashboard", {"request": {"identifier": 1, alias: value}}
+ )
+ assert result.isError is True
+ assert "request.dashboard_title: Text must not be empty" in
result.content[0].text
+
+
[email protected]("alias", ["chart_name", "name", "title",
"slice_name"])
[email protected]("script_only", [True, False])
+def test_update_chart_sanitization_matches_generate(
+ alias: str, script_only: bool
+) -> None:
+ """Update rejects stripped-only names and warns on partially stripped
names."""
+ if script_only:
+ with pytest.raises(ValidationError) as exc:
+ UpdateChartRequest.model_validate(
+ {"identifier": 1, alias: "<script>alert(1)</script>"}
+ )
+ assert exc.value.errors()[0]["type"] == "unsafe_text"
+ assert exc.value.errors()[0]["loc"] == ("chart_name",)
+ return
+ request = UpdateChartRequest.model_validate(
+ {
+ "identifier": 1,
+ alias: "Weekly KPIs <img src=x onerror=alert(1)>",
+ "sanitization_warnings": ["untrusted"],
+ }
+ )
+ assert request.chart_name == "Weekly KPIs"
+ assert len(request.sanitization_warnings) == 1
+ assert "untrusted" not in request.sanitization_warnings
+ assert "sanitization_warnings" not in json.dumps(
+ UpdateChartRequest.model_json_schema()
+ )
+
+
+def test_duplicate_dataset_error_is_readable() -> None:
+ """Duplicate errors use a plain string and preserve display-name
characters."""
+ from superset.commands.dataset.exceptions import
DatasetExistsValidationError
+ from superset.sql.parse import Table
+
+ name = "Sales “Q3” & Growth"
+ error = DatasetExistsValidationError(
+ Table(name, schema="my schema", catalog="my catalog")
+ )
+ assert error.normalized_messages() == {
+ "table": [f"Dataset my catalog.my schema.{name} already exists"]
+ }
+ assert type(error.messages[0]) is str
+
+
[email protected](
+ "kind,limit", [("dataset", 250), ("saved_query", 256), ("tab", 256)]
+)
[email protected]("extra", [0, 1])
+def test_padded_sibling_name_boundary(kind: str, limit: int, extra: int) ->
None:
+ """Persisted names that trim must check the length after trimming."""
+ from superset.mcp_service.sql_lab.schemas import OpenSqlLabRequest
+
+ cases: dict[str, tuple[type[BaseModel], dict[str, Any], str]] = {
+ "dataset": (
+ CreateVirtualDatasetRequest,
+ {"database_id": 1, "sql": "select 1"},
+ "dataset_name",
+ ),
+ "saved_query": (
+ SaveSqlQueryRequest,
+ {"database_id": 1, "sql": "select 1"},
+ "label",
+ ),
+ "tab": (OpenSqlLabRequest, {"database_id": 1}, "title"),
+ }
+ model, payload, field = cases[kind]
+ value = "x" * (limit + extra)
+ if not extra:
+ assert (
+ getattr(model.model_validate({**payload, field: f" {value} "}),
field)
+ == value
+ )
+ else:
+ with pytest.raises(ValidationError) as exc:
+ model.model_validate({**payload, field: f" {value} "})
+ assert exc.value.errors()[0]["type"] == "string_too_long"
+
+
[email protected](
+ "model,payload",
+ [
+ (GenerateDashboardRequest, {"chart_ids": [1]}),
+ (UpdateDashboardRequest, {"identifier": 1}),
+ (DuplicateDashboardRequest, {"dashboard_id": 1}),
+ ],
+)
[email protected]("alias", ["dashboard_title", "title", "name"])
[email protected]("length", [500, 501])
+def test_padded_dashboard_title_boundary(
+ model: type[BaseModel], payload: dict[str, Any], alias: str, length: int
+) -> None:
+ """Dashboard aliases check the trimmed title against the column cap."""
+ if length == 500:
+ assert (
+ model.model_validate(
+ {**payload, alias: " " + "x" * length + " "}
+ ).dashboard_title
+ == "x" * length
+ )
+ else:
+ with pytest.raises(ValidationError) as exc:
+ model.model_validate({**payload, alias: " " + "x" * length + "
"})
+ assert exc.value.errors()[0]["type"] == "string_too_long"
+
+
+def test_padded_dashboard_slug_boundary() -> None:
+ """Update slug normalization precedes its persisted column limit."""
+ request = UpdateDashboardRequest(identifier=1, slug=" " + "x" * 255 + "
")
+ assert request.slug == "x" * 255
+ with pytest.raises(ValidationError):
+ UpdateDashboardRequest(identifier=1, slug=" " + "x" * 256 + " ")
+
+
+_TITLE_SANITIZATION_CASES = [
+ (
+ GenerateChartRequest,
+ {
+ "dataset_id": 1,
+ "config": {"chart_type": "table", "columns": [{"name": "a"}]},
+ "save_chart": False,
+ },
+ "chart_name",
+ ),
+ (
+ GenerateChartRequest,
+ {
+ "dataset_id": 1,
+ "config": {"chart_type": "table", "columns": [{"name": "a"}]},
+ },
+ "chart_name",
+ ),
+ (UpdateChartRequest, {"identifier": 1}, "chart_name"),
+ (GenerateDashboardRequest, {"chart_ids": [1]}, "dashboard_title"),
+ (UpdateDashboardRequest, {"identifier": 1}, "dashboard_title"),
+ (DuplicateDashboardRequest, {"dashboard_id": 1}, "dashboard_title"),
+]
+
+
[email protected]("model,payload,field", _TITLE_SANITIZATION_CASES)
[email protected](
+ "value",
+ [
+ "Users aged <30",
+ "Revenue <$1M",
+ "Love <3 charts",
+ "Users aged <30 & “VIP” 📈",
+ "Users aged >30",
+ "Revenue\u00a0>\u00a0$1M",
+ ],
+)
+def test_literal_less_than_title_round_trip(
+ model: type[BaseModel], payload: dict[str, Any], field: str, value: str
+) -> None:
+ """Non-tag less-than signs survive validation and JSON readback
unchanged."""
+ request = model.model_validate({**payload, field: value})
+ assert getattr(request, field) == value
+ assert request.sanitization_warnings == []
+ assert json.loads(request.model_dump_json())[field] == value
+
+
[email protected]("model,payload,field", _TITLE_SANITIZATION_CASES)
[email protected](
+ "value", ["Weekly <img src=x>KPIs", "Q3 <b>revenue</b> & costs"]
+)
+def test_title_sanitized_once(
+ model: type[BaseModel], payload: dict[str, Any], field: str, value: str
+) -> None:
+ """Persist the single sanitizer result, never a second-pass empty
fallback."""
+ from unittest.mock import patch
+
+ from superset.mcp_service.utils.sanitization import sanitize_user_input
+
+ expected = sanitize_user_input(value, "Title", max_length=500,
allow_empty=True)
+ assert expected
+ with (
+ patch(
+ "superset.mcp_service.utils.sanitization.sanitize_user_input",
+ wraps=sanitize_user_input,
+ ) as sanitize,
+ patch("superset.mcp_service.chart.schemas.sanitize_user_input",
sanitize),
+ patch("superset.mcp_service.dashboard.schemas.sanitize_user_input",
sanitize),
+ ):
+ request = model.model_validate({**payload, field: value})
+ calls = [
+ c
+ for c in sanitize.call_args_list
+ if c.args[1] in ("Chart name", "Dashboard title")
+ ]
+ assert len(calls) == 1
+ assert getattr(request, field) == expected
+ assert len(request.sanitization_warnings) == 1
+
+
[email protected]("model,payload,field", _TITLE_SANITIZATION_CASES)
[email protected](
+ "value", ["<<</b>x", "&lt;</script>click= <b>📈"]
+)
+def test_reconstructed_tag_title_rejected(
+ model: type[BaseModel], payload: dict[str, Any], field: str, value: str
+) -> None:
+ """Encoded tag text is rejected instead of stored or auto-named."""
+ with pytest.raises(ValidationError) as exc:
+ model.model_validate({**payload, field: value})
+ assert exc.value.errors()[0]["loc"] == (field,)
+ assert exc.value.errors()[0]["type"] == "unsafe_text"
+
+
[email protected]("model,payload,field", _TITLE_SANITIZATION_CASES)
+def test_empty_sanitized_title_never_uses_fallback(
+ model: type[BaseModel], payload: dict[str, Any], field: str
+) -> None:
+ """Encoded script-only names retain the existing disallowed-content
reason."""
+ with pytest.raises(ValidationError) as exc:
+ model.model_validate({**payload, field:
"&lt;script>alert(1)</script>"})
+ assert exc.value.errors()[0]["loc"] == (field,)
+ assert exc.value.errors()[0]["type"] == "unsafe_text"
+
+
+_SIBLING_SANITIZATION_CASES = [
+ (ManageDashboardCertificationRequest, {"identifier": 1}, "certified_by"),
+ (
+ ManageDashboardCertificationRequest,
+ {"identifier": 1},
+ "certification_details",
+ ),
+ (BaseChartConfig, {}, "temporal_column"),
+ (ColumnRef, {}, "name"),
+ (ColumnRef, {"name": "amount"}, "label"),
+ (FilterConfig, {"op": "=", "value": "safe"}, "column"),
+ (FilterConfig, {"column": "region", "op": "="}, "value"),
+ (GanttSortByConfig, {}, "column"),
+ (
+ SunburstChartConfig,
+ {
+ "hierarchy": [{"name": "region"}],
+ "metric": {"name": "amount", "aggregate": "SUM"},
+ },
+ "time_range",
+ ),
+ (
+ InteractivePivotChartConfig,
+ {
+ "rows": [{"name": "region"}],
+ "metrics": [{"name": "amount", "aggregate": "SUM"}],
+ },
+ "comparison_period",
+ ),
+ (
+ BulletChartConfig,
+ {"metric": {"name": "amount", "aggregate": "SUM"}},
+ "time_range",
+ ),
+ (
+ BulletChartConfig,
+ {"metric": {"name": "amount", "aggregate": "SUM"}},
+ "range_labels",
+ ),
+ (
+ BulletChartConfig,
+ {"metric": {"name": "amount", "aggregate": "SUM"}},
+ "marker_labels",
+ ),
+ (
+ BulletChartConfig,
+ {"metric": {"name": "amount", "aggregate": "SUM"}},
+ "marker_line_labels",
+ ),
+]
+
+
[email protected]("model,payload,field", _SIBLING_SANITIZATION_CASES)
[email protected](
+ "value", ["Users aged <30", "Users aged >30", "Users\u00a0>30"]
+)
+def test_sibling_fields_preserve_literal_comparison(
+ model: type[BaseModel], payload: dict[str, Any], field: str, value: str
+) -> None:
+ """Every sibling accepts a non-tag comparison and preserves its literal
text."""
+ if model is InteractivePivotChartConfig:
+ payload = {**payload, "comparison_type": "values"}
+ expected = [value] if field.endswith("_labels") else value
+ request = model.model_validate({**payload, field: expected})
+ assert getattr(request, field) == expected
+
+
[email protected]("model,payload,field", _SIBLING_SANITIZATION_CASES)
[email protected](
+ "value", ["<<</b>x", "&lt;</script>click= <b>📈"]
+)
+def test_sibling_fields_reject_reconstructed_tag(
+ model: type[BaseModel], payload: dict[str, Any], field: str, value: str
+) -> None:
+ """Reconstructed tag openers fail with a canonical field and typed
reason."""
+ if model is InteractivePivotChartConfig:
+ payload = {**payload, "comparison_type": "values"}
+ with pytest.raises(ValidationError) as exc:
+ model.model_validate(
+ {**payload, field: [value] if field.endswith("_labels") else value}
+ )
+ assert [(error["type"], error["loc"]) for error in exc.value.errors()] == [
+ ("unsafe_text", (field,))
+ ]
+
+
[email protected](
+ "model,payload,field",
+ [
+ (
+ CreateVirtualDatasetRequest,
+ {"database_id": 1, "sql": "select 1"},
+ "dataset_name",
+ ),
+ (UpdateDatasetRequest, {"dataset_id": 1}, "table_name"),
+ (
+ CreateDatasetMetricRequest,
+ {"dataset_id": 1, "expression": "SUM(x)"},
+ "metric_name",
+ ),
+ (UpdateDatasetMetricRequest, {"dataset_id": 1, "metric": 1},
"metric_name"),
+ (CreateThemeRequest, {"json_data": {}}, "theme_name"),
+ (SaveSqlQueryRequest, {"database_id": 1, "sql": "select 1"}, "label"),
+ (
+ CreateVirtualDatasetRequest,
+ {"database_id": 1, "sql": "select 1", "dataset_name": "ds"},
+ "description",
+ ),
+ (UpdateDatasetRequest, {"dataset_id": 1}, "description"),
+ (
+ SaveSqlQueryRequest,
+ {"database_id": 1, "sql": "select 1", "label": "query"},
+ "description",
+ ),
+ (UpdateDashboardRequest, {"identifier": 1}, "description"),
+ ],
+)
[email protected](
+ "value",
+ [
+ "Users aged <30",
+ "Revenue <$1M",
+ "Love <3 charts",
+ "<<</b>x",
+ "&lt;</script>click= <b>📈",
+ ],
+)
+def test_unsanitized_sibling_names_remain_literal(
+ model: type[BaseModel], payload: dict[str, Any], field: str, value: str
+) -> None:
+ """Non-sanitizer paths retain their existing literal identifier/text
contract."""
+ request = model.model_validate({**payload, field: value})
+ assert getattr(request, field) == value
+
+
[email protected](
+ "value",
+ [
+ "Users aged <30",
+ "Revenue <$1M",
+ "Love <3 charts",
+ "<<</b>x",
+ "&lt;</script>click= <b>📈",
+ ],
+)
+def test_sql_lab_title_remains_literal(value: str) -> None:
+ """SQL Lab tab titles trim whitespace without HTML sanitization."""
+ from superset.mcp_service.sql_lab.schemas import OpenSqlLabRequest
+
+ assert OpenSqlLabRequest(database_id=1, title=value).title == value
+
+
[email protected]
[email protected]("field", ["dashboard_title", "title", "name"])
[email protected](
+ "rejected", ["<script>alert(1)</script>", "Weekly
&am<b></b>p;lt;b&gt;KPIs"]
+)
+async def test_duplicate_dashboard_removed_or_unstable_title_wire_reason(
+ field: str, rejected: str
+) -> None:
+ """Removed or unstable markup is unsafe content, not an empty input."""
+ async with Client(make_title_server()) as client:
+ result = await client.call_tool_mcp(
+ "duplicate_dashboard", {"request": {"dashboard_id": 1, field:
rejected}}
+ )
+ assert result.isError is True
+ assert result.content == [
+ TextContent(
+ type="text",
+ text="Error: Validation error in duplicate_dashboard: "
+ "request.dashboard_title: Contains disallowed content "
+ "(HTML, script or data: URL, event handler, or SQL keyword); "
+ "a 'data:' label must not be followed by a comma",
+ )
+ ]
+
+
[email protected]("model,payload,field", _TITLE_SANITIZATION_CASES)
[email protected]("character", [">", "\u00a0"])
[email protected]("at_limit", [True, False])
+def test_title_length_counts_plain_text(
+ model: type[BaseModel],
+ payload: dict[str, Any],
+ field: str,
+ character: str,
+ at_limit: bool,
+) -> None:
+ """Entity serialization must not inflate persisted titles or reject valid
ones."""
+ limit = 250 if field == "chart_name" else 500
+ value = "x" * (limit - 2 + (not at_limit)) + character + "y"
+ if not at_limit:
+ with pytest.raises(ValidationError) as exc:
+ model.model_validate({**payload, field: value})
+ assert exc.value.errors()[0]["type"] == "string_too_long"
+ return
+ request = model.model_validate({**payload, field: value})
+ assert getattr(request, field) == value
+ assert len(getattr(request, field)) == limit
+ assert request.sanitization_warnings == []
+ assert getattr(model.model_validate(request.model_dump()), field) == value
+
+
+def test_builtin_name_validation_documented_for_administrators() -> None:
+ """Built-in validation guidance belongs with server administration."""
+ from pathlib import Path
+
+ root = Path(__file__).resolve().parents[3]
+ admin = (root / "docs/admin_docs/configuration/mcp-server.mdx").read_text()
+ extensions = (root / "docs/developer_docs/extensions/mcp.md").read_text()
+ for heading in ("Chart and dashboard titles", "Object-name validation"):
+ assert heading in admin
+ assert heading not in extensions
+ assert "Theme names retain their whitespace" in admin
+ assert "length validation applies after trimming" in admin
+
+
+def test_name_contract_changes_have_upgrade_guidance() -> None:
+ """Clients need migration guidance for blank renames and tighter name
caps."""
+ from pathlib import Path
+
+ root = Path(__file__).resolve().parents[3]
+ next_release = (
+ (root / "UPDATING.md").read_text().split("## Next", 1)[1].split("\n##
", 1)[0]
+ )
+ assert (
+ "`update_dashboard` rejects blank or whitespace-only `dashboard_title`"
+ in next_release
+ )
+ assert "chart names and theme names are capped at 250 characters" in
next_release
+ assert "omit the field to keep the stored title" in next_release
+
+
Review Comment:
**🔴 High — `test_name_contract_changes_have_upgrade_guidance` will fail at
the next release cut**
This test reads only the `## Next` section of UPDATING.md. At a release cut,
a `## x.y.z` heading is inserted above the entries that ship; d76b896c9c did
this for 6.1.0. Once this entry ships it is no longer under `## Next`, so the
test fails in the release PR. No other unit test reads UPDATING.md, and the
contract is already pinned by `test_update_dashboard_blank_title_wire_reason`,
`test_chart_name_model_column_boundary` and
`test_theme_name_model_column_boundary`. I suggest dropping it:
```suggestion
```
##########
superset/mcp_service/utils/sanitization.py:
##########
@@ -70,11 +70,36 @@ def _strip_html_tags(value: str) -> str:
# url_schemes=set() blocks all URL schemes in any remaining attributes
cleaned = nh3.clean(decoded, tags=set(), url_schemes=set())
- # Only restore & → & to preserve ampersands in display text (e.g. "A &
B").
- # Do NOT use html.unescape() here: nh3 may pass through HTML entities from
- # the input (e.g. <script>), and a full unescape would re-introduce
- # raw angle brackets, creating an XSS vector.
- return cleaned.replace("&", "&")
+ # Restore brackets only before comparison punctuation, numbers or
whitespace.
+ # Non-ASCII letters can case-fold into tag names in downstream renderers;
+ # leave those and reconstructed tag openers escaped.
+ # Normalize removable characters before deciding whether a bracket is safe;
+ # otherwise a later Unicode strip could turn <\u200bscript into markup.
+ cleaned = _remove_dangerous_unicode(cleaned)
+ # Greater-than signs and NBSP cannot open tags. Restore them so plain text
+ # neither changes on display nor outgrows the model-column length limits.
+ return (
+ re.sub(r"<(?=[\d\s$€£¥%+=.,:;)\]&-]|$)", "<", cleaned)
Review Comment:
**🔴 High — `<` before control whitespace now forms Slack control sequences**
The `\s` in the allowlist I suggested is too broad. It restores `<`
before TAB/LF/CR, and `>` is now raw. Report titles go through `sanitize_title`
(`superset/utils/file.py`), which strips control characters. So an MCP chart
name `Q3 <\t!channel>` is stored as-is, and the Slack report message contains a
live `<!channel>`. `<@U…>`, `<#C…|…>` and `<https://evil.example|Open
dashboard>` behave the same way. At the previous head `>` stayed `>`, so
none of these closed.
The fix keeps `<` escaped before control whitespace, unless only whitespace
follows. Literal comparisons and the case-fold cases are unchanged, and the
touched MCP tests pass with it:
```suggestion
re.sub(
# Control whitespace stays escaped: report titles strip it, which
# could join "<" to a Slack control sequence such as <!channel>.
r"<(?=[\d$€£¥%+=.,:;)\]&-]|[^\S\x00-\x1f\x7f-\x9f]|(?:\s| )*$)",
"<",
cleaned,
)
```
--
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]
