文章背景

每隔幾年，開(kāi)放式Web應(yīng)用程序安全項(xiàng)目就會(huì)對(duì)最關(guān)鍵的Web應(yīng)用程序安全風(fēng)險(xiǎn)進(jìn)行排名。自第一次報(bào)告以來(lái)，注入風(fēng)險(xiǎn)高居其位！在所有注入類型中，SQL注入是最常見(jiàn)的攻擊手段之一，而且是最危險(xiǎn)的。由于Python是世界上最流行的編程語(yǔ)言之一，因此了解如何防止Python SQL注入對(duì)于我們來(lái)說(shuō)還是比較重要的

那么在寫(xiě)這篇文章的時(shí)候我也是查詢了國(guó)內(nèi)外很多資料，最后帶著問(wèn)題去完善總結(jié)：

• 什么是Python SQL注入以及如何防止注入
• 如何使用文字和標(biāo)識(shí)符作為參數(shù)組合查詢
• 如何安全地執(zhí)行數(shù)據(jù)庫(kù)中的查詢

文章演示的操作適用于所有數(shù)據(jù)庫(kù)，這里的示例使用的是PG，但是效果跟過(guò)程可以在其他數(shù)據(jù)庫(kù)（例如SQLite，MySQL，Oracle等等系統(tǒng)中）重現(xiàn)

1. 了解Python SQL注入

  SQL注入攻擊是一種常見(jiàn)的安全漏洞。在我們?nèi)粘９ぷ髦猩珊蛨?zhí)行SQL查詢也同樣是一項(xiàng)常見(jiàn)的任務(wù)。但是，有時(shí)候在編寫(xiě)SQL語(yǔ)句時(shí)常常會(huì)犯下可怕錯(cuò)誤

當(dāng)我們使用Python將這些查詢直接執(zhí)行到數(shù)據(jù)庫(kù)中時(shí)，很可能會(huì)損害到系統(tǒng)。所以如何成功實(shí)現(xiàn)組成動(dòng)態(tài)SQL查詢的函數(shù)，而又不會(huì)使系統(tǒng)遭受Python SQL注入的威脅呢？

2. 設(shè)置數(shù)據(jù)庫(kù)

首先，建立一個(gè)新的PostgreSQL數(shù)據(jù)庫(kù)并用數(shù)據(jù)填充它。在文章中，將使用該數(shù)據(jù)庫(kù)直接見(jiàn)證Python SQL注入的工作方式及基本操作

2.1 創(chuàng)建數(shù)據(jù)庫(kù)

打開(kāi)你的shell工具并創(chuàng)建一個(gè)用戶擁有的新PostgreSQL數(shù)據(jù)庫(kù)：

$ createdb -O postgres psycopgtest

在這里，使用了命令行選項(xiàng)-O將數(shù)據(jù)庫(kù)的所有者設(shè)置為用戶postgres。還指定了數(shù)據(jù)庫(kù)的名稱psycopgtest

postgres是一個(gè)特殊用戶，通常將保留該用戶用于管理任務(wù)，但是對(duì)于本文章而言，可以使用postgres。但是，在實(shí)際系統(tǒng)中，應(yīng)該創(chuàng)建一個(gè)單獨(dú)的用戶作為數(shù)據(jù)庫(kù)的所有者

新數(shù)據(jù)庫(kù)已準(zhǔn)備就緒！現(xiàn)在我們連接它：

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

現(xiàn)在，可以看到以psycopgtest用戶身份連接到數(shù)據(jù)庫(kù)postgres。該用戶也是數(shù)據(jù)庫(kù)所有者，因此將具有數(shù)據(jù)庫(kù)中每個(gè)表的讀取權(quán)限

2.2 構(gòu)造數(shù)據(jù)創(chuàng)建表

這里我們需要?jiǎng)?chuàng)建一個(gè)包含一些用戶信息的表，并向其中添加一些數(shù)據(jù)：

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兩個(gè)列。該admin列指示用戶是否具有管理特權(quán)。我們的目標(biāo)是瞄準(zhǔn)該admin領(lǐng)域并嘗試濫用它

2.3 設(shè)置Python虛擬環(huán)境

現(xiàn)在我們已經(jīng)有了一個(gè)數(shù)據(jù)庫(kù)，是時(shí)候設(shè)置Python環(huán)境。在新目錄中創(chuàng)建虛擬環(huán)境：

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

運(yùn)行此命令后，venv將創(chuàng)建一個(gè)名為的新目錄。該目錄將存儲(chǔ)在虛擬環(huán)境中安裝的所有軟件包

2.4 使用Python連接數(shù)據(jù)庫(kù)

再使用Python連接PostgreSQL數(shù)據(jù)庫(kù)時(shí)需要確保我們的環(huán)境是否安裝了psycopg2，如果沒(méi)有使用pip安裝psycopg2：

pip install psycopg2

安裝完之后，我們編寫(xiě)創(chuàng)建與數(shù)據(jù)庫(kù)連接的代碼：

import psycopg2

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

psycopg2.connect()函數(shù)用來(lái)創(chuàng)建與數(shù)據(jù)庫(kù)的連接且接受以下參數(shù)：

• host是數(shù)據(jù)庫(kù)所在服務(wù)器的IP地址
• database是要連接的數(shù)據(jù)庫(kù)的名稱
• user是具有數(shù)據(jù)庫(kù)權(quán)限的用戶
• password連接數(shù)據(jù)庫(kù)的密碼

我們?cè)O(shè)置完連接后，使用配置了會(huì)話autocommit=True。激活autocommit意味著不必通過(guò)發(fā)出commit或來(lái)手動(dòng)管理rollback。這是 大多數(shù)ORM中的默認(rèn) 行為。也可以在這里使用此行為，以便可以專注于編寫(xiě)SQL查詢而不是管理事務(wù)

2.5 執(zhí)行查詢

現(xiàn)在我們已經(jīng)連接到了數(shù)據(jù)庫(kù)，開(kāi)始執(zhí)行我們的查詢：

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

使用該connection對(duì)象創(chuàng)建了一個(gè)cursor。就像Python中的文件操作一樣，cursor是作為上下文管理器實(shí)現(xiàn)的。創(chuàng)建上下文時(shí)，將cursor打開(kāi)一個(gè)供使用以將命令發(fā)送到數(shù)據(jù)庫(kù)。當(dāng)上下文退出時(shí)，將cursor關(guān)閉，將無(wú)法再使用它

Python with語(yǔ)句的實(shí)現(xiàn)感興趣的朋友可以自己查詢一下

在上下文中時(shí)，曾經(jīng)cursor執(zhí)行查詢并獲取結(jié)果。在這種情況下，發(fā)出查詢以對(duì)users表中的行進(jìn)行計(jì)數(shù)。要從查詢中獲取結(jié)果，執(zhí)行cursor.fetchone()并接收了一個(gè)元組。由于查詢只能返回一個(gè)結(jié)果，因此使用fetchone()。如果查詢返回的結(jié)果不止一個(gè)，那么我們就需要迭代cursor

3. 在SQL中使用查詢參數(shù)

現(xiàn)在我們創(chuàng)建了數(shù)據(jù)庫(kù)并且建立了與數(shù)據(jù)庫(kù)的連接，并執(zhí)行了查詢。但是我們使用的查詢是靜態(tài)的。換句話說(shuō)，它沒(méi)有參數(shù)?，F(xiàn)在，將開(kāi)始在查詢中使用參數(shù)

首先，將實(shí)現(xiàn)一個(gè)檢查用戶是否為管理員的功能。is_admin()接受用戶名并返回該用戶的管理員狀態(tài)：

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

此函數(shù)執(zhí)行查詢以獲取admin給定用戶名的列的值。曾經(jīng)fetchone()返回一個(gè)具有單個(gè)結(jié)果的元組。然后，將此元組解壓縮到變量中admin。要測(cè)試的功能，請(qǐng)檢查用戶名：

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

到目前為止，一切都是正常的。該函數(shù)返回了兩個(gè)用戶的預(yù)期結(jié)果。但是我們?nèi)绻榭床淮嬖诘挠脩裟?？看下?huì)怎樣：

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

當(dāng)用戶不存在時(shí)可以看到出現(xiàn)了異常，這是因?yàn)槿绻也坏浇Y(jié)果，則.fetchone()返回None，導(dǎo)致引發(fā)TypeError

要處理不存在的用戶，我們可以創(chuàng)建一個(gè)特例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不存在，則該函數(shù)應(yīng)返回False。再次在某些用戶上測(cè)試該功能：

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

可以發(fā)現(xiàn)這個(gè)函數(shù)現(xiàn)在已經(jīng)可以處理不存在的用戶名

4. 使用Python SQL注入利用查詢參數(shù)

在上一個(gè)示例中，使用了字符串插值來(lái)生成查詢。然后，執(zhí)行查詢并將結(jié)果字符串直接發(fā)送到數(shù)據(jù)庫(kù)。但是，在此過(guò)程中可能會(huì)忽略一些事情

回想一下username傳遞給is_admin()。這個(gè)變量究竟代表什么？我們可能會(huì)認(rèn)為這username只是代表實(shí)際用戶名的字符串。但是，正如我們將要看到的，入侵者可以通過(guò)執(zhí)行Python SQL注入輕松利用這種監(jiān)督并造成破壞

嘗試檢查以下用戶是否是管理員：

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

等等…發(fā)生了什么事？

讓我們?cè)倏匆幌聦?shí)現(xiàn)。打印出數(shù)據(jù)庫(kù)中正在執(zhí)行的實(shí)際查詢：

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

結(jié)果文本包含三個(gè)語(yǔ)句。為了確切地了解Python SQL注入的工作原理，需要單獨(dú)檢查每個(gè)部分。第一條語(yǔ)句如下：

select admin from users where username = '';

這是我們想要的查詢。分號(hào)（;）終止查詢，因此該查詢的結(jié)果無(wú)關(guān)緊要。接下來(lái)是第二個(gè)語(yǔ)句：

select true;

這是入侵者構(gòu)造的。它旨在始終返回True。

最后，我們會(huì)看到這段簡(jiǎn)短的代碼：

--'

該代碼片段可消除其后的所有內(nèi)容。入侵者添加了注釋符號(hào)（–），以將我們可能在最后一個(gè)占位符之后輸入的所有內(nèi)容轉(zhuǎn)換為注釋

使用此參數(shù)執(zhí)行函數(shù)時(shí)，它將始終返回True。例如，如果我們?cè)诘卿涰?yè)面中使用此功能，則入侵者可以使用用戶名登錄'; select true; --，并將被授予訪問(wèn)權(quán)限。

如果我們認(rèn)為這很難受，則可能會(huì)變得更難受！了解表結(jié)構(gòu)的入侵者可以使用Python SQL注入造成永久性破壞。例如，入侵者可以注入一條更新語(yǔ)句來(lái)更改數(shù)據(jù)庫(kù)中的信息：

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

讓我們?cè)俅畏纸猓?/p>

';

就像之前的注入一樣，此代碼段終止了查詢。下一條語(yǔ)句如下：

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

更新admin到true用戶lisi

最后，有以下代碼片段：

select true; --

與前面的示例一樣，該片段返回true并注釋掉其后的所有內(nèi)容。

如果入侵者設(shè)法使用此輸入執(zhí)行功能，則用戶lisi將成為管理員：

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

入侵者可以使用用戶名登錄lisi。（如果入侵者確實(shí)想破壞，那么可以使用DROP DATABASE命令）

現(xiàn)在我們恢復(fù)lisi的原始狀態(tài)：

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

4.1 制作安全查詢參數(shù)

了解了入侵者如何通過(guò)使用精心設(shè)計(jì)的字符串來(lái)利用系統(tǒng)并獲得管理員權(quán)限。問(wèn)題是我們?cè)试S從客戶端傳遞的值直接執(zhí)行到數(shù)據(jù)庫(kù)，而無(wú)需執(zhí)行任何類型的檢查或驗(yàn)證。SQL注入依賴于這種類型的漏洞

每當(dāng)在數(shù)據(jù)庫(kù)查詢中使用用戶輸入時(shí)，SQL注入就可能存在漏洞。防止Python SQL注入的關(guān)鍵是確保該值已按我們開(kāi)發(fā)的預(yù)期使用。在上一個(gè)示例中，username用作了字符串。實(shí)際上，它被用作原始SQL語(yǔ)句

為了確保我們按預(yù)期使用值，需要對(duì)值進(jìn)行轉(zhuǎn)義。例如，為防止入侵者將原始SQL替換為字符串參數(shù)，可以對(duì)引號(hào)進(jìn)行轉(zhuǎn)義：

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

這只是一個(gè)例子。嘗試防止Python SQL注入時(shí)，有很多特殊字符和場(chǎng)景需要考慮?，F(xiàn)代的數(shù)據(jù)庫(kù)適配器隨附了一些內(nèi)置工具，這些工具可通過(guò)使用查詢參數(shù)來(lái)防止Python SQL注入。使用這些參數(shù)代替普通字符串插值可組成帶有參數(shù)的查詢

現(xiàn)在，我們已經(jīng)對(duì)該漏洞有了一個(gè)明確的知曉，可以使用查詢參數(shù)而不是字符串插值來(lái)重寫(xiě)該函數(shù)：

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

我們使用了一個(gè)命名參數(shù)username來(lái)指示用戶名應(yīng)該去哪里

將值username作為第二個(gè)參數(shù)傳遞給cursor.execute()。username在數(shù)據(jù)庫(kù)中執(zhí)行查詢時(shí)，連接將使用的類型和值
要測(cè)試此功能，我們先嘗試一些有效以及無(wú)效的值跟一些有隱患的字符串：

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

跟我們想象的一毛一樣！該函數(shù)返回所有值的預(yù)期結(jié)果。并且，隱患的字符串不再起作用。要了解原因，可以檢查由生成的查詢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視為字符串，并轉(zhuǎn)義了可能終止該字符串的所有字符并引入了Python SQL注入

4.2 傳遞安全查詢參數(shù)

數(shù)據(jù)庫(kù)適配器通常提供幾種傳遞查詢參數(shù)的方法。命名占位符通常是可讀性最好的，但是某些實(shí)現(xiàn)可能會(huì)受益于使用其他選項(xiàng)

讓我們快速看一下使用查詢參數(shù)的一些對(duì)與錯(cuò)方法。以下代碼塊顯示了我們需要避免的查詢類型：

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

這些語(yǔ)句中的每條語(yǔ)句都username直接從客戶端傳遞到數(shù)據(jù)庫(kù)，而無(wú)需執(zhí)行任何類型的檢查或驗(yàn)證。這類代碼已經(jīng)可以達(dá)到Python SQL注入

相比上面，以下類型的查詢可以安全地執(zhí)行：

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

在這些語(yǔ)句中，username作為命名參數(shù)傳遞?，F(xiàn)在，數(shù)據(jù)庫(kù)將username在執(zhí)行查詢時(shí)使用指定的類型和值，從而提供針對(duì)Python SQL注入的保護(hù)

5. 使用SQL組合

但是，如果我們有一個(gè)用例需要編寫(xiě)一個(gè)不同的查詢（該參數(shù)是其他參數(shù)，例如表或列名），該怎么辦？

繼上一個(gè)列子，我們實(shí)現(xiàn)一個(gè)函數(shù)，該函數(shù)接受表的名稱并返回該表中的行數(shù)：

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

嘗試在用戶表上執(zhí)行該功能：

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

該命令無(wú)法生成SQL。數(shù)據(jù)庫(kù)適配器將變量視為字符串或文字。但是，表名不是純字符串。這就是SQL組合的用武之地

我們已經(jīng)知道使用字符串插值來(lái)編寫(xiě)SQL是不安全的。psycopg提供了一個(gè)名為的模塊psycopg.sql，可以幫助我們安全地編寫(xiě)SQL查詢。讓我們使用psycopg.sql.SQL()以下代碼重寫(xiě)該函數(shù)：

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

此實(shí)現(xiàn)有兩個(gè)區(qū)別。sql.SQL()組成查詢。sql.Identifier()對(duì)參數(shù)值進(jìn)行注釋table_name（標(biāo)識(shí)符是列或表的名稱）

現(xiàn)在，我們嘗試在users表上執(zhí)行該函數(shù)：

>>> count_rows('users')
2

接下來(lái)，讓我們看看表不存在時(shí)會(huì)發(fā)生什么：

>>> 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"
                        ^

該函數(shù)引發(fā)UndefinedTable異常。將使用此異常來(lái)表明我們的函數(shù)可以安全地免受Python SQL注入攻擊

要將所有內(nèi)容放在一起，添加一個(gè)選項(xiàng)以對(duì)表中的行進(jìn)行計(jì)數(shù)，直到達(dá)到特定限制。對(duì)于非常大的表，這個(gè)功能很有用。要實(shí)現(xiàn)這個(gè)操作，LIMIT在查詢中添加一個(gè)子句，以及該限制值的查詢參數(shù)：

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使用簡(jiǎn)單方法時(shí)，會(huì)將所有查詢參數(shù)綁定為文字。但是，使用時(shí)sql.SQL()，需要使用sql.Identifier()或顯式注釋每個(gè)參數(shù)sql.Literal()

不幸的是，Python API規(guī)范不解決標(biāo)識(shí)符的綁定，僅處理文字。Psycopg是唯一流行的適配器，它添加了使用文字和標(biāo)識(shí)符安全地組合SQL的功能。這個(gè)事實(shí)使得在綁定標(biāo)識(shí)符時(shí)要特別注意

執(zhí)行該函數(shù)以確保其起作用：

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

現(xiàn)在我們已經(jīng)看到該函數(shù)正在運(yùn)行，檢查它是否安全：

>>> 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轉(zhuǎn)義了該值，并且數(shù)據(jù)庫(kù)將其視為表名。由于不存在具有該名稱的表，因此UndefinedTable引發(fā)了異常所以是安全的！

6. 結(jié)論

通過(guò)實(shí)現(xiàn)組成動(dòng)態(tài)SQL，可與你使我們有效的規(guī)避系統(tǒng)遭受Python SQL注入的威脅！在查詢過(guò)程中同時(shí)使用文字和標(biāo)識(shí)符，并不會(huì)影響安全性

7. 致謝

到此這篇關(guān)于使用Python防止SQL注入攻擊的實(shí)現(xiàn)示例的文章就介紹到這了,更多相關(guān)Python防止SQL注入攻擊內(nèi)容請(qǐng)搜索腳本之家以前的文章或繼續(xù)瀏覽下面的相關(guān)文章希望大家以后多多支持腳本之家！

最新評(píng)論