如何避免 SQL 被 "SELECT * FROM {table_name}" 注入?

How to avoid SQL injection with "SELECT * FROM {table_name}"?

在 Python 中使用带有以下代码的 Psycopg2:

import psycopg2

import getpass

conn = psycopg2.connect("dbname=mydb user=%s" % getpass.getuser())
cursor = conn.cursor()

tables = ["user", "group", "partner", "product"]
for table in tables:

    # with sql injection
    cursor.execute("SELECT name FROM %s LIMIT 1" % (table,))
    print "table", table, "result", len(cursor.fetchone())

    # without sql injection
    cursor.execute("SELECT name FROM %s LIMIT 1", (table,))
    print "table", table, "result", len(cursor.fetchone())

输出为:

table res_partner result 1
Traceback (most recent call last):
  File "my_psycopg2_example.py", line 16, in <module>
    cursor.execute("SELECT name FROM %s LIMIT 1", (table,))
psycopg2.ProgrammingError: syntax error at or near "'res_partner'"
LINE 1: SELECT name FROM 'res_partner' LIMIT 1

使用 SQL 注入效果很好。

但我们不想造成安全问题。

我们阅读了 this documentation 并在其中发现了以下评论:

Only variable values should be bound via this method: it shouldn’t be used to set table or field names. For these elements, ordinary string formatting should be used before running execute().

但是如果我们使用 "ordinary string formatting",我们也会有 SQL 注入。

管理这种特殊情况并避免 SQL 注入的好方法是什么?

我认为您混淆了 SQL 注入的定义。 SQL 注入是对您的软件的 攻击 ,有人会导致您的 SQL 查询执行您不希望它执行的操作。字符串插值不是 SQL 注入。字符串插值 有时可以 启用 SQL 注入,但并非总是如此。要了解字符串插值并不总是不安全的,请考虑以下哪项最安全:

  1. sql = 'SELECT name FROM user'
  2. sql = 'SELECT name FROM ' + 'user'
  3. sql = 'SELECT name FROM %s' % ['user']
  4. sql = 'SELECT name FROM {}'.format('user')

这些代码行中的每一行都做完全相同的事情,因此 none 其中的代码可能比其他代码更安全或更不安全。在您的确切示例中,没有 SQL 注入的危险,因为您只是在构建硬编码的 SQL 查询字符串。

另一方面,如果您的 table 值来自用户,则可能存在安全问题:

  • 如果他们传递了一个存在的 table 的名称,但您不想让他们查询怎么办?

    table = 'secrets'
    sql = 'SELECT name FROM %s LIMIT 1' % table
    

    结果:

    SELECT name FROM secrets LIMIT 1
    
  • 如果他们传递的 something 实际上不是 table 名称怎么办?

    table = 'product; DROP TABLE user; --'
    sql = 'SELECT name FROM %s LIMIT 1' % table
    

    结果:

    SELECT name FROM product;
    DROP TABLE user;
    -- LIMIT 1
    

您可以通过检查是否允许 table 名称来防止这种情况发生:

if table.lower() not in ["user", "group", "partner", "product"]:
    raise Something('Bad table name: %r' % table)

execute函数中使用psycopg2查询参数是最安全的,当参数以字面量形式使用时使用起来更方便。

cursor.mogrify("select * from foo where bar = %s", ('example',))
# yields "select * from foo where bar = 'example'"

(注意 cursor.mogrify() 的行为类似于execute,只是显示格式化后的SQL,并没有真正执行它)

但是,当您希望参数是 table、模式或其他标识符时,这会有点棘手。您可以使用 AsIs 来包装您的参数,但这仍然为 SQL 注入敞开大门。

from psycopg2.extensions import AsIs

cur.mogrify('select %s from foo;', (AsIs('* from dual; drop table students; --'),))
# yields 'select * from dual; drop table students; -- from foo;'

看起来新开发的 psycopg2 (>=2.7) 将有一个标识符 class,您可以将参数包装在其中并希望是安全的。如果它尚未发布,或者您没有,可以通过以下方式创建您自己的 class。我将在下面给出一些片段,但你也可以看到 my gist

import re
import psycopg2.extensions

class NotSqlIdentifierError(Exception):
    pass

valid_pattern = r'^[a-zA-Z_][a-zA-Z0-9_$]*$'

class QuotedIdentifier(object):
    def __init__(self, obj_str):
        self.obj_str = obj_str

    def getquoted(self):
        if re.match(valid_pattern, self.obj_str):
            return self.obj_str
        else:
            raise NotSqlIdentifierError(repr(self.obj_str))

psycopg2.extensions.register_adapter(QuotedIdentifier, lambda x: x)

如果你已经有一个 psycopg2 游标实例,你可以test/use这样:

# Test that a valid identifier formats into string
cursor.mogrify('select %s from foo;', (QuotedIdentifier('bar'),))
# returns 'select bar from foo;'

# Test formatting both an identifier and a literal
cursor.mogrify(
    'select * from foo where %s = %s;', 
    (
        QuotedIdentifier('bar'),
        'example'
    )
)
# returns "select * from foo where bar = 'example';"

# Test that a non-valid identifier fails with exception
cursor.mogrify('select %s from foo;', (QuotedIdentifier('* from dummy; drop table students; --'),))
"""Returns following:
---------------------------------------------------------------------------
NotSqlIdentifierError                     Traceback (most recent call last)
<ipython-input-14-d6a960dc458a> in <module>()
----> 1 cur.mogrify('select %s from foo;', (QuotedIdentifier('* from dummy; drop table students; --'),))
<ipython-input-12-0a1327cbaf78> in getquoted(self)
     18             return self.obj_str
     19         else:
---> 20             raise NotSqlIdentifierError(repr(self.obj_str))
     21 
     22 psycopg2.extensions.register_adapter(QuotedIdentifier, lambda x: x)
NotSqlIdentifierError: '* from dummy; drop table students; --'
"""

有关自定义 classes 包装 SQL 参数的机制的更多信息,请参阅文档中的 this section

这是一篇关于注入和 python 代码的好文章 https://realpython.com/prevent-python-sql-injection/

你可以变身

tables = ["user", "group", "partner", "product"]
for table in tables:    
    cursor.execute("SELECT name FROM %s LIMIT 1" % (table,))

from psycopg2 import sql

tables = ["user", "group", "partner", "product"]
for table_name in tables:        
    stmt = sql.SQL("SELECT name FROM {table_name} LIMIT 1").format(
        table_name = sql.Identifier(table_name),
    )
    cursor.execute(stmt)