BUG#37145655: MySQL Connector/Python Configuration Files RCE

Change-Id: I7e41ba87775e8e7ef871e797ab2cd5adfd48645f

Author: SubCoder1

CHANGES.txt

@@ -12,6 +12,7 @@ v9.2.0
 ======
 
 - WL16285: Remake Multi Statement Execution
+- BUG#37145655: MySQL Connector/Python Configuration Files RCE
 - BUG#36126909: "Unread result found" exception/bad MySQLCursor.statement when query text contains code comments
 - BUG#35810050: Executing multiple statements fails when importing Sakila
 

mysql-connector-python/lib/mysql/connector/optionfiles.py

@@ -30,6 +30,7 @@
 
 """Implements parser to parse MySQL option files."""
 
+import ast
 import codecs
 import io
 import os
@@ -74,9 +75,11 @@ def read_option_files(**config: Union[str, List[str]]) -> Dict[str, Any]:
         for group in groups:
             try:
                 for option, value in config_from_file[group].items():
+                    value += (group,)
                     try:
                         if option == "socket":
                             option = "unix_socket"
+                            option_parser.set(group, "unix_socket", value[0])
 
                         if option not in CNX_POOL_ARGS and option != "failover":
                             _ = DEFAULT_CONFIGURATION[option]
@@ -94,17 +97,22 @@ def read_option_files(**config: Union[str, List[str]]) -> Dict[str, Any]:
             except KeyError:
                 continue
 
-        not_evaluate = ("password", "passwd")
-        for option, value in config_options.items():
-            if option not in config:
-                try:
-                    if option in not_evaluate:
-                        config[option] = value[0]
-                    else:
-                        config[option] = eval(value[0])  # pylint: disable=eval-used
-                except (NameError, SyntaxError):
-                    config[option] = value[0]
-
+        for option, values in config_options.items():
+            value, _, section = values
+            if (
+                option not in config
+                and option_parser.has_section(section)
+                and option_parser.has_option(section, option)
+            ):
+                if option in ("password", "passwd"):  # keep the value as string
+                    config[option] = str(value)
+                else:
+                    try:
+                        config[option] = ast.literal_eval(value)
+                    except (ValueError, TypeError, SyntaxError):
+                        config[option] = value
+        if "socket" in config:
+            config["unix_socket"] = config.pop("socket")
     return config
 
 

mysql-connector-python/tests/data/option_files/bug_37145655.cnf

@@ -0,0 +1,27 @@
+[correct_config]
+user=root
+password=mypass
+database=cpydata
+port=10000
+failover=[{"host": "127.0.1.2", "port": 3306, "user": "root", "priority": 80},{"host": "127.0.1.5", "port": 2450, "user": "root_123", "priority": 10},]
+
+[correct_config_with_unix_socket]
+user=root
+password=mypass
+database=cpydata
+port=10000
+unix_socket=unix_socket_path_for_bug37145655
+
+[correct_config_with_socket]
+user=root
+password=mypass
+database=cpydata
+port=10000
+socket=unix_socket_path_for_bug37145655
+
+[incorrect_config]
+user=root
+password=mypass
+database=cpydata
+port=int(10000)
+allow_local_infile=__import__('os').system('whoami')

mysql-connector-python/tests/test_bugs.py

@@ -3527,10 +3527,8 @@ def test_unsupported_arguments(self):
             )
         }
         exp.update(config)
-
         self.assertEqual(exp, new_config)
 
-
 class BugOra21530100(tests.MySQLConnectorTests):
     """BUG#21530100: CONNECT FAILS WHEN USING MULTIPLE OPTION_GROUPS WITH
     PYTHON 3.3
@@ -8911,3 +8909,53 @@ def test_incorrect_host_err_msg(self):
                 str(err),
                 "Error message cannot contain unstructured bind address",
             )
+
+class BugOra37145655(tests.MySQLConnectorTests):
+    """BUG#37145655: MySQL Connector/Python Configuration Files RCE"""
+
+    def test_read_optionfiles(self):
+        option_file_dir = os.path.join("tests", "data", "option_files")
+        opt_file = os.path.join(option_file_dir, "bug_37145655.cnf")
+        exp = {
+            "user": "root",
+            "password": "mypass",
+            "database": "cpydata",
+            "port": 10000,
+            "failover": [
+                {
+                    "host": "127.0.1.2",
+                    "port": 3306,
+                    "user": "root",
+                    "priority": 80,
+                },
+                {
+                    "host": "127.0.1.5",
+                    "port": 2450,
+                    "user": "root_123",
+                    "priority": 10,
+                },
+            ],
+        }
+        self.assertEqual(
+            exp, read_option_files(option_files=opt_file, option_groups=["correct_config"])
+        )
+        exp.pop("failover")
+        exp["unix_socket"] = "unix_socket_path_for_bug37145655"
+        self.assertEqual(
+            exp, read_option_files(option_files=opt_file, option_groups=["correct_config_with_unix_socket"])
+        )
+        self.assertEqual(
+            exp, read_option_files(option_files=opt_file, option_groups=["correct_config_with_socket"])
+        )
+        del exp["unix_socket"]
+        exp["port"] = "int(10000)"
+        exp["allow_local_infile"] = "__import__('os').system('whoami')"
+        self.assertEqual(
+            exp, read_option_files(option_files=opt_file, option_groups=["incorrect_config"])
+        )
+        with self.assertRaises((errors.InterfaceError, TypeError),) as _:
+            with mysql.connector.connect(
+                read_default_file=opt_file,
+                option_groups=["incorrect_config"],
+            ) as _:
+                pass