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

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