使用 Python 防止 SQL 注入攻击

目录

每隔几年，开放 Web 应用程序安全项目 (OWASP) 就会对最关键的Web 应用程序安全风险进行排名。自第一份报告以来，注入风险一直居于首位。在所有注入类型中，SQL 注入是最常见的攻击媒介之一，可以说是最危险的。由于 Python 是世界上最流行的编程语言之一，因此了解如何防止 Python SQL 注入至关重要。

在本教程中，您将学习：

• 什么是Python SQL 注入以及如何防止它
• 如何使用文字和标识符作为参数组合查询
• 如何安全地在数据库中执行查询

本教程适用于所有数据库引擎的用户。这里的例子使用的是 PostgreSQL，但结果可以在其他数据库管理系统（如SQLite、MySQL、 Microsoft SQL Server 、 Oracle 等）中重现。

了解 Python SQL 注入

SQL注入攻击是一个如此常见的安全漏洞，传奇的xkcd网络漫画专门为它写了一个漫画：

生成和执行SQL 查询是一项常见任务。然而，世界各地的公司在编写 SQL 语句时经常犯可怕的错误。虽然ORM 层通常组成 SQL 查询，但有时您必须自己编写。

当您使用 Python 直接在数据库中执行这些查询时，您可能会犯可能危及系统的错误。在本教程中，您将学习如何成功实现组成动态 SQL 查询的函数，而不会使您的系统面临 Python SQL 注入的风险。

设置数据库

首先，您将设置一个新的 PostgreSQL 数据库并用数据填充它。在整个教程中，您将使用该数据库亲眼目睹 Python SQL 注入的工作原理。

$ createdb -O postgres psycopgtest

$ psql -U postgres -d psycopgtest
psql (11.2, server 10.5)
Type "help" for help.

psycopgtest=# CREATE TABLE users (
    username varchar(30),
    admin boolean
);
CREATE TABLE

psycopgtest=# INSERT INTO users
    (username, admin)
VALUES
    ('ran', true),
    ('haki', false);
INSERT 0 2

psycopgtest=# SELECT * FROM users;
 username | admin
----------+-------
 ran      | t
 haki     | f
(2 rows)

(~/src) $ mkdir psycopgtest
(~/src) $ cd psycopgtest
(~/src/psycopgtest) $ python3 -m venv venv

(~/src/psycopgtest) $ source venv/bin/activate
(~/src/psycopgtest) $ python -m pip install psycopg2>=2.8.0
Collecting psycopg2
  Using cached https://....
  psycopg2-2.8.2.tar.gz
Installing collected packages: psycopg2
  Running setup.py install for psycopg2 ... done
Successfully installed psycopg2-2.8.2

import psycopg2

connection = psycopg2.connect(
    host="localhost",
    database="psycopgtest",
    user="postgres",
    password=None,
)
connection.set_session(autocommit=True)

from django.db import connection

>>> with connection.cursor() as cursor:
...     cursor.execute('SELECT COUNT(*) FROM users')
...     result = cursor.fetchone()
... print(result)
(2,)

# BAD EXAMPLE. DON'T DO THIS!
def is_admin(username: str) -> bool:
    with connection.cursor() as cursor:
        cursor.execute("""
            SELECT
                admin
            FROM
                users
            WHERE
                username = '%s'
        """ % username)
        result = cursor.fetchone()
    admin, = result
    return admin

>>> is_admin('haki')
False
>>> is_admin('ran')
True

>>> is_admin('foo')
Traceback (most recent call last):
  File "<stdin>", line 1, in <module>
  File "<stdin>", line 12, in is_admin
TypeError: cannot unpack non-iterable NoneType object

# BAD EXAMPLE. DON'T DO THIS!
def is_admin(username: str) -> bool:
    with connection.cursor() as cursor:
        cursor.execute("""
            SELECT
                admin
            FROM
                users
            WHERE
                username = '%s'
        """ % username)
        result = cursor.fetchone()

    if result is None:
        # User does not exist
        return False

    admin, = result
    return admin

>>> is_admin('haki')
False
>>> is_admin('ran')
True
>>> is_admin('foo')
False

使用 Python SQL 注入利用查询参数

在前面的示例中，您使用字符串插值来生成查询。然后，您执行查询并将结果字符串直接发送到数据库。但是，在此过程中您可能忽略了某些内容。

回想一下username你传递给的论点is_admin()。这个变量到底代表什么？您可能认为这username只是一个表示实际用户名的字符串。但是，正如您即将看到的，入侵者可以轻松利用这种疏忽，并通过执行 Python SQL 注入造成重大危害。

尝试检查以下用户是否是管理员：

>>> is_admin("'; select true; --")
True

等等……刚刚发生了什么？

让我们再看一下实现。打印出数据库中正在执行的实际查询：

>>> print("select admin from users where username = '%s'" % "'; select true; --")
select admin from users where username = ''; select true; --'

结果文本包含三个语句。要准确了解 Python SQL 注入的工作原理，您需要单独检查每个部分。第一条语句如下：

select admin from users where username = '';

这是您想要的查询。分号 ( ;) 终止查询，因此此查询的结果无关紧要。接下来是第二个语句：

select true;

这个语句是由入侵者构建的。它旨在始终返回True。

最后，您会看到这段简短的代码：

--'

此代码段消除了它之后的任何内容。入侵者添加了注释符号 ( --) 以将您可能在最后一个占位符之后放置的所有内容转换为注释。

当您使用此参数执行函数时，它将始终返回True。例如，如果您在登录页面中使用此功能，入侵者可以使用 username 登录'; select true; --，并且他们将被授予访问权限。

如果你认为这很糟糕，它可能会变得更糟！了解您的表结构的入侵者可以使用 Python SQL 注入来造成永久性损坏。例如，入侵者可以注入一条更新语句来改变数据库中的信息：

>>> is_admin('haki')
False
>>> is_admin("'; update users set admin = 'true' where username = 'haki'; select true; --")
True
>>> is_admin('haki')
True

让我们再次分解它：

';

这个片段终止了查询，就像之前的注入一样。接下来的声明如下：

update users set admin = 'true' where username = 'haki';

本节更新admin到true用户haki。

最后，有这个代码片段：

select true; --

与前面的示例一样，这一段返回true并注释掉它后面的所有内容。

为什么这更糟？好吧，如果入侵者设法使用此输入执行该功能，那么用户haki将成为管理员：

psycopgtest=# select * from users;
 username | admin
----------+-------
 ran      | t
 haki     | t
(2 rows)

入侵者不再需要使用 hack。他们只需使用用户名登录即可haki。（如果入侵者真的想造成伤害，那么他们甚至可以发出DROP DATABASE命令。）

在你忘记之前，恢复haki到原来的状态：

psycopgtest=# update users set admin = false where username = 'haki';
UPDATE 1

那么，为什么会发生这种情况？那么，你对username争论有什么了解？您知道它应该是一个表示用户名的字符串，但您实际上并未检查或强制执行此断言。这可能很危险！这正是攻击者试图入侵您的系统时所寻找的。

>>> # BAD EXAMPLE. DON'T DO THIS!
>>> username = username.replace("'", "''")

 1def is_admin(username: str) -> bool:
 2    with connection.cursor() as cursor:
 3        cursor.execute("""
 4            SELECT
 5                admin
 6            FROM
 7                users
 8            WHERE
 9                username = %(username)s
10        """, {
11            'username': username
12        })
13        result = cursor.fetchone()
14
15    if result is None:
16        # User does not exist
17        return False
18
19    admin, = result
20    return admin

>>> is_admin('haki')
False
>>> is_admin('ran')
True
>>> is_admin('foo')
False
>>> is_admin("'; select true; --")
False

>>> with connection.cursor() as cursor:
...    cursor.execute("""
...        SELECT
...            admin
...        FROM
...            users
...        WHERE
...            username = %(username)s
...    """, {
...        'username': "'; select true; --"
...    })
...    print(cursor.query.decode('utf-8'))
SELECT
    admin
FROM
    users
WHERE
    username = '''; select true; --'

# BAD EXAMPLES. DON'T DO THIS!
cursor.execute("SELECT admin FROM users WHERE username = '" + username + '");
cursor.execute("SELECT admin FROM users WHERE username = '%s' % username);
cursor.execute("SELECT admin FROM users WHERE username = '{}'".format(username));
cursor.execute(f"SELECT admin FROM users WHERE username = '{username}'");

# SAFE EXAMPLES. DO THIS!
cursor.execute("SELECT admin FROM users WHERE username = %s'", (username, ));
cursor.execute("SELECT admin FROM users WHERE username = %(username)s", {'username': username});

使用 SQL 组合

到目前为止，您已经将参数用于文字。文字是数字、字符串和日期等值。但是，如果您有一个需要编写不同查询的用例（其中参数是其他内容，例如表名或列名），该怎么办？

受上一个示例的启发，让我们实现一个函数，该函数接受表名并返回该表中的行数：

# BAD EXAMPLE. DON'T DO THIS!
def count_rows(table_name: str) -> int:
    with connection.cursor() as cursor:
        cursor.execute("""
            SELECT
                count(*)
            FROM
                %(table_name)s
        """, {
            'table_name': table_name,
        })
        result = cursor.fetchone()

    rowcount, = result
    return rowcount

尝试在您的用户表上执行该函数：

Traceback (most recent call last):
  File "<stdin>", line 1, in <module>
  File "<stdin>", line 9, in count_rows
psycopg2.errors.SyntaxError: syntax error at or near "'users'"
LINE 5:                 'users'
                        ^

该命令未能生成 SQL。正如您已经看到的，数据库适配器将变量视为字符串或文字。但是，表名不是纯字符串。这就是SQL 组合的用武之地。

您已经知道使用字符串插值来编写 SQL 是不安全的。幸运的是，Psycopg 提供了一个模块psycopg.sql来帮助您安全地编写SQL 查询。让我们使用psycopg.sql.SQL()以下方法重写函数：

from psycopg2 import sql

def count_rows(table_name: str) -> int:
    with connection.cursor() as cursor:
        stmt = sql.SQL("""
            SELECT
                count(*)
            FROM
                {table_name}
        """).format(
            table_name = sql.Identifier(table_name),
        )
        cursor.execute(stmt)
        result = cursor.fetchone()

    rowcount, = result
    return rowcount

此实现有两个不同之处。首先，您用于sql.SQL()编写查询。然后，您用来sql.Identifier()注释参数 value table_name。（标识符是列名或表名。）

现在，尝试执行users表上的函数：

>>> count_rows('users')
2

伟大的！接下来，让我们看看当表不存在时会发生什么：

>>> count_rows('foo')
Traceback (most recent call last):
  File "<stdin>", line 1, in <module>
  File "<stdin>", line 11, in count_rows
psycopg2.errors.UndefinedTable: relation "foo" does not exist
LINE 5:                 "foo"
                        ^

该函数抛出UndefinedTable异常。在以下步骤中，您将使用此异常作为您的函数免受 Python SQL 注入攻击的指示。

总而言之，添加一个选项来计算表中的行数，直到达到一定的限制。此功能可能对非常大的表很有用。要实现这一点，请LIMIT向查询添加一个子句，以及限制值的查询参数：

from psycopg2 import sql

def count_rows(table_name: str, limit: int) -> int:
    with connection.cursor() as cursor:
        stmt = sql.SQL("""
            SELECT
                COUNT(*)
            FROM (
                SELECT
                    1
                FROM
                    {table_name}
                LIMIT
                    {limit}
            ) AS limit_query
        """).format(
            table_name = sql.Identifier(table_name),
            limit = sql.Literal(limit),
        )
        cursor.execute(stmt)
        result = cursor.fetchone()

    rowcount, = result
    return rowcount

在此代码块中，您limit使用sql.Literal(). 与前面的示例一样，psycopg使用简单方法时，将所有查询参数绑定为文字。但是，在使用 时sql.SQL()，您需要使用sql.Identifier()或显式注释每个参数sql.Literal()。

执行该函数以确保其正常工作：

>>> count_rows('users', 1)
1
>>> count_rows('users', 10)
2

现在您看到该函数正在运行，请确保它也是安全的：

>>> count_rows("(select 1) as foo; update users set admin = true where name = 'haki'; --", 1)
Traceback (most recent call last):
  File "<stdin>", line 1, in <module>
  File "<stdin>", line 18, in count_rows
psycopg2.errors.UndefinedTable: relation "(select 1) as foo; update users set admin = true where name = '" does not exist
LINE 8:                     "(select 1) as foo; update users set adm...
                            ^

此回溯显示psycopg转义值，并且数据库将其视为表名。由于不存在具有此名称的表，因此UndefinedTable引发了异常并且您没有被黑！

结论

您已经成功实现了一个组合动态 SQL 的函数，而不会使您的系统面临 Python SQL 注入的风险！您在查询中同时使用了文字和标识符，而没有影响安全性。

你已经学到：

• 什么是Python SQL 注入以及如何利用它
• 如何使用查询参数防止 Python SQL 注入
• 如何安全地编写使用文字和标识符作为参数的SQL 语句

您现在可以创建可以抵御外部攻击的程序。前进并挫败黑客！