fix(cli): stop interpreting table content as console markup

The CLI passed table and view content straight to Rich as plain
strings, so square-bracket sequences in a property key or value, a
column name or doc, a ref name, an identifier or a path were parsed
as style tags instead of text. Content therefore controlled the
styling of the output, and unbalanced tags could alter how
surrounding rows rendered.

Disable markup on the consoles the CLI writes through, rather than
escaping each call site, so every current and future render is
covered.

Closes #3984

Author: ArulJerald

pyiceberg/cli/output.py

@@ -87,18 +87,23 @@ def __init__(self, **properties: Any) -> None:
     def _table(self) -> RichTable:
         return RichTable.grid(padding=(0, 2))
 
+    def _console(self, *, stderr: bool = False, soft_wrap: bool = False) -> Console:
+        # Identifiers, properties, schemas and paths all originate outside the CLI,
+        # so console markup stays disabled to keep them from styling the output.
+        return Console(markup=False, stderr=stderr, soft_wrap=soft_wrap)
+
     def exception(self, ex: Exception) -> None:
         if self.verbose:
-            Console(stderr=True).print_exception()
+            self._console(stderr=True).print_exception()
         else:
-            Console(stderr=True).print(ex)
+            self._console(stderr=True).print(ex)
 
     def identifiers(self, identifiers: list[Identifier]) -> None:
         table = self._table
         for identifier in identifiers:
             table.add_row(".".join(identifier))
 
-        Console().print(table)
+        self._console().print(table)
 
     def describe_table(self, table: Table) -> None:
         metadata = table.metadata
@@ -126,7 +131,7 @@ def describe_table(self, table: Table) -> None:
         output_table.add_row("Current snapshot", str(table.current_snapshot()))
         output_table.add_row("Snapshots", snapshot_tree)
         output_table.add_row("Properties", table_properties)
-        Console().print(output_table)
+        self._console().print(output_table)
 
     def describe_view(self, view: View) -> None:
         metadata = view.metadata
@@ -151,7 +156,7 @@ def describe_view(self, view: View) -> None:
         output_table.add_row("Current schema", schema_tree)
         output_table.add_row("SQL", representations_tree)
         output_table.add_row("Properties", view_properties)
-        Console().print(output_table)
+        self._console().print(output_table)
 
     def files(self, table: Table, history: bool) -> None:
         if history:
@@ -175,31 +180,31 @@ def files(self, table: Table, history: bool) -> None:
                 manifest_tree = list_tree.add(f"Manifest: {manifest.manifest_path}")
                 for manifest_entry in manifest.fetch_manifest_entry(io, discard_deleted=False):
                     manifest_tree.add(f"Datafile: {manifest_entry.data_file.file_path}")
-        Console().print(snapshot_tree)
+        self._console().print(snapshot_tree)
 
     def describe_properties(self, properties: Properties) -> None:
         output_table = self._table
         for k, v in properties.items():
             output_table.add_row(k, v)
-        Console().print(output_table)
+        self._console().print(output_table)
 
     def text(self, response: str) -> None:
-        Console(soft_wrap=True).print(response)
+        self._console(soft_wrap=True).print(response)
 
     def schema(self, schema: Schema) -> None:
         output_table = self._table
         for field in schema.fields:
             output_table.add_row(field.name, str(field.field_type), field.doc or "")
-        Console().print(output_table)
+        self._console().print(output_table)
 
     def spec(self, spec: PartitionSpec) -> None:
-        Console().print(str(spec))
+        self._console().print(str(spec))
 
     def uuid(self, uuid: UUID | None) -> None:
-        Console().print(str(uuid) if uuid else "missing")
+        self._console().print(str(uuid) if uuid else "missing")
 
     def version(self, version: str) -> None:
-        Console().print(version)
+        self._console().print(version)
 
     def describe_refs(self, ref_details: list[tuple[str, SnapshotRefType, dict[str, str]]]) -> None:
         refs_table = RichTable(title="Snapshot Refs")
@@ -212,7 +217,7 @@ def describe_refs(self, ref_details: list[tuple[str, SnapshotRefType, dict[str,
             refs_table.add_row(
                 name, type, ref_detail["max_ref_age_ms"], ref_detail["min_snapshots_to_keep"], ref_detail["max_snapshot_age_ms"]
             )
-        Console().print(refs_table)
+        self._console().print(refs_table)
 
 
 class JsonOutput(Output):

tests/cli/test_console.py

@@ -184,6 +184,17 @@ def test_describe_namespace_does_not_exists(catalog: InMemoryCatalog) -> None:
     assert result.output == "Namespace doesnotexist does not exists\n"
 
 
+def test_describe_namespace_property_with_rich_markup_is_rendered_literally(catalog: InMemoryCatalog) -> None:
+    malicious_value = "[bold red]injected[/]"
+    catalog.create_namespace(TEST_TABLE_NAMESPACE, {"malicious": malicious_value})
+
+    runner = CliRunner()
+    result = runner.invoke(run, ["describe", "--entity", "namespace", "default"])
+
+    assert result.exit_code == 0
+    assert malicious_value in result.output
+
+
 @pytest.fixture()
 def test_describe_table(catalog: InMemoryCatalog, mock_datetime_now: None) -> None:
     catalog.create_table(
@@ -227,6 +238,25 @@ def test_describe_table_does_not_exists(catalog: InMemoryCatalog) -> None:
     assert result.output == "Table, view, or namespace does not exist: default.doesnotexist\n"
 
 
+def test_describe_table_property_with_rich_markup_is_rendered_literally(
+    catalog: InMemoryCatalog, mock_datetime_now: None
+) -> None:
+    malicious_value = "[bold red]injected[/]"
+    catalog.create_namespace(TEST_TABLE_NAMESPACE)
+    catalog.create_table(
+        identifier=TEST_TABLE_IDENTIFIER,
+        schema=TEST_TABLE_SCHEMA,
+        partition_spec=TEST_TABLE_PARTITION_SPEC,
+        properties={"malicious": malicious_value},
+    )
+
+    runner = CliRunner()
+    result = runner.invoke(run, ["describe", "default.my_table"])
+
+    assert result.exit_code == 0
+    assert malicious_value in result.output
+
+
 @pytest.mark.parametrize("entity_args", [[], ["--entity", "table"]], ids=["any", "table"])
 def test_describe_table_entity_detection(catalog: InMemoryCatalog, mock_datetime_now: None, entity_args: list[str]) -> None:
     catalog.create_namespace(TEST_TABLE_NAMESPACE)
@@ -277,6 +307,38 @@ def test_schema(catalog: InMemoryCatalog) -> None:
     )
 
 
+def test_schema_field_with_rich_markup_is_rendered_literally(catalog: InMemoryCatalog) -> None:
+    markup = "[bold red]injected[/]"
+    catalog.create_namespace(TEST_TABLE_NAMESPACE)
+    catalog.create_table(
+        identifier=TEST_TABLE_IDENTIFIER,
+        schema=Schema(NestedField(1, markup, LongType(), required=False, doc=markup)),
+    )
+
+    runner = CliRunner()
+    result = runner.invoke(run, ["schema", "default.my_table"])
+
+    assert result.exit_code == 0
+    assert result.output.count(markup) == 2
+
+
+def test_describe_table_schema_field_with_rich_markup_is_rendered_literally(
+    catalog: InMemoryCatalog, mock_datetime_now: None
+) -> None:
+    markup = "[bold red]injected[/]"
+    catalog.create_namespace(TEST_TABLE_NAMESPACE)
+    catalog.create_table(
+        identifier=TEST_TABLE_IDENTIFIER,
+        schema=Schema(NestedField(1, markup, LongType(), required=False)),
+    )
+
+    runner = CliRunner()
+    result = runner.invoke(run, ["describe", "default.my_table"])
+
+    assert result.exit_code == 0
+    assert markup in result.output
+
+
 def test_schema_does_not_exists(catalog: InMemoryCatalog) -> None:
     # pylint: disable=unused-argument
 
@@ -1242,6 +1304,19 @@ def test_describe_view(catalog_with_view: tuple[InMemoryCatalog, View]) -> None:
     assert "spark: SELECT * FROM my_table" in result.output
 
 
+def test_describe_view_property_with_rich_markup_is_rendered_literally(catalog: InMemoryCatalog) -> None:
+    malicious_value = "[bold red]injected[/]"
+    view_metadata = {**TEST_VIEW_METADATA, "properties": {"malicious": malicious_value}}
+    view = View(TEST_VIEW_IDENTIFIER, ViewMetadata.model_validate(view_metadata))
+    catalog.load_view = MagicMock(return_value=view)  # type: ignore
+
+    runner = CliRunner()
+    result = runner.invoke(run, ["describe", "--entity=view", "default.my_view"])
+
+    assert result.exit_code == 0
+    assert malicious_value in result.output
+
+
 def test_describe_view_does_not_exist(catalog: InMemoryCatalog) -> None:
     catalog.load_view = MagicMock(side_effect=NoSuchViewError("View does not exist: default.doesnotexist"))  # type: ignore