pymysql 语句谁是正确的？

Question

我创建了一个名为 test 的测试数据库，其中有一个名为 testTable 的表，其中包含一个自动增量 id 值和一个采用 varchar(30) 的名称字段。

PREPARE 语句查询（其中 4 个）在复制到 phpmyadmin 时执行良好，但我收到错误 ???? 2021-01-08 18:26:53,022 (MainThread) [ERROR] (1064, "您的 SQL 语法有错误；请查看与您的 MySQL 服务器版本相对应的手册以获取在 'SET 附近使用的正确语法\n @name = 'fred';\nEXECUTE\n 语句 USING @name;\nDEALLOCATE\nPREPARE\n ' 在第 5 行")

测试代码：

import pymysql
import logging


class TestClass():

    def __init__(self):
        # mysqlconnections

        self.mySQLHostName = "localhost" 
        self.mySQLHostPort = 3306
        self.mySQLuserName = "userName"
        self.mySQLpassword = "pass"
        self.MySQLauthchandb = "mysql"
        
    def QueryMYSQL (self, query):
        try:
            #logging.info("QueryMYSQL  : " + str( query)) # Uncomment to print all mysql queries sent
            conn = pymysql.connect(host=self.mySQLHostName, port=self.mySQLHostPort, user=self.mySQLuserName, passwd=self.mySQLpassword, db=self.MySQLauthchandb, charset='utf8')
            conn.autocommit(True)
            cursor = conn.cursor()
            if cursor:
                returnSuccess = cursor.execute(query)
                
            if cursor:
                returnValue = cursor.fetchall()
            #logging.info ("return value : " + str(returnValue)) # Uncomment to print all returned mysql queries
            if cursor:
                cursor.close()
            if conn:
                conn.close()
            return returnValue
        except Exception as e:
            logging.error("Problem in ConnectTomySQL")
            logging.error(query)
            logging.error(e)
            return False

# Default error logging log file location:
logging.basicConfig(format='%(asctime)s (%(threadName)-10s) [%(levelname)s] %(message)s', filename= 'ERROR.log',filemode = "w", level=logging.DEBUG)
logging.info("Logging Started")

test = TestClass()
result = test.QueryMYSQL("Describe test.testTable")
print(result)
query = """
PREPARE
    statement
FROM
    'INSERT INTO test.testTable (id, name) VALUES (NULL , ?)';
SET
    @name = 'fred';
EXECUTE
    statement USING @name;
DEALLOCATE
PREPARE
    statement;
    """
result = test.QueryMYSQL(query)
print(result)

我假设这是一个库问题而不是 mysql 问题？我正在尝试使用准备好的语句来防止来自用户输入的代码注入，因为我知道这种准备好的语句是最好的方法，而不是尝试预先过滤用户输入并遗漏一些东西。

我在 github 上问过这个问题，但其中一位作者（甲烷 Inada Naoki）回答说：

========

当存在查询注入漏洞时，攻击者可以使用多语句。所以默认是关闭的。

as I understand this prepared statements are the best way

你完全错了。您使用准备好的语句根本不能保护您免受 SQL 注入。如果启用多语句，您的“准备语句”可能会受到 SQL 注入的攻击。

但我不是免费的技术支持，也不是免费的老师。 OSS 维护者不是。请不要在这里问。

他关闭了这个问题。

他说的对吗？

我正在阅读 Robin Nixon 的作者书，“Learning PHP, MySQL and JavaScript” O'Reilly 第 5 版。他似乎被误解了，我在第 260 页的使用占位符部分引用了“让我介绍与 MySQL 交互的最佳和推荐方式，这在安全性方面几乎是防弹的”。他错了吗？

因为我买这本书是为了改进我的安全实践，现在我不确定什么是正确的。

Answer 1

我从pymysql的开发者那里得知该库不支持PREPARE mysql语句。另外pymysql库默认不执行多语句。

我了解，如果启用了多语句，我第一次尝试将值替换到 INSERT 语句中本质上是不安全的。这可以通过在连接构造函数中使用 client_flag=pymysql.constants.CLIENT.MULTI_STATEMENTS 来完成。

然而，pymysql 库允许使用 cursor.execute(query, (tuple)) 方法在 MySQL 查询中使用占位符。

为了证明这一点，我编写了以下测试代码示例。

import pymysql
import logging

class TestClass():

    def __init__(self):
        # mysqlconnections

        self.mySQLHostName = "localhost"
        self.mySQLHostPort = 3306
        self.mySQLuserName = "name"
        self.mySQLpassword = "pw"
        self.MySQLauthchandb = "mysql"
        
    def QueryMYSQL (self, query, data = ()):
        try:
            logging.info("QueryMYSQL  : " + str( query)) # Uncomment to print all mysql queries sent
            conn = pymysql.connect(host=self.mySQLHostName, port=self.mySQLHostPort, user=self.mySQLuserName, passwd=self.mySQLpassword, db=self.MySQLauthchandb, charset='utf8', client_flag=pymysql.constants.CLIENT.MULTI_STATEMENTS) #code injection requires multistatements to be allowed this is off in pymysql by default and has to be set on manually. 
            conn.autocommit(True)
            cursor = conn.cursor()
            if cursor:
                if data:
                    returnSuccess = cursor.execute(query, data)
                else:
                    returnSuccess = cursor.execute(query)
                
            if cursor:
                returnValue = cursor.fetchall()
            logging.info ("return value : " + str(returnValue)) # Uncomment to print all returned mysql queries
            if cursor:
                cursor.close()
            if conn:
                conn.close()
            return returnValue
        except Exception as e:
            logging.error("Problem in ConnectTomySQL")
            logging.error(e)
            logging.error(query)
            if data:
                logging.error("Data {}".format(str(data)))
            return False

# Default error logging log file location:
logging.basicConfig(format='%(asctime)s (%(threadName)-10s) [%(levelname)s] %(message)s', filename= 'ERROR.log',filemode = "w", level=logging.DEBUG)
logging.info("Logging Started")

def usePlaceholder(userInput):
    query = "INSERT INTO test.testTable (id, name) VALUES (NULL , %s)"
    data = (userInput,) 
    result = test.QueryMYSQL(query,data)
    print(result)

def useSubstitution(userInput):
    query = "INSERT INTO test.testTable (id, name) VALUES (NULL , '{}')".format(userInput) # this is unsafe.
    result = test.QueryMYSQL(query)
    print(result)

test = TestClass()

#Create the test database and testTable.
query = "CREATE DATABASE test"
test.QueryMYSQL(query)
query = "CREATE TABLE `test`.`testTable` ( `id` INT NOT NULL AUTO_INCREMENT , `name` VARCHAR(256) NULL DEFAULT NULL , PRIMARY KEY (`id`)) ENGINE = InnoDB;"
test.QueryMYSQL(query)

#Simulated user input.
legitUserEntry = "Ringo"
injectionAttempt = "333' ); INSERT INTO test.testTable (id, name) VALUES (NULL , 666);#" #A simulated user sql injection attempt.

useSubstitution(legitUserEntry) # this will also insert Ringo - but could be unsafe.
usePlaceholder(legitUserEntry) # this will insert Ringo - but is safer.

useSubstitution(injectionAttempt) # this will inject the input code and execute it. 
usePlaceholder(injectionAttempt) # this will insert the input into the database without executing the injected code.

因此，从此练习中，我将通过将多语句设置为关闭（默认）并使用占位符和数据元组而不是替换来提高我的安全性。