文章背景

每隔幾年，開放式Web應用程式安全專案就會對最關鍵的Web應用程式安全風險進行排名。自第一次報告以來，注入風險高居其位！在所有注入型別中，SQL注入是最常見的攻擊手段之一，而且是最危險的。由於Python是世界上最流行的程式語言之一，因此瞭解如何防止Python SQL注入對於我們來說還是比較重要的

那麼在寫這篇文章的時候我也是查詢了國內外很多資料，最後帶著問題去完善總結：

• 什麼是Python SQL注入以及如何防止注入
• 如何使用文字和識別符號作為引數組合查詢
• 如何安全地執行資料庫中的查詢

文章演示的操作適用於所有資料庫，這裡的示例使用的是PG，但是效果跟過程可以在其他資料庫（例如SQLite，MySQL，Oracle等等系統中）重現

1. 瞭解Python SQL注入

  SQL注入攻擊是一種常見的安全漏洞。在我們日常工作中生成和執行SQL查詢也同樣是一項常見的任務。但是，有時候在編寫SQL語句時常常會犯下可怕錯誤

當我們使用Python將這些查詢直接執行到資料庫中時，很可能會損害到系統。所以如何成功實現組成動態SQL查詢的函式，而又不會使系統遭受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>",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 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),= 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 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防止SQL注入攻擊的實現示例的文章就介紹到這了,更多相關Python防止SQL注入攻擊內容請搜尋我們以前的文章或繼續瀏覽下面的相關文章希望大家以後多多支援我們！