LyScript 通过PEB结构解析堆基址

举报
微软技术分享 发表于 2022/11/20 17:14:11 2022/11/20
【摘要】 LyScript中默认并没有提供获取进程堆基址的函数，不过却提供了获取PEB/TEB的函数，以PEB获取为例，可以调用`dbg.get_peb_address(local_pid)`用户传入当前进程的PID号，通常PID号可以使用`dbg.get_process_id()`函数得到，当得到了PEB进程环境块的基地址，那么获取堆基址就变得很简单了。

LyScript中默认并没有提供获取进程堆基址的函数，不过却提供了获取PEB/TEB的函数，以PEB获取为例，可以调用dbg.get_peb_address(local_pid)用户传入当前进程的PID号，通常PID号可以使用dbg.get_process_id()函数得到，当得到了PEB进程环境块的基地址，那么获取堆基址就变得很简单了。

首先以获取kernel32.dll模块基地址为例，如果使用汇编获取则代码是这样的，根据这段代码我们举一反三。

_asm
{
	push esi
	mov esi, dword ptr fs : [0x30]    // PEB地址
	mov esi, [esi + 0x0C]             // PEB_LDR_DATA
	mov esi, [esi + 0x1C]             // InInitializationOrderModuleList
	mov esi, [esi]                    //
	mov eax, [esi + 0x08]             // 模块基址
	pop esi
}

使用lyscript得到地址的代码就变得很简单了，只需要多次读取指针变量即可得到。

from LyScript32 import MyDebug

if __name__ == "__main__":
    dbg = MyDebug()
    conn = dbg.connect()

    # 内置函数得到进程PEB
    local_pid = dbg.get_process_id()
    peb = dbg.get_peb_address(local_pid)
    print("进程PEB: {}".format(hex(peb)))

    # esi = PEB_LDR_DATA结构体的地址
    ptr = peb + 0x0c

    # 读取内存指针
    PEB_LDR_DATA = dbg.read_memory_ptr(ptr)
    print("读入PEB_LDR_DATA里面的地址: {}".format(hex(PEB_LDR_DATA)))

    # esi = 模块链表指针InInitializationOrderModuleList
    ptr = PEB_LDR_DATA + 0x1c
    InInitializationOrderModuleList = dbg.read_memory_ptr(ptr)
    print("读入InInitializationOrderModuleList里面的地址: {}".format(hex(InInitializationOrderModuleList)))

    # 取出kernel32.dll模块基址
    ptr = InInitializationOrderModuleList + 0x08
    modbase = dbg.read_memory_ptr(ptr)
    print("kernel32.dll = {}".format(hex(modbase)))

    dbg.close()

读取效果如下：

image.png

如上kernel模块基地址的获取已经实现了，那么堆基址的获取也就非常简单了，我们只需要找到peb+0x90的位置，将其读取出来即可。

0:000> dt _peb @$peb
ntdll!_PEB+0x090 ProcessHeaps

0:000> dd 7c99ffe0 l8
7c99ffe0  00150000 00250000 00260000 00000000
7c99fff0  00000000 00000000 00000000 00000000

读取内存指针即可得到堆地址，将堆地址获取封装成getHeapsAddress()函数方便后续调用。

from LyScript32 import MyDebug

# 获取模块基址
def getKernelModuleBase(dbg):
    # 内置函数得到进程PEB
    local_pid = dbg.get_process_id()
    peb = dbg.get_peb_address(local_pid)
    # print("进程PEB: {}".format(hex(peb)))

    # esi = PEB_LDR_DATA结构体的地址
    ptr = peb + 0x0c

    # 读取内存指针
    PEB_LDR_DATA = dbg.read_memory_ptr(ptr)
    # print("读入PEB_LDR_DATA里面的地址: {}".format(hex(PEB_LDR_DATA)))

    # esi = 模块链表指针InInitializationOrderModuleList
    ptr = PEB_LDR_DATA + 0x1c
    InInitializationOrderModuleList = dbg.read_memory_ptr(ptr)
    # print("读入InInitializationOrderModuleList里面的地址: {}".format(hex(InInitializationOrderModuleList)))

    # 取出kernel32.dll模块基址
    ptr = InInitializationOrderModuleList + 0x08
    modbase = dbg.read_memory_ptr(ptr)
    # print("kernel32.dll = {}".format(hex(modbase)))
    return modbase

# 获取进程堆基址
def getHeapsAddress(dbg):
    # 内置函数得到进程PEB
    local_pid = dbg.get_process_id()
    peb = dbg.get_peb_address(local_pid)
    # print("进程PEB: {}".format(hex(peb)))

    # 读取堆分配地址
    ptr = peb + 0x90
    peb_address = dbg.read_memory_ptr(ptr)
    # print("读取堆分配: {}".format(hex(peb_address)))

    heap_address = dbg.read_memory_ptr(peb_address)
    # print("当前进程堆基址: {}".format(hex(heap_address)))
    return heap_address

if __name__ == "__main__":
    dbg = MyDebug()
    conn = dbg.connect()

    k32 = getKernelModuleBase(dbg)
    print("kernel32 = {}".format(hex(k32)))

    heap = getHeapsAddress(dbg)
    print("heap 堆地址: {}".format(hex(heap)))

    dbg.close()

读取效果如下：

image.png

结构的解析可以封装成一个PEB类，读入前488字节，并解析。

from LyScript32 import MyDebug
import struct

class _PEB():
    def __init__(self, dbg):
        # 内置函数得到进程PEB
        self.base = dbg.get_peb_address(dbg.get_process_id())
        self.PEB = bytearray()

        # 填充前488字节
        for index in range(0,488):
            readbyte = dbg.read_memory_byte(self.base + index)
            self.PEB.append(readbyte)

        """
        0:000> !kdex2x86.strct PEB
        Loaded kdex2x86 extension DLL
        struct   _PEB (sizeof=488)
        +000 byte     InheritedAddressSpace
        +001 byte     ReadImageFileExecOptions
        +002 byte     BeingDebugged
        +003 byte     SpareBool
        +004 void     *Mutant
        +008 void     *ImageBaseAddress
        +00c struct   _PEB_LDR_DATA *Ldr
        +010 struct   _RTL_USER_PROCESS_PARAMETERS *ProcessParameters
        +014 void     *SubSystemData
        +018 void     *ProcessHeap
        +01c void     *FastPebLock
        +020 void     *FastPebLockRoutine
        +024 void     *FastPebUnlockRoutine
        +028 uint32   EnvironmentUpdateCount
        +02c void     *KernelCallbackTable
        +030 uint32   SystemReserved[2]
        +038 struct   _PEB_FREE_BLOCK *FreeList
        +03c uint32   TlsExpansionCounter
        +040 void     *TlsBitmap
        +044 uint32   TlsBitmapBits[2]
        +04c void     *ReadOnlySharedMemoryBase
        +050 void     *ReadOnlySharedMemoryHeap
        +054 void     **ReadOnlyStaticServerData
        +058 void     *AnsiCodePageData
        +05c void     *OemCodePageData
        +060 void     *UnicodeCaseTableData
        +064 uint32   NumberOfProcessors
        +068 uint32   NtGlobalFlag
        +070 union    _LARGE_INTEGER CriticalSectionTimeout
        +070 uint32   LowPart
        +074 int32    HighPart
        +070 struct   __unnamed3 u
        +070 uint32   LowPart
        +074 int32    HighPart
        +070 int64    QuadPart
        +078 uint32   HeapSegmentReserve
        +07c uint32   HeapSegmentCommit
        +080 uint32   HeapDeCommitTotalFreeThreshold
        +084 uint32   HeapDeCommitFreeBlockThreshold
        +088 uint32   NumberOfHeaps
        +08c uint32   MaximumNumberOfHeaps
        +090 void     **ProcessHeaps
        +094 void     *GdiSharedHandleTable
        +098 void     *ProcessStarterHelper
        +09c uint32   GdiDCAttributeList
        +0a0 void     *LoaderLock
        +0a4 uint32   OSMajorVersion
        +0a8 uint32   OSMinorVersion
        +0ac uint16   OSBuildNumber
        +0ae uint16   OSCSDVersion
        +0b0 uint32   OSPlatformId
        +0b4 uint32   ImageSubsystem
        +0b8 uint32   ImageSubsystemMajorVersion
        +0bc uint32   ImageSubsystemMinorVersion
        +0c0 uint32   ImageProcessAffinityMask
        +0c4 uint32   GdiHandleBuffer[34]
        +14c function *PostProcessInitRoutine
        +150 void     *TlsExpansionBitmap
        +154 uint32   TlsExpansionBitmapBits[32]
        +1d4 uint32   SessionId
        +1d8 void     *AppCompatInfo
        +1dc struct   _UNICODE_STRING CSDVersion
        +1dc uint16   Length
        +1de uint16   MaximumLength
        +1e0 uint16   *Buffer
        """

        # 初始化PEB
        index = 0x000
        self.InheritedAddressSpace = self.PEB[index]
        index = 0x001
        self.ReadImageFileExecOptions = self.PEB[index]
        index = 0x002
        self.BeingDebugged = self.PEB[index]
        index = 0x003
        self.SpareBool = self.PEB[index]
        index = 0x004
        self.Mutant = self.PEB[index:index+4]
        index = 0x008
        self.ImageBaseAddress = self.PEB[index:index+4]
        index = 0x00c
        self.Ldr = self.PEB[index:index+4]
        index = 0x010
        self.ProcessParameters = self.PEB[index:index+4]
        index = 0x014
        self.SubSystemData = self.PEB[index:index+4]
        index = 0x018
        self.ProcessHeap = self.PEB[index:index+4]

        index = 0x01c
        self.FastPebLock = self.PEB[index:index+4]
        index = 0x020
        self.FastPebLockRoutine = self.PEB[index:index+4]
        index = 0x024
        self.FastPebUnlockRoutine = self.PEB[index:index+4]
        index = 0x028
        self.EnviromentUpdateCount = self.PEB[index:index+4]
        index = 0x02c
        self.KernelCallbackTable = self.PEB[index:index+4]
        index = 0x030

        self.SystemReserved = []
        for i in range(0,2):
            self.SystemReserved.append(self.PEB[index:index+4])
            index += 4

        index = 0x038
        self.FreeList = self.PEB[index:index+4]
        index = 0x03c
        self.TlsExpansionCounter = self.PEB[index:index+4]
        index = 0x040
        self.TlsBitmap = self.PEB[index:index+4]
        index = 0x044

        self.TlsBitmapBits = []
        for i in range(0,2):
            self.TlsBitmapBits.append(self.PEB[index:index+4])
            index += 4
        index = 0x04c
        self.ReadOnlySharedMemoryBase = self.PEB[index:index+4]
        index = 0x050
        self.ReadOnlySharedMemoryheap = self.PEB[index:index+4]
        index = 0x054
        self.ReadOnlyStaticServerData = self.PEB[index:index+4]
        index = 0x058
        self.AnsiCodePageData = self.PEB[index:index+4]
        index = 0x05c
        self.OemCodePageData = self.PEB[index:index+4]
        index = 0x060
        self.UnicodeCaseTableData = self.PEB[index:index+4]
        index = 0x064
        self.NumberOfProcessors = self.PEB[index:index+4]
        index = 0x068
        self.NtGlobalFlag = self.PEB[index:index+4]

        # 这里的4个字节会发生什么 ?
        index = 0x070
        self.CriticalSectionTimeout_LowPart = self.PEB[index:index+4]
        index = 0x074
        self.CriticalSectionTimeout_HighPart = self.PEB[index:index+4]
        index = 0x078
        self.HeapSegmentReserve = self.PEB[index:index+4]
        index = 0x07c
        self.HeapSegmentCommit = self.PEB[index:index+4]
        index = 0x080
        self.HeapDeCommitTotalFreeThreshold = self.PEB[index:index+4]
        index = 0x084
        self.HeapDeCommitFreeBlockThreshold = self.PEB[index:index+4]
        index = 0x088
        self.NumberOfHeaps = self.PEB[index:index+4]
        index = 0x08c
        self.MaximumNumberOfHeaps = self.PEB[index:index+4]
        index = 0x090
        self.ProcessHeaps = self.PEB[index:index+4]

        index = 0x094
        self.GdiSharedHandleTable = self.PEB[index:index+4]
        index = 0x098
        self.ProcessStarterHelper = self.PEB[index:index+4]
        index = 0x09c
        self.GdiDCAttributeList = self.PEB[index:index+4]
        index = 0x0a0
        self.LoaderLock = self.PEB[index:index+4]
        index = 0x0a4
        self.OSMajorVersion = self.PEB[index:index+4]
        index = 0x0a8
        self.OSMinorVersion = self.PEB[index:index+4]
        index = 0x0ac
        self.OSBuildNumber = self.PEB[index:index+2]
        index = 0x0ae
        self.OSCSDVersion = self.PEB[index:index+2]
        index = 0x0b0
        self.OSPlatformId = self.PEB[index:index+4]
        index = 0x0b4
        self.ImageSubsystem = self.PEB[index:index+4]
        index = 0x0b8
        self.ImageSubsystemMajorVersion = self.PEB[index:index+4]
        index = 0x0bc
        self.ImageSubsystemMinorVersion = self.PEB[index:index+4]
        index = 0x0c0
        self.ImageProcessAffinityMask = self.PEB[index:index+4]
        index = 0x0c4

        # uint32 GdiHandleBuffer[34]
        self.GdiHandleBuffer = []
        for i in range(0,34):
            self.GdiHandleBuffer.append(self.PEB[index:index+4])
            index += 4
        index = 0x14c
        self.PostProcessInitRoutine = self.PEB[index:index+4]
        index = 0x150
        self.TlsExpansionBitmap = self.PEB[index:index+4]
        index = 0x154

        # uint32 TlsExpansionBitmapBits[32]
        self.TlsExpansionBitmapBits = []
        for i in range(0,32):
            self.TlsExpansionBitmapBits.append(self.PEB[index:index+4])
            index += 4
        index = 0x1d4
        self.SessionId = self.PEB[index:index+4]
        index = 0x1d8
        self.AppCompatInfo = self.PEB[index:index+4]
        index = 0x1dc

        # struct _UNICODE_STRING CSDVersion
        self.CSDVersion_Length = self.PEB[index:index+2]
        index += 2
        self.CSDVersion_MaximumLength = self.PEB[index:index+2]
        index += 2
        self.CSDVersion_Buffer = self.PEB[index:index+2]
        index += 2

    def get_BeingDebugged(self):
        return self.BeingDebugged

    def get_ProcessHeaps(self):
        pack = struct.unpack('<L', bytes(self.ProcessHeap))
        return hex(pack[0])

if __name__ == "__main__":
    dbg = MyDebug()
    connect = dbg.connect()

    # 初始化PEB填充结构
    peb = _PEB(dbg)

    # 获取进程调试状态
    is_debug = peb.get_BeingDebugged()
    print("是否被调试: {}".format(is_debug))

    heap = peb.get_ProcessHeaps()
    print("堆地址: {}".format(heap))

    dbg.close()

解析效果如下：

image.png

当我们得到了堆的起始地址以后，那么对堆地址进行深度解析就变得很容易了，只需要填充特定的结构体，即可。

from LyScript32 import MyDebug
import struct
import string

DEBUG = False

class _PEB():
    def __init__(self, dbg):
        # 内置函数得到进程PEB
        self.base = dbg.get_peb_address(dbg.get_process_id())
        self.PEB = bytearray()

        # 填充前488字节
        for index in range(0, 488):
            readbyte = dbg.read_memory_byte(self.base + index)
            self.PEB.append(readbyte)

        index = 0x018
        self.ProcessHeap = self.PEB[index:index + 4]

    def get_ProcessHeaps(self):
        pack = struct.unpack('<L', bytes(self.ProcessHeap))
        return pack[0]

class GrabHeap():
    def __init__(self, dbg, heap_addr):
        # 内置函数得到进程PEB
        self.base = dbg.get_peb_address(dbg.get_process_id())
        self.address = heap_addr
        self.buffer = bytearray()

    def grapHeap(self):
        for idx in range(0, 1416):
            readbyte = dbg.read_memory_byte(self.address + idx)
            self.buffer.append(readbyte)

        index = 0x8
        (self.Signature, self.Flags, self.ForceFlags, self.VirtualMemoryThreshold, \
         self.SegmentReserve, self.SegmentCommit, self.DeCommitFreeBlockThreshold, self.DeCommitTotalBlockThreshold, \
         self.TotalFreeSize, self.MaximumAllocationSize, self.ProcessHeapListIndex, self.HeaderValidateLength, \
         self.HeaderValidateCopy, self.NextAvailableTagIndex, self.MaximumTagIndex, self.TagEntries, \
         self.UCRSegments, self.UnusedUnCommittedRanges, self.AlignRound, self.AlignMask) = \
            struct.unpack("LLLLLLLLLLHHLHHLLLLL", self.buffer[index:index + (0x50 - 8)])

        index += 0x50 - 8
        self.VirtualAllocedBlock = struct.unpack("LL", self.buffer[index:index + 8])

        index += 8
        self._Segments = struct.unpack("L" * 64, self.buffer[index:index + 64 * 4])
        index += 64 * 4
        self.FreeListInUseLong = struct.unpack("LLLL", self.buffer[index:index + 16])
        index += 16
        (self.FreeListInUseTerminate, self.AllocatorBackTraceIndex) = struct.unpack("HH", self.buffer[index: index + 4])
        index += 4
        (self.Reserved1, self.LargeBlocksIndex) = struct.unpack("LL", self.buffer[index: index + 8])

    def get_Signature(self):
        return self.Signature

# 段
class Segment():
    def __init__(self, dbg, heap_addr):
        self.address = heap_addr
        self.buffer = bytearray()

        # AVOID THE ENTRY ITSELF
        self.address += 8
        for idx in range(0, 0x34):
            readbyte = dbg.read_memory_byte(self.address + idx)
            self.buffer.append(readbyte)

        (self.Signature, self.Flags, self.Heap, self.LargestUnCommitedRange, self.BaseAddress, \
         self.NumberOfPages, self.FirstEntry, self.LastValidEntry, self.NumberOfUnCommittedPages, \
         self.NumberOfUnCommittedRanges, self.UnCommittedRanges, self.AllocatorBackTraceIndex, \
         self.Reserved, self.LastEntryInSegment) = struct.unpack("LLLLLLLLLLLHHL", self.buffer)

        if DEBUG == True:
            print("SEGMENT: {} Sig: {}".format(hex(self.address), hex(self.Signature)))

            print("Heap: {} LargetUncommit {} Base: {}"
                  .format(hex(self.Heap), hex(self.LargestUnCommitedRange), hex(self.BaseAddress)))

            print("NumberOfPages {} FirstEntry: {} LastValid: {}"
                  .format(hex(self.NumberOfPages), hex(self.FirstEntry), hex(self.LastValidEntry)))

            print("Uncommited: {}".format(self.UnCommittedRanges))

            Pages = []
            Items = bytearray()
            if self.UnCommittedRanges:
                addr = self.UnCommittedRanges
                if addr != 0:
                    # 读入内存
                    for idx in range(0, 0x10):
                        readbyte = dbg.read_memory_byte(self.address + idx)
                        Items.append(readbyte)

                    (C_Next, C_Addr, C_Size, C_Filler) = struct.unpack("LLLL", Items)
                    print("Memory: {} Address: {} (a: {}) Size: {}"
                          .format(hex(self.address), hex(C_Next), C_Addr, C_Size))
                    Pages.append(C_Addr + C_Size)
                    addr = C_Next

if __name__ == "__main__":
    dbg = MyDebug()
    connect = dbg.connect()

    # 初始化PEB填充结构
    peb = _PEB(dbg)

    # 堆地址
    process_heap = peb.get_ProcessHeaps()
    print("堆地址: {}".format(hex(process_heap)))

    # 定义Segment
    heap = Segment(dbg, process_heap)

    # 输出内容
    print("Signature = {}".format(heap.Signature))
    print("Flags = {}".format(heap.Flags))
    print("Heap = {}".format(heap.Heap))

    # 初始化堆
    heap = GrabHeap(dbg, process_heap)
    heap.grapHeap()

    # 获取Signature
    Signature = heap.get_Signature()
    print("Signature = {}".format(hex(Signature)))

    dbg.close()

解析堆内的段：

image.png

低内存堆的输出也可以使用如上方法实现，只是在输出是需要解析的结构体程序稍多一些，但总体上原理与上方代码一致。

from LyScript32 import MyDebug
import struct
import string

# 读内存
def readMemory(address,size):
    ref_buffer = bytearray()
    for idx in range(0, size):
        readbyte = dbg.read_memory_byte(address + idx)
        ref_buffer.append(readbyte)
    return ref_buffer

def readLong(address):
    return dbg.read_memory_dword(address)

# 得到进程PEB
class _PEB():
    def __init__(self, dbg):
        # 内置函数得到进程PEB
        self.base = dbg.get_peb_address(dbg.get_process_id())
        self.PEB = bytearray()
        self.PEB = readMemory(self.base,488)

        # 通过偏移找到ProcessHeap
        index = 0x018
        self.ProcessHeap = self.PEB[index:index + 4]

    def get_ProcessHeaps(self):
        pack = struct.unpack('<L', bytes(self.ProcessHeap))
        return pack[0]

class UserMemoryCache():
    def __init__(self, addr, mem):
        self.address = addr
        (self.Next, self.Depth, self.Sequence, self.AvailableBlocks,\
         self.Reserved) = struct.unpack("LHHLL", mem[ 0 : 16 ])

class Bucket():
    def __init__(self, addr, mem):
        self.address = addr
        (self.BlockUnits, self.SizeIndex, Flag) =\
         struct.unpack("HBB", mem[:4])

        # 从理论上讲，这是标志的分离方式
        self.UseAffinity = Flag & 0x1
        self.DebugFlags  = (Flag >1) & 0x3

# 低内存堆
class LFHeap():
    def __init__(self, addr):
        mem = readMemory(addr, 0x300)
        index = 0
        self.address = addr

        (self.Lock, self.field_4, self.field_8, self.field_c,\
         self.field_10, field_14, self.SubSegmentZone_Flink,
         self.SubSegmentZone_Blink, self.ZoneBlockSize,\
         self.Heap, self.SegmentChange, self.SegmentCreate,\
         self.SegmentInsertInFree, self.SegmentDelete, self.CacheAllocs,\
         self.CacheFrees) = struct.unpack("L" * 0x10, mem[index:index+0x40])

        index += 0x40
        self.UserBlockCache = []
        for a in range(0,12):
            umc = UserMemoryCache(addr + index, mem[index:index + 0x10])
            index += 0x10
            self.UserBlockCache.append(umc)

        self.Buckets = []
        for a in range(0, 128):
            entry = mem[index: index + 4]
            b = Bucket(addr + index, entry)
            index = index + 4
            self.Buckets.append(b)

if __name__ == "__main__":
    dbg = MyDebug()
    connect = dbg.connect()

    # 初始化PEB填充结构
    peb = _PEB(dbg)

    # 堆地址
    process_heap = peb.get_ProcessHeaps()
    print("堆地址: {}".format(hex(process_heap)))

    # 定义低内存堆类
    lf_heap = LFHeap(process_heap)

    print("堆内存锁: {}".format(hex(lf_heap.Lock)))
    print("堆地址: {}".format(hex(lf_heap.Heap)))
    print("堆分配: {}".format(hex(lf_heap.CacheAllocs)))

    # 循环输出block
    for index in lf_heap.UserBlockCache:
        print("地址: {} --> 下一个地址: {}".format(hex(index.address),hex(index.Next)))

    for index in lf_heap.Buckets:
        print(index.SizeIndex,index.DebugFlags)

    dbg.close()

输出效果：

image.png

【版权声明】本文为华为云社区用户原创内容，转载时必须标注文章的来源（华为云社区）、文章链接、文章作者等基本信息， 否则作者和本社区有权追究责任。如果您发现本社区中有涉嫌抄袭的内容，欢迎发送邮件进行举报，并提供相关证据，一经查实，本社区将立刻删除涉嫌侵权内容，举报邮箱： cloudbbs@huaweicloud.com
任务调度
  • 点赞
  • 收藏
  • 关注作者

评论（0）

0/1000
抱歉，系统识别当前为高风险访问，暂不支持该操作

全部回复

上滑加载中

设置昵称

在此一键设置昵称，即可参与社区互动！

*长度不超过10个汉字或20个英文字符，设置后3个月内不可修改。

*长度不超过10个汉字或20个英文字符，设置后3个月内不可修改。

确认 取消

加入云驻计划，成为创作者

  • 华为云周边好礼
  • 免费体验产品
  • 特殊身份标识
  • 线下官方门票
  • 内部专家零距离
  • 与10000+优质创作者共同成长
立即加入