关于python:LyScript-通过PEB结构解析堆基址

37次阅读

共计 14557 个字符，预计需要花费 37 分钟才能阅读完成。

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

  • 插件地址：https://github.com/lyshark/Ly…

首先以获取 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()

读取成果如下：

如上 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()

读取成果如下：

构造的解析能够封装成一个 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()

解析成果如下：

当咱们失去了堆的起始地址当前，那么对堆地址进行深度解析就变得很容易了，只须要填充特定的构造体，即可。

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()

解析堆内的段：

低内存堆的输入也能够应用如上办法实现，只是在输入是须要解析的构造体程序稍多一些，但总体上原理与上方代码统一。

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()

输入成果：

正文完
python
发表至： python
2023-02-10
 0
关于python:20张高清数据分析全知识地图拿去吧你
关于python:用Python制作一个简单的球球大作战
关于python:逼自己玩命学了3个多月吃透了Python技术核心分享给你让你今年进个大厂
关于python:Python随笔1int对象池