脚本专栏 
首页 > 脚本专栏 > 浏览文章

使用Python防止SQL注入攻击的实现示例

(编辑：jimmy 日期: 2024/11/18 浏览：3 次 )

文章背景

每隔几年，开放式Web应用程序安全项目就会对最关键的Web应用程序安全风险进行排名。自第一次报告以来，注入风险高居其位！在所有注入类型中，SQL注入是最常见的攻击手段之一，而且是最危险的。由于Python是世界上最流行的编程语言之一，因此了解如何防止Python SQL注入对于我们来说还是比较重要的

那么在写这篇文章的时候我也是查询了国内外很多资料，最后带着问题去完善总结：

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

文章演示的操作适用于所有数据库，这里的示例使用的是PG，但是效果跟过程可以在其他数据库（例如SQLite，MySQL，Oracle等等系统中）重现

1. 了解Python SQL注入

"text-align: center">使用Python防止SQL注入攻击的实现示例

2. 设置数据库

首先，建立一个新的PostgreSQL数据库并用数据填充它。在文章中，将使用该数据库直接见证Python SQL注入的工作方式及基本操作

2.1 创建数据库

打开你的shell工具并创建一个用户拥有的新PostgreSQL数据库：

$ createdb -O postgres psycopgtest

在这里，使用了命令行选项-O将数据库的所有者设置为用户postgres。还指定了数据库的名称psycopgtest

postgres是一个特殊用户，通常将保留该用户用于管理任务，但是对于本文章而言，可以使用postgres。但是，在实际系统中，应该创建一个单独的用户作为数据库的所有者

新数据库已准备就绪！现在我们连接它：

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

现在，可以看到以psycopgtest用户身份连接到数据库postgres。该用户也是数据库所有者，因此将具有数据库中每个表的读取权限

2.2 构造数据创建表

这里我们需要创建一个包含一些用户信息的表，并向其中添加一些数据：

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

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

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

我们添加了username和admin两个列。该admin列指示用户是否具有管理特权。我们的目标是瞄准该admin领域并尝试滥用它

2.3 设置Python虚拟环境

现在我们已经有了一个数据库，是时候设置Python环境。在新目录中创建虚拟环境：

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

运行此命令后，venv将创建一个名为的新目录。该目录将存储在虚拟环境中安装的所有软件包

2.4 使用Python连接数据库

再使用Python连接PostgreSQL数据库时需要确保我们的环境是否安装了psycopg2，如果没有使用pip安装psycopg2：

pip install psycopg2

安装完之后，我们编写创建与数据库连接的代码：

import psycopg2

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

psycopg2.connect()函数用来创建与数据库的连接且接受以下参数：

  • host是数据库所在服务器的IP地址
  • database是要连接的数据库的名称
  • user是具有数据库权限的用户
  • password连接数据库的密码

我们设置完连接后，使用配置了会话autocommit=True。激活autocommit意味着不必通过发出commit或来手动管理rollback。这是 大多数ORM中的默认 行为。也可以在这里使用此行为，以便可以专注于编写SQL查询而不是管理事务

2.5 执行查询

现在我们已经连接到了数据库，开始执行我们的查询：

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

使用该connection对象创建了一个cursor。就像Python中的文件操作一样，cursor是作为上下文管理器实现的。创建上下文时，将cursor打开一个供使用以将命令发送到数据库。当上下文退出时，将cursor关闭，将无法再使用它

Python with语句的实现感兴趣的朋友可以自己查询一下

在上下文中时，曾经cursor执行查询并获取结果。在这种情况下，发出查询以对users表中的行进行计数。要从查询中获取结果，执行cursor.fetchone()并接收了一个元组。由于查询只能返回一个结果，因此使用fetchone()。如果查询返回的结果不止一个，那么我们就需要迭代cursor

3. 在SQL中使用查询参数

现在我们创建了数据库并且建立了与数据库的连接，并执行了查询。但是我们使用的查询是静态的。换句话说，它没有参数。现在，将开始在查询中使用参数

首先，将实现一个检查用户是否为管理员的功能。is_admin()接受用户名并返回该用户的管理员状态：

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

此函数执行查询以获取admin给定用户名的列的值。曾经fetchone()返回一个具有单个结果的元组。然后，将此元组解压缩到变量中admin。要测试的功能，请检查用户名：

> is_admin('lisi')
False
> is_admin('zhangsan')
True

到目前为止，一切都是正常的。该函数返回了两个用户的预期结果。但是我们如果查看不存在的用户呢？看下会怎样：

> is_admin('wangwu')
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

当用户不存在时可以看到出现了异常，这是因为如果找不到结果，则.fetchone()返回None，导致引发TypeError

要处理不存在的用户，我们可以创建一个特例None：

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:
    return False

  admin, = result
  return admin

在这里，添加了处理的特殊情况None。如果username不存在，则该函数应返回False。再次在某些用户上测试该功能：

> is_admin('lisi')
False
> is_admin('zhangsan')
True
> is_admin('wangwu')
False

可以发现这个函数现在已经可以处理不存在的用户名

4. 使用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。例如，如果我们在登录页面中使用此功能，则入侵者可以使用用户名登录'; select true; --，并将被授予访问权限。

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

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

让我们再次分解：

';

就像之前的注入一样，此代码段终止了查询。下一条语句如下：

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

更新admin到true用户lisi

最后，有以下代码片段：

select true; --

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

如果入侵者设法使用此输入执行功能，则用户lisi将成为管理员：

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

入侵者可以使用用户名登录lisi。（如果入侵者确实想破坏，那么可以使用DROP DATABASE命令）

现在我们恢复lisi的原始状态：

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

4.1 制作安全查询参数

了解了入侵者如何通过使用精心设计的字符串来利用系统并获得管理员权限。问题是我们允许从客户端传递的值直接执行到数据库，而无需执行任何类型的检查或验证。SQL注入依赖于这种类型的漏洞

每当在数据库查询中使用用户输入时，SQL注入就可能存在漏洞。防止Python SQL注入的关键是确保该值已按我们开发的预期使用。在上一个示例中，username用作了字符串。实际上，它被用作原始SQL语句

为了确保我们按预期使用值，需要对值进行转义。例如，为防止入侵者将原始SQL替换为字符串参数，可以对引号进行转义：

> username = username.replace("'", "''")

这只是一个例子。尝试防止Python SQL注入时，有很多特殊字符和场景需要考虑。现代的数据库适配器随附了一些内置工具，这些工具可通过使用查询参数来防止Python SQL注入。使用这些参数代替普通字符串插值可组成带有参数的查询

现在，我们已经对该漏洞有了一个明确的知晓，可以使用查询参数而不是字符串插值来重写该函数：

def is_admin(username: str) -> bool:
  with connection.cursor() as cursor:
    cursor.execute("""
      SELECT
        admin
      FROM
        users
      WHERE
        username = %(username)s
    """, {
      'username': username
    })
    result = cursor.fetchone()

  if result is None:
    return False

  admin, = result
  return admin

我们使用了一个命名参数username来指示用户名应该去哪里

将值username作为第二个参数传递给cursor.execute()。username在数据库中执行查询时，连接将使用的类型和值
要测试此功能，我们先尝试一些有效以及无效的值跟一些有隐患的字符串：

> is_admin('lisi')
False
> is_admin('zhangsan')
True
> is_admin('wangwu')
False
> is_admin("'; select true; --")
False

跟我们想象的一毛一样！该函数返回所有值的预期结果。并且，隐患的字符串不再起作用。要了解原因，可以检查由生成的查询execute()：

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; --'

该连接将值username视为字符串，并转义了可能终止该字符串的所有字符并引入了Python SQL注入

4.2 传递安全查询参数

数据库适配器通常提供几种传递查询参数的方法。命名占位符通常是可读性最好的，但是某些实现可能会受益于使用其他选项

让我们快速看一下使用查询参数的一些对与错方法。以下代码块显示了我们需要避免的查询类型：

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}'");

这些语句中的每条语句都username直接从客户端传递到数据库，而无需执行任何类型的检查或验证。这类代码已经可以达到Python SQL注入

相比上面，以下类型的查询可以安全地执行：

cursor.execute("SELECT admin FROM users WHERE username = %s'", (username, ));
cursor.execute("SELECT admin FROM users WHERE username = %(username)s", {'username': username});

在这些语句中，username作为命名参数传递。现在，数据库将username在执行查询时使用指定的类型和值，从而提供针对Python SQL注入的保护

5. 使用SQL组合

但是，如果我们有一个用例需要编写一个不同的查询（该参数是其他参数，例如表或列名），该怎么办？

继上一个列子，我们实现一个函数，该函数接受表的名称并返回该表中的行数：

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()对参数值进行注释table_name（标识符是列或表的名称）

现在，我们尝试在users表上执行该函数：

> count_rows('users')
2

接下来，让我们看看表不存在时会发生什么：

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

该函数引发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()

不幸的是，Python API规范不解决标识符的绑定，仅处理文字。Psycopg是唯一流行的适配器，它添加了使用文字和标识符安全地组合SQL的功能。这个事实使得在绑定标识符时要特别注意

执行该函数以确保其起作用：

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

现在我们已经看到该函数正在运行，检查它是否安全：

> count_rows("(select 1) as wangwu; update users set admin = true where name = 'lisi'; --", 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 wangwu; update users set admin = true where name = '" does not exist
LINE 8:                     "(select 1) as wangwu; update users set adm...
                            ^                     

异常显示psycopg转义了该值，并且数据库将其视为表名。由于不存在具有该名称的表，因此UndefinedTable引发了异常所以是安全的！

6. 结论

通过实现组成动态SQL，可与你使我们有效的规避系统遭受Python SQL注入的威胁！在查询过程中同时使用文字和标识符，并不会影响安全性

7. 致谢

上一篇：Python logging模块写入中文出现乱码
下一篇：django的403/404/500错误自定义页面的配置方式
友情链接:杰晶网络 DDR爱好者之家 南强小屋 黑松山资源网 白云城资源网 SiteMap