Sanitize filenames in MySQLHook

airflow/providers/mysql/hooks/mysql.py

@@ -20,6 +20,7 @@
 
 import json
 import logging
+import re
 from typing import TYPE_CHECKING, Any, Union
 
 from airflow.exceptions import AirflowOptionalProviderFeatureException
@@ -164,6 +165,10 @@ def _get_conn_config_mysql_connector_python(self, conn: Connection) -> dict:
 
         return conn_config
 
+    @staticmethod
+    def _sanitize_filename(filename: str) -> str:
+        return re.sub(r"(;.*)$", "", filename)
+
     def get_conn(self) -> MySQLConnectionTypes:
         """
         Connection to a MySQL database.
@@ -208,7 +213,7 @@ def bulk_load(self, table: str, tmp_file: str) -> None:
         cur = conn.cursor()
         cur.execute(
             f"""
-            LOAD DATA LOCAL INFILE '{tmp_file}'
+            LOAD DATA LOCAL INFILE '{self._sanitize_filename(tmp_file)}'
             INTO TABLE {table}
             """
         )
@@ -221,7 +226,7 @@ def bulk_dump(self, table: str, tmp_file: str) -> None:
         cur = conn.cursor()
         cur.execute(
             f"""
-            SELECT * INTO OUTFILE '{tmp_file}'
+            SELECT * INTO OUTFILE '{self._sanitize_filename(tmp_file)}'
             FROM {table}
             """
         )
@@ -288,7 +293,7 @@ def bulk_load_custom(
 
         cursor.execute(
             f"""
-            LOAD DATA LOCAL INFILE '{tmp_file}'
+            LOAD DATA LOCAL INFILE '{self._sanitize_filename(tmp_file)}'
             {duplicate_key_handling}
             INTO TABLE {table}
             {extra_options}

tests/providers/mysql/hooks/test_mysql.py

@@ -279,6 +279,15 @@ def test_bulk_load(self):
             """
         )
 
+    def test_bulk_load_with_semicolon_in_filename(self):
+        self.db_hook.bulk_load("table", "/tmp/file; SELECT * FROM DUAL")
+        self.cur.execute.assert_called_once_with(
+            """
+            LOAD DATA LOCAL INFILE '/tmp/file'
+            INTO TABLE table
+            """
+        )
+
     def test_bulk_dump(self):
         self.db_hook.bulk_dump("table", "/tmp/file")
         self.cur.execute.assert_called_once_with(
@@ -288,6 +297,15 @@ def test_bulk_dump(self):
             """
         )
 
+    def test_bulk_dump_with_semicolon_in_filename(self):
+        self.db_hook.bulk_dump("table", "/tmp/file; SELECT * FROM DUAL")
+        self.cur.execute.assert_called_once_with(
+            """
+            SELECT * INTO OUTFILE '/tmp/file'
+            FROM table
+            """
+        )
+
     def test_serialize_cell(self):
         assert "foo" == self.db_hook._serialize_cell("foo", None)
 
@@ -311,6 +329,26 @@ def test_bulk_load_custom(self):
             """
         )
 
+    def test_bulk_load_custom_with_semicolon_in_filename(self):
+        self.db_hook.bulk_load_custom(
+            "table",
+            "/tmp/file; SELECT * FROM DUAL",
+            "IGNORE",
+            """FIELDS TERMINATED BY ';'
+            OPTIONALLY ENCLOSED BY '"'
+            IGNORE 1 LINES""",
+        )
+        self.cur.execute.assert_called_once_with(
+            """
+            LOAD DATA LOCAL INFILE '/tmp/file'
+            IGNORE
+            INTO TABLE table
+            FIELDS TERMINATED BY ';'
+            OPTIONALLY ENCLOSED BY '"'
+            IGNORE 1 LINES
+            """
+        )
+
 
 DEFAULT_DATE = timezone.datetime(2015, 1, 1)
 DEFAULT_DATE_ISO = DEFAULT_DATE.isoformat()