from asyncio.log import logger
from datetime import datetime
import hashlib
import json
from pathlib import Path
from typing import TYPE_CHECKING, Dict, List
import sys
import zlib
import os
if TYPE_CHECKING:
from lldb import *
import lldb
black_list = {
'startswith': ['/dev', '/system/fonts', '/dmabuf'],
'endswith': ['(deleted)', '.apk', '.odex', '.vdex', '.dex', '.jar', '.art', '.oat', '.art]'],
'includes': [],
}
dump_path = Path("./dump/")
def __lldb_init_module(debugger: 'SBDebugger', internal_dict: dict):
debugger.HandleCommand('command script add -f lldb_dumper.dumpMem dumpmem')
def dumpMem(debugger: 'SBDebugger', command: str, exe_ctx: 'SBExecutionContext', result: 'SBCommandReturnObject', internal_dict: dict):
target = exe_ctx.GetTarget()
arch_long = dump_arch_info(target)
frame = exe_ctx.GetFrame()
regs = dump_regs(frame)
target = exe_ctx.GetTarget()
sections = dump_memory_info(target)
max_seg_size = 64 * 1024 * 1024
process = exe_ctx.GetProcess()
segments = dump_memory(process, dump_path, black_list, max_seg_size)
(dump_path / "regs.json").write_text(json.dumps(regs))
(dump_path / "sections.json").write_text(json.dumps(sections))
(dump_path / "segments.json").write_text(json.dumps(segments))
def dump_arch_info(target: 'SBTarget'):
triple = target.GetTriple()
print(f'[dump_arch_info] triple => {triple}')
arch, vendor, sys, abi = triple.split('-')
if arch == 'aarch64' or arch == 'arm64':
return 'arm64le'
elif arch == 'aarch64_be':
return 'arm64be'
elif arch == 'armeb':
return 'armbe'
elif arch == 'arm':
return 'armle'
else:
return ''
def dump_regs(frame: 'SBFrame'):
regs = {}
registers = None
for registers in frame.GetRegisters():
print(f'registers name => {registers.GetName()}')
for register in registers:
register_name = register.GetName()
register.SetFormat(lldb.eFormatHex)
register_value = register.GetValue()
regs[register_name] = register_value
print(f'regs => {json.dumps(regs, ensure_ascii=False, indent=4)}')
return regs
def dump_memory_info(target: 'SBTarget'):
logger.debug('start dump_memory_info')
sections = []
for module in target.module_iter():
module: SBModule
for section in module.section_iter():
section: SBSection
module_name = module.file.GetFilename()
start, end, size, name = get_section_info(target, section)
section_info = {
'module': module_name,
'start': start,
'end': end,
'size': size,
'name': name,
}
print(f'Appending: {name}')
sections.append(section_info)
return sections
def get_section_info(tar, sec):
name = sec.name if sec.name is not None else ""
if sec.GetParent().name is not None:
name = sec.GetParent().name + "." + sec.name
module_name = sec.addr.module.file.GetFilename()
module_name = module_name if module_name is not None else ""
long_name = module_name + "." + name
return sec.GetLoadAddress(tar), (sec.GetLoadAddress(tar) + sec.size), sec.size, long_name
def dump_memory(process: 'SBProcess', dump_path: Path, black_list: Dict[str, List[str]], max_seg_size: int):
logger.debug('start dump memory')
memory_list = []
mem_info = lldb.SBMemoryRegionInfo()
start_addr = -1
next_region_addr = 0
while next_region_addr > start_addr:
err = process.GetMemoryRegionInfo(next_region_addr, mem_info)
if not err.success:
logger.warning(f'GetMemoryRegionInfo failed, {err}, break')
break
next_region_addr = mem_info.GetRegionEnd()
if next_region_addr >= sys.maxsize:
logger.info(f'next_region_addr:0x{next_region_addr:x} >= sys.maxsize, break')
break
start = mem_info.GetRegionBase()
end = mem_info.GetRegionEnd()
region_name = 'UNKNOWN'
if mem_info.IsMapped():
name = mem_info.GetName()
if name is None:
name = ''
mem_info_obj = {
'start': start,
'end': end,
'name': name,
'permissions': {
'r': mem_info.IsReadable(),
'w': mem_info.IsWritable(),
'x': mem_info.IsExecutable(),
},
'content_file': '',
}
memory_list.append(mem_info_obj)
for seg_info in memory_list:
try:
start_addr = seg_info['start']
end_addr = seg_info['end']
region_name = seg_info['name']
permissions = seg_info['permissions']
if seg_info['permissions']['r'] is False:
logger.warning(f'Skip dump {region_name} permissions => {permissions}')
continue
predicted_size = end_addr - start_addr
if predicted_size > max_seg_size:
logger.warning(f'Skip dump {region_name} size:0x{predicted_size:x}')
continue
skip_dump = False
for rule in black_list['startswith']:
if region_name.startswith(rule):
skip_dump = True
logger.warning(f'Skip dump {region_name} hit startswith rule:{rule}')
if skip_dump: continue
for rule in black_list['endswith']:
if region_name.endswith(rule):
skip_dump = True
logger.warning(f'Skip dump {region_name} hit endswith rule:{rule}')
if skip_dump: continue
for rule in black_list['includes']:
if rule in region_name:
skip_dump = True
logger.warning(f'Skip dump {region_name} hit includes rule:{rule}')
if skip_dump: continue
ts = datetime.now()
err = lldb.SBError()
seg_content = process.ReadMemory(start_addr, predicted_size, err)
tm = (datetime.now() - ts).total_seconds()
if seg_content is None:
logger.debug(f'Segment empty: @0x{start_addr:016x} {region_name} => {err}')
else:
logger.info(f'Dumping @0x{start_addr:016x} {tm:.2f}s size:0x{len(seg_content):x}: {region_name} {permissions}')
compressed_seg_content = zlib.compress(seg_content)
md5_sum = hashlib.md5(compressed_seg_content).hexdigest() + '.bin'
seg_info['content_file'] = md5_sum
(dump_path / md5_sum).write_bytes(compressed_seg_content)
except Exception as e:
logger.error(f'Exception reading segment {region_name}', exc_info=e)
return memory_list
