sadpandajoe commented on code in PR #45575:
URL: https://github.com/apache/superset/pull/45575#discussion_r4366160751
##########
tests/unit_tests/mcp_service/test_sanitization_warning_inputs.py:
##########
@@ -524,3 +579,789 @@ 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
+
+
[email protected]("model,payload,field", _TITLE_SANITIZATION_CASES)
+def test_escaped_title_still_fits_model_column(
+ model: type[BaseModel], payload: dict[str, Any], field: str
+) -> None:
+ """Conservative bracket escaping cannot bypass a persisted title's length
cap."""
+ limit = 250 if field == "chart_name" else 500
+ value = "x" * (limit - 2) + "<ſ"
+ with pytest.raises(ValidationError) as exc:
+ model.model_validate({**payload, field: value})
+ assert exc.value.errors()[0]["type"] == "string_too_long"
+
+
[email protected]("model,payload,field", _TITLE_SANITIZATION_CASES)
[email protected]("whitespace", ["\t", "\n", "\r", "\v", "\f", "\x85"])
[email protected](
+ "control",
+ ["!channel", "@U123", "#C123|general", "https://example.invalid|Open
dashboard"],
+)
+def test_title_control_whitespace_cannot_form_slack_markup(
+ model: type[BaseModel],
+ payload: dict[str, Any],
+ field: str,
+ whitespace: str,
+ control: str,
+) -> None:
+ """Report title normalization must not join brackets into Slack
controls."""
+ from superset.utils.file import sanitize_title
+
+ value = f"Q3 <{whitespace}{control}>"
+ request = model.model_validate({**payload, field: value})
+ stored = getattr(request, field)
+ assert "<" not in sanitize_title(stored)
+ assert getattr(model.model_validate(request.model_dump()), field) == stored
+
+
[email protected]("model,payload,field", _TITLE_SANITIZATION_CASES)
[email protected]("alias_index", range(4))
[email protected]("length_delta", [-5, 0, 1])
[email protected]("padding", ["", " ", " ", "\u00a0"])
+def test_markup_title_length_uses_sanitized_text(
+ model: type[BaseModel],
+ payload: dict[str, Any],
+ field: str,
+ alias_index: int,
+ length_delta: int,
+ padding: str,
+) -> None:
+ """Markup is not part of the persisted title's character budget."""
+ aliases = (
+ ("chart_name", "name", "title", "slice_name")
+ if field == "chart_name"
+ else ("dashboard_title", "title", "name", "dashboard_title")
+ )
+ limit = 250 if field == "chart_name" else 500
+ text = "x" * (limit + length_delta)
+ value = "<b>" + padding + text + padding + "</b>"
+ if length_delta > 0:
+ with pytest.raises(ValidationError) as exc:
+ model.model_validate({**payload, aliases[alias_index]: value})
+ assert exc.value.errors()[0]["type"] == "string_too_long"
+ else:
+ request = model.model_validate({**payload, aliases[alias_index]:
value})
+ assert getattr(request, field) == text
+ assert len(request.sanitization_warnings) == 1
+
+
[email protected](
+ "model,payload,field",
+ [case for case in _SIBLING_SANITIZATION_CASES if case[2] != "value"],
+)
[email protected]("extra", [0, 1])
[email protected]("padding", ["", " ", " ", "\u00a0"])
+def test_sibling_sanitized_length_budget(
+ model: type[BaseModel],
+ payload: dict[str, Any],
+ field: str,
+ extra: int,
+ padding: str,
+) -> None:
+ """All callers of the shared text sanitizer enforce the final text cap."""
+ limits = {
+ "certified_by": 500,
+ "certification_details": 5000,
+ "temporal_column": 255,
+ "name": 255,
+ "label": 500,
+ "column": 255,
+ "time_range": 1000,
+ "comparison_period": 100,
+ "range_labels": 200,
+ "marker_labels": 200,
+ "marker_line_labels": 200,
+ }
+ if model is InteractivePivotChartConfig:
+ payload = {**payload, "comparison_type": "values"}
+ text = "x" * (limits[field] + extra)
+ value = "<b>" + padding + text + padding + "</b>"
+ if extra:
+ with pytest.raises(ValidationError) as exc:
+ model.model_validate(
+ {**payload, field: [value] if field.endswith("_labels") else
value}
+ )
+ assert exc.value.errors()[0]["type"] == "string_too_long"
+ else:
+ request = model.model_validate(
+ {**payload, field: [value] if field.endswith("_labels") else value}
+ )
+ assert getattr(request, field) == (
+ [text] if field.endswith("_labels") else text
+ )
+
+
[email protected](
+ "model,payload,field",
+ [
+ case
+ for case in _SIBLING_SANITIZATION_CASES
+ if case[2] != "value" and not case[2].endswith("_labels")
+ ],
+)
[email protected]("value", [b"<b>Safe</b>", bytearray(b"<b>Safe</b>")])
Review Comment:
This parametrization covers sibling configuration and certification fields,
but not the chart name or title models. `UpdateChartRequest(identifier=1,
chart_name=b"Safe <b>title</b>")` skips the string-only `before` validator,
pydantic then decodes the bytes, and `normalize_blank_chart_name` only checks
blankness, so the stored name keeps `<b>` (it was sanitized to `Safe title`
before this change). Could `chart_name` (and the dashboard title) be added with
bytes/bytearray inputs asserting the sanitized value, or the after-validator
sanitize as well?
##########
superset/mcp_service/chart/schemas.py:
##########
@@ -6396,16 +6416,24 @@ def _detect_chart_name_sanitization(cls, data: Any) ->
Any:
@field_validator("chart_name")
@classmethod
- def sanitize_chart_name(cls, v: str | None) -> str | None:
- """Sanitize chart name to prevent XSS attacks."""
- try:
- return sanitize_user_input(
- v, "Chart name", max_length=255, allow_empty=True
- )
- except ValueError as ex:
- raise PydanticCustomError(
- "unsafe_text", "{error}", {"error": str(ex)}
- ) from ex
+ def normalize_blank_chart_name(cls, v: str | None) -> str | None:
+ """Keep the optional-name fallback; nonblank names are sanitized
above."""
+ return v if v and v.strip() else None
Review Comment:
A whitespace-only `chart_name` longer than 250 characters now fails with
`string_too_long`: `Field(max_length=250)` runs before this validator, and the
model-level sanitizer returns early for blank input. For example
`generate_chart` with `chart_name=" " * 251` used to normalize to `None` and
fall back to the generated name under the old 255 cap. Should blank values be
collapsed to `None` in the `mode="before"` validator, ahead of the length
constraint, so the documented trim-before-length behavior holds here too?
--
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]
