Security advisory PYSEC-2025-51: Failure to Sanitize Special Elements into a Different Plane (Special Element Injection) vulnerability in Apache Airflow Providers Snowflake.

This issue affects Apache Airflow Providers Snowflake: before 6.4.0.

Sanitation of table and stage parameters were added in CopyFromExternalStageToSnowflakeOperator to prevent SQL injection
Users are recommended to upgrade to version 6.4.0, which fixes the issue.
Package: apache-airflow-providers-snowflake (pypi), vulnerable version: 6.3.1, fixed in: 6.4.0
Files changed by the fix:
providers/snowflake/src/airflow/providers/snowflake/transfers/copy_into_snowflake.py

Fix diff:
diff --git a/providers/snowflake/src/airflow/providers/snowflake/transfers/copy_into_snowflake.py b/providers/snowflake/src/airflow/providers/snowflake/transfers/copy_into_snowflake.py
index 0a29ecd9974b7..e8b70344fbf5c 100644
--- a/providers/snowflake/src/airflow/providers/snowflake/transfers/copy_into_snowflake.py
+++ b/providers/snowflake/src/airflow/providers/snowflake/transfers/copy_into_snowflake.py
@@ -27,6 +27,15 @@
 from airflow.providers.snowflake.utils.common import enclose_param
 
 
+def _validate_parameter(param_name: str, value: str | None) -> str | None:
+    """Validate that the parameter doesn't contain any invalid pattern."""
+    if value is None:
+        return None
+    if ";" in value:
+        raise ValueError(f"Invalid {param_name}: semicolons (;) not allowed.")
+    return value
+
+
 class CopyFromExternalStageToSnowflakeOperator(BaseOperator):
     """
     Executes a COPY INTO command to load files from an external stage from clouds to Snowflake.
@@ -91,8 +100,8 @@ def __init__(
     ):
         super().__init__(**kwargs)
         self.files = files
-        self.table = table
-        self.stage = stage
+        self.table = _validate_parameter("table", table)
+        self.stage = _validate_parameter("stage", stage)
         self.prefix = prefix
         self.file_format = file_format
         self.schema = schema
@@ -126,7 +135,7 @@ def execute(self, context: Any) -> None:
         if self.schema:
             into = f"{self.schema}.{self.table}"
         else:
-            into = self.table
+            into = self.table  # type: ignore[assignment]
 
         if self.columns_array:
             into = f"{into}({', '.join(self.columns_array)})"
diff --git a/providers/snowflake/tests/unit/snowflake/transfers/test_copy_into_snowflake.py b/providers/snowflake/tests/unit/snowflake/transfers/test_copy_into_snowflake.py
index 5b0bf27f600b0..ed0bd3b2dc7bc 100644
--- a/providers/snowflake/tests/unit/snowflake/transfers/test_copy_into_snowflake.py
+++ b/providers/snowflake/tests/unit/snowflake/transfers/test_copy_into_snowflake.py
@@ -549,3 +549,26 @@ def test_get_openlineage_facets_on_complete_with_empty_directory(self, mock_hook
             },
             job_facets={"sql": SQLJobFacet(query=expected_sql)},
         )
+
+    def test_init_with_invalid_parameters(self):
+        # Test by passing invalid parameters to the operator
+        import re
+
+        with pytest.raises(ValueError, match=re.escape("Invalid table: semicolons (;) not allowed.")):
+            _ = CopyFromExternalStageToSnowflakeOperator(
+                task_id="test",
+                table="table; some_new_table",
+                stage="stage",
+                database="",
+                schema="schema",
+                file_format="CSV",
+            )
+        with pytest.raises(ValueError, match=re.escape("Invalid stage: semicolons (;) not allowed.")):
+            _ = CopyFromExternalStageToSnowflakeOperator(
+                task_id="test",
+                table="table",
+                stage="stage; some_new_stabge",
+                database="",
+                schema="schema",
+                file_format="CSV",
+            )