fix(ingestion): prevent avro parser from leaking file content in warning logs (#24798)

Author: rajattiwari-stack

ingestion/src/metadata/parsers/avro_parser.py

@@ -16,6 +16,7 @@
 import traceback
 from typing import List, Optional, Tuple, Type, Union
 
+import avro.errors
 import avro.schema as avroschema
 from avro.schema import ArraySchema, RecordSchema, Schema, UnionSchema
 from pydantic import BaseModel
@@ -260,9 +261,11 @@ def parse_avro_schema(
             )
         ]
         return models
+    except avro.errors.SchemaParseException:
+        logger.warning("Unable to parse the avro schema: SchemaParseException")
     except Exception as exc:  # pylint: disable=broad-except
         logger.debug(traceback.format_exc())
-        logger.warning(f"Unable to parse the avro schema: {exc}")
+        logger.warning(f"Unable to parse the avro schema: {type(exc).__name__}")
     return None
 
 
@@ -303,7 +306,9 @@ def get_avro_fields(
                 )
             else:
                 field_models.append(parse_single_field(field, cls=cls))
+        except avro.errors.SchemaParseException:
+            logger.warning("Unable to parse the avro schema into models: SchemaParseException")
         except Exception as exc:  # pylint: disable=broad-except
             logger.debug(traceback.format_exc())
-            logger.warning(f"Unable to parse the avro schema into models: {exc}")
+            logger.warning(f"Unable to parse the avro schema into models: {type(exc).__name__}")
     return field_models

ingestion/src/metadata/readers/dataframe/avro.py

@@ -12,7 +12,6 @@
 """
 Avro DataFrame reader - streams records in batches to avoid OOM
 """
-import traceback
 from functools import singledispatchmethod
 from typing import Iterator, List, Optional
 
@@ -94,9 +93,8 @@ def _get_avro_columns(file_obj) -> Optional[List[Column]]:
                     writer_schema = json.dumps(reader.writer_schema)
 
                 return parse_avro_schema(schema=writer_schema, cls=Column)
-        except Exception as warn:
-            logger.warning(f"Error reading Avro schema: {warn}")
-            logger.debug(traceback.format_exc())
+        except Exception:
+            logger.warning("Error reading Avro schema")
         return None
 
     @singledispatchmethod

ingestion/tests/unit/test_avro_parser.py

@@ -12,6 +12,7 @@
 """
 Avro parser tests
 """
+import logging
 from unittest import TestCase
 
 from metadata.parsers.avro_parser import parse_avro_schema
@@ -823,3 +824,14 @@ def test_recursive_issue_parsing(self):
             .children[0]
             .children
         )
+
+    def test_parse_failure_does_not_leak_schema_content(self):
+        sensitive_schema = '{"secret_key": "super_secret_value_12345"}'
+
+        with self.assertLogs("metadata.Ingestion", level="DEBUG") as log_ctx:
+            result = parse_avro_schema(sensitive_schema)
+
+        self.assertIsNone(result)
+        for log_line in log_ctx.output:
+            self.assertNotIn("super_secret_value_12345", log_line)
+            self.assertNotIn("secret_key", log_line)